Rt indexer bulk

README.md

@@ -118,6 +118,20 @@ $table->addDocuments([
     ]); 
 ```
 
+### Fast bulk insert (indexer-assisted):
+
+Requires a Manticore build that supports `/bulk?indexer_rt_bulk=1`. Documents must include explicit non-zero numeric ids.
+
+```php
+$result = $table->addDocumentsStreaming([
+        ['id'=>2,'title'=>'Interstellar','plot'=>'A team of explorers travel through a wormhole in space in an attempt to ensure humanity\'s survival.','_year'=>2014,'rating'=>8.5],
+        ['id'=>3,'title'=>'Inception','plot'=>'A thief who steals corporate secrets through the use of dream-sharing technology is given the inverse task of planting an idea into the mind of a C.E.O.','_year'=>2010,'rating'=>8.8],
+    ], [
+        // omit workers to auto-tune 1,2,4,... up to server workers_total
+        'max_workers' => 8,
+    ]);
+```
+
 ### Perform a search:
 
 ```php

docs/configuration.md

@@ -48,6 +48,14 @@ Connection and transport options:
 * timeout - query timeout
 * connection_timeout - connection timeout
 * proxy - proxy definition as host:port
+
+`Client::indexerBulk()` / `Table::addDocumentsStreaming()` require the curl-based `Http` or `Https` transport. They open additional curl handles for concurrent chunked uploads that share one document iterator and drain it until empty (no batching on the main path).
+
+For large assisted bulk loads:
+
+* Prefer raising the connection `timeout` (default 300s) if indexer builds are slow
+* Server `max_packet_size` is bypassed by chunked `/bulk` uploads, but reverse proxies may still impose their own body limits
+* Cap client concurrency with `max_workers`; the client also reads `workers_total` from `SHOW STATUS`
 * username - username for HTTP Basic authentication
 * password - password for HTTP Basic authentication
 * bearer_token - bearer token used in the `Authorization` header

docs/lowlevelclient.md

