EQL: Add infra for planning and query folding (#52065)

Actual folding not yet in place (TBD)

(cherry picked from commit d52b96f)

Author: costin

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

@@ -6,6 +6,7 @@
 
 package org.elasticsearch.xpack.eql.analysis;
 
+import org.elasticsearch.xpack.eql.common.Failure;
 import org.elasticsearch.xpack.ql.expression.Attribute;
 import org.elasticsearch.xpack.ql.expression.NamedExpression;
 import org.elasticsearch.xpack.ql.expression.UnresolvedAttribute;

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

@@ -7,23 +7,14 @@
 
 import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.xpack.eql.EqlClientException;
-import org.elasticsearch.xpack.ql.tree.Location;
-import org.elasticsearch.xpack.ql.util.StringUtils;
+import org.elasticsearch.xpack.eql.common.Failure;
 
 import java.util.Collection;
-import java.util.stream.Collectors;
 
 public class VerificationException extends EqlClientException {
 
     protected VerificationException(Collection<Failure> sources) {
-        super(asMessage(sources));
-    }
-
-    private static String asMessage(Collection<Failure> failures) {
-        return failures.stream().map(f -> {
-            Location l = f.node().source().source();
-            return "line " + l.getLineNumber() + ":" + l.getColumnNumber() + ": " + f.message();
-        }).collect(Collectors.joining(StringUtils.NEW_LINE, "Found " + failures.size() + " problem(s)\n", StringUtils.EMPTY));
+        super(Failure.failMessage(sources));
     }
 
     @Override

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

@@ -6,6 +6,7 @@
 
 package org.elasticsearch.xpack.eql.analysis;
 
+import org.elasticsearch.xpack.eql.common.Failure;
 import org.elasticsearch.xpack.ql.capabilities.Unresolvable;
 import org.elasticsearch.xpack.ql.expression.Attribute;
 import org.elasticsearch.xpack.ql.expression.UnresolvedAttribute;
@@ -22,7 +23,7 @@
 import java.util.Set;
 
 import static java.util.stream.Collectors.toMap;
-import static org.elasticsearch.xpack.eql.analysis.Failure.fail;
+import static org.elasticsearch.xpack.eql.common.Failure.fail;
 
 /**
  * The verifier has the role of checking the analyzed tree for failures and build a list of failures following this check.

x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/analysis/Failure.java renamed to x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/common/Failure.java

@@ -4,29 +4,33 @@
  * you may not use this file except in compliance with the Elastic License.
  */
 
-package org.elasticsearch.xpack.eql.analysis;
+package org.elasticsearch.xpack.eql.common;
 
+import org.elasticsearch.xpack.ql.tree.Location;
 import org.elasticsearch.xpack.ql.tree.Node;
+import org.elasticsearch.xpack.ql.util.StringUtils;
 
+import java.util.Collection;
 import java.util.Objects;
+import java.util.stream.Collectors;
 
 import static org.elasticsearch.common.logging.LoggerMessageFormat.format;
 
-class Failure {
+public class Failure {
 
     private final Node<?> node;
     private final String message;
 
-    Failure(Node<?> node, String message) {
+    public Failure(Node<?> node, String message) {
         this.node = node;
         this.message = message;
     }
 
-    Node<?> node() {
+    public Node<?> node() {
         return node;
     }
 
-    String message() {
+    public String message() {
         return message;
     }
 
@@ -54,7 +58,16 @@ public String toString() {
         return message;
     }
 
-    static Failure fail(Node<?> source, String message, Object... args) {
+    public static Failure fail(Node<?> source, String message, Object... args) {
         return new Failure(source, format(message, args));
     }
+
+    public static String failMessage(Collection<Failure> failures) {
+        return failures.stream().map(f -> {
+            Location l = f.node().source().source();
+            return "line " + l.getLineNumber() + ":" + l.getColumnNumber() + ": " + f.message();
+        }).collect(Collectors.joining(StringUtils.NEW_LINE,
+                format("Found {} problem{}\n", failures.size(), failures.size() > 1 ? "s" : StringUtils.EMPTY), 
+                StringUtils.EMPTY));
+    }
 }

x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/plan/physical/EsQueryExec.java

@@ -0,0 +1,80 @@
+/*
+ * 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.plan.physical;
+
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.xpack.eql.querydsl.container.QueryContainer;
+import org.elasticsearch.xpack.eql.session.EqlSession;
+import org.elasticsearch.xpack.eql.session.Results;
+import org.elasticsearch.xpack.ql.expression.Attribute;
+import org.elasticsearch.xpack.ql.tree.NodeInfo;
+import org.elasticsearch.xpack.ql.tree.Source;
+
+import java.util.List;
+import java.util.Objects;
+
+public class EsQueryExec extends LeafExec {
+
+    private final String index;
+    private final List<Attribute> output;
+    private final QueryContainer queryContainer;
+
+    public EsQueryExec(Source source, String index, List<Attribute> output, QueryContainer queryContainer) {
+        super(source);
+        this.index = index;
+        this.output = output;
+        this.queryContainer = queryContainer;
+    }
+
+    @Override
+    protected NodeInfo<EsQueryExec> info() {
+        return NodeInfo.create(this, EsQueryExec::new, index, output, queryContainer);
+    }
+
+    public EsQueryExec with(QueryContainer queryContainer) {
+        return new EsQueryExec(source(), index, output, queryContainer);
+    }
+
+    public String index() {
+        return index;
+    }
+
+    @Override
+    public List<Attribute> output() {
+        return output;
+    }
+
+    @Override
+    public void execute(EqlSession session, ActionListener<Results> listener) {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(index, queryContainer, output);
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (this == obj) {
+            return true;
+        }
+
+        if (obj == null || getClass() != obj.getClass()) {
+            return false;
+        }
+
+        EsQueryExec other = (EsQueryExec) obj;
+        return Objects.equals(index, other.index)
+                && Objects.equals(queryContainer, other.queryContainer)
+                && Objects.equals(output, other.output);
+    }
+
+    @Override
+    public String nodeString() {
+        return nodeName() + "[" + index + "," + queryContainer + "]";
+    }
+}

x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/plan/physical/FilterExec.java

@@ -0,0 +1,74 @@
+/*
+ * 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.plan.physical;
+
+import org.elasticsearch.xpack.ql.expression.Attribute;
+import org.elasticsearch.xpack.ql.expression.Expression;
+import org.elasticsearch.xpack.ql.tree.NodeInfo;
+import org.elasticsearch.xpack.ql.tree.Source;
+
+import java.util.List;
+import java.util.Objects;
+
+public class FilterExec extends UnaryExec implements Unexecutable {
+
+    private final Expression condition;
+    private final boolean onAggs;
+
+    public FilterExec(Source source, PhysicalPlan child, Expression condition) {
+        this(source, child, condition, false);
+    }
+
+    public FilterExec(Source source, PhysicalPlan child, Expression condition, boolean onAggs) {
+        super(source, child);
+        this.condition = condition;
+        this.onAggs = onAggs;
+    }
+
+    @Override
+    protected NodeInfo<FilterExec> info() {
+        return NodeInfo.create(this, FilterExec::new, child(), condition, onAggs);
+    }
+
+    @Override
+    protected FilterExec replaceChild(PhysicalPlan newChild) {
+        return new FilterExec(source(), newChild, condition, onAggs);
+    }
+
+    public Expression condition() {
+        return condition;
+    }
+
+    public boolean onAggs() {
+        return onAggs;
+    }
+
+    @Override
+    public List<Attribute> output() {
+        return child().output();
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(condition, onAggs, child());
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (this == obj) {
+            return true;
+        }
+
+        if (obj == null || getClass() != obj.getClass()) {
+            return false;
+        }
+
+        FilterExec other = (FilterExec) obj;
+        return onAggs == other.onAggs
+                && Objects.equals(condition, other.condition)
+                && Objects.equals(child(), other.child());
+    }
+}

x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/plan/physical/LeafExec.java

@@ -0,0 +1,23 @@
+/*
+ * 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.plan.physical;
+
+import org.elasticsearch.xpack.ql.tree.Source;
+
+import java.util.Collections;
+import java.util.List;
+
+public abstract class LeafExec extends PhysicalPlan {
+
+    protected LeafExec(Source source) {
+        super(source, Collections.emptyList());
+    }
+
+    @Override
+    public final LeafExec replaceChildren(List<PhysicalPlan> newChildren) {
+        throw new UnsupportedOperationException("this type of node doesn't have any children to replace");
+    }
+}

x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/plan/physical/LimitExec.java

@@ -0,0 +1,56 @@
+/*
+ * 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.plan.physical;
+
+import org.elasticsearch.xpack.ql.expression.Expression;
+import org.elasticsearch.xpack.ql.tree.NodeInfo;
+import org.elasticsearch.xpack.ql.tree.Source;
+
+import java.util.Objects;
+
+public class LimitExec extends UnaryExec implements Unexecutable {
+
+    private final Expression limit;
+
+    public LimitExec(Source source, PhysicalPlan child, Expression limit) {
+        super(source, child);
+        this.limit = limit;
+    }
+
+    @Override
+    protected NodeInfo<LimitExec> info() {
+        return NodeInfo.create(this, LimitExec::new, child(), limit);
+    }
+
+    @Override
+    protected LimitExec replaceChild(PhysicalPlan newChild) {
+        return new LimitExec(source(), newChild, limit);
+    }
+
+    public Expression limit() {
+        return limit;
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(limit, child());
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (this == obj) {
+            return true;
+        }
+
+        if (obj == null || getClass() != obj.getClass()) {
+            return false;
+        }
+
+        LimitExec other = (LimitExec) obj;
+        return Objects.equals(limit, other.limit)
+                && Objects.equals(child(), other.child());
+    }
+}