EQL: Introduce support for sequences (#56300)

Initial support for EQL sequences
The current algorithm is focused on correctness and does not contain
any optimization which is left for the future.

The current implementation uses a state machine approach which moves
ascending and runs each query one after the other working on computing
sequences as the data comes in.
For each result, the key and its timestamp are being extracted which are
then used for matching/building a sequence.

x-pack/plugin/eql/qa/common/src/main/java/org/elasticsearch/test/eql/CommonEqlActionTestCase.java

@@ -7,15 +7,19 @@
 package org.elasticsearch.test.eql;
 
 import com.carrotsearch.randomizedtesting.annotations.ParametersFactory;
+
 import org.elasticsearch.Build;
 import org.elasticsearch.action.bulk.BulkRequest;
 import org.elasticsearch.action.bulk.BulkResponse;
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.action.support.WriteRequest;
+import org.elasticsearch.client.EqlClient;
 import org.elasticsearch.client.RequestOptions;
 import org.elasticsearch.client.RestHighLevelClient;
 import org.elasticsearch.client.eql.EqlSearchRequest;
 import org.elasticsearch.client.eql.EqlSearchResponse;
+import org.elasticsearch.client.eql.EqlSearchResponse.Hits;
+import org.elasticsearch.client.eql.EqlSearchResponse.Sequence;
 import org.elasticsearch.client.indices.CreateIndexRequest;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.io.Streams;
@@ -32,7 +36,9 @@
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
 
+import static java.util.stream.Collectors.toList;
 import static org.hamcrest.Matchers.instanceOf;
 
 public abstract class CommonEqlActionTestCase extends ESRestTestCase {
@@ -121,14 +127,15 @@ public void cleanup() throws Exception {
 
     @ParametersFactory(shuffle = false, argumentFormatting = PARAM_FORMATTING)
     public static List<Object[]> readTestSpecs() throws Exception {
-        List<Object[]> testSpecs = new ArrayList<>();
 
         // Load EQL validation specs
         List<EqlSpec> specs = EqlSpecLoader.load("/test_queries.toml", true);
         specs.addAll(EqlSpecLoader.load("/test_queries_supported.toml", true));
         List<EqlSpec> unsupportedSpecs = EqlSpecLoader.load("/test_queries_unsupported.toml", false);
 
         // Validate only currently supported specs
+        List<EqlSpec> filteredSpecs = new ArrayList<>();
+
         for (EqlSpec spec : specs) {
             boolean supported = true;
             // Check if spec is supported, simple iteration, cause the list is short.
@@ -140,18 +147,25 @@ public static List<Object[]> readTestSpecs() throws Exception {
             }
 
             if (supported) {
-                String name = spec.description();
-                if (Strings.isNullOrEmpty(name)) {
-                    name = spec.note();
-                }
-                if (Strings.isNullOrEmpty(name)) {
-                    name = spec.query();
-                }
-
-                testSpecs.add(new Object[]{++counter, name, spec});
+                filteredSpecs.add(spec);
             }
         }
-        return testSpecs;
+        counter = specs.size();
+        return asArray(filteredSpecs);
+    }
+
+    public static List<Object[]> asArray(List<EqlSpec> specs) {
+        AtomicInteger counter = new AtomicInteger();
+        return specs.stream().map(spec -> {
+            String name = spec.description();
+            if (Strings.isNullOrEmpty(name)) {
+                name = spec.note();
+            }
+            if (Strings.isNullOrEmpty(name)) {
+                name = spec.query();
+            }
+            return new Object[] { counter.incrementAndGet(), name, spec };
+        }).collect(toList());
     }
 
     private final int num;
@@ -165,9 +179,29 @@ public CommonEqlActionTestCase(int num, String name, EqlSpec spec) {
     }
 
     public void test() throws Exception {
-        EqlSearchRequest request = new EqlSearchRequest(testIndexName, spec.query());
-        EqlSearchResponse response = highLevelClient().eql().search(request, RequestOptions.DEFAULT);
-        assertSpec(response.hits().events());
+        assertResponse(runQuery(testIndexName, spec.query()));
+    }
+
+    protected void assertResponse(EqlSearchResponse response) {
+        Hits hits = response.hits();
+        if (hits.events() != null) {
+            assertSearchHits(hits.events());
+        }
+        else if (hits.sequences() != null) {
+            assertSequences(hits.sequences());
+        }
+        else {
+            fail("No events or sequences found");
+        }
+    }
+
+    protected EqlSearchResponse runQuery(String index, String query) throws Exception {
+        EqlSearchRequest request = new EqlSearchRequest(testIndexName, query);
+        return eqlClient().search(request, RequestOptions.DEFAULT);
+    }
+
+    protected EqlClient eqlClient() {
+        return highLevelClient().eql();
     }
 
     private static long[] extractIds(List<SearchHit> events) {
@@ -179,11 +213,18 @@ private static long[] extractIds(List<SearchHit> events) {
         return ids;
     }
 
-    private void assertSpec(List<SearchHit> events) {
+    protected void assertSearchHits(List<SearchHit> events) {
         assertNotNull(events);
         assertArrayEquals("unexpected result for spec: [" + spec.toString() + "]", spec.expectedEventIds(), extractIds(events));
     }
 
+    protected void assertSequences(List<Sequence> sequences) {
+        List<SearchHit> events = sequences.stream()
+                .flatMap(s -> s.events().stream())
+                .collect(toList());
+        assertSearchHits(events);
+    }
+
     private RestHighLevelClient highLevelClient() {
         if (highLevelClient == null) {
             highLevelClient = new RestHighLevelClient(

x-pack/plugin/eql/qa/common/src/main/resources/test_queries_supported.toml

@@ -124,3 +124,26 @@ query = "file where 66.0 / serial_event_id == 1"
 [[queries]]
 expected_event_ids = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 46]
 query = "process where serial_event_id + ((1 + 3) * 2 / (3 - 1)) * 2 == 54 or 70 + serial_event_id < 100"
+
+[[queries]]
+query = '''
+sequence
+  [process where serial_event_id = 1]
+  [process where serial_event_id = 2]
+'''
+expected_event_ids  = [1, 2]
+
+[[queries]]
+query = '''
+sequence
+  [process where serial_event_id=1] by unique_pid
+  [process where true] by unique_ppid'''
+expected_event_ids  = [1, 2]
+
+[[queries]]
+query = '''
+sequence
+  [process where serial_event_id<3] by unique_pid
+  [process where true] by unique_ppid
+'''
+expected_event_ids  = [1, 2, 2, 3]

x-pack/plugin/eql/qa/common/src/main/resources/test_queries_unsupported.toml

@@ -234,14 +234,6 @@ process where true
 | sort md5 event_subtype_full null_field process_name
 | sort serial_event_id'''
 
-[[queries]]
-query = '''
-sequence
-  [process where serial_event_id = 1]
-  [process where serial_event_id = 2]
-'''
-expected_event_ids  = [1, 2]
-
 [[queries]]
 query = '''
 sequence
@@ -250,13 +242,6 @@ sequence
 '''
 expected_event_ids  = [4, 5]
 
-[[queries]]
-query = '''
-sequence
-  [process where serial_event_id=1] by unique_pid
-  [process where true] by unique_ppid'''
-expected_event_ids  = [1, 2]
-
 [[queries]]
 query = '''
 sequence

x-pack/plugin/eql/src/main/antlr/EqlBase.g4

@@ -59,10 +59,14 @@ sequenceTerm
    ;
 
 subquery
-    : LB eventQuery RB
+    : LB eventFilter RB
     ;
 
 eventQuery
+    : eventFilter
+    ;
+
+eventFilter
     : (ANY | event=identifier) WHERE expression
     ;
 

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

@@ -13,10 +13,10 @@
 import org.elasticsearch.xpack.ql.type.InvalidMappedField;
 import org.elasticsearch.xpack.ql.type.UnsupportedEsField;
 
-import java.util.ArrayList;
 import java.util.Collection;
-import java.util.List;
+import java.util.LinkedHashSet;
 import java.util.Objects;
+import java.util.Set;
 
 import static java.util.stream.Collectors.toList;
 
@@ -32,7 +32,7 @@ static Attribute resolveAgainstList(UnresolvedAttribute u, Collection<Attribute>
     }
 
     static Attribute resolveAgainstList(UnresolvedAttribute u, Collection<Attribute> attrList, boolean allowCompound) {
-        List<Attribute> matches = new ArrayList<>();
+        Set<Attribute> matches = new LinkedHashSet<>();
 
         // first take into account the qualified version
         boolean qualified = u.qualifier() != null;
@@ -57,7 +57,7 @@ static Attribute resolveAgainstList(UnresolvedAttribute u, Collection<Attribute>
         }
 
         if (matches.size() == 1) {
-            return handleSpecialFields(u, matches.get(0), allowCompound);
+            return handleSpecialFields(u, matches.iterator().next(), allowCompound);
         }
 
         return u.withUnresolvedMessage(

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

@@ -18,9 +18,8 @@
 import org.elasticsearch.xpack.ql.rule.RuleExecutor;
 import org.elasticsearch.xpack.ql.session.Configuration;
 
-import java.util.ArrayList;
 import java.util.Collection;
-import java.util.List;
+import java.util.LinkedHashSet;
 
 import static java.util.Collections.singletonList;
 import static org.elasticsearch.xpack.eql.analysis.AnalysisUtils.resolveAgainstList;
@@ -75,7 +74,7 @@ protected LogicalPlan rule(LogicalPlan plan) {
             return plan.transformExpressionsUp(e -> {
                 if (e instanceof UnresolvedAttribute) {
                     UnresolvedAttribute u = (UnresolvedAttribute) e;
-                    List<Attribute> childrenOutput = new ArrayList<>();
+                    Collection<Attribute> childrenOutput = new LinkedHashSet<>();
                     for (LogicalPlan child : plan.children()) {
                         childrenOutput.addAll(child.output());
                     }

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

@@ -15,8 +15,9 @@ public class PreAnalyzer {
 
     public LogicalPlan preAnalyze(LogicalPlan plan, IndexResolution indices) {
         if (plan.analyzed() == false) {
+            final EsRelation esRelation = new EsRelation(plan.source(), indices.get(), false);
             // FIXME: includeFrozen needs to be set already
-            plan = plan.transformUp(r -> new EsRelation(r.source(), indices.get(), false), UnresolvedRelation.class);
+            plan = plan.transformUp(r -> esRelation, UnresolvedRelation.class);
             plan.forEachUp(LogicalPlan::setPreAnalyzed);
         }
         return plan;

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

@@ -0,0 +1,42 @@
+/*
+ * 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.execution.assembler;
+
+import org.elasticsearch.search.builder.SearchSourceBuilder;
+import org.elasticsearch.xpack.ql.execution.search.extractor.HitExtractor;
+
+import java.util.List;
+
+public class Criterion {
+
+    private final SearchSourceBuilder searchSource;
+    private final List<HitExtractor> keyExtractors;
+    private final HitExtractor timestampExtractor;
+
+    public Criterion(SearchSourceBuilder searchSource, List<HitExtractor> searchAfterExractors, HitExtractor timestampExtractor) {
+        this.searchSource = searchSource;
+        this.keyExtractors = searchAfterExractors;
+        this.timestampExtractor = timestampExtractor;
+    }
+
+    public SearchSourceBuilder searchSource() {
+        return searchSource;
+    }
+
+    public List<HitExtractor> keyExtractors() {
+        return keyExtractors;
+    }
+
+    public HitExtractor timestampExtractor() {
+        return timestampExtractor;
+    }
+
+    public void fromTimestamp(long timestampMarker) {
+        // TODO: this is likely to be rewritten afterwards
+        searchSource.searchAfter(new Object[] { timestampMarker });
+    }
+}

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

@@ -0,0 +1,15 @@
+/*
+ * 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.execution.assembler;
+
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.xpack.eql.session.Results;
+
+public interface Executable {
+
+    void execute(ActionListener<Results> resultsListener);
+}