Support Fields API in conditional ingest processors

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

@@ -74,3 +74,150 @@ teardown:
   - match: { _source.bytes_source_field: "1kb" }
   - match: { _source.conditional_field: "bar" }
   - is_false: _source.bytes_target_field
+
+---
+"Test conditional processor with fields API":
+  - do:
+      ingest.put_pipeline:
+        id: "my_pipeline"
+        body:
+          description: "_description"
+          processors:
+            - set:
+                if: "field('get.field').get('') == 'one'"
+                field: "one"
+                value: 1
+            - set:
+                if: "field('get.field').get('') == 'two'"
+                field: "missing"
+                value: "missing"
+            - set:
+                if: " /* avoid yaml stash */ $('get.field', 'one') == 'one'"
+                field: "dollar"
+                value: true
+            - set:
+                if: "field('missing.field').get('fallback') == 'fallback'"
+                field: "fallback"
+                value: "fallback"
+            - set:
+                if: "field('nested.array.get.with.index.field').get(1, null) == 'two'"
+                field: "two"
+                value: 2
+            - set:
+                if: "field('getName.field').getName() == 'getName.field'"
+                field: "three"
+                value: 3
+            - set:
+                if: "field('existing.field').exists()"
+                field: "four"
+                value: 4
+            - set:
+                if: "!field('empty.field').isEmpty()"
+                field: "missing"
+                value: "missing"
+            - set:
+                if: "field('size.field').size() == 2"
+                field: "five"
+                value: 5
+            - set:
+                if: >
+                  def iterator = field('iterator.field').iterator();
+                  def sum = 0;
+                  while (iterator.hasNext()) {
+                    sum += iterator.next();
+                  }
+                  return sum == 6;
+                field: "six"
+                value: 6
+            - set:
+                if: "field('hasValue.field').hasValue(v -> v == 'two')"
+                field: "seven"
+                value: 7
+  - match: { acknowledged: true }
+
+  - do:
+      index:
+        index: test
+        id: "1"
+        pipeline: "my_pipeline"
+        body:
+          get.field: "one"
+          nested:
+            array:
+              get.with.index.field: ["one", "two", "three"]
+          getName.field: "my_name"
+          existing.field: "indeed"
+          empty.field: []
+          size.field: ["one", "two"]
+          iterator.field: [1, 2, 3]
+          hasValue.field: ["one", "two", "three"]
+
+  - do:
+      get:
+        index: test
+        id: "1"
+  - match: { _source.get\.field: "one" }
+  - match: { _source.one: 1 }
+  - is_false: _source.missing
+  - is_true: _source.dollar
+  - match: { _source.fallback: "fallback" }
+  - match: { _source.nested.array.get\.with\.index\.field: ["one", "two", "three"] }
+  - match: { _source.two: 2 }
+  - match: { _source.three: 3 }
+  - match: { _source.four: 4 }
+  - match: { _source.five: 5 }
+  - match: { _source.six: 6 }
+  - match: { _source.seven: 7 }
+
+---
+"Test fields iterator is unmodifiable":
+  - do:
+      ingest.put_pipeline:
+        id: "my_pipeline"
+        body:
+          description: "_description"
+          processors:
+            - set:
+                if: >
+                  def iterator = field('iterator.field').iterator();
+                  def sum = 0;
+                    while (iterator.hasNext()) {
+                      sum += iterator.next();
+                      iterator.remove();
+                    }
+                  return sum == 6;
+                field: "sum"
+                value: 6
+  - match: { acknowledged: true }
+
+  - do:
+      index:
+        index: test
+        id: "1"
+        pipeline: "my_pipeline"
+        body:
+          test.field: [1, 2, 3]
+
+  - do:
+      index:
+        index: test
+        id: "2"
+        pipeline: "my_pipeline"
+        body:
+          iterator.field: [1, 2, 3]
+      catch: bad_request
+  - match: { error.root_cause.0.type: "script_exception" }
+  - match: { error.root_cause.0.reason: "runtime error" }
+
+  - do:
+      get:
+        index: test
+        id: "1"
+  - match: { _source.test\.field: [1, 2, 3] }
+  - is_false: _source.sum
+
+  - do:
+      get:
+        index: test
+        id: "2"
+      catch: missing

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

@@ -0,0 +1,25 @@
+#
+ # 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 {
+    SourceMapField field(String)
+}
+
+class org.elasticsearch.script.field.SourceMapField {
+    String getName()
+    boolean exists()
+    boolean isEmpty()
+    int size()
+    Iterator iterator()
+    def get(def)
+    def get(int, def)
+    boolean hasValue(Predicate)
+}

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 SourceMapFieldScript {
 
-    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/main/java/org/elasticsearch/script/SourceMapFieldScript.java

@@ -0,0 +1,40 @@
+/*
+ * 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".
+ */
+
+package org.elasticsearch.script;
+
+import org.elasticsearch.script.field.SourceMapField;
+
+import java.util.Map;
+
+/**
+ * Abstract base class for scripts that read field values.
+ * These scripts provide {@code ctx} for backwards compatibility and expose {@link Metadata}.
+ */
+public abstract class SourceMapFieldScript {
+    protected final CtxMap<?> ctxMap;
+
+    public SourceMapFieldScript(CtxMap<?> ctxMap) {
+        this.ctxMap = ctxMap;
+    }
+
+    /** Provides backwards compatibility access to ctx */
+    public Map<String, Object> getCtx() {
+        return ctxMap;
+    }
+
+    /** Return the metadata for this script */
+    public Metadata metadata() {
+        return ctxMap.getMetadata();
+    }
+
+    public SourceMapField field(String path) {
+        return new SourceMapField(path, ctxMap::getSource);
+    }
+}