return status and error information

miguelgrinberg · miguelgrinberg · commit b57901acb456 · 2026-04-21T16:42:51.000+01:00
diff --git a/docs/reference/bulk.md b/docs/reference/bulk.md
@@ -40,7 +40,7 @@ The bulk helper can be called as follows:
 ```php
 use Elastic\Elasticsearch\Helper\Bulk;
 
-$count = Bulk::bulk($client, $index, $actions, $chunk_size = 500, $max_chunk_bytes = 100 * 1024 * 1024);
+$response = Bulk::bulk($client, $index, $actions, $statS_only = false, $chunk_size = 500, $max_chunk_bytes = 100 * 1024 * 1024);
 ```
 
 This function has three required arguments:
@@ -50,15 +50,23 @@ This function has three required arguments:
 - `$actions` is the iterable that yields the bulk actions, normally implemented
    as a generator.
 
+The `$stats_only` optional argument controls wether details of each individual
+operation are included in the response. A value of `true` can be passed to only
+return total amount of operations processed and count of errors.
+
 The two optional arguments, `$chunk_size` and`$max_chunk_bytes`, determine when
 a Bulk API request is issued. The helper accumulates actions and submits a Bulk
 API request when the action count reaches `$chunk_size` or the payload size
 reaches `$max_chunk_bytes`, whichever happens first. The application can
 trigger a Bulk API request to be sent at specific times by yielding a `flush`
 action from its generator.
 
-The return value of the `bulk()` function is the total number of actions
-successfully processed.
+The return value of the `bulk()` function is an array with three elements:
+
+- The total number of actions successfully processed
+- The count of errors
+- An array with the details of each operation, as returned by the bulk API. Note that
+  this array is omitted when the `$stats_only` argument is set to `true`.
 
 ## Bulk actions
 
diff --git a/examples/dense_vector_benchmark.php b/examples/dense_vector_benchmark.php
@@ -76,18 +76,20 @@ function upload($client, $index, $dataset, $chunk_size, $repetitions, $packed) {
     $len = sizeof($dataset);
     $body = [];
     $start = microtime(true);
-    Bulk::bulk($client, $index, get_next_document($dataset, $repetitions, $packed), $chunk_size);
+    $response = Bulk::bulk($client, $index, get_next_document($dataset, $repetitions, $packed), true, $chunk_size);
+    assert ($response[0] == sizeof($dataset));  // make sure all items were ingested
+    assert ($response[1] == 0);  // make sure there were no errors
     return microtime(true) - $start;
 }
 
 $opts = getopt('s:r:', array('url:', 'json', 'runs:', 'help'), $rest_index);
 if (array_key_exists('help', $opts)) {
-    echo 'Usage: ' . $argv[0] . '[-s CHUNK_SIZES] [-r REPETITIONS] [--url URL] [--json] [--runs RUNS] DATASET_FILE\n';
-    echo '  -s CHUNK_SIZES List of chunk sizes to use, separated by commas (default: 100,250,500,1000)\n';
-    echo '  -r REPETITIONS Number of times the dataset is repeated (default: 20)\n';
-    echo '  --url URL      The Elasticsearch connection URL\n';
-    echo '  --json         Output benchmark results in JSON format\n';
-    echo '  --runs         Number of runs that are averaged for each chunk size (default: 3)\n';
+    echo "Usage: " . $argv[0] . "[-s CHUNK_SIZES] [-r REPETITIONS] [--url URL] [--json] [--runs RUNS] DATASET_FILE\n";
+    echo "  -s CHUNK_SIZES List of chunk sizes to use, separated by commas (default: 100,250,500,1000)\n";
+    echo "  -r REPETITIONS Number of times the dataset is repeated (default: 20)\n";
+    echo "  --url URL      The Elasticsearch connection URL\n";
+    echo "  --json         Output benchmark results in JSON format\n";
+    echo "  --runs         Number of runs that are averaged for each chunk size (default: 3)\n";
     exit(0);
 }
 if (!array_key_exists('url', $opts)) {
diff --git a/src/Helper/Bulk.php b/src/Helper/Bulk.php
@@ -28,14 +28,27 @@ class Bulk
      * @param $client Elasticsearch client
      * @param $defaultIndex The index that the bulk operation applies to
      * @param $actions Iterator that provides actions to include in the bulk requests
+     * @param $stats_only set to true to omit the individual operation results.
      * @param $chunkSize the maximum number of operations in a bulk request (DEFAULT: 500)
      * @param $maxChunkBytes the maximum size of a bulk request (DEFAULT: 10MB)
-     * @throws BulkHelperException
+     * @return array<mixed> an array where the first and second elements are the 
+     *     total number of operations processed and the total number of errors.
+     *     If $stats_only is false, a third element is returned, with an array of
+     *     individual operation results, as returned by the bulk API.
      * @throws ClientResponseException
      * @throws ServerResponseException
      */
-    public static function bulk(Client $client, string $index, Iterator $actions, int $chunk_size = 500, int $max_chunk_bytes = 100 * 1024 * 1024): int {
+    public static function bulk(
+        Client $client,
+        string $index,
+        Iterator $actions,
+        bool $stats_only = false,
+        int $chunk_size = 500,
+        int $max_chunk_bytes = 100 * 1024 * 1024,
+    ): array {
         $totalCount = 0;
+        $totalErrors = 0;
+        $results = [];
         $chunkCount = 0;
         $chunkBytes = 0;
         $body = [];
@@ -51,10 +64,14 @@ public static function bulk(Client $client, string $index, Iterator $actions, in
             }
             if (count($action) == 0 || $chunkCount >= $chunk_size || $chunkBytes >= $max_chunk_bytes) {
                 $response = $client->bulk(['index' => $index, 'body' => $body]);
-                if ($response['errors']) {
-                    $error = new BulkHelperException('Bulk upload error');
-                    $error->setResponse($response);
-                    throw $error;
+                foreach ($response['items'] as $item) {
+                    $status = $item[array_key_first($item)]['status'];
+                    if ($status < 200 || $status >= 300) {
+                        $totalErrors += 1;
+                    }
+                    if (!$stats_only) {
+                        $results[] = $item;
+                    }
                 }
                 $body = [];
                 $chunkCount = 0;
@@ -64,15 +81,17 @@ public static function bulk(Client $client, string $index, Iterator $actions, in
         }
         if (!empty($body)) {
             $response = $client->bulk(['index' => $index, 'body' => $body]);
-            if ($response['errors']) {
-                $error = new BulkHelperException('Bulk upload error');
-                $error->setResponse($response);
-                throw $error;
+            foreach ($response['items'] as $item) {
+                $status = $item[array_key_first($item)]['status'];
+                if ($status < 200 || $status >= 300) {
+                    $totalErrors += 1;
+                }
+                $results[] = $item;
             }
             unset($body);
             unset($response);
         }
-        return $totalCount;
+        return [$totalCount, $totalErrors, $results];
     }
 
     private static function action(string $action, array $metadata, ?array $document): array {
diff --git a/tests/Integration/BulkTest.php b/tests/Integration/BulkTest.php
@@ -177,11 +177,13 @@ function flushByCountActions($client, $index) {
             assert($response['hits']['total']['value'] == 3);
         }
 
-        $count = Bulk::bulk(
+        $response = Bulk::bulk(
             $this->client, self::TEST_INDEX,
-            flushByCountActions($this->client, self::TEST_INDEX), 2
+            flushByCountActions($this->client, self::TEST_INDEX), true, 2
         );
-        $this->assertEquals($count, 6);
+        $this->assertEquals($response[0], 6);
+        $this->assertEquals($response[1], 0);
+        $this->assertEquals($response[2], []);  // only stats in this response
 
         $response = readIndex($this->client, self::TEST_INDEX);
         $this->assertEquals(200, $response->getStatusCode());
@@ -232,11 +234,15 @@ function flushBySizeActions($client, $index) {
             assert($response['hits']['total']['value'] == 4);
         }
 
-        $count = Bulk::bulk(
+        $response = Bulk::bulk(
             $this->client, self::TEST_INDEX,
-            flushBySizeActions($this->client, self::TEST_INDEX), 500, 40
+            flushBySizeActions($this->client, self::TEST_INDEX),
+            false,  // include individual item results
+            500, 40,
         );
-        $this->assertEquals($count, 6);
+        $this->assertEquals($response[0], 6);
+        $this->assertEquals($response[1], 0);
+        $this->assertEquals(sizeof($response[2]), 6);
 
         $response = readIndex($this->client, self::TEST_INDEX);
         $this->assertEquals(200, $response->getStatusCode());
@@ -294,11 +300,14 @@ function explicitFlushActions($client, $index) {
             assert($response['hits']['total']['value'] == 3);
         }
 
-        $count = Bulk::bulk(
+        $response = Bulk::bulk(
             $this->client, self::TEST_INDEX,
-            explicitFlushActions($this->client, self::TEST_INDEX)
+            explicitFlushActions($this->client, self::TEST_INDEX),
+            true,  // stats only
         );
-        $this->assertEquals($count, 6);
+        $this->assertEquals($response[0], 6);
+        $this->assertEquals($response[1], 0);
+        $this->assertEquals($response[2], []);
 
         $response = readIndex($this->client, self::TEST_INDEX);
         $this->assertEquals(200, $response->getStatusCode());