@@ -402,4 +402,28 @@ $doc = [
 $response = $client->bulk($doc);
 ```
 
-<!-- proofread -->
+### Indexer-assisted bulk (`indexerBulk`)
+
+Experimental high-speed append path that posts to `/bulk?indexer_rt_bulk=1`. Prefer the Table helper `addDocumentsStreaming()` for document lists; use `indexerBulk()` for raw insert operations.
+
+```php
+$result = $client->indexerBulk([
+    ['insert' => ['table' => 'products', 'id' => 101, 'doc' => ['title' => 'Bag', 'price' => 9.99]]],
+    ['insert' => ['table' => 'products', 'id' => 102, 'doc' => ['title' => 'Baz', 'price' => 4.5]]],
+], [
+    'workers' => 2,
+]);
+```
+
+Notes:
+
+- Requires curl `Http`/`Https` transport. `PhpHttp` is not supported for this path.
+- IDs must be non-zero PHP integers or decimal digit strings in the uint64 range. Pass IDs above `PHP_INT_MAX` as strings, for example `'18446744073709551615'`; floats are rejected.
+- Opens up to `workers` concurrent chunked streams that share one iterator and pull until every document is inserted (no batching on the main path).
+- `probe_docs` applies only to auto-tune probe uploads when `workers` is omitted (default 1000 docs per worker).
+- PHP `post_max_size` / `upload_max_filesize` do not apply to outgoing client requests; `memory_limit`, proxies, and timeouts still do.
+- When `workers` is omitted, concurrency is probed as 1, 2, 4, … up to `workers_total` from `SHOW STATUS` (or `max_workers`).
+- Do not retry an ambiguous network failure after a request may have been accepted: each assisted request is atomic, but a lost response does not tell you whether the disk chunk was attached.
+- Server support depends on a Manticore build that includes indexer-assisted RT bulk; see upstream documentation for the released version that ships the feature.
+
+<!-- proofread -->

docs/tableclass.md

@@ -137,6 +137,46 @@ Returns an array response with:
 - errors - stating whether an error occurred
 - items - response status for each document.
 
+### addDocumentsStreaming()
+
+High-throughput append-only insert using Manticore's indexer-assisted bulk mode (`/bulk?indexer_rt_bulk=1`).
+
+Each request is streamed as NDJSON with HTTP chunked transfer encoding (documents are encoded on demand into a small buffer).
+
+With `workers = N`, the client opens up to N concurrent streams that share one document iterator and pull until every document is inserted. Each finished stream commits as one disk chunk. 
+
+Requirements and limits (server-side):
+
+- Local real-time table only (not distributed, sharded, replicated, percolate, or plain)
+- Insert only; replace/update/delete are rejected
+- Clustered tables are not supported
+- Every document must have an explicit non-zero numeric `id`
+
+IDs can be PHP integers or decimal digit strings up to `18446744073709551615`. Pass IDs above `PHP_INT_MAX` as strings to preserve their exact value; floats are rejected. Leading zeros are removed before duplicate detection.
+
+Options:
+
+- `workers` - fixed concurrency; when omitted, the client probes `1,2,4,...` until throughput stops improving
+- `max_workers` - upper bound (also capped by Manticore's `workers_total` setting)
+- `probe_docs` - documents per worker during each auto-tune probe (default 1000)
+- `gain_threshold` - minimum relative throughput gain to keep doubling workers (default `0`; stop when the next step is not better)
+- `timeout` - per-request timeout override
+- `fallback` - if true, fall back to regular `/bulk` when assisted mode is unavailable
+
+Example:
+
+```php
+$result = $table->addDocumentsStreaming([
+    ['id' => 1, 'title' => 'Crossbody Bag', 'gid' => 10],
+    ['id' => 2, 'title' => 'Sheet Set', 'gid' => 11],
+], [
+    // 'workers' => 4, // set explicitly to skip auto-tuning
+    'max_workers' => 8,
+]);
+
+// Persist $result['selected_workers'] for later runs if auto-tuning was used.
+```
+
 ### getDocumentById()
 
 Get an existing document by its ID.
@@ -698,4 +738,4 @@ Array
         )
 )
 ```
-<!-- proofread -->
+<!-- proofread -->

src/Manticoresearch/Bulk/AdaptiveTuner.php

@@ -0,0 +1,123 @@
+<?php
+
+// Copyright (c) Manticore Software LTD (https://manticoresearch.com)
+//
+// This source code is licensed under the MIT license found in the
+// LICENSE file in the root directory of this source tree.
+
+namespace Manticoresearch\Bulk;
+
+/**
+ * Chooses concurrent request count by exponential probing of throughput.
+ */
+class AdaptiveTuner
+{
+	/** @var float */
+	private $gainThreshold;
+
+	/** @var int */
+	private $maxWorkers;
+
+	/**
+	 * @param float $gainThreshold
+	 * @param int $maxWorkers
+	 */
+	public function __construct(float $gainThreshold, int $maxWorkers) {
+		$this->gainThreshold = $gainThreshold;
+		$this->maxWorkers = max(1, $maxWorkers);
+	}
+
+	/**
+	 * Yield candidate worker counts: 1, 2, 4, ... up to max.
+	 *
+	 * @return int[]
+	 */
+	public function candidates(): array {
+		$values = [];
+		for ($n = 1; $n <= $this->maxWorkers; $n *= 2) {
+			$values[] = $n;
+			if ($n > intdiv($this->maxWorkers, 2)) {
+				break;
+			}
+		}
+		if ($values[sizeof($values) - 1] !== $this->maxWorkers) {
+			$values[] = $this->maxWorkers;
+		}
+		return array_values(array_unique($values));
+	}
+
+	/**
+	 * Decide whether a newer probe is better enough to keep searching.
+	 *
+	 * @param float $previousThroughput docs/sec (or bytes/sec)
+	 * @param float $currentThroughput
+	 * @return bool true when current is an improvement worth continuing from
+	 */
+	public function isImprovement(float $previousThroughput, float $currentThroughput): bool {
+		if ($previousThroughput <= 0) {
+			return $currentThroughput > 0;
+		}
+		return ($currentThroughput - $previousThroughput) / $previousThroughput >= $this->gainThreshold;
+	}
+
+	/**
+	 * Pick the worker count with the best throughput; ties prefer fewer workers.
+	 *
+	 * @param array<int,float> $throughputByWorkers worker => throughput
+	 * @return int
+	 */
+	public function selectBest(array $throughputByWorkers): int {
+		if ($throughputByWorkers === []) {
+			return 1;
+		}
+		$bestWorkers = 1;
+		$bestRate = -1.0;
+		ksort($throughputByWorkers);
+		foreach ($throughputByWorkers as $workers => $rate) {
+			if ($rate <= $bestRate) {
+				continue;
+			}
+
+			$bestRate = $rate;
+			$bestWorkers = (int)$workers;
+		}
+		return max(1, min($bestWorkers, $this->maxWorkers));
+	}
+
+	/**
+	 * Walk candidates using a callable probe: fn(int $workers): float throughput.
+	 * Stops early when improvement falls below the gain threshold.
+	 *
+	 * @param callable $probe
+	 * @return array{selected_workers:int,stages:array}
+	 */
+	public function tune(callable $probe): array {
+		$stages = [];
+		$rates = [];
+		$previous = null;
+
+		foreach ($this->candidates() as $workers) {
+			$rate = (float)$probe($workers);
+			$rates[$workers] = $rate;
+			$stages[] = [
+				'workers' => $workers,
+				'throughput' => $rate,
+			];
+			if ($previous === null) {
+				$previous = $rate;
+				continue;
+			}
+			if ($this->isImprovement($previous, $rate)) {
+				$previous = $rate;
+				continue;
+			}
+			// No meaningful gain — stop probing higher concurrency
+			break;
+		}
+
+		return [
+			'selected_workers' => $this->selectBest($rates),
+			'stages' => $stages,
+		];
+	}
+}