Bugfixes

Author: julian-elastic

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/ExpressionQueryList.java

@@ -13,8 +13,13 @@
 
 import java.util.List;
 
+/**
+ * A {@link LookupEnrichQueryGenerator} that combines multiple {@link QueryList}s into a single query.
+ * Each query in the resulting query will be a conjunction of all queries from the input lists at the same position.
+ * In the future we can extend this to support more complex expressions, such as disjunctions or negations.
+ */
 public class ExpressionQueryList implements LookupEnrichQueryGenerator {
-    List<QueryList> queryLists;
+    private final List<QueryList> queryLists;
 
     public ExpressionQueryList(List<QueryList> queryLists) {
         if (queryLists.size() < 2) {
@@ -25,13 +30,10 @@ public ExpressionQueryList(List<QueryList> queryLists) {
 
     @Override
     public Query getQuery(int position) {
-        // JULIAN: Do we have a guarantee that the positions will match across all QueryLists?
-        // for now we only support AND of the queries in the lists
         BooleanQuery.Builder builder = new BooleanQuery.Builder();
         for (QueryList queryList : queryLists) {
             Query q = queryList.getQuery(position);
-            builder.add(q, BooleanClause.Occur.MUST);
-
+            builder.add(q, BooleanClause.Occur.FILTER);
         }
         return builder.build();
     }
@@ -49,6 +51,6 @@ public int getPositionCount() {
                 );
             }
         }
-        return queryLists.get(0).getPositionCount();
+        return positionCount;
     }
 }

x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/LookupFromIndexIT.java

@@ -224,16 +224,16 @@ private void runLookup(DataType keyType, PopulateIndices populateIndices) throws
                 TEST_REQUEST_TIMEOUT
             );
             final String finalNodeWithShard = nodeWithShard;
+            List<LookupFromIndexOperator.MatchConfig> matchFields = new ArrayList<>();
+            matchFields.add(new LookupFromIndexOperator.MatchConfig(new FieldAttribute.FieldName("key"), 1, keyType));
             LookupFromIndexOperator.Factory lookup = new LookupFromIndexOperator.Factory(
+                matchFields,
                 "test",
                 parentTask,
                 QueryPragmas.ENRICH_MAX_WORKERS.get(Settings.EMPTY),
-                1,
                 ctx -> internalCluster().getInstance(TransportEsqlQueryAction.class, finalNodeWithShard).getLookupFromIndexService(),
-                keyType,
                 "lookup",
                 "lookup",
-                new FieldAttribute.FieldName("key"),
                 List.of(new Alias(Source.EMPTY, "l", new ReferenceAttribute(Source.EMPTY, "l", DataType.LONG))),
                 Source.EMPTY
             );

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/AbstractLookupService.java

@@ -309,7 +309,6 @@ private void doLookup(T request, CancellableTask task, ActionListener<List<Page>
             final int[] mergingChannels = IntStream.range(0, request.extractFields.size()).map(i -> i + 2).toArray();
             final Operator finishPages;
             final OrdinalBytesRefBlock ordinalsBytesRefBlock;
-            // JULIAN: SHOULD WE DO THE NEXT CODE FOR EACH BLOCK IN THE PAGE?
             Block inputBlock = request.inputPage.getBlock(0);
             if (mergePages  // TODO fix this optimization for Lookup.
                 && inputBlock instanceof BytesRefBlock bytesRefBlock

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/LookupFromIndexOperator.java

@@ -110,7 +110,7 @@ public Operator get(DriverContext driverContext) {
     private final String lookupIndex;
     private final List<NamedExpression> loadFields;
     private final Source source;
-    private long totalTerms = 0L;
+    private long totalRows = 0L;
     private List<MatchConfig> matchFields;
     /**
      * Total number of pages emitted by this {@link Operator}.
@@ -146,14 +146,15 @@ public LookupFromIndexOperator(
 
     @Override
     protected void performAsync(Page inputPage, ActionListener<OngoingJoin> listener) {
-        // JULIAN: should I be getting multiple blocks, and send them using the LookupFromIndexService.Request
-        // is the totalTerms supposed to be the total number of terms in all blocks combined?
         Block[] inputBlockArray = new Block[matchFields.size()];
         for (int i = 0; i < matchFields.size(); i++) {
             MatchConfig matchField = matchFields.get(i);
             int inputChannel = matchField.channel;
             final Block inputBlock = inputPage.getBlock(inputChannel);
-            totalTerms += inputBlock.getTotalValueCount();
+            if (i == 0) {
+                // we only add to the totalRows once, so we can use the first block
+                totalRows += inputBlock.getTotalValueCount();
+            }
             inputBlockArray[i] = inputBlock;
         }
         LookupFromIndexService.Request request = new LookupFromIndexService.Request(
@@ -246,7 +247,7 @@ protected void doClose() {
 
     @Override
     protected Operator.Status status(long receivedPages, long completedPages, long processNanos) {
-        return new LookupFromIndexOperator.Status(receivedPages, completedPages, processNanos, totalTerms, emittedPages);
+        return new LookupFromIndexOperator.Status(receivedPages, completedPages, processNanos, totalRows, emittedPages);
     }
 
     public static class Status extends AsyncOperator.Status {