Fix script upsert on bulk requests (#1974)

According to the [documentation](https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-update.html#scripted_upsert), it's required to add the `scripted_upsert=true` for running a upsert with script. If the flag is not set, the document is added, but without running the script. I've added a test that replicates the issue and then fixed it.
ruflin · Oct 25, 2021 · e479379 · e479379
1 parent 702427b
commit e479379
Show file tree

Hide file tree

Showing 4 changed files with 86 additions and 0 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -21,6 +21,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 * Added `Elastica\Aggregation\NormalizeAggregation` [#1956](https://github.com/ruflin/Elastica/pull/1956)
 * Added `Elastica\Suggest\Phrase::addDirectGenerator` to align with ES specification [#1964](https://github.com/ruflin/Elastica/pull/1964)
 * Added support for `psr/log` 2.0 and 3.0 [#1971](https://github.com/ruflin/Elastica/pull/1971)
+* Added support for scripted upsert on bulk requests [#1974](https://github.com/ruflin/Elastica/pull/1974)
 * Added new optional 'case_insensitive' option to `Elastica\Query\Wildcard` [#1894](https://github.com/ruflin/Elastica/pull/1894)
 * Added `Elastica\Result::getSort()` fetching the "sort" property of results [#1979](https://github.com/ruflin/Elastica/pull/1979)
 * Added exposure of Point-In-Time ID for search responses in `Elastica\ResultSet::getPointInTimeId()` [#1991](https://github.com/ruflin/Elastica/pull/1991)

diff --git a/src/Bulk/Action/UpdateDocument.php b/src/Bulk/Action/UpdateDocument.php
@@ -51,6 +51,10 @@ public function setScript(AbstractScript $script): AbstractDocument
 
             if (!empty($upsert)) {
                 $source['upsert'] = $upsert;
+
+                if ($script->getScriptedUpsert()) {
+                    $source['scripted_upsert'] = true;
+                }
             }
         }
 

diff --git a/src/Script/AbstractScript.php b/src/Script/AbstractScript.php
@@ -27,6 +27,11 @@ abstract class AbstractScript extends AbstractUpdateAction
      */
     private $_lang;
 
+    /**
+     * @var bool|null
+     */
+    private $scriptedUpsert;
+
     /**
      * @param string|null $lang       Script language, see constants
      * @param string|null $documentId Document ID the script action should be performed on (only relevant in update context)
@@ -136,4 +141,14 @@ private static function _createFromArray(array $data)
 
         throw new InvalidException('Failed to create script. Invalid data passed.');
     }
+
+    public function setScriptedUpsert(bool $scriptedUpsert): void
+    {
+        $this->scriptedUpsert = $scriptedUpsert;
+    }
+
+    public function getScriptedUpsert(): ?bool
+    {
+        return $this->scriptedUpsert;
+    }
 }
diff --git a/tests/BulkTest.php b/tests/BulkTest.php
@@ -583,6 +583,72 @@ public function testUpsert(): void
         $this->assertEquals('Maradona', $docData['name']);
     }
 
+    /**
+     * @group functional
+     */
+    public function testScriptUpsert(): void
+    {
+        $index = $this->_createIndex();
+        $client = $index->getClient();
+
+        $bulk = new Bulk($client);
+        $bulk->setIndex($index);
+
+        $id = 1;
+        $field = 123456789;
+        $subField = ['field' => 'a'];
+        $subField2 = ['field_2' => 'b'];
+
+        $defaultData = [
+            'field' => null,
+            'sub_field' => [],
+            'sub_field_2' => [],
+        ];
+
+        //insert doc and update field
+        $script = new Script('ctx._source.field = params.field', ['field' => $field]);
+        $script->setUpsert($defaultData);
+        $script->setId($id);
+        $script->setScriptedUpsert(true);
+
+        $action = AbstractDocument::create($script, Action::OP_TYPE_UPDATE);
+        $bulk->addAction($action);
+
+        //update sub_field
+        $script = new Script('if ( !ctx._source.sub_field.contains(params) ) ctx._source.sub_field.add(params)', $subField);
+        $script->setUpsert($defaultData);
+        $script->setId($id);
+        $script->setScriptedUpsert(true);
+
+        $action = AbstractDocument::create($script, Action::OP_TYPE_UPDATE);
+        $bulk->addAction($action);
+
+        //update sub_field_2
+        $script = new Script('if ( !ctx._source.sub_field_2.contains(params) ) ctx._source.sub_field_2.add(params)', $subField2);
+        $script->setUpsert($defaultData);
+        $script->setId($id);
+        $script->setScriptedUpsert(true);
+
+        $action = AbstractDocument::create($script, Action::OP_TYPE_UPDATE);
+        $bulk->addAction($action);
+
+        $response = $bulk->send();
+
+        $this->assertTrue($response->isOk());
+
+        $index->refresh();
+
+        $doc = $index->getDocument($id);
+
+        $this->assertEquals($field, $doc->getData()['field']);
+
+        $this->assertCount(1, $doc->getData()['sub_field']);
+        $this->assertCount(1, $doc->getData()['sub_field_2']);
+
+        $this->assertEquals($subField['field'], $doc->getData()['sub_field'][0]['field']);
+        $this->assertEquals($subField2['field_2'], $doc->getData()['sub_field_2'][0]['field_2']);
+    }
+
     /**
      * @group unit
      */