EQL: Add option for returning results from the tail of the stream

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 {