Bulk Insert bulkInsert()

Return ES data

Author: pdphilip

src/Connection.php

@@ -535,15 +535,15 @@ public function delete($query, $bindings = [])
      *
      * @throws BulkInsertQueryException
      */
-    public function insert($query, $bindings = []): Elasticsearch
+    public function insert($query, $bindings = [], $continueWithErrors = false): Elasticsearch
     {
         $result = $this->run(
             $this->addClientParams($query),
             $bindings,
             $this->connection->bulk(...)
         );
 
-        if (! empty($result['errors'])) {
+        if (! $continueWithErrors && ! empty($result['errors'])) {
             throw new BulkInsertQueryException($result);
         }
 

src/Eloquent/Builder.php

@@ -70,6 +70,7 @@ class Builder extends BaseEloquentBuilder
         'bucket',
         'bucketaggregation',
         'openpit',
+        'bulkinsert',
     ];
 
     /**

src/Query/Builder.php

@@ -686,6 +686,14 @@ public function insert(array $values): MetaDTO|bool
         return $this->processor->processInsert($this, $this->connection->insert($this->grammar->compileInsert($this, $values)));
     }
 
+    /**
+     * Same as insert, but returns the ES data of the operation.
+     */
+    public function bulkInsert(array $values): array
+    {
+        return $this->processor->processBulkInsert($this, $this->connection->insert($this->grammar->compileInsert($this, $values), [], true));
+    }
+
     /**
      * {@inheritdoc}
      */

src/Query/Processor.php

@@ -427,6 +427,46 @@ public function processInsert(Builder $query, Elasticsearch $result): bool
         return ! $this->getRawResponse()['errors'];
     }
 
+    public function processBulkInsert(Builder $query, Elasticsearch $result): array
+    {
+        $this->rawResponse = $result;
+        $this->query = $query;
+
+        $process = $result->asArray();
+
+        $outcome = [
+            'hasErrors' => $process['errors'],
+            'total' => count($process['items']),
+            'took' => $process['took'],
+            'success' => 0,
+            'created' => 0,
+            'modified' => 0,
+            'failed' => 0,
+            'errors' => [],
+        ];
+        if (! empty($process['items'])) {
+            foreach ($process['items'] as $item) {
+                if (! empty($item['index']['error'])) {
+                    $outcome['errors'][] = [
+                        'id' => $item['index']['_id'],
+                        'type' => $item['index']['error']['type'],
+                        'reason' => $item['index']['error']['reason'],
+                    ];
+                    $outcome['failed']++;
+                } else {
+                    $outcome['success']++;
+                    if ($item['index']['status'] == 201) {
+                        $outcome['created']++;
+                    } else {
+                        $outcome['modified']++;
+                    }
+                }
+            }
+        }
+
+        return $outcome;
+    }
+
     public function processRaw($query, $response)
     {
         $this->rawResponse = $response;