EQL: Introduce until functionality

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

@@ -25,6 +25,7 @@
 import org.junit.BeforeClass;
 
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
@@ -171,7 +172,10 @@ private RestHighLevelClient highLevelClient() {
 
     protected void assertSearchHits(List<SearchHit> events) {
         assertNotNull(events);
-        assertArrayEquals("unexpected result for spec: [" + spec.toString() + "]", spec.expectedEventIds(), extractIds(events));
+        long[] expected = spec.expectedEventIds();
+        long[] actual = extractIds(events);
+        assertArrayEquals("unexpected result for spec: [" + spec.toString() + "]" + Arrays.toString(expected) + " vs " + Arrays.toString(
+                actual), expected, actual);
     }
 
     private static long[] extractIds(List<SearchHit> events) {

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

@@ -345,36 +345,9 @@ process where true
 | sort serial_event_id
 '''
 
-
 [[queries]]
 name = "fourSequencesByPidWithUntil1"
 query = '''
-sequence
-  [process where opcode == 1] by unique_pid
-  [file where opcode == 0]    by unique_pid
-  [file where opcode == 0]    by unique_pid
-  [file where opcode == 0]    by unique_pid
-until
-  [file where opcode == 2]    by unique_pid
-'''
-expected_event_ids  = []
-
-[[queries]]
-name = "fourSequencesByPidWithUntil2"
-query = '''
-sequence
-  [process where opcode == 1] by unique_pid
-  [file where opcode == 0]    by unique_pid
-  [file where opcode == 0]    by unique_pid
-  [file where opcode == 0]    by unique_pid
-until
-  [file where opcode == 200]  by unique_pid
-'''
-expected_event_ids  = [54, 55, 61, 67]
-
-[[queries]]
-name = "doubleSameSequenceWithByAndFilter"
-query = '''
 sequence
   [file where opcode=0] by unique_pid
   [file where opcode=0] by unique_pid
@@ -385,17 +358,6 @@ expected_event_ids  = [87, 92]
 [[queries]]
 name = "doubleSameSequenceWithByUntilAndHead2"
 query = '''
-sequence
-  [file where opcode=0 and file_name="*.exe"] by unique_pid
-  [file where opcode=0 and file_name="*.exe"] by unique_pid
-until [process where opcode=1] by unique_ppid
-| head 1
-'''
-expected_event_ids = []
-
-[[queries]]
-name = "doubleJoinWithByUntilAndHead"
-query = '''
 join
   [file where opcode=0 and file_name="*.exe"] by unique_pid
   [file where opcode=2 and file_name="*.exe"] by unique_pid
@@ -698,29 +660,6 @@ expected_event_ids  = [48, 50, 51, 54, 93]
 [[queries]]
 name = "twoSequencesWithTwoKeysAndUntil"
 query = '''
-sequence by user_name
-  [file where opcode=0] by pid,file_path
-  [file where opcode=2] by pid,file_path
-until
-  [process where opcode == 2] by ppid,process_path
-'''
-expected_event_ids = []
-
-[[queries]]
-name = "twoSequencesWithUntil"
-query = '''
-sequence by user_name
-  [file where opcode=0] by pid,file_path
-  [file where opcode=2] by pid,file_path
-until
-  [process where opcode == 5] by ppid,process_path
-| head 2
-'''
-expected_event_ids  = [55, 59, 61, 65]
-
-[[queries]]
-name = "twoSequencesWithHead"
-query = '''
 join by user_name
   [file where true] by pid,file_path
   [process where true] by ppid,process_path

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

@@ -75,6 +75,10 @@ public BoxedQueryRequest from(Ordinal begin) {
         return this;
     }
 
+    public Ordinal after() {
+        return after;
+    }
+
     public Ordinal from() {
         return from;
     }

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

@@ -20,6 +20,14 @@ public class KeyAndOrdinal {
         this.ordinal = ordinal;
     }
 
+    public SequenceKey key() {
+        return key;
+    }
+
+    public Ordinal ordinal() {
+        return ordinal;
+    }
+
     @Override
     public int hashCode() {
         return Objects.hash(key, ordinal);

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

@@ -5,11 +5,12 @@
  */
 package org.elasticsearch.xpack.eql.execution.assembler;
 
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.search.SearchHit;
 import org.elasticsearch.xpack.eql.execution.search.Limit;
-import org.elasticsearch.xpack.eql.execution.search.Ordinal;
 import org.elasticsearch.xpack.eql.execution.sequence.Sequence;
 import org.elasticsearch.xpack.eql.execution.sequence.SequenceStateMachine;
 import org.elasticsearch.xpack.eql.session.Payload;
@@ -21,6 +22,8 @@
  */
 class Matcher {
 
+    private final Logger log = LogManager.getLogger(Matcher.class);
+
     // NB: just like in a list, this represents the total number of stages yet counting starts at 0
     private final SequenceStateMachine stateMachine;
     private final int numberOfStages;
@@ -48,27 +51,33 @@ boolean match(int stage, Iterable<Tuple<KeyAndOrdinal, SearchHit>> hits) {
                 // early skip in case of reaching the limit
                 // check the last stage to avoid calling the state machine in other stages
                 if (stateMachine.reachedLimit()) {
+                    log.trace("Limit reached {}", stateMachine.stats());
                     return false;
                 }
             }
         }
+        log.trace("{}", stateMachine.stats());
         return true;
     }
 
-    boolean until(Iterable<Ordinal> markers) {
-        // no-op so far
-
-        return false;
+    void until(Iterable<KeyAndOrdinal> markers) {
+        stateMachine.until(markers);
     }
 
-    public boolean hasCandidates(int stage) {
+    boolean hasCandidates(int stage) {
         return stateMachine.hasCandidates(stage);
     }
 
+    void dropUntil() {
+        stateMachine.dropUntil();
+    }
+
     Payload payload(long startTime) {
         List<Sequence> completed = stateMachine.completeSequences();
         TimeValue tookTime = new TimeValue(System.currentTimeMillis() - startTime);
-        return new SequencePayload(completed, false, tookTime);
+        Payload p = new SequencePayload(completed, false, tookTime);
+        stateMachine.clear();
+        return p;
     }
 
     @Override