EQL: Add Substring function with Python semantics (#53688)

Does not reuse substring from SQL due to the difference in semantics and
the accepted arguments.
Currently it is missing full integration tests as, due to the usage of 
scripting, requires an actual integration test against a proper cluster (and 
likely its own QA project).

Author: costin

x-pack/plugin/eql/build.gradle

@@ -46,12 +46,6 @@ dependencies {
 
   // TOML parser for EqlActionIT tests
   testCompile 'io.ous:jtoml:2.0.0'
-
-  // JSON parser for tests input data
-  testCompile "com.fasterxml.jackson.core:jackson-core:${versions.jackson}"
-  testCompile "com.fasterxml.jackson.core:jackson-annotations:${versions.jackson}"
-  testCompile "com.fasterxml.jackson.core:jackson-databind:${versions.jackson}"
-
 }
 
 

x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/analysis/Analyzer.java

@@ -10,9 +10,11 @@
 import org.elasticsearch.xpack.ql.expression.Attribute;
 import org.elasticsearch.xpack.ql.expression.NamedExpression;
 import org.elasticsearch.xpack.ql.expression.UnresolvedAttribute;
+import org.elasticsearch.xpack.ql.expression.function.Function;
+import org.elasticsearch.xpack.ql.expression.function.FunctionDefinition;
 import org.elasticsearch.xpack.ql.expression.function.FunctionRegistry;
+import org.elasticsearch.xpack.ql.expression.function.UnresolvedFunction;
 import org.elasticsearch.xpack.ql.plan.logical.LogicalPlan;
-import org.elasticsearch.xpack.ql.rule.Rule;
 import org.elasticsearch.xpack.ql.rule.RuleExecutor;
 
 import java.util.ArrayList;
@@ -35,7 +37,8 @@ public Analyzer(FunctionRegistry functionRegistry, Verifier verifier) {
     @Override
     protected Iterable<RuleExecutor<LogicalPlan>.Batch> batches() {
         Batch resolution = new Batch("Resolution",
-                new ResolveRefs());
+                new ResolveRefs(),
+                new ResolveFunctions());
 
         return asList(resolution);
     }
@@ -52,7 +55,7 @@ private LogicalPlan verify(LogicalPlan plan) {
         return plan;
     }
 
-    private static class ResolveRefs extends AnalyzeRule<LogicalPlan> {
+    private static class ResolveRefs extends AnalyzerRule<LogicalPlan> {
 
         @Override
         protected LogicalPlan rule(LogicalPlan plan) {
@@ -87,20 +90,34 @@ protected LogicalPlan rule(LogicalPlan plan) {
         }
     }
 
-    abstract static class AnalyzeRule<SubPlan extends LogicalPlan> extends Rule<SubPlan, LogicalPlan> {
+    private class ResolveFunctions extends AnalyzerRule<LogicalPlan> {
 
-        // transformUp (post-order) - that is first children and then the node
-        // but with a twist; only if the tree is not resolved or analyzed
         @Override
-        public final LogicalPlan apply(LogicalPlan plan) {
-            return plan.transformUp(t -> t.analyzed() || skipResolved() && t.resolved() ? t : rule(t), typeToken());
-        }
+        protected LogicalPlan rule(LogicalPlan plan) {
+            return plan.transformExpressionsUp(e -> {
+                if (e instanceof UnresolvedFunction) {
+                    UnresolvedFunction uf = (UnresolvedFunction) e;
 
-        @Override
-        protected abstract LogicalPlan rule(SubPlan plan);
+                    if (uf.analyzed()) {
+                        return uf;
+                    }
+
+                    String name = uf.name();
 
-        protected boolean skipResolved() {
-            return true;
+                    if (uf.childrenResolved() == false) {
+                        return uf;
+                    }
+
+                    String functionName = functionRegistry.resolveAlias(name);
+                    if (functionRegistry.functionExists(functionName) == false) {
+                        return uf.missing(functionName, functionRegistry.listFunctions());
+                    }
+                    FunctionDefinition def = functionRegistry.resolveFunction(functionName);
+                    Function f = uf.buildResolved(null, def);
+                    return f;
+                }
+                return e;
+            });
         }
     }
 }

x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/execution/PlanExecutor.java

@@ -12,6 +12,7 @@
 import org.elasticsearch.xpack.eql.analysis.Analyzer;
 import org.elasticsearch.xpack.eql.analysis.PreAnalyzer;
 import org.elasticsearch.xpack.eql.analysis.Verifier;
+import org.elasticsearch.xpack.eql.expression.function.EqlFunctionRegistry;
 import org.elasticsearch.xpack.eql.optimizer.Optimizer;
 import org.elasticsearch.xpack.eql.parser.ParserParams;
 import org.elasticsearch.xpack.eql.planner.Planner;
@@ -44,7 +45,7 @@ public PlanExecutor(Client client, IndexResolver indexResolver, NamedWriteableRe
         this.writableRegistry = writeableRegistry;
 
         this.indexResolver = indexResolver;
-        this.functionRegistry = null;
+        this.functionRegistry = new EqlFunctionRegistry();
 
         this.metrics = new Metrics();
 

x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/expression/function/EqlFunctionRegistry.java

@@ -6,10 +6,30 @@
 
 package org.elasticsearch.xpack.eql.expression.function;
 
+import org.elasticsearch.xpack.eql.expression.function.scalar.string.Substring;
+import org.elasticsearch.xpack.ql.expression.function.FunctionDefinition;
 import org.elasticsearch.xpack.ql.expression.function.FunctionRegistry;
 
+import java.util.Locale;
+
 public class EqlFunctionRegistry extends FunctionRegistry {
 
     public EqlFunctionRegistry() {
+        super(functions());
+    }
+    
+    private static FunctionDefinition[][] functions() {
+        return new FunctionDefinition[][] {
+        // Scalar functions
+        // String
+            new FunctionDefinition[] {
+                def(Substring.class, Substring::new, "substring"),
+            },
+        };
+    }
+
+    @Override
+    protected String normalize(String name) {
+        return name.toLowerCase(Locale.ROOT);
     }
 }

x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/expression/function/scalar/string/StringUtils.java

@@ -0,0 +1,50 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.xpack.eql.expression.function.scalar.string;
+
+import org.elasticsearch.common.Strings;
+
+import static org.elasticsearch.common.Strings.hasLength;
+
+final class StringUtils {
+
+    private StringUtils() {}
+
+    /**
+     * Returns a substring using the Python slice semantics, meaning
+     * start and end can be negative
+     */
+    static String substringSlice(String string, int start, int end) {
+        if (hasLength(string) == false) {
+            return string;
+        }
+
+        int length = string.length();
+        
+        // handle first negative values
+        if (start < 0) {
+            start += length;
+        }
+        if (start < 0) {
+            start = 0;
+        }
+        if (end < 0) {
+            end += length;
+        }
+        if (end < 0) {
+            end = 0;
+        } else if (end > length) {
+            end = length;
+        }
+
+        if (start >= end) {
+            return org.elasticsearch.xpack.ql.util.StringUtils.EMPTY;
+        }
+
+        return Strings.substring(string, start, end);
+    }
+}

x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/expression/function/scalar/string/Substring.java

@@ -0,0 +1,129 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.xpack.eql.expression.function.scalar.string;
+
+import org.elasticsearch.xpack.ql.expression.Expression;
+import org.elasticsearch.xpack.ql.expression.Expressions;
+import org.elasticsearch.xpack.ql.expression.Expressions.ParamOrdinal;
+import org.elasticsearch.xpack.ql.expression.FieldAttribute;
+import org.elasticsearch.xpack.ql.expression.Literal;
+import org.elasticsearch.xpack.ql.expression.function.OptionalArgument;
+import org.elasticsearch.xpack.ql.expression.function.scalar.ScalarFunction;
+import org.elasticsearch.xpack.ql.expression.gen.pipeline.Pipe;
+import org.elasticsearch.xpack.ql.expression.gen.script.ScriptTemplate;
+import org.elasticsearch.xpack.ql.tree.NodeInfo;
+import org.elasticsearch.xpack.ql.tree.Source;
+import org.elasticsearch.xpack.ql.type.DataType;
+import org.elasticsearch.xpack.ql.type.DataTypes;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Locale;
+
+import static java.lang.String.format;
+import static org.elasticsearch.xpack.eql.expression.function.scalar.string.SubstringFunctionProcessor.doProcess;
+import static org.elasticsearch.xpack.ql.expression.TypeResolutions.isInteger;
+import static org.elasticsearch.xpack.ql.expression.TypeResolutions.isStringAndExact;
+import static org.elasticsearch.xpack.ql.expression.gen.script.ParamsBuilder.paramsBuilder;
+
+/**
+ * EQL specific substring function - similar to the one in Python.
+ * Note this is different than the one in SQL.
+ */
+public class Substring extends ScalarFunction implements OptionalArgument {
+
+    private final Expression source, start, end;
+
+    public Substring(Source source, Expression src, Expression start, Expression end) {
+        super(source, Arrays.asList(src, start, end != null ? end : new Literal(source, null, DataTypes.NULL)));
+        this.source = src;
+        this.start = start;
+        this.end = arguments().get(2);
+    }
+
+    @Override
+    protected TypeResolution resolveType() {
+        if (!childrenResolved()) {
+            return new TypeResolution("Unresolved children");
+        }
+
+        TypeResolution sourceResolution = isStringAndExact(source, sourceText(), ParamOrdinal.FIRST);
+        if (sourceResolution.unresolved()) {
+            return sourceResolution;
+        }
+
+        TypeResolution startResolution = isInteger(start, sourceText(), ParamOrdinal.SECOND);
+        if (startResolution.unresolved()) {
+            return startResolution;
+        }
+
+        return isInteger(end, sourceText(), ParamOrdinal.THIRD);
+    }
+
+    @Override
+    protected Pipe makePipe() {
+        return new SubstringFunctionPipe(source(), this, Expressions.pipe(source), Expressions.pipe(start), Expressions.pipe(end));
+    }
+
+    @Override
+    public boolean foldable() {
+        return source.foldable() && start.foldable() && end.foldable();
+    }
+
+    @Override
+    public Object fold() {
+        return doProcess(source.fold(), start.fold(), end.fold());
+    }
+
+    @Override
+    protected NodeInfo<? extends Expression> info() {
+        return NodeInfo.create(this, Substring::new, source, start, end);
+    }
+
+    @Override
+    public ScriptTemplate asScript() {
+        ScriptTemplate sourceScript = asScript(source);
+        ScriptTemplate startScript = asScript(start);
+        ScriptTemplate endScript = asScript(end);
+
+        return asScriptFrom(sourceScript, startScript, endScript);
+    }
+
+    protected ScriptTemplate asScriptFrom(ScriptTemplate sourceScript, ScriptTemplate startScript, ScriptTemplate endScript) {
+        return new ScriptTemplate(format(Locale.ROOT, formatTemplate("{eql}.%s(%s,%s,%s)"),
+                "substring",
+                sourceScript.template(),
+                startScript.template(),
+                endScript.template()),
+                paramsBuilder()
+                    .script(sourceScript.params())
+                    .script(startScript.params())
+                    .script(endScript.params())
+                    .build(), dataType());
+    }
+
+    @Override
+    public ScriptTemplate scriptWithField(FieldAttribute field) {
+        return new ScriptTemplate(processScript("doc[{}].value"),
+                paramsBuilder().variable(field.exactAttribute().name()).build(),
+                dataType());
+    }
+
+    @Override
+    public DataType dataType() {
+        return DataTypes.KEYWORD;
+    }
+
+    @Override
+    public Expression replaceChildren(List<Expression> newChildren) {
+        if (newChildren.size() != 3) {
+            throw new IllegalArgumentException("expected [3] children but received [" + newChildren.size() + "]");
+        }
+
+        return new Substring(source(), newChildren.get(0), newChildren.get(1), newChildren.get(2));
+    }
+}