EQL: Avoid filtering on tiebreakers (#63415)

Do not filter by tiebreaker while searching sequence matches as
it's not monotonic and thus can filter out valid data.
Add handling for data 'near' the boundary that has the same timestamp
but different tie-breaker and thus can be just outside the window.

Fix #62781
Relates #63215

(cherry picked from commit 36f8346)
(cherry picked from commit 72a2ce8)
(cherry picked from commit 2ab5f22)

Author: costin

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

@@ -6,48 +6,38 @@
 
 package org.elasticsearch.xpack.eql.execution.assembler;
 
-import org.elasticsearch.index.query.BoolQueryBuilder;
 import org.elasticsearch.index.query.RangeQueryBuilder;
 import org.elasticsearch.search.builder.SearchSourceBuilder;
 import org.elasticsearch.xpack.eql.execution.search.Ordinal;
 import org.elasticsearch.xpack.eql.execution.search.QueryRequest;
+import org.elasticsearch.xpack.eql.execution.search.RuntimeUtils;
 
-import static org.elasticsearch.index.query.QueryBuilders.boolQuery;
 import static org.elasticsearch.index.query.QueryBuilders.rangeQuery;
 
 /**
  * Ranged or boxed query. Provides a beginning or end to the current query.
  * The query moves between them through search_after.
- * 
+ *
  * Note that the range is not set at once on purpose since each query tends to have
  * its own number of results separate from the others.
- * As such, each query starts where it lefts to reach the current in-progress window
+ * As such, each query starts from where it left off to reach the current in-progress window
  * as oppose to always operating with the exact same window.
  */
 public class BoxedQueryRequest implements QueryRequest {
 
     private final RangeQueryBuilder timestampRange;
-    private final RangeQueryBuilder tiebreakerRange;
 
     private final SearchSourceBuilder searchSource;
 
     private Ordinal from, to;
     private Ordinal after;
 
-    public BoxedQueryRequest(QueryRequest original, String timestamp, String tiebreaker) {
+    public BoxedQueryRequest(QueryRequest original, String timestamp) {
         searchSource = original.searchSource();
 
         // setup range queries and preserve their reference to simplify the update
         timestampRange = rangeQuery(timestamp).timeZone("UTC").format("epoch_millis");
-        BoolQueryBuilder filter = boolQuery().filter(timestampRange);
-        if (tiebreaker != null) {
-            tiebreakerRange = rangeQuery(tiebreaker);
-            filter.filter(tiebreakerRange);
-        } else {
-            tiebreakerRange = null;
-        }
-        // add ranges to existing query
-        searchSource.query(filter.must(searchSource.query()));
+        RuntimeUtils.addFilter(timestampRange, searchSource);
     }
 
     @Override
@@ -69,9 +59,6 @@ public void nextAfter(Ordinal ordinal) {
     public BoxedQueryRequest from(Ordinal begin) {
         from = begin;
         timestampRange.gte(begin != null ? begin.timestamp() : null);
-        if (tiebreakerRange != null) {
-            tiebreakerRange.gte(begin != null ? begin.tiebreaker() : null);
-        }
         return this;
     }
 
@@ -85,14 +72,10 @@ public Ordinal from() {
 
     /**
      * Sets the upper boundary for the query (inclusive).
-     * Can be removed (when the query in unbounded) through null.
      */
     public BoxedQueryRequest to(Ordinal end) {
         to = end;
         timestampRange.lte(end != null ? end.timestamp() : null);
-        if (tiebreakerRange != null) {
-            tiebreakerRange.lte(end != null ? end.tiebreaker() : null);
-        }
         return this;
     }
 
@@ -104,4 +87,4 @@ public String toString() {
     private static String string(Ordinal o) {
         return o != null ? o.toString() : "<none>";
     }
-}
+}

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

@@ -72,7 +72,7 @@ public Executable assemble(List<List<Attribute>> listOfKeys,
             if (query instanceof EsQueryExec) {
                 SearchSourceBuilder source = ((EsQueryExec) query).source(session);
                 QueryRequest original = () -> source;
-                BoxedQueryRequest boxedRequest = new BoxedQueryRequest(original, timestampName, tiebreakerName);
+                BoxedQueryRequest boxedRequest = new BoxedQueryRequest(original, timestampName);
                 Criterion<BoxedQueryRequest> criterion =
                         new Criterion<>(i, boxedRequest, keyExtractors, tsExtractor, tbExtractor, i == 0 && descending);
                 criteria.add(criterion);

x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/execution/search/Ordinal.java

@@ -17,7 +17,7 @@ public Ordinal(long timestamp, Comparable<Object> tiebreaker) {
         this.timestamp = timestamp;
         this.tiebreaker = tiebreaker;
     }
-    
+
     public long timestamp() {
         return timestamp;
     }
@@ -36,11 +36,11 @@ public boolean equals(Object obj) {
         if (this == obj) {
             return true;
         }
-        
+
         if (obj == null || getClass() != obj.getClass()) {
             return false;
         }
-        
+
         Ordinal other = (Ordinal) obj;
         return Objects.equals(timestamp, other.timestamp)
                 && Objects.equals(tiebreaker, other.tiebreaker);
@@ -81,7 +81,23 @@ public boolean between(Ordinal left, Ordinal right) {
         return (compareTo(left) <= 0 && compareTo(right) >= 0) || (compareTo(right) <= 0 && compareTo(left) >= 0);
     }
 
+    public boolean before(Ordinal other) {
+        return compareTo(other) < 0;
+    }
+
+    public boolean beforeOrAt(Ordinal other) {
+        return compareTo(other) <= 0;
+    }
+
+    public boolean after(Ordinal other) {
+        return compareTo(other) > 0;
+    }
+
+    public boolean afterOrAt(Ordinal other) {
+        return compareTo(other) >= 0;
+    }
+
     public Object[] toArray() {
         return tiebreaker != null ? new Object[] { timestamp, tiebreaker } : new Object[] { timestamp };
     }
-}
+}

x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/execution/search/RuntimeUtils.java

@@ -11,6 +11,9 @@
 import org.elasticsearch.action.search.SearchRequest;
 import org.elasticsearch.action.search.SearchResponse;
 import org.elasticsearch.client.Client;
+import org.elasticsearch.index.query.BoolQueryBuilder;
+import org.elasticsearch.index.query.QueryBuilder;
+import org.elasticsearch.search.SearchHit;
 import org.elasticsearch.search.aggregations.Aggregation;
 import org.elasticsearch.search.builder.SearchSourceBuilder;
 import org.elasticsearch.xpack.eql.EqlIllegalArgumentException;
@@ -27,11 +30,14 @@
 import org.elasticsearch.xpack.ql.index.IndexResolver;
 
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Set;
 
+import static org.elasticsearch.index.query.QueryBuilders.boolQuery;
+
 public final class RuntimeUtils {
 
     static final Logger QUERY_LOG = LogManager.getLogger(QueryClient.class);
@@ -48,10 +54,10 @@ static void logSearchResponse(SearchResponse response, Logger logger) {
             aggsNames.append(aggs.get(i).getName() + (i + 1 == aggs.size() ? "" : ", "));
         }
 
-        logger.trace("Got search response [hits {} {}, {} aggregations: [{}], {} failed shards, {} skipped shards, "
-                + "{} successful shards, {} total shards, took {}, timed out [{}]]", response.getHits().getTotalHits().relation.toString(),
-                response.getHits().getTotalHits().value, aggs.size(), aggsNames, response.getFailedShards(), response.getSkippedShards(),
-                response.getSuccessfulShards(), response.getTotalShards(), response.getTook(), response.isTimedOut());
+        logger.trace("Got search response [hits {}, {} aggregations: [{}], {} failed shards, {} skipped shards, "
+                + "{} successful shards, {} total shards, took {}, timed out [{}]]", response.getHits().getTotalHits(), aggs.size(),
+                aggsNames, response.getFailedShards(), response.getSkippedShards(), response.getSuccessfulShards(),
+                response.getTotalShards(), response.getTook(), response.isTimedOut());
     }
 
     public static List<HitExtractor> createExtractor(List<FieldExtraction> fields, EqlConfiguration cfg) {
@@ -62,7 +68,7 @@ public static List<HitExtractor> createExtractor(List<FieldExtraction> fields, E
         }
         return extractors;
     }
-    
+
     public static HitExtractor createExtractor(FieldExtraction ref, EqlConfiguration cfg) {
         if (ref instanceof SearchHitFieldRef) {
             SearchHitFieldRef f = (SearchHitFieldRef) ref;
@@ -92,7 +98,7 @@ public static HitExtractor createExtractor(FieldExtraction ref, EqlConfiguration
 
         throw new EqlIllegalArgumentException("Unexpected value reference {}", ref.getClass());
     }
-    
+
 
     public static SearchRequest prepareRequest(Client client,
                                                SearchSourceBuilder source,
@@ -105,4 +111,34 @@ public static SearchRequest prepareRequest(Client client,
                         includeFrozen ? IndexResolver.FIELD_CAPS_FROZEN_INDICES_OPTIONS : IndexResolver.FIELD_CAPS_INDICES_OPTIONS)
                 .request();
     }
-}
+
+    public static List<SearchHit> searchHits(SearchResponse response) {
+        return Arrays.asList(response.getHits().getHits());
+    }
+
+    // optimized method that adds filter to existing bool queries without additional wrapping
+    // additionally checks whether the given query exists for safe decoration
+    public static SearchSourceBuilder addFilter(QueryBuilder filter, SearchSourceBuilder source) {
+        BoolQueryBuilder bool = null;
+        QueryBuilder query = source.query();
+
+        if (query instanceof BoolQueryBuilder) {
+            bool = (BoolQueryBuilder) query;
+            if (filter != null && bool.filter().contains(filter) == false) {
+                bool.filter(filter);
+            }
+        }
+        else {
+            bool = boolQuery();
+            if (query != null) {
+                bool.filter(query);
+            }
+            if (filter != null) {
+                bool.filter(filter);
+            }
+
+            source.query(bool);
+        }
+        return source;
+    }
+}

x-pack/plugin/eql/src/main/java/org/elasticsearch/xpack/eql/execution/search/SourceGenerator.java

@@ -76,6 +76,8 @@ public static SearchSourceBuilder sourceBuilder(QueryContainer container, QueryB
             }
         }
 
+        optimize(container, source);
+
         return source;
     }
 
@@ -94,7 +96,7 @@ private static void sorting(QueryContainer container, SearchSourceBuilder source
                     sortBuilder = fieldSort(fa.name())
                             .missing(as.missing().position())
                             .unmappedType(fa.dataType().esType());
-                    
+
                     if (fa.isNested()) {
                         FieldSortBuilder fieldSort = fieldSort(fa.name())
                                 .missing(as.missing().position())
@@ -134,8 +136,6 @@ private static void sorting(QueryContainer container, SearchSourceBuilder source
     }
 
     private static void optimize(QueryContainer query, SearchSourceBuilder builder) {
-        if (query.shouldTrackHits()) {
-            builder.trackTotalHits(true);
-        }
+        builder.trackTotalHits(query.shouldTrackHits());
     }
-}
+}