EQL: Add option for returning results from the tail of the stream (elastic#64869)

Introduce option for specifying whether the results are returned from
the tail (end) of the stream or the head (beginning).
Improve sequencing algorithm by significantly eliminating the number
of in-flight sequences for spare datasets.
Refactor the sequence class by eliminating some of the redundant code.
Change matching behavior for tail sequences.
Return results based on their first entry ordinal instead of
insertion order (which was ordered on the last match ordinal).
Randomize results position inside test suite.

Close elastic#58646

Author: costin

client/rest-high-level/src/main/java/org/elasticsearch/client/eql/EqlSearchRequest.java

@@ -39,6 +39,7 @@ public class EqlSearchRequest implements Validatable, ToXContentObject {
     private QueryBuilder filter = null;
     private String timestampField = "@timestamp";
     private String eventCategoryField = "event.category";
+    private String resultPosition = "head";
 
     private int size = 10;
     private int fetchSize = 1000;
@@ -57,6 +58,7 @@ public class EqlSearchRequest implements Validatable, ToXContentObject {
     static final String KEY_SIZE = "size";
     static final String KEY_FETCH_SIZE = "fetch_size";
     static final String KEY_QUERY = "query";
+    static final String KEY_RESULT_POSITION = "result_position";
     static final String KEY_WAIT_FOR_COMPLETION_TIMEOUT = "wait_for_completion_timeout";
     static final String KEY_KEEP_ALIVE = "keep_alive";
     static final String KEY_KEEP_ON_COMPLETION = "keep_on_completion";
@@ -79,6 +81,7 @@ public XContentBuilder toXContent(XContentBuilder builder, ToXContent.Params par
         builder.field(KEY_EVENT_CATEGORY_FIELD, eventCategoryField());
         builder.field(KEY_SIZE, size());
         builder.field(KEY_FETCH_SIZE, fetchSize());
+        builder.field(KEY_RESULT_POSITION, resultPosition());
 
         builder.field(KEY_QUERY, query);
         if (waitForCompletionTimeout != null) {
@@ -140,6 +143,19 @@ public EqlSearchRequest eventCategoryField(String eventCategoryField) {
         return this;
     }
 
+    public String resultPosition() {
+        return resultPosition;
+    }
+
+    public EqlSearchRequest resultPosition(String position) {
+        if ("head".equals(position) || "tail".equals(position)) {
+            resultPosition = position;
+        } else {
+            throw new IllegalArgumentException("result position needs to be 'head' or 'tail', received '" + position + "'");
+        }
+        return this;
+    }
+
     public int size() {
         return this.size;
     }
@@ -211,6 +227,7 @@ public boolean equals(Object o) {
         EqlSearchRequest that = (EqlSearchRequest) o;
         return size == that.size &&
                 fetchSize == that.fetchSize &&
+                resultPosition == that.resultPosition &&
                 Arrays.equals(indices, that.indices) &&
                 Objects.equals(indicesOptions, that.indicesOptions) &&
                 Objects.equals(filter, that.filter) &&
@@ -237,6 +254,7 @@ public int hashCode() {
             tiebreakerField,
             eventCategoryField,
             query,
+            resultPosition,
             waitForCompletionTimeout,
             keepAlive,
             keepOnCompletion);

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

@@ -7,6 +7,7 @@
 package org.elasticsearch.test.eql;
 
 import org.apache.http.HttpHost;
+import org.apache.http.client.config.RequestConfig;
 import org.elasticsearch.client.EqlClient;
 import org.elasticsearch.client.Request;
 import org.elasticsearch.client.RequestOptions;
@@ -118,11 +119,19 @@ protected EqlSearchResponse runQuery(String index, String query) throws Exceptio
         // some queries return more than 10 results
         request.size(50);
         request.fetchSize(randomIntBetween(2, 50));
+        request.resultPosition(randomBoolean() ? "head" : "tail");
         return runRequest(eqlClient(), request);
     }
 
     protected  EqlSearchResponse runRequest(EqlClient eqlClient, EqlSearchRequest request) throws IOException {
-        return eqlClient.search(request, RequestOptions.DEFAULT);
+        int timeout = Math.toIntExact(timeout().millis());
+
+        RequestConfig config = RequestConfig.copy(RequestConfig.DEFAULT)
+            .setConnectionRequestTimeout(timeout)
+            .setConnectTimeout(timeout)
+            .setSocketTimeout(timeout)
+            .build();
+        return eqlClient.search(request, RequestOptions.DEFAULT.toBuilder().setRequestConfig(config).build());
     }
 
     protected EqlClient eqlClient() {

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

@@ -268,7 +268,8 @@ sequence by unique_pid
   [any where true]
   [any where serial_event_id < 72]
 '''
-expected_event_ids  = [54, 55, 59,
+expected_event_ids  = [
+                       54, 55, 59,
                        55, 59, 61,
                        59, 61, 65,
                        16, 60, 66,

x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/action/EqlSearchRequest.java

@@ -49,6 +49,7 @@ public class EqlSearchRequest extends ActionRequest implements IndicesRequest.Re
     private int size = RequestDefaults.SIZE;
     private int fetchSize = RequestDefaults.FETCH_SIZE;
     private String query;
+    private String resultPosition = "head";
 
     // Async settings
     private TimeValue waitForCompletionTimeout = null;
@@ -65,6 +66,7 @@ public class EqlSearchRequest extends ActionRequest implements IndicesRequest.Re
     static final String KEY_WAIT_FOR_COMPLETION_TIMEOUT = "wait_for_completion_timeout";
     static final String KEY_KEEP_ALIVE = "keep_alive";
     static final String KEY_KEEP_ON_COMPLETION = "keep_on_completion";
+    static final String KEY_RESULT_POSITION = "result_position";
 
     static final ParseField FILTER = new ParseField(KEY_FILTER);
     static final ParseField TIMESTAMP_FIELD = new ParseField(KEY_TIMESTAMP_FIELD);
@@ -76,6 +78,7 @@ public class EqlSearchRequest extends ActionRequest implements IndicesRequest.Re
     static final ParseField WAIT_FOR_COMPLETION_TIMEOUT = new ParseField(KEY_WAIT_FOR_COMPLETION_TIMEOUT);
     static final ParseField KEEP_ALIVE = new ParseField(KEY_KEEP_ALIVE);
     static final ParseField KEEP_ON_COMPLETION = new ParseField(KEY_KEEP_ON_COMPLETION);
+    static final ParseField RESULT_POSITION = new ParseField(KEY_RESULT_POSITION);
 
     private static final ObjectParser<EqlSearchRequest, Void> PARSER = objectParser(EqlSearchRequest::new);
 
@@ -99,6 +102,9 @@ public EqlSearchRequest(StreamInput in) throws IOException {
             this.keepAlive = in.readOptionalTimeValue();
             this.keepOnCompletion = in.readBoolean();
         }
+        if (in.getVersion().onOrAfter(Version.V_7_10_0)) {
+            resultPosition = in.readString();
+        }
     }
 
     @Override
@@ -168,6 +174,7 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
             builder.field(KEY_KEEP_ALIVE, keepAlive);
         }
         builder.field(KEY_KEEP_ON_COMPLETION, keepOnCompletion);
+        builder.field(KEY_RESULT_POSITION, resultPosition);
 
         return builder;
     }
@@ -192,6 +199,7 @@ protected static <R extends EqlSearchRequest> ObjectParser<R, Void> objectParser
         parser.declareField(EqlSearchRequest::keepAlive,
             (p, c) -> TimeValue.parseTimeValue(p.text(), KEY_KEEP_ALIVE), KEEP_ALIVE, ObjectParser.ValueType.VALUE);
         parser.declareBoolean(EqlSearchRequest::keepOnCompletion, KEEP_ON_COMPLETION);
+        parser.declareString(EqlSearchRequest::resultPosition, RESULT_POSITION);
         return parser;
     }
 
@@ -281,6 +289,19 @@ public EqlSearchRequest keepOnCompletion(boolean keepOnCompletion) {
         return this;
     }
 
+    public String resultPosition() {
+        return resultPosition;
+    }
+
+    public EqlSearchRequest resultPosition(String position) {
+        if ("head".equals(position) || "tail".equals(position)) {
+            resultPosition = position;
+        } else {
+            throw new IllegalArgumentException("result position needs to be 'head' or 'tail', received '" + position + "'");
+        }
+        return this;
+    }
+
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
@@ -298,6 +319,10 @@ public void writeTo(StreamOutput out) throws IOException {
             out.writeOptionalTimeValue(keepAlive);
             out.writeBoolean(keepOnCompletion);
         }
+
+        if (out.getVersion().onOrAfter(Version.V_7_10_0)) { // TODO: Remove after backport
+            out.writeString(resultPosition);
+        }
     }
 
     @Override
@@ -321,7 +346,8 @@ public boolean equals(Object o) {
                 Objects.equals(eventCategoryField, that.eventCategoryField) &&
                 Objects.equals(query, that.query) &&
                 Objects.equals(waitForCompletionTimeout, that.waitForCompletionTimeout) &&
-                Objects.equals(keepAlive, that.keepAlive);
+                Objects.equals(keepAlive, that.keepAlive) &&
+                Objects.equals(resultPosition, that.resultPosition);
     }
 
 
@@ -338,7 +364,8 @@ public int hashCode() {
             eventCategoryField,
             query,
             waitForCompletionTimeout,
-            keepAlive);
+            keepAlive,
+            resultPosition);
     }
 
     @Override

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

@@ -26,7 +26,6 @@
 public class BoxedQueryRequest implements QueryRequest {
 
     private final RangeQueryBuilder timestampRange;
-
     private final SearchSourceBuilder searchSource;
 
     private Ordinal from, to;
@@ -61,6 +60,16 @@ public BoxedQueryRequest from(Ordinal begin) {
         return this;
     }
 
+    /**
+     * Sets the upper boundary for the query (inclusive).
+     * Can be removed through null.
+     */
+    public BoxedQueryRequest to(Ordinal end) {
+        to = end;
+        timestampRange.lte(end != null ? end.timestamp() : null);
+        return this;
+    }
+
     public Ordinal after() {
         return after;
     }
@@ -69,13 +78,8 @@ public Ordinal from() {
         return from;
     }
 
-    /**
-     * Sets the upper boundary for the query (inclusive).
-     */
-    public BoxedQueryRequest to(Ordinal end) {
-        to = end;
-        timestampRange.lte(end != null ? end.timestamp() : null);
-        return this;
+    public Ordinal to() {
+        return to;
     }
 
     @Override

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

@@ -23,35 +23,39 @@ public class Criterion<Q extends QueryRequest> {
     private final HitExtractor timestamp;
     private final HitExtractor tiebreaker;
 
-    private final boolean reverse;
+    private final boolean descending;
 
     Criterion(int stage,
               Q queryRequest,
               List<HitExtractor> keys,
               HitExtractor timestamp,
               HitExtractor tiebreaker,
-              boolean reverse) {
+              boolean descending) {
         this.stage = stage;
         this.queryRequest = queryRequest;
         this.keys = keys;
         this.timestamp = timestamp;
         this.tiebreaker = tiebreaker;
 
-        this.reverse = reverse;
+        this.descending = descending;
     }
 
     public int stage() {
         return stage;
     }
 
-    public boolean reverse() {
-        return reverse;
+    public boolean descending() {
+        return descending;
     }
 
     public Q queryRequest() {
         return queryRequest;
     }
 
+    public int keySize() {
+        return keys.size();
+    }
+
     public SequenceKey key(SearchHit hit) {
         SequenceKey key;
         if (keys.isEmpty()) {
@@ -89,6 +93,6 @@ public Ordinal ordinal(SearchHit hit) {
 
     @Override
     public String toString() {
-        return "[" + stage + "][" + reverse + "]";
+        return "[" + stage + "][" + descending + "]";
     }
-}
+}

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

@@ -57,7 +57,6 @@ public Executable assemble(List<List<Attribute>> listOfKeys,
         HitExtractor tbExtractor = Expressions.isPresent(tiebreaker) ? hitExtractor(tiebreaker, extractorRegistry) : null;
         // NB: since there's no aliasing inside EQL, the attribute name is the same as the underlying field name
         String timestampName = Expressions.name(timestamp);
-        String tiebreakerName = Expressions.isPresent(tiebreaker) ? Expressions.name(tiebreaker) : null;
 
         // secondary criteria
         List<Criterion<BoxedQueryRequest>> criteria = new ArrayList<>(plans.size() - 1);

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

@@ -13,9 +13,10 @@
 import java.util.Map;
 import java.util.Map.Entry;
 
-/** Dedicated collection for mapping a key to a list of sequences */
-/** The list represents the sequence for each stage (based on its index) and is fixed in size */
-
+/**
+ * Dedicated collection for mapping a key to a list of sequences
+ * The list represents the sequence for each stage (based on its index) and is fixed in size
+ */
 class KeyToSequences {
 
     private final int listSize;
@@ -52,24 +53,6 @@ void add(int stage, Sequence sequence) {
         groups[stage].add(sequence);
     }
 
-    void resetGroupInsertPosition() {
-        for (SequenceGroup[] groups : keyToSequences.values()) {
-            for (SequenceGroup group : groups) {
-                if (group != null) {
-                    group.resetInsertPosition();
-                }
-            }
-        }
-    }
-
-    void resetUntilInsertPosition() {
-        for (UntilGroup until : keyToUntil.values()) {
-            if (until != null) {
-                until.resetInsertPosition();
-            }
-        }
-    }
-
     void until(Iterable<KeyAndOrdinal> until) {
         for (KeyAndOrdinal keyAndOrdinal : until) {
             // ignore unknown keys
@@ -116,17 +99,26 @@ void dropUntil() {
         keyToUntil.clear();
     }
 
+    /**
+     * Remove all matches expect the latest.
+     */
+    void trimToTail() {
+        for (SequenceGroup[] groups : keyToSequences.values()) {
+            for (SequenceGroup group : groups) {
+                if (group != null) {
+                    group.trimToLast();
+                }
+            }
+        }
+    }
+
     public void clear() {
         keyToSequences.clear();
         keyToUntil.clear();
     }
 
-    int numberOfKeys() {
-        return keyToSequences.size();
-    }
-
     @Override
     public String toString() {
         return LoggerMessageFormat.format(null, "Keys=[{}], Until=[{}]", keyToSequences.size(), keyToUntil.size());
     }
-}
+}

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

@@ -12,7 +12,7 @@
 import java.util.Objects;
 
 /**
- * A match within a sequence, holding the result and occurrance time.
+ * A match within a sequence, holding the result and occurrence time.
  */
 class Match {