Support Fields API in conditional ingest processors

Author: eyalkoren

modules/ingest-common/src/yamlRestTest/resources/rest-api-spec/test/ingest/210_conditional_processor.yml

@@ -74,3 +74,39 @@ teardown:
   - match: { _source.bytes_source_field: "1kb" }
   - match: { _source.conditional_field: "bar" }
   - is_false: _source.bytes_target_field
+
+---
+"Test conditional processor with dotted field":
+  - do:
+      ingest.put_pipeline:
+        id: "my_pipeline"
+        body:  >
+          {
+            "description": "_description",
+            "processors": [
+              {
+                "bytes" : {
+                  "if" : "field('conditional.field').get('') == 'bar'",
+                  "field" : "bytes_source_field",
+                  "target_field" : "bytes_target_field"
+                }
+              }
+            ]
+          }
+  - match: { acknowledged: true }
+
+  - do:
+      index:
+        index: test
+        id: "1"
+        pipeline: "my_pipeline"
+        body: {bytes_source_field: "1kb", conditional.field: "bar"}
+
+  - do:
+      get:
+        index: test
+        id: "1"
+  - match: { _source.bytes_source_field: "1kb" }
+  - match: { _source.conditional.field: "bar" }
+  - match: { _source.bytes_target_field: 1024 }
+

modules/lang-painless/src/main/resources/org/elasticsearch/painless/org.elasticsearch.script.processor_conditional.txt

@@ -0,0 +1,37 @@
+#
+ # Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ # or more contributor license agreements. Licensed under the "Elastic License
+ # 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ # Public License v 1"; you may not use this file except in compliance with, at
+ # your election, the "Elastic License 2.0", the "GNU Affero General Public
+ # License v3.0 only", or the "Server Side Public License, v 1".
+#
+
+# This file contains a whitelist for conditional ingest scripts
+
+class org.elasticsearch.script.IngestConditionalScript {
+    WriteField field(String)
+}
+
+class org.elasticsearch.script.field.WriteField {
+    String getName()
+    boolean exists()
+    WriteField move(def)
+    WriteField overwrite(def)
+    void remove()
+    WriteField set(def)
+    WriteField append(def)
+    boolean isEmpty()
+    int size()
+    Iterator iterator()
+    def get(def)
+    def get(int, def)
+    boolean hasValue(Predicate)
+    WriteField transform(Function)
+    WriteField deduplicate()
+    WriteField removeValuesIf(Predicate)
+    WriteField removeValue(int)
+    NestedDocument doc()
+    NestedDocument doc(int)
+    Iterable docs()
+}

server/src/main/java/org/elasticsearch/ingest/ConditionalProcessor.java

@@ -56,7 +56,7 @@ public class ConditionalProcessor extends AbstractProcessor implements WrappingP
     private final Processor processor;
     private final IngestMetric metric;
     private final LongSupplier relativeTimeProvider;
-    private final IngestConditionalScript precompiledConditionScript;
+    private final IngestConditionalScript.Factory precompiledConditionScriptFactory;
 
     ConditionalProcessor(String tag, String description, Script script, ScriptService scriptService, Processor processor) {
         this(tag, description, script, scriptService, processor, System::nanoTime);
@@ -78,12 +78,10 @@ public class ConditionalProcessor extends AbstractProcessor implements WrappingP
         this.relativeTimeProvider = relativeTimeProvider;
 
         try {
-            final IngestConditionalScript.Factory factory = scriptService.compile(script, IngestConditionalScript.CONTEXT);
             if (ScriptType.INLINE.equals(script.getType())) {
-                precompiledConditionScript = factory.newInstance(script.getParams());
+                precompiledConditionScriptFactory = scriptService.compile(script, IngestConditionalScript.CONTEXT);
             } else {
-                // stored script, so will have to compile at runtime
-                precompiledConditionScript = null;
+                precompiledConditionScriptFactory = null;
             }
         } catch (ScriptException e) {
             throw newConfigurationException(TYPE, tag, null, e);
@@ -141,12 +139,12 @@ public void execute(IngestDocument ingestDocument, BiConsumer<IngestDocument, Ex
     }
 
     boolean evaluate(IngestDocument ingestDocument) {
-        IngestConditionalScript script = precompiledConditionScript;
-        if (script == null) {
-            IngestConditionalScript.Factory factory = scriptService.compile(condition, IngestConditionalScript.CONTEXT);
-            script = factory.newInstance(condition.getParams());
+        IngestConditionalScript.Factory factory = precompiledConditionScriptFactory;
+        if (factory == null) {
+            factory = scriptService.compile(condition, IngestConditionalScript.CONTEXT);
         }
-        return script.execute(new UnmodifiableIngestData(new DynamicMap(ingestDocument.getSourceAndMetadata(), FUNCTIONS)));
+        return factory.newInstance(condition.getParams(), ingestDocument.getCtxMap())
+            .execute(new UnmodifiableIngestData(new DynamicMap(ingestDocument.getSourceAndMetadata(), FUNCTIONS)));
     }
 
     public Processor getInnerProcessor() {

server/src/main/java/org/elasticsearch/script/IngestConditionalScript.java

@@ -16,9 +16,9 @@
 /**
  * A script used by {@link org.elasticsearch.ingest.ConditionalProcessor}.
  */
-public abstract class IngestConditionalScript {
+public abstract class IngestConditionalScript extends WriteScript {
 
-    public static final String[] PARAMETERS = { "ctx" };
+    public static final String[] PARAMETERS = { "ctxdsf" /*todo change param name*/ };
 
     /** The context used to compile {@link IngestConditionalScript} factories. */
     public static final ScriptContext<Factory> CONTEXT = new ScriptContext<>(
@@ -33,7 +33,8 @@ public abstract class IngestConditionalScript {
     /** The generic runtime parameters for the script. */
     private final Map<String, Object> params;
 
-    public IngestConditionalScript(Map<String, Object> params) {
+    public IngestConditionalScript(Map<String, Object> params, CtxMap<?> ctxMap) {
+        super(ctxMap);
         this.params = params;
     }
 
@@ -42,9 +43,9 @@ public Map<String, Object> getParams() {
         return params;
     }
 
-    public abstract boolean execute(Map<String, Object> ctx);
+    public abstract boolean execute(Map<String, Object> params);
 
     public interface Factory {
-        IngestConditionalScript newInstance(Map<String, Object> params);
+        IngestConditionalScript newInstance(Map<String, Object> params, CtxMap<?> ctxMap);
     }
 }

server/src/test/java/org/elasticsearch/ingest/ConditionalProcessorTests.java

@@ -206,9 +206,9 @@ public void testRuntimeCompileError() {
             if (fail.get()) {
                 throw new ScriptException("bad script", new ParseException("error", 0), List.of(), "", "lang", null);
             } else {
-                return params -> new IngestConditionalScript(params) {
+                return (params, ctxMap) -> new IngestConditionalScript(params, ctxMap) {
                     @Override
-                    public boolean execute(Map<String, Object> ctx) {
+                    public boolean execute(Map<String, Object> params) {
                         return false;
                     }
                 };
@@ -226,9 +226,9 @@ public boolean execute(Map<String, Object> ctx) {
     public void testRuntimeError() {
         ScriptService scriptService = MockScriptService.singleContext(
             IngestConditionalScript.CONTEXT,
-            code -> params -> new IngestConditionalScript(params) {
+            code -> (params, ctxMap) -> new IngestConditionalScript(params, ctxMap) {
                 @Override
-                public boolean execute(Map<String, Object> ctx) {
+                public boolean execute(Map<String, Object> params) {
                     throw new IllegalArgumentException("runtime problem");
                 }
             },

test/framework/src/main/java/org/elasticsearch/script/MockScriptEngine.java

@@ -163,10 +163,10 @@ public void execute() {
         } else if (context.instanceClazz.equals(AggregationScript.class)) {
             return context.factoryClazz.cast(new MockAggregationScript(script));
         } else if (context.instanceClazz.equals(IngestConditionalScript.class)) {
-            IngestConditionalScript.Factory factory = parameters -> new IngestConditionalScript(parameters) {
+            IngestConditionalScript.Factory factory = (parameters, ctxMap) -> new IngestConditionalScript(parameters, ctxMap) {
                 @Override
-                public boolean execute(Map<String, Object> ctx) {
-                    return (boolean) script.apply(ctx);
+                public boolean execute(Map<String, Object> params) {
+                    return (boolean) script.apply(params);
                 }
             };
             return context.factoryClazz.cast(factory);