EQL: Introduce support for sequences

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

@@ -118,3 +118,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

@@ -17,9 +17,8 @@
 import org.elasticsearch.xpack.ql.plan.logical.LogicalPlan;
 import org.elasticsearch.xpack.ql.rule.RuleExecutor;
 
-import java.util.ArrayList;
 import java.util.Collection;
-import java.util.List;
+import java.util.LinkedHashSet;
 
 import static java.util.Arrays.asList;
 import static org.elasticsearch.xpack.eql.analysis.AnalysisUtils.resolveAgainstList;
@@ -72,7 +71,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);
+}