Deprecate performing update operation with default pipeline or final …

…pipeline (#16712) * Deprecate performing update operation with default pipeline or final pipeline Signed-off-by: Gao Binlong <[email protected]> * Modify the warning message Signed-off-by: Gao Binlong <[email protected]> * Modify changelog Signed-off-by: Gao Binlong <[email protected]> * Fix test issue Signed-off-by: Gao Binlong <[email protected]> --------- Signed-off-by: Gao Binlong <[email protected]>
opensearch-project · Dec 2, 2024 · d2a1477 · d2a1477
1 parent b75f27a
commit d2a1477
Show file tree

Hide file tree

Showing 3 changed files with 116 additions and 1 deletion.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -43,6 +43,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Indexed IP field supports `terms_query` with more than 1025 IP masks [#16391](https://github.com/opensearch-project/OpenSearch/pull/16391)
 
 ### Deprecated
+- Performing update operation with default pipeline or final pipeline is deprecated ([#16712](https://github.com/opensearch-project/OpenSearch/pull/16712))
 
 ### Removed
 

diff --git a/modules/ingest-common/src/yamlRestTest/resources/rest-api-spec/test/ingest/75_update.yml b/modules/ingest-common/src/yamlRestTest/resources/rest-api-spec/test/ingest/75_update.yml
@@ -0,0 +1,102 @@
+setup:
+  - do:
+      ingest.put_pipeline:
+        id: "pipeline1"
+        body:  >
+          {
+            "description": "_description",
+            "processors": [
+              {
+                "set" : {
+                  "field" : "field1",
+                  "value": "value1"
+                }
+              }
+            ]
+          }
+  - do:
+      indices.create:
+        index: test_1
+        body:
+          settings:
+            index.default_pipeline: "pipeline1"
+  - do:
+      indices.create:
+        index: test_2
+        body:
+          settings:
+            index.final_pipeline: "pipeline1"
+---
+teardown:
+  - do:
+      ingest.delete_pipeline:
+        id: "pipeline1"
+        ignore: 404
+
+  - do:
+      indices.delete:
+        index: test_1
+  - do:
+      indices.delete:
+        index: test_2
+---
+"update operation with predefined default or final pipeline returns warning header":
+  - skip:
+      version:  " - 2.99.99"
+      reason:   "this change is added in 3.0.0"
+      features: allowed_warnings
+  - do:
+      index:
+        index: test_1
+        id:    1
+        body:  { foo: bar }
+
+  - match: { _seq_no:       0       }
+  - match: { _version:      1       }
+  - match: { _primary_term: 1       }
+  - match: { result:        created }
+
+  - do:
+      allowed_warnings:
+        - "the index [test_1] has a default ingest pipeline or a final ingest pipeline, the support of the ingest pipelines for update operation causes unexpected result and will be removed in 3.0.0"
+      update:
+        index:            test_1
+        id:               1
+        _source:          true
+        body:
+          doc:            { foo: bar1 }
+
+  - match: { _seq_no:       1       }
+  - match: { _primary_term: 1       }
+  - match: { _version:      2       }
+  - match: { result:        updated }
+  - match: { get._source.foo:        bar1 }
+  - match: { get._source.field1:        value1 }
+
+  - do:
+      index:
+        index: test_2
+        id:    1
+        body:  { foo: bar }
+
+  - match: { _seq_no:       0       }
+  - match: { _version:      1       }
+  - match: { _primary_term: 1       }
+  - match: { result:        created }
+
+  - do:
+      allowed_warnings:
+        - "the index [test_2] has a default ingest pipeline or a final ingest pipeline, the support of the ingest pipelines for update operation causes unexpected result and will be removed in 3.0.0"
+      update:
+        index:            test_2
+        id:               1
+        _source:          true
+        body:
+          doc:            { foo: bar1 }
+
+  - match: { _seq_no:       1       }
+  - match: { _primary_term: 1       }
+  - match: { _version:      2       }
+  - match: { result:        updated }
+  - match: { get._source.foo:        bar1 }
+  - match: { get._source.field1:        value1 }
diff --git a/server/src/main/java/org/opensearch/action/update/TransportUpdateAction.java b/server/src/main/java/org/opensearch/action/update/TransportUpdateAction.java
@@ -57,6 +57,8 @@
 import org.opensearch.cluster.service.ClusterService;
 import org.opensearch.common.collect.Tuple;
 import org.opensearch.common.inject.Inject;
+import org.opensearch.common.logging.DeprecationLogger;
+import org.opensearch.common.settings.Settings;
 import org.opensearch.common.xcontent.XContentHelper;
 import org.opensearch.core.action.ActionListener;
 import org.opensearch.core.common.bytes.BytesReference;
@@ -67,6 +69,7 @@
 import org.opensearch.core.xcontent.MediaType;
 import org.opensearch.index.IndexNotFoundException;
 import org.opensearch.index.IndexService;
+import org.opensearch.index.IndexSettings;
 import org.opensearch.index.engine.VersionConflictEngineException;
 import org.opensearch.index.shard.IndexShard;
 import org.opensearch.index.shard.IndexingStats.Stats.DocStatusStats;
@@ -90,7 +93,7 @@
  * @opensearch.internal
  */
 public class TransportUpdateAction extends TransportInstanceSingleOperationAction<UpdateRequest, UpdateResponse> {
-
+    private static final DeprecationLogger deprecationLogger = DeprecationLogger.getLogger(TransportUpdateAction.class);
     private final AutoCreateIndex autoCreateIndex;
     private final UpdateHelper updateHelper;
     private final IndicesService indicesService;
@@ -276,6 +279,15 @@ protected void shardOperation(final UpdateRequest request, final ActionListener<
                 IndexRequest indexRequest = result.action();
                 // we fetch it from the index request so we don't generate the bytes twice, its already done in the index request
                 final BytesReference indexSourceBytes = indexRequest.source();
+                final Settings indexSettings = indexService.getIndexSettings().getSettings();
+                if (IndexSettings.DEFAULT_PIPELINE.exists(indexSettings) || IndexSettings.FINAL_PIPELINE.exists(indexSettings)) {
+                    deprecationLogger.deprecate(
+                        "update_operation_with_ingest_pipeline",
+                        "the index ["
+                            + indexRequest.index()
+                            + "] has a default ingest pipeline or a final ingest pipeline, the support of the ingest pipelines for update operation causes unexpected result and will be removed in 3.0.0"
+                    );
+                }
                 client.bulk(toSingleItemBulkRequest(indexRequest), wrapBulkResponse(ActionListener.<IndexResponse>wrap(response -> {
                     UpdateResponse update = new UpdateResponse(
                         response.getShardInfo(),