From 2eb3d30b9f79290820a9c7d38127c34dea2d45f6 Mon Sep 17 00:00:00 2001 From: Julian Kiryakov Date: Wed, 22 Oct 2025 14:55:04 -0400 Subject: [PATCH 01/11] POC BulkKeywordQueryList (cherry picked from commit 9b73572ac21a0a26f67e28453b874a5dc381d279) # Conflicts: # x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/ExpressionQueryList.java --- .../operator/lookup/BulkKeywordQueryList.java | 101 ++++++++++++++++++ .../lookup/EnrichQuerySourceOperator.java | 15 +++ .../lookup/LookupEnrichQueryGenerator.java | 7 ++ .../esql/enrich/ExpressionQueryList.java | 63 +++++++++++ 4 files changed, 186 insertions(+) create mode 100644 x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/BulkKeywordQueryList.java diff --git a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/BulkKeywordQueryList.java b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/BulkKeywordQueryList.java new file mode 100644 index 0000000000000..9d966c4bfefc3 --- /dev/null +++ b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/BulkKeywordQueryList.java @@ -0,0 +1,101 @@ +/* + * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one + * or more contributor license agreements. Licensed under the Elastic License + * 2.0; you may not use this file except in compliance with the Elastic License + * 2.0. + */ + +package org.elasticsearch.compute.operator.lookup; + +import org.apache.lucene.index.IndexReader; +import org.apache.lucene.index.LeafReaderContext; +import org.apache.lucene.index.PostingsEnum; +import org.apache.lucene.index.TermsEnum; +import org.apache.lucene.util.BytesRef; +import org.elasticsearch.cluster.service.ClusterService; +import org.elasticsearch.compute.data.Block; +import org.elasticsearch.compute.data.BytesRefBlock; +import org.elasticsearch.compute.data.IntVector; +import org.elasticsearch.compute.operator.Warnings; +import org.elasticsearch.index.mapper.MappedFieldType; +import org.elasticsearch.index.query.SearchExecutionContext; +import org.elasticsearch.search.internal.AliasFilter; + +import java.util.function.IntFunction; + +public class BulkKeywordQueryList { + private final MappedFieldType rightFieldType; + private final SearchExecutionContext context; + private final BytesRefBlock block; + private final ClusterService clusterService; + private final AliasFilter aliasFilter; + private final Warnings warnings; + private final String fieldName; + private final IntFunction blockValueReader; + + public BulkKeywordQueryList( + MappedFieldType rightFieldType, + SearchExecutionContext context, + Block block, + ClusterService clusterService, + AliasFilter aliasFilter, + Warnings warnings + ) { + this.rightFieldType = rightFieldType; + this.context = context; + this.block = (BytesRefBlock) block; + this.clusterService = clusterService; + this.aliasFilter = aliasFilter; + this.warnings = warnings; + this.fieldName = rightFieldType.name(); + this.blockValueReader = QueryList.createBlockValueReader(block); + + } + + /** + * Process a single query at the given position using direct Lucene index access. + * This method bypasses Lucene's query framework entirely and directly accesses + * the inverted index using TermsEnum and PostingsEnum for maximum performance. + */ + public int processQuery( + int position, + IndexReader indexReader, + IntVector.Builder docsBuilder, + IntVector.Builder segmentsBuilder, + IntVector.Builder positionsBuilder + ) { + try { + final int valueCount = block.getValueCount(position); + if (valueCount != 1) { + return 0; // Skip multi-value positions and null positions + } + final int firstValueIndex = block.getFirstValueIndex(position); + BytesRef termBytes = block.getBytesRef(firstValueIndex, new BytesRef()); + int totalMatches = 0; + for (LeafReaderContext leafContext : indexReader.leaves()) { + final TermsEnum termsEnum = leafContext.reader().terms(fieldName).iterator(); + if (termsEnum.seekExact(termBytes) == false) { + continue; // Term doesn't exist in this segment + } + PostingsEnum postings = termsEnum.postings(null, 0); + int docId; + while ((docId = postings.nextDoc()) != PostingsEnum.NO_MORE_DOCS) { + docsBuilder.appendInt(docId); + if (segmentsBuilder != null) { + segmentsBuilder.appendInt(leafContext.ord); + } + positionsBuilder.appendInt(position); + totalMatches++; + } + } + return totalMatches; + } catch (Exception e) { + warnings.registerException(e); + return 0; + } + } + + public int getPositionCount() { + return block.getPositionCount(); + } +} diff --git a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/EnrichQuerySourceOperator.java b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/EnrichQuerySourceOperator.java index 256d1b7544a60..794b5842cf8cb 100644 --- a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/EnrichQuerySourceOperator.java +++ b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/EnrichQuerySourceOperator.java @@ -146,6 +146,9 @@ public Page getOutput() { if (indexReader.leaves().size() > 1) { segmentsBuilder = blockFactory.newIntVectorBuilder(estimatedSize); } + if (queryList.getBulkQueryList() != null) { + return processBulkQueries(positionsBuilder, segmentsBuilder, docsBuilder); + } int totalMatches = 0; do { Query query; @@ -193,6 +196,18 @@ public Page getOutput() { } } + private Page processBulkQueries(IntVector.Builder positionsBuilder, IntVector.Builder segmentsBuilder, IntVector.Builder docsBuilder) { + queryPosition++; + BulkKeywordQueryList bulkQueryList = queryList.getBulkQueryList(); + int totalMatches = 0; + while (queryPosition < queryList.getPositionCount() && totalMatches < maxPageSize) { + int matches = bulkQueryList.processQuery(queryPosition, indexReader, docsBuilder, segmentsBuilder, positionsBuilder); + totalMatches += matches; + queryPosition++; + } + return buildPage(totalMatches, positionsBuilder, segmentsBuilder, docsBuilder); + } + Page buildPage(int positions, IntVector.Builder positionsBuilder, IntVector.Builder segmentsBuilder, IntVector.Builder docsBuilder) { IntVector positionsVector = null; IntVector shardsVector = null; diff --git a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/LookupEnrichQueryGenerator.java b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/LookupEnrichQueryGenerator.java index 897c2aba4711b..adc879ec27983 100644 --- a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/LookupEnrichQueryGenerator.java +++ b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/LookupEnrichQueryGenerator.java @@ -29,4 +29,11 @@ public interface LookupEnrichQueryGenerator { */ int getPositionCount(Page inputPage); + /** + * Returns a BulkKeywordQueryList if applicable, null otherwise. + */ + default BulkKeywordQueryList getBulkQueryList() { + return null; + }; + } diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/ExpressionQueryList.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/ExpressionQueryList.java index 890172942697f..fdbe75001c20c 100644 --- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/ExpressionQueryList.java +++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/ExpressionQueryList.java @@ -14,8 +14,10 @@ import org.elasticsearch.compute.data.ElementType; import org.elasticsearch.compute.data.Page; import org.elasticsearch.compute.operator.Warnings; +import org.elasticsearch.compute.operator.lookup.BulkKeywordQueryList; import org.elasticsearch.compute.operator.lookup.LookupEnrichQueryGenerator; import org.elasticsearch.compute.operator.lookup.QueryList; +import org.elasticsearch.index.mapper.KeywordFieldMapper; import org.elasticsearch.index.mapper.MappedFieldType; import org.elasticsearch.index.query.QueryBuilder; import org.elasticsearch.index.query.Rewriteable; @@ -144,6 +146,11 @@ public static ExpressionQueryList expressionBasedJoin( return expressionQueryList; } + @Override + public BulkKeywordQueryList getBulkQueryList() { + return bulkKeywordQueryList; + }; + private void buildJoinOnForExpressionJoin( Expression joinOnConditions, List matchFields, @@ -152,6 +159,10 @@ private void buildJoinOnForExpressionJoin( Warnings warnings ) { List expressions = Predicates.splitAnd(joinOnConditions); + if (applyAsFastKeywordFilter(expressions, matchFields, inputPage, clusterService, warnings)) { + // we managed to apply the whole condition as a fast keyword filter + return; + } for (Expression expr : expressions) { boolean applied = applyAsLeftRightBinaryComparison(expr, matchFields, context, clusterService, warnings); if (applied == false) { @@ -163,6 +174,55 @@ private void buildJoinOnForExpressionJoin( } } + private boolean applyAsFastKeywordFilter( + List expressions, + List matchFields, + Page inputPage, + ClusterService clusterService, + Warnings warnings + ) { + if (expressions.size() == 1) { + Expression expr = expressions.get(0); + if (expr instanceof EsqlBinaryComparison binaryComparison + && binaryComparison.left() instanceof Attribute leftAttribute + && binaryComparison.right() instanceof Attribute rightAttribute) { + // the left side comes from the page that was sent to the lookup node + // the right side is the field from the lookup index + // check if the left side is in the matchFields + // if it is its corresponding page is the corresponding number in inputPage + Block block = null; + DataType dataType = null; + for (int i = 0; i < matchFields.size(); i++) { + if (matchFields.get(i).fieldName().equals(leftAttribute.name())) { + block = inputPage.getBlock(i); + dataType = matchFields.get(i).type(); + break; + } + } + MappedFieldType rightFieldType = context.getFieldType(rightAttribute.name()); + if (rightFieldType instanceof KeywordFieldMapper.KeywordFieldType == false) { + return false; + } + if (block != null && rightFieldType != null && dataType != null) { + // special handle Equals operator on keyword fields + // we can apply as a BulkKeywordQueryList for better performance + if (binaryComparison instanceof Equals equals) { + bulkKeywordQueryList = new BulkKeywordQueryList( + rightFieldType, + context, + block, + clusterService, + aliasFilter, + warnings + ); + return true; + } + } + } + } + return false; + } + private boolean applyAsRightSidePushableFilter( Expression filter, SearchExecutionContext context, @@ -316,6 +376,9 @@ public Query getQuery(int position, Page inputPage, SearchExecutionContext searc */ @Override public int getPositionCount(Page inputPage) { + if (bulkKeywordQueryList != null) { + return bulkKeywordQueryList.getPositionCount(inputPage); + } int positionCount = queryLists.get(0).getPositionCount(inputPage); for (QueryList queryList : queryLists) { if (queryList.getPositionCount(inputPage) != positionCount) { From 8bf5914142d79c01dfeb820ba3202985ea99c2d8 Mon Sep 17 00:00:00 2001 From: Julian Kiryakov Date: Wed, 22 Oct 2025 21:11:30 -0400 Subject: [PATCH 02/11] Bugfix (cherry picked from commit 4ac3891374dde27600fd0bb2a3e2b511499ff4c2) --- .../compute/operator/lookup/EnrichQuerySourceOperator.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/EnrichQuerySourceOperator.java b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/EnrichQuerySourceOperator.java index 794b5842cf8cb..4318957fe084e 100644 --- a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/EnrichQuerySourceOperator.java +++ b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/EnrichQuerySourceOperator.java @@ -200,7 +200,7 @@ private Page processBulkQueries(IntVector.Builder positionsBuilder, IntVector.Bu queryPosition++; BulkKeywordQueryList bulkQueryList = queryList.getBulkQueryList(); int totalMatches = 0; - while (queryPosition < queryList.getPositionCount() && totalMatches < maxPageSize) { + while (queryPosition < queryList.getPositionCount()) { int matches = bulkQueryList.processQuery(queryPosition, indexReader, docsBuilder, segmentsBuilder, positionsBuilder); totalMatches += matches; queryPosition++; From f17ed79c7f674150b2812e3d53f957192a66bcfe Mon Sep 17 00:00:00 2001 From: Julian Kiryakov Date: Thu, 23 Oct 2025 11:36:18 -0400 Subject: [PATCH 03/11] Address code review feedback (cherry picked from commit 75d4a7502a787ee9c52caeb7cc2d2bb043d04c74) --- .../operator/lookup/BulkKeywordQueryList.java | 36 +++++++++++++++++-- 1 file changed, 34 insertions(+), 2 deletions(-) diff --git a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/BulkKeywordQueryList.java b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/BulkKeywordQueryList.java index 9d966c4bfefc3..2b00b77cbbf63 100644 --- a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/BulkKeywordQueryList.java +++ b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/BulkKeywordQueryList.java @@ -10,7 +10,9 @@ import org.apache.lucene.index.IndexReader; import org.apache.lucene.index.LeafReaderContext; import org.apache.lucene.index.PostingsEnum; +import org.apache.lucene.index.Terms; import org.apache.lucene.index.TermsEnum; +import org.apache.lucene.util.Bits; import org.apache.lucene.util.BytesRef; import org.elasticsearch.cluster.service.ClusterService; import org.elasticsearch.compute.data.Block; @@ -21,6 +23,10 @@ import org.elasticsearch.index.query.SearchExecutionContext; import org.elasticsearch.search.internal.AliasFilter; +import java.io.IOException; +import java.io.UncheckedIOException; +import java.util.HashMap; +import java.util.Map; import java.util.function.IntFunction; public class BulkKeywordQueryList { @@ -33,6 +39,9 @@ public class BulkKeywordQueryList { private final String fieldName; private final IntFunction blockValueReader; + private final Map termsEnumCache = new HashMap<>(); + private final Map postingsCache = new HashMap<>(); + public BulkKeywordQueryList( MappedFieldType rightFieldType, SearchExecutionContext context, @@ -73,13 +82,36 @@ public int processQuery( BytesRef termBytes = block.getBytesRef(firstValueIndex, new BytesRef()); int totalMatches = 0; for (LeafReaderContext leafContext : indexReader.leaves()) { - final TermsEnum termsEnum = leafContext.reader().terms(fieldName).iterator(); + TermsEnum termsEnum = termsEnumCache.computeIfAbsent(leafContext, ctx -> { + try { + Terms terms = ctx.reader().terms(fieldName); + return terms != null ? terms.iterator() : null; + } catch (IOException e) { + throw new UncheckedIOException(e); + } + }); + if (termsEnum.seekExact(termBytes) == false) { continue; // Term doesn't exist in this segment } - PostingsEnum postings = termsEnum.postings(null, 0); + PostingsEnum postings = postingsCache.computeIfAbsent(leafContext, ctx -> { + try { + return termsEnum.postings(null, 0); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + }); + + // Reset the postings to the current term (reuse the cached PostingsEnum) + postings = termsEnum.postings(postings, 0); + + Bits liveDocs = leafContext.reader().getLiveDocs(); int docId; while ((docId = postings.nextDoc()) != PostingsEnum.NO_MORE_DOCS) { + // Check if document is not deleted + if (liveDocs != null && liveDocs.get(docId) == false) { + continue; // Skip deleted documents + } docsBuilder.appendInt(docId); if (segmentsBuilder != null) { segmentsBuilder.appendInt(leafContext.ord); From e4d63c518a8a23877bc68c0a8fdcf584bde3a576 Mon Sep 17 00:00:00 2001 From: Julian Kiryakov Date: Thu, 23 Oct 2025 14:33:51 -0400 Subject: [PATCH 04/11] Address more code review comments (cherry picked from commit 118fbecf02bd1d0c2cb8cb030a1b248c81c6df26) --- .../operator/lookup/BulkKeywordQueryList.java | 52 +++++++++++-------- .../lookup/EnrichQuerySourceOperator.java | 4 +- 2 files changed, 33 insertions(+), 23 deletions(-) diff --git a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/BulkKeywordQueryList.java b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/BulkKeywordQueryList.java index 2b00b77cbbf63..0aa977a6c9b35 100644 --- a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/BulkKeywordQueryList.java +++ b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/BulkKeywordQueryList.java @@ -24,9 +24,6 @@ import org.elasticsearch.search.internal.AliasFilter; import java.io.IOException; -import java.io.UncheckedIOException; -import java.util.HashMap; -import java.util.Map; import java.util.function.IntFunction; public class BulkKeywordQueryList { @@ -39,8 +36,9 @@ public class BulkKeywordQueryList { private final String fieldName; private final IntFunction blockValueReader; - private final Map termsEnumCache = new HashMap<>(); - private final Map postingsCache = new HashMap<>(); + private TermsEnum[] termsEnumCache = null; + private PostingsEnum[] postingsCache = null; + private final BytesRef scratch = new BytesRef(); public BulkKeywordQueryList( MappedFieldType rightFieldType, @@ -79,28 +77,19 @@ public int processQuery( return 0; // Skip multi-value positions and null positions } final int firstValueIndex = block.getFirstValueIndex(position); - BytesRef termBytes = block.getBytesRef(firstValueIndex, new BytesRef()); + BytesRef termBytes = block.getBytesRef(firstValueIndex, scratch); int totalMatches = 0; for (LeafReaderContext leafContext : indexReader.leaves()) { - TermsEnum termsEnum = termsEnumCache.computeIfAbsent(leafContext, ctx -> { - try { - Terms terms = ctx.reader().terms(fieldName); - return terms != null ? terms.iterator() : null; - } catch (IOException e) { - throw new UncheckedIOException(e); - } - }); - + int leafOrd = leafContext.ord; + TermsEnum termsEnum = termsEnumCache[leafOrd]; if (termsEnum.seekExact(termBytes) == false) { continue; // Term doesn't exist in this segment } - PostingsEnum postings = postingsCache.computeIfAbsent(leafContext, ctx -> { - try { - return termsEnum.postings(null, 0); - } catch (IOException e) { - throw new UncheckedIOException(e); - } - }); + PostingsEnum postings = postingsCache[leafOrd]; + if (postings == null) { + postings = termsEnum.postings(null, 0); + postingsCache[leafOrd] = postings; + } // Reset the postings to the current term (reuse the cached PostingsEnum) postings = termsEnum.postings(postings, 0); @@ -130,4 +119,23 @@ public int processQuery( public int getPositionCount() { return block.getPositionCount(); } + + /** + * Initialize caches for the given index reader. This should be called once + * before the first processQuery call for a given index reader. + */ + public void initializeCaches(IndexReader indexReader) throws IOException { + if (termsEnumCache == null) { + int numLeaves = indexReader.leaves().size(); + termsEnumCache = new TermsEnum[numLeaves]; + postingsCache = new PostingsEnum[numLeaves]; + + // Pre-populate caches with TermsEnum for each leaf + for (int i = 0; i < numLeaves; i++) { + LeafReaderContext leafContext = indexReader.leaves().get(i); + Terms terms = leafContext.reader().terms(fieldName); + termsEnumCache[i] = terms.iterator(); + } + } + } } diff --git a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/EnrichQuerySourceOperator.java b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/EnrichQuerySourceOperator.java index 4318957fe084e..c129b3f411ba3 100644 --- a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/EnrichQuerySourceOperator.java +++ b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/EnrichQuerySourceOperator.java @@ -196,10 +196,12 @@ public Page getOutput() { } } - private Page processBulkQueries(IntVector.Builder positionsBuilder, IntVector.Builder segmentsBuilder, IntVector.Builder docsBuilder) { + private Page processBulkQueries(IntVector.Builder positionsBuilder, IntVector.Builder segmentsBuilder, IntVector.Builder docsBuilder) + throws IOException { queryPosition++; BulkKeywordQueryList bulkQueryList = queryList.getBulkQueryList(); int totalMatches = 0; + bulkQueryList.initializeCaches(indexReader); while (queryPosition < queryList.getPositionCount()) { int matches = bulkQueryList.processQuery(queryPosition, indexReader, docsBuilder, segmentsBuilder, positionsBuilder); totalMatches += matches; From 8c92cd9007fd0a6f9a674af473a9588c77bd66da Mon Sep 17 00:00:00 2001 From: cimequinox Date: Wed, 11 Feb 2026 12:52:19 -0800 Subject: [PATCH 05/11] Resolve rebase issues ExpressionQueryList * applyAsFastKeywordFilter - only applies optimization for KEYWORD fields - accepts context and follows pattern used by applyAsLeftRightBinaryComparison to lookup dataType and channelOffset before passing to BulkKeywordQueryList constructor. * getPageCount passes inputPage when delegating to bulkKeywordQueryList EnrichQuerySourceOperator * inputPage passed to bulkQueryList methods BulkKeywordQueryList * holds channelOffset * processQuery and getPositionCount take inputPage and look up block using channelOffset --- .../operator/lookup/BulkKeywordQueryList.java | 25 ++++++---- .../lookup/EnrichQuerySourceOperator.java | 20 ++++++-- .../esql/enrich/ExpressionQueryList.java | 46 +++++++++---------- 3 files changed, 53 insertions(+), 38 deletions(-) diff --git a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/BulkKeywordQueryList.java b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/BulkKeywordQueryList.java index 0aa977a6c9b35..e5c4ca36bf1f9 100644 --- a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/BulkKeywordQueryList.java +++ b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/BulkKeywordQueryList.java @@ -17,24 +17,26 @@ import org.elasticsearch.cluster.service.ClusterService; import org.elasticsearch.compute.data.Block; import org.elasticsearch.compute.data.BytesRefBlock; +import org.elasticsearch.compute.data.ElementType; import org.elasticsearch.compute.data.IntVector; +import org.elasticsearch.compute.data.Page; import org.elasticsearch.compute.operator.Warnings; import org.elasticsearch.index.mapper.MappedFieldType; import org.elasticsearch.index.query.SearchExecutionContext; import org.elasticsearch.search.internal.AliasFilter; import java.io.IOException; -import java.util.function.IntFunction; +import java.util.function.BiFunction; public class BulkKeywordQueryList { private final MappedFieldType rightFieldType; + private final int channelOffset; private final SearchExecutionContext context; - private final BytesRefBlock block; private final ClusterService clusterService; private final AliasFilter aliasFilter; private final Warnings warnings; private final String fieldName; - private final IntFunction blockValueReader; + private final BiFunction blockValueReader; private TermsEnum[] termsEnumCache = null; private PostingsEnum[] postingsCache = null; @@ -42,21 +44,21 @@ public class BulkKeywordQueryList { public BulkKeywordQueryList( MappedFieldType rightFieldType, + ElementType leftElementType, SearchExecutionContext context, - Block block, + int channelOffset, ClusterService clusterService, AliasFilter aliasFilter, Warnings warnings ) { this.rightFieldType = rightFieldType; this.context = context; - this.block = (BytesRefBlock) block; + this.channelOffset = channelOffset; this.clusterService = clusterService; this.aliasFilter = aliasFilter; this.warnings = warnings; this.fieldName = rightFieldType.name(); - this.blockValueReader = QueryList.createBlockValueReader(block); - + this.blockValueReader = QueryList.createBlockValueReaderForType(leftElementType); } /** @@ -65,6 +67,7 @@ public BulkKeywordQueryList( * the inverted index using TermsEnum and PostingsEnum for maximum performance. */ public int processQuery( + Page inputPage, int position, IndexReader indexReader, IntVector.Builder docsBuilder, @@ -72,12 +75,13 @@ public int processQuery( IntVector.Builder positionsBuilder ) { try { + final BytesRefBlock block = inputPage.getBlock(channelOffset); final int valueCount = block.getValueCount(position); if (valueCount != 1) { return 0; // Skip multi-value positions and null positions } final int firstValueIndex = block.getFirstValueIndex(position); - BytesRef termBytes = block.getBytesRef(firstValueIndex, scratch); + final BytesRef termBytes = block.getBytesRef(firstValueIndex, scratch); int totalMatches = 0; for (LeafReaderContext leafContext : indexReader.leaves()) { int leafOrd = leafContext.ord; @@ -116,7 +120,8 @@ public int processQuery( } } - public int getPositionCount() { + public int getPositionCount(Page inputPage) { + final Block block = inputPage.getBlock(channelOffset); return block.getPositionCount(); } @@ -126,7 +131,7 @@ public int getPositionCount() { */ public void initializeCaches(IndexReader indexReader) throws IOException { if (termsEnumCache == null) { - int numLeaves = indexReader.leaves().size(); + final int numLeaves = indexReader.leaves().size(); termsEnumCache = new TermsEnum[numLeaves]; postingsCache = new PostingsEnum[numLeaves]; diff --git a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/EnrichQuerySourceOperator.java b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/EnrichQuerySourceOperator.java index c129b3f411ba3..18284437a0d9b 100644 --- a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/EnrichQuerySourceOperator.java +++ b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/EnrichQuerySourceOperator.java @@ -147,7 +147,7 @@ public Page getOutput() { segmentsBuilder = blockFactory.newIntVectorBuilder(estimatedSize); } if (queryList.getBulkQueryList() != null) { - return processBulkQueries(positionsBuilder, segmentsBuilder, docsBuilder); + return processBulkQueries(inputPage, positionsBuilder, segmentsBuilder, docsBuilder); } int totalMatches = 0; do { @@ -196,14 +196,24 @@ public Page getOutput() { } } - private Page processBulkQueries(IntVector.Builder positionsBuilder, IntVector.Builder segmentsBuilder, IntVector.Builder docsBuilder) - throws IOException { + private Page processBulkQueries(Page inputPage, + IntVector.Builder positionsBuilder, + IntVector.Builder segmentsBuilder, + IntVector.Builder docsBuilder + ) throws IOException { queryPosition++; BulkKeywordQueryList bulkQueryList = queryList.getBulkQueryList(); int totalMatches = 0; bulkQueryList.initializeCaches(indexReader); - while (queryPosition < queryList.getPositionCount()) { - int matches = bulkQueryList.processQuery(queryPosition, indexReader, docsBuilder, segmentsBuilder, positionsBuilder); + while (queryPosition < queryList.getPositionCount(inputPage)) { + int matches = bulkQueryList.processQuery( + inputPage, + queryPosition, + indexReader, + docsBuilder, + segmentsBuilder, + positionsBuilder + ); totalMatches += matches; queryPosition++; } diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/ExpressionQueryList.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/ExpressionQueryList.java index fdbe75001c20c..4ac9c6089e5cf 100644 --- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/ExpressionQueryList.java +++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/ExpressionQueryList.java @@ -17,7 +17,6 @@ import org.elasticsearch.compute.operator.lookup.BulkKeywordQueryList; import org.elasticsearch.compute.operator.lookup.LookupEnrichQueryGenerator; import org.elasticsearch.compute.operator.lookup.QueryList; -import org.elasticsearch.index.mapper.KeywordFieldMapper; import org.elasticsearch.index.mapper.MappedFieldType; import org.elasticsearch.index.query.QueryBuilder; import org.elasticsearch.index.query.Rewriteable; @@ -64,6 +63,7 @@ public class ExpressionQueryList implements LookupEnrichQueryGenerator { private final List lucenePushableFilterBuilders = new ArrayList<>(); private final AliasFilter aliasFilter; private final ClusterService clusterService; + private BulkKeywordQueryList bulkKeywordQueryList = null; private ExpressionQueryList( List queryLists, @@ -159,7 +159,7 @@ private void buildJoinOnForExpressionJoin( Warnings warnings ) { List expressions = Predicates.splitAnd(joinOnConditions); - if (applyAsFastKeywordFilter(expressions, matchFields, inputPage, clusterService, warnings)) { + if (applyAsFastKeywordFilter(expressions, matchFields, context, clusterService, warnings)) { // we managed to apply the whole condition as a fast keyword filter return; } @@ -177,7 +177,7 @@ private void buildJoinOnForExpressionJoin( private boolean applyAsFastKeywordFilter( List expressions, List matchFields, - Page inputPage, + SearchExecutionContext context, ClusterService clusterService, Warnings warnings ) { @@ -189,33 +189,33 @@ private boolean applyAsFastKeywordFilter( // the left side comes from the page that was sent to the lookup node // the right side is the field from the lookup index // check if the left side is in the matchFields - // if it is its corresponding page is the corresponding number in inputPage - Block block = null; + int channelOffset = -1; DataType dataType = null; for (int i = 0; i < matchFields.size(); i++) { if (matchFields.get(i).fieldName().equals(leftAttribute.name())) { - block = inputPage.getBlock(i); + channelOffset = i; dataType = matchFields.get(i).type(); break; } } - MappedFieldType rightFieldType = context.getFieldType(rightAttribute.name()); - if (rightFieldType instanceof KeywordFieldMapper.KeywordFieldType == false) { - return false; - } - if (block != null && rightFieldType != null && dataType != null) { - // special handle Equals operator on keyword fields - // we can apply as a BulkKeywordQueryList for better performance - if (binaryComparison instanceof Equals equals) { - bulkKeywordQueryList = new BulkKeywordQueryList( - rightFieldType, - context, - block, - clusterService, - aliasFilter, - warnings - ); - return true; + if (channelOffset != -1 && dataType == DataType.KEYWORD) { + MappedFieldType rightFieldType = context.getFieldType(rightAttribute.name()); + if (rightFieldType != null) { + // special handle Equals operator on keyword fields + // we can apply as a BulkKeywordQueryList for better performance + if (binaryComparison instanceof Equals) { + ElementType leftElementType = PlannerUtils.toElementType(dataType); + bulkKeywordQueryList = new BulkKeywordQueryList( + rightFieldType, + leftElementType, + context, + channelOffset, + clusterService, + aliasFilter, + warnings + ); + return true; + } } } } From 08e125618230dbf0f8121849b801518de0a29dec Mon Sep 17 00:00:00 2001 From: cimequinox Date: Wed, 11 Feb 2026 13:51:22 -0800 Subject: [PATCH 06/11] Rename BulkKeywordQueryList to BulkKeywordLookup The class is really a specialized helper for the EnrichQuerySourceOperator (and eventually EnrichQueryFromExchangeOperator). It doesn't inherit or extend QueryList or implement the QueryList apis such as doGetQuery. --- ...wordQueryList.java => BulkKeywordLookup.java} | 4 ++-- .../lookup/EnrichQuerySourceOperator.java | 12 +++++++----- .../lookup/LookupEnrichQueryGenerator.java | 4 ++-- .../xpack/esql/enrich/ExpressionQueryList.java | 16 ++++++++-------- 4 files changed, 19 insertions(+), 17 deletions(-) rename x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/{BulkKeywordQueryList.java => BulkKeywordLookup.java} (98%) diff --git a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/BulkKeywordQueryList.java b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/BulkKeywordLookup.java similarity index 98% rename from x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/BulkKeywordQueryList.java rename to x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/BulkKeywordLookup.java index e5c4ca36bf1f9..b8007d69c116a 100644 --- a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/BulkKeywordQueryList.java +++ b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/BulkKeywordLookup.java @@ -28,7 +28,7 @@ import java.io.IOException; import java.util.function.BiFunction; -public class BulkKeywordQueryList { +public class BulkKeywordLookup { private final MappedFieldType rightFieldType; private final int channelOffset; private final SearchExecutionContext context; @@ -42,7 +42,7 @@ public class BulkKeywordQueryList { private PostingsEnum[] postingsCache = null; private final BytesRef scratch = new BytesRef(); - public BulkKeywordQueryList( + public BulkKeywordLookup( MappedFieldType rightFieldType, ElementType leftElementType, SearchExecutionContext context, diff --git a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/EnrichQuerySourceOperator.java b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/EnrichQuerySourceOperator.java index 18284437a0d9b..3a404d585f133 100644 --- a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/EnrichQuerySourceOperator.java +++ b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/EnrichQuerySourceOperator.java @@ -146,7 +146,7 @@ public Page getOutput() { if (indexReader.leaves().size() > 1) { segmentsBuilder = blockFactory.newIntVectorBuilder(estimatedSize); } - if (queryList.getBulkQueryList() != null) { + if (queryList.getBulkKeywordLookup() != null) { return processBulkQueries(inputPage, positionsBuilder, segmentsBuilder, docsBuilder); } int totalMatches = 0; @@ -202,11 +202,11 @@ private Page processBulkQueries(Page inputPage, IntVector.Builder docsBuilder ) throws IOException { queryPosition++; - BulkKeywordQueryList bulkQueryList = queryList.getBulkQueryList(); + BulkKeywordLookup bulkKeywordLookup = queryList.getBulkKeywordLookup(); int totalMatches = 0; - bulkQueryList.initializeCaches(indexReader); + bulkKeywordLookup.initializeCaches(indexReader); while (queryPosition < queryList.getPositionCount(inputPage)) { - int matches = bulkQueryList.processQuery( + int matches = bulkKeywordLookup.processQuery( inputPage, queryPosition, indexReader, @@ -217,7 +217,9 @@ private Page processBulkQueries(Page inputPage, totalMatches += matches; queryPosition++; } - return buildPage(totalMatches, positionsBuilder, segmentsBuilder, docsBuilder); + final Page result = buildPage(totalMatches, positionsBuilder, segmentsBuilder, docsBuilder); + + return result; } Page buildPage(int positions, IntVector.Builder positionsBuilder, IntVector.Builder segmentsBuilder, IntVector.Builder docsBuilder) { diff --git a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/LookupEnrichQueryGenerator.java b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/LookupEnrichQueryGenerator.java index adc879ec27983..7ca4e672933f9 100644 --- a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/LookupEnrichQueryGenerator.java +++ b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/LookupEnrichQueryGenerator.java @@ -30,9 +30,9 @@ public interface LookupEnrichQueryGenerator { int getPositionCount(Page inputPage); /** - * Returns a BulkKeywordQueryList if applicable, null otherwise. + * Returns a BulkKeywordLookup if applicable, null otherwise. */ - default BulkKeywordQueryList getBulkQueryList() { + default BulkKeywordLookup getBulkKeywordLookup() { return null; }; diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/ExpressionQueryList.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/ExpressionQueryList.java index 4ac9c6089e5cf..61c694ac49add 100644 --- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/ExpressionQueryList.java +++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/ExpressionQueryList.java @@ -14,7 +14,7 @@ import org.elasticsearch.compute.data.ElementType; import org.elasticsearch.compute.data.Page; import org.elasticsearch.compute.operator.Warnings; -import org.elasticsearch.compute.operator.lookup.BulkKeywordQueryList; +import org.elasticsearch.compute.operator.lookup.BulkKeywordLookup; import org.elasticsearch.compute.operator.lookup.LookupEnrichQueryGenerator; import org.elasticsearch.compute.operator.lookup.QueryList; import org.elasticsearch.index.mapper.MappedFieldType; @@ -63,7 +63,7 @@ public class ExpressionQueryList implements LookupEnrichQueryGenerator { private final List lucenePushableFilterBuilders = new ArrayList<>(); private final AliasFilter aliasFilter; private final ClusterService clusterService; - private BulkKeywordQueryList bulkKeywordQueryList = null; + private BulkKeywordLookup bulkKeywordLookup = null; private ExpressionQueryList( List queryLists, @@ -147,8 +147,8 @@ public static ExpressionQueryList expressionBasedJoin( } @Override - public BulkKeywordQueryList getBulkQueryList() { - return bulkKeywordQueryList; + public BulkKeywordLookup getBulkKeywordLookup() { + return bulkKeywordLookup; }; private void buildJoinOnForExpressionJoin( @@ -202,10 +202,10 @@ private boolean applyAsFastKeywordFilter( MappedFieldType rightFieldType = context.getFieldType(rightAttribute.name()); if (rightFieldType != null) { // special handle Equals operator on keyword fields - // we can apply as a BulkKeywordQueryList for better performance + // we can apply as a BulkKeywordLookup for better performance if (binaryComparison instanceof Equals) { ElementType leftElementType = PlannerUtils.toElementType(dataType); - bulkKeywordQueryList = new BulkKeywordQueryList( + bulkKeywordLookup = new BulkKeywordLookup( rightFieldType, leftElementType, context, @@ -376,8 +376,8 @@ public Query getQuery(int position, Page inputPage, SearchExecutionContext searc */ @Override public int getPositionCount(Page inputPage) { - if (bulkKeywordQueryList != null) { - return bulkKeywordQueryList.getPositionCount(inputPage); + if (bulkKeywordLookup != null) { + return bulkKeywordLookup.getPositionCount(inputPage); } int positionCount = queryLists.get(0).getPositionCount(inputPage); for (QueryList queryList : queryLists) { From 7f1f9655b6893346d202ebcb7b086dfcd48e6903 Mon Sep 17 00:00:00 2001 From: cimequinox Date: Fri, 6 Mar 2026 13:58:07 -0800 Subject: [PATCH 07/11] Identify extractChannelOffset in BulkKeywordLookup Also produce warning when encountering multivalues on left --- .../operator/lookup/BulkKeywordLookup.java | 27 +++++++++++----- .../esql/enrich/ExpressionQueryList.java | 31 +++++++++++++++---- .../esql/enrich/LookupFromIndexService.java | 4 +++ 3 files changed, 49 insertions(+), 13 deletions(-) diff --git a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/BulkKeywordLookup.java b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/BulkKeywordLookup.java index b8007d69c116a..39f61ce149fcf 100644 --- a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/BulkKeywordLookup.java +++ b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/BulkKeywordLookup.java @@ -30,7 +30,8 @@ public class BulkKeywordLookup { private final MappedFieldType rightFieldType; - private final int channelOffset; + private final int matchChannelOffset; + private final int extractChannelOffset; private final SearchExecutionContext context; private final ClusterService clusterService; private final AliasFilter aliasFilter; @@ -46,14 +47,16 @@ public BulkKeywordLookup( MappedFieldType rightFieldType, ElementType leftElementType, SearchExecutionContext context, - int channelOffset, + int matchChannelOffset, + int extractChannelOffset, ClusterService clusterService, AliasFilter aliasFilter, Warnings warnings ) { this.rightFieldType = rightFieldType; this.context = context; - this.channelOffset = channelOffset; + this.matchChannelOffset = matchChannelOffset; // offset of field in left (input) page + this.extractChannelOffset = extractChannelOffset; // offset of field in right (output) page this.clusterService = clusterService; this.aliasFilter = aliasFilter; this.warnings = warnings; @@ -75,10 +78,16 @@ public int processQuery( IntVector.Builder positionsBuilder ) { try { - final BytesRefBlock block = inputPage.getBlock(channelOffset); + final BytesRefBlock block = inputPage.getBlock(matchChannelOffset); final int valueCount = block.getValueCount(position); - if (valueCount != 1) { - return 0; // Skip multi-value positions and null positions + if (valueCount > 1) { + warnings.registerException( + new IllegalArgumentException("LOOKUP JOIN encountered multi-value") + ); + return 0; // Skip multi-value positions + } + if (valueCount < 1) { + return 0; // Skip null positions } final int firstValueIndex = block.getFirstValueIndex(position); final BytesRef termBytes = block.getBytesRef(firstValueIndex, scratch); @@ -121,10 +130,14 @@ public int processQuery( } public int getPositionCount(Page inputPage) { - final Block block = inputPage.getBlock(channelOffset); + final Block block = inputPage.getBlock(matchChannelOffset); return block.getPositionCount(); } + public int getExtractChannelOffset() { + return extractChannelOffset; + } + /** * Initialize caches for the given index reader. This should be called once * before the first processQuery call for a given index reader. diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/ExpressionQueryList.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/ExpressionQueryList.java index 61c694ac49add..bc9ca91aaef29 100644 --- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/ExpressionQueryList.java +++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/ExpressionQueryList.java @@ -25,6 +25,7 @@ import org.elasticsearch.xpack.esql.capabilities.TranslationAware; import org.elasticsearch.xpack.esql.core.expression.Attribute; import org.elasticsearch.xpack.esql.core.expression.Expression; +import org.elasticsearch.xpack.esql.core.expression.NamedExpression; import org.elasticsearch.xpack.esql.core.type.DataType; import org.elasticsearch.xpack.esql.expression.predicate.Predicates; import org.elasticsearch.xpack.esql.expression.predicate.operator.comparison.Equals; @@ -139,6 +140,7 @@ public static ExpressionQueryList expressionBasedJoin( expressionQueryList.buildJoinOnForExpressionJoin( request.getJoinOnConditions(), request.getMatchFields(), + request.getExtractFields(), context, lucenePushdownPredicates, warnings @@ -154,12 +156,13 @@ public BulkKeywordLookup getBulkKeywordLookup() { private void buildJoinOnForExpressionJoin( Expression joinOnConditions, List matchFields, + List extractFields, SearchExecutionContext context, LucenePushdownPredicates lucenePushdownPredicates, Warnings warnings ) { List expressions = Predicates.splitAnd(joinOnConditions); - if (applyAsFastKeywordFilter(expressions, matchFields, context, clusterService, warnings)) { + if (applyAsFastKeywordFilter(expressions, matchFields, extractFields, context, clusterService, warnings)) { // we managed to apply the whole condition as a fast keyword filter return; } @@ -177,6 +180,7 @@ private void buildJoinOnForExpressionJoin( private boolean applyAsFastKeywordFilter( List expressions, List matchFields, + List extractFields, SearchExecutionContext context, ClusterService clusterService, Warnings warnings @@ -186,21 +190,35 @@ private boolean applyAsFastKeywordFilter( if (expr instanceof EsqlBinaryComparison binaryComparison && binaryComparison.left() instanceof Attribute leftAttribute && binaryComparison.right() instanceof Attribute rightAttribute) { + // the left side comes from the page that was sent to the lookup node // the right side is the field from the lookup index // check if the left side is in the matchFields - int channelOffset = -1; + int matchChannelOffset = -1; DataType dataType = null; for (int i = 0; i < matchFields.size(); i++) { if (matchFields.get(i).fieldName().equals(leftAttribute.name())) { - channelOffset = i; + matchChannelOffset = i; dataType = matchFields.get(i).type(); break; } } - if (channelOffset != -1 && dataType == DataType.KEYWORD) { + + if (matchChannelOffset != -1 && dataType == DataType.KEYWORD) { + + // BulkLookupMvFilterOperator needs the extractChannelOffset later + // when filtering out false-positive multivalue matches + // + int extractChannelOffset = -1; + for (int i = 0; i < extractFields.size(); i++) { + if (extractFields.get(i).name().equals(rightAttribute.name())) { + extractChannelOffset = i; + break; + } + } MappedFieldType rightFieldType = context.getFieldType(rightAttribute.name()); - if (rightFieldType != null) { + + if (extractChannelOffset != -1 && rightFieldType != null) { // special handle Equals operator on keyword fields // we can apply as a BulkKeywordLookup for better performance if (binaryComparison instanceof Equals) { @@ -209,7 +227,8 @@ private boolean applyAsFastKeywordFilter( rightFieldType, leftElementType, context, - channelOffset, + matchChannelOffset, + extractChannelOffset, clusterService, aliasFilter, warnings diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/LookupFromIndexService.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/LookupFromIndexService.java index 139a2f8a8fabb..c8d39fbec926d 100644 --- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/LookupFromIndexService.java +++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/LookupFromIndexService.java @@ -308,6 +308,10 @@ public List getMatchFields() { return matchFields; } + public List getExtractFields() { + return extractFields; + } + @Override public void writeTo(StreamOutput out) throws IOException { super.writeTo(out); From 487acfa8eadb708fdc03c7cb8d4f323a77a13472 Mon Sep 17 00:00:00 2001 From: cimequinox Date: Mon, 9 Mar 2026 07:35:30 -0700 Subject: [PATCH 08/11] BulkLookupSingleValued Add evaluator to filter out false-positive multivalue matches Also generates warning when multivalues are encountered --- .../lookup/BulkLookupSingleValued.java | 63 ++++++++++++ .../lookup/BulkLookupSingleValuedTests.java | 98 +++++++++++++++++++ 2 files changed, 161 insertions(+) create mode 100644 x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/BulkLookupSingleValued.java create mode 100644 x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/lookup/BulkLookupSingleValuedTests.java diff --git a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/BulkLookupSingleValued.java b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/BulkLookupSingleValued.java new file mode 100644 index 0000000000000..b12ad850d0ad2 --- /dev/null +++ b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/BulkLookupSingleValued.java @@ -0,0 +1,63 @@ +/* + * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one + * or more contributor license agreements. Licensed under the Elastic License + * 2.0; you may not use this file except in compliance with the Elastic License + * 2.0. + */ + +package org.elasticsearch.compute.operator.lookup; + +import org.apache.lucene.util.RamUsageEstimator; +import org.elasticsearch.compute.data.Block; +import org.elasticsearch.compute.data.BooleanVector; +import org.elasticsearch.compute.data.Page; +import org.elasticsearch.compute.operator.DriverContext; +import org.elasticsearch.compute.operator.EvalOperator; +import org.elasticsearch.compute.operator.Warnings; + +/** + * Emit page of boolean values where corresponding position in specified channel contains a single value. + * Used in AbstractLookupService to filter out false-positive matches when using BulkKeywordLookup optimization. + */ +public record BulkLookupSingleValued(DriverContext context, int channelOffset, Warnings warnings) + implements EvalOperator.ExpressionEvaluator +{ + private static final long BASE_RAM_BYTES_USED = RamUsageEstimator.shallowSizeOfInstance(BulkLookupSingleValued.class); + + @Override + public Block eval(Page page) { + final Block block = page.getBlock(channelOffset); + final int positionCount = block.getPositionCount(); + final BooleanVector.FixedBuilder singles = context.blockFactory().newBooleanVectorFixedBuilder(positionCount); + + boolean encounteredMultiValue = false; + for (int p = 0; p < positionCount; p++) { + final int valueCount = block.getValueCount(p); + if (valueCount > 1) { + encounteredMultiValue = true; + } + singles.appendBoolean(valueCount == 1); + } + if (encounteredMultiValue) { + warnings.registerException( + new IllegalArgumentException("LOOKUP JOIN encountered multi-value") + ); + } + + final Block result = singles.build().asBlock(); + return result; + } + + @Override + public long baseRamBytesUsed() { + return BASE_RAM_BYTES_USED; + } + + @Override + public String toString() { + return "BulkLookupSingleValued[channelOffset=" + channelOffset + ']'; + } + + @Override + public void close() {} +} diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/lookup/BulkLookupSingleValuedTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/lookup/BulkLookupSingleValuedTests.java new file mode 100644 index 0000000000000..656ee558be848 --- /dev/null +++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/lookup/BulkLookupSingleValuedTests.java @@ -0,0 +1,98 @@ +/* + * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one + * or more contributor license agreements. Licensed under the Elastic License + * 2.0; you may not use this file except in compliance with the Elastic License + * 2.0. + */ + +package org.elasticsearch.compute.operator.lookup; + +import org.apache.lucene.util.BytesRef; +import org.elasticsearch.compute.data.BlockFactory; +import org.elasticsearch.compute.data.BooleanBlock; +import org.elasticsearch.compute.data.BytesRefBlock; +import org.elasticsearch.compute.data.ElementType; +import org.elasticsearch.compute.data.Page; +import org.elasticsearch.compute.operator.DriverContext; +import org.elasticsearch.compute.operator.EvalOperator; +import org.elasticsearch.compute.operator.FilterOperator; +import org.elasticsearch.compute.operator.Operator; +import org.elasticsearch.compute.operator.SourceOperator; +import org.elasticsearch.compute.test.OperatorTestCase; +import org.elasticsearch.compute.test.operator.blocksource.ListRowsBlockSourceOperator; +import org.hamcrest.Matcher; + +import java.util.List; +import java.util.stream.IntStream; + +import static org.hamcrest.Matchers.equalTo; + +// based on FilterOperatorTests +public class BulkLookupSingleValuedTests extends OperatorTestCase { + + @Override + protected SourceOperator simpleInput(BlockFactory blockFactory, int size) { + + final Object[] possibilities = { + "single", + List.of("multiple", "values") + }; + + // returns pages with two blocks + // in first block even rows have single values, odd rows have multi values + // in second block even rows have value == true, odd rows have value == false + // + return new ListRowsBlockSourceOperator( + blockFactory, + List.of(ElementType.BYTES_REF, ElementType.BOOLEAN), + IntStream + .range(0, size) + .mapToObj(l -> List.of(possibilities[l % 2], (l % 2) == 0)) + .toList() + ); + } + + @Override + protected void assertSimpleOutput(List input, List results) { + final BytesRef expected = new BytesRef("single"); + final BytesRef scratch = new BytesRef(); + for (var page : results) { + final BytesRefBlock b0 = page.getBlock(0); + final BooleanBlock b1 = page.getBlock(1); + for (int p = 0; p < page.getPositionCount(); p++) { + final BytesRef bytesValue = b0.getBytesRef(p, scratch); + final Boolean boolValue = b1.getBoolean(p); + + // only the single values should pass the filter + assertThat(bytesValue, equalTo(expected)); + assertThat(boolValue, equalTo(true)); + } + } + } + + @Override + protected Operator.OperatorFactory simple(SimpleOptions options) { + return new FilterOperator.FilterOperatorFactory(new EvalOperator.ExpressionEvaluator.Factory() { + + @Override + public EvalOperator.ExpressionEvaluator get(DriverContext context) { + return new BulkLookupSingleValued(context, 0, null); + } + + @Override + public String toString() { + return "BulkLookupSingleValued[channelOffset=0]"; + } + }); + } + + @Override + protected Matcher expectedDescriptionOfSimple() { + return equalTo("FilterOperator[evaluator=BulkLookupSingleValued[channelOffset=0]]"); + } + + @Override + protected Matcher expectedToStringOfSimple() { + return expectedDescriptionOfSimple(); + } +} From 735feb2318e6d6460aecc3a4fcbabbf9b99635b9 Mon Sep 17 00:00:00 2001 From: cimequinox Date: Mon, 9 Mar 2026 07:39:19 -0700 Subject: [PATCH 09/11] AbstractLookupService removes false positive matches --- .../esql/enrich/AbstractLookupService.java | 46 ++++++++++++++++++- 1 file changed, 44 insertions(+), 2 deletions(-) diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/AbstractLookupService.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/AbstractLookupService.java index 93dd6480c79c5..aa272b6a297b0 100644 --- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/AbstractLookupService.java +++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/AbstractLookupService.java @@ -37,11 +37,14 @@ import org.elasticsearch.compute.lucene.read.ValuesSourceReaderOperator; import org.elasticsearch.compute.operator.Driver; import org.elasticsearch.compute.operator.DriverContext; +import org.elasticsearch.compute.operator.FilterOperator; import org.elasticsearch.compute.operator.Operator; import org.elasticsearch.compute.operator.OutputOperator; import org.elasticsearch.compute.operator.ProjectOperator; import org.elasticsearch.compute.operator.Warnings; import org.elasticsearch.compute.operator.lookup.BlockOptimization; +import org.elasticsearch.compute.operator.lookup.BulkKeywordLookup; +import org.elasticsearch.compute.operator.lookup.BulkLookupSingleValued; import org.elasticsearch.compute.operator.lookup.EnrichQuerySourceOperator; import org.elasticsearch.compute.operator.lookup.LookupEnrichQueryGenerator; import org.elasticsearch.compute.operator.lookup.MergePositionsOperator; @@ -101,7 +104,7 @@ *

*

* The join process spawns a {@link Driver} per incoming page which runs in - * two or three stages: + * two, three or four stages: *

*

* Stage 1: Finding matching document IDs for the input page. This stage is done @@ -114,7 +117,11 @@ * {@code [DocVector, IntBlock: positions, Block: field1, Block: field2,...]}. *

*

- * Stage 3: Optionally this combines the extracted values based on positions and filling + * Stage 3: Optionally the BulkLookupMvFilterOperator removes false-positive + * multivalue matches when the {@link BulkKeywordLookup} optimization. + *

+ *

+ * Stage 4: Optionally this combines the extracted values based on positions and filling * nulls for positions without matches. This is done by {@link MergePositionsOperator}. * The output page is represented as {@code [Block: field1, Block: field2,...]}. *

@@ -345,6 +352,8 @@ protected void doLookup(T request, CancellableTask task, ActionListener operators = new ArrayList<>(); if (request.extractFields.isEmpty() == false) { var extractFieldsOperator = extractFieldsOperator(shardContext.context, driverContext, request.extractFields); releasables.add(extractFieldsOperator); operators.add(extractFieldsOperator); } + + // Stage 3 - 137269 + Operator bulkLookupMvFilterOperator = bulkLookupMvFilterOperator(queryList, driverContext, warnings); + if (bulkLookupMvFilterOperator != null) { + operators.add(bulkLookupMvFilterOperator); + } + + // Stage 4 operators.add(finishPages); /* @@ -493,6 +511,30 @@ private Operator dropDocBlockOperator(List extractFields) { return new ProjectOperator(projection); } + /** + * Returns an operator to remove false-positive multivalue matches from + * BulkKeywordLookup or null when that optimization is not used. + */ + private static Operator bulkLookupMvFilterOperator(LookupEnrichQueryGenerator queryList, DriverContext driverContext, Warnings warnings) + { + final BulkKeywordLookup bulkLookup = queryList.getBulkKeywordLookup(); + if (bulkLookup != null) { + + // at this point the output page [DocVector, IntBlock: positions, Block: field1, Block: field2,...] + // get the channel ignoreing the DocVector and IntBlock + // + final int channelOffset = 2 + bulkLookup.getExtractChannelOffset(); + return new FilterOperator( + new BulkLookupSingleValued( + driverContext, + channelOffset, + warnings + ) + ); + } + return null; + } + protected Page createNullResponse(int positionCount, List extractFields) { final Block[] blocks = new Block[extractFields.size()]; try { From 993c37d0c98b7b409d9324e1c9335bebda14e6c0 Mon Sep 17 00:00:00 2001 From: cimequinox Date: Mon, 9 Mar 2026 07:49:57 -0700 Subject: [PATCH 10/11] Update csv tests Add near-duplicate row with multivalues for testing --- .../data/languages_non_unique_key.csv | 1 + .../src/main/resources/inlinestats.csv-spec | 3 +- .../src/main/resources/lookup-join.csv-spec | 35 +++++++++++++++++++ .../src/main/resources/views.csv-spec | 6 ++-- 4 files changed, 42 insertions(+), 3 deletions(-) diff --git a/x-pack/plugin/esql/qa/testFixtures/src/main/resources/data/languages_non_unique_key.csv b/x-pack/plugin/esql/qa/testFixtures/src/main/resources/data/languages_non_unique_key.csv index d6381b174d739..6d4d34c4d84ea 100644 --- a/x-pack/plugin/esql/qa/testFixtures/src/main/resources/data/languages_non_unique_key.csv +++ b/x-pack/plugin/esql/qa/testFixtures/src/main/resources/data/languages_non_unique_key.csv @@ -4,6 +4,7 @@ language_code:integer,language_name:keyword,country:keyword 1,,United Kingdom 1,English,United States of America 2,German,[Germany,Austria] +2,German,[Germany,Belgium] 2,German,Switzerland 2,German, 4,Quenya, diff --git a/x-pack/plugin/esql/qa/testFixtures/src/main/resources/inlinestats.csv-spec b/x-pack/plugin/esql/qa/testFixtures/src/main/resources/inlinestats.csv-spec index 336d198fd58bc..6a8f065b05127 100644 --- a/x-pack/plugin/esql/qa/testFixtures/src/main/resources/inlinestats.csv-spec +++ b/x-pack/plugin/esql/qa/testFixtures/src/main/resources/inlinestats.csv-spec @@ -4180,12 +4180,13 @@ FROM languages_lookup_non_unique_key | EVAL language_code = null::integer | INLINE STATS MAX(language_code) BY language_code | SORT country -| LIMIT 5 +| LIMIT 6 ; country:keyword |language_name:keyword |MAX(language_code):integer |language_code:integer Atlantis |null |null |null [Austria, Germany]|German |null |null +[Belgium, Germany]|German |null |null Canada |English |null |null Mv-Land |Mv-Lang |null |null Mv-Land2 |Mv-Lang2 |null |null diff --git a/x-pack/plugin/esql/qa/testFixtures/src/main/resources/lookup-join.csv-spec b/x-pack/plugin/esql/qa/testFixtures/src/main/resources/lookup-join.csv-spec index 62d17789ce0f0..89dae2fc85694 100644 --- a/x-pack/plugin/esql/qa/testFixtures/src/main/resources/lookup-join.csv-spec +++ b/x-pack/plugin/esql/qa/testFixtures/src/main/resources/lookup-join.csv-spec @@ -592,6 +592,41 @@ language_code:integer | language_name:keyword | country:text 8 | null | null ; +mvKeywordJoinWarningsFromLeftSide +required_capability: join_lookup_v12 +required_capability: async_operator_warnings_fix + +ROW name = ["English", "Spanish"] +| LOOKUP JOIN languages_lookup_non_unique_key ON name == language_name +| KEEP name, country.keyword, language_name +| LIMIT 1 +; + +warning:Line 2:3: evaluation of [LOOKUP JOIN languages_lookup_non_unique_key ON name == language_name] failed, treating result as null. Only first 20 failures recorded. +warning:Line 2:3: java.lang.IllegalArgumentException: LOOKUP JOIN encountered multi-value + +name:keyword | country.keyword:keyword | language_name:keyword +[English, Spanish] | null | null +; + +mvKeywordJoinWarningsFromRightSide +required_capability: join_lookup_v12 +required_capability: async_operator_warnings_fix + +ROW name = "Germany" +| LOOKUP JOIN languages_lookup_non_unique_key ON name == country.keyword +| KEEP name, country.keyword, language_name +| LIMIT 1 +; + +warning:Line 2:3: evaluation of [LOOKUP JOIN languages_lookup_non_unique_key ON name == language_name] failed, treating result as null. Only first 20 failures recorded. +warning:Line 2:3: java.lang.IllegalArgumentException: LOOKUP JOIN encountered multi-value + +name:keyword | country.keyword:keyword | language_name:keyword +Germany | null | null +; + + ############################################### # Filtering tests with languages_lookup index ############################################### diff --git a/x-pack/plugin/esql/qa/testFixtures/src/main/resources/views.csv-spec b/x-pack/plugin/esql/qa/testFixtures/src/main/resources/views.csv-spec index 22a160b641663..e38f6210d7d7e 100644 --- a/x-pack/plugin/esql/qa/testFixtures/src/main/resources/views.csv-spec +++ b/x-pack/plugin/esql/qa/testFixtures/src/main/resources/views.csv-spec @@ -159,8 +159,9 @@ ignoreOrder:true count:long | country:keyword 1 | Atlantis 1 | Austria +1 | Belgium 1 | Canada -1 | Germany +2 | Germany 1 | Switzerland 1 | United Kingdom 1 | United States @@ -176,8 +177,9 @@ ignoreOrder:true count:long | country:keyword 1 | Atlantis 1 | Austria +1 | Belgium 1 | Canada -1 | Germany +2 | Germany 1 | Switzerland 1 | United Kingdom 1 | United States From 42db6bd96757f5774e5e513dea640dba1e3323b6 Mon Sep 17 00:00:00 2001 From: elasticsearchmachine Date: Mon, 9 Mar 2026 15:16:30 +0000 Subject: [PATCH 11/11] [CI] Auto commit changes from spotless --- .../operator/lookup/BulkKeywordLookup.java | 6 ++--- .../lookup/BulkLookupSingleValued.java | 8 +++---- .../lookup/EnrichQuerySourceOperator.java | 23 ++++++++++--------- .../lookup/BulkLookupSingleValuedTests.java | 14 ++++------- .../esql/enrich/AbstractLookupService.java | 15 +++++------- .../esql/enrich/ExpressionQueryList.java | 2 +- 6 files changed, 28 insertions(+), 40 deletions(-) diff --git a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/BulkKeywordLookup.java b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/BulkKeywordLookup.java index 39f61ce149fcf..bc0f1f1e8ec4f 100644 --- a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/BulkKeywordLookup.java +++ b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/BulkKeywordLookup.java @@ -55,7 +55,7 @@ public BulkKeywordLookup( ) { this.rightFieldType = rightFieldType; this.context = context; - this.matchChannelOffset = matchChannelOffset; // offset of field in left (input) page + this.matchChannelOffset = matchChannelOffset; // offset of field in left (input) page this.extractChannelOffset = extractChannelOffset; // offset of field in right (output) page this.clusterService = clusterService; this.aliasFilter = aliasFilter; @@ -81,9 +81,7 @@ public int processQuery( final BytesRefBlock block = inputPage.getBlock(matchChannelOffset); final int valueCount = block.getValueCount(position); if (valueCount > 1) { - warnings.registerException( - new IllegalArgumentException("LOOKUP JOIN encountered multi-value") - ); + warnings.registerException(new IllegalArgumentException("LOOKUP JOIN encountered multi-value")); return 0; // Skip multi-value positions } if (valueCount < 1) { diff --git a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/BulkLookupSingleValued.java b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/BulkLookupSingleValued.java index b12ad850d0ad2..1203551180945 100644 --- a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/BulkLookupSingleValued.java +++ b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/BulkLookupSingleValued.java @@ -20,8 +20,8 @@ * Used in AbstractLookupService to filter out false-positive matches when using BulkKeywordLookup optimization. */ public record BulkLookupSingleValued(DriverContext context, int channelOffset, Warnings warnings) - implements EvalOperator.ExpressionEvaluator -{ + implements + EvalOperator.ExpressionEvaluator { private static final long BASE_RAM_BYTES_USED = RamUsageEstimator.shallowSizeOfInstance(BulkLookupSingleValued.class); @Override @@ -39,9 +39,7 @@ public Block eval(Page page) { singles.appendBoolean(valueCount == 1); } if (encounteredMultiValue) { - warnings.registerException( - new IllegalArgumentException("LOOKUP JOIN encountered multi-value") - ); + warnings.registerException(new IllegalArgumentException("LOOKUP JOIN encountered multi-value")); } final Block result = singles.build().asBlock(); diff --git a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/EnrichQuerySourceOperator.java b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/EnrichQuerySourceOperator.java index 3a404d585f133..87d17ba3b3e59 100644 --- a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/EnrichQuerySourceOperator.java +++ b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/EnrichQuerySourceOperator.java @@ -196,10 +196,11 @@ public Page getOutput() { } } - private Page processBulkQueries(Page inputPage, - IntVector.Builder positionsBuilder, - IntVector.Builder segmentsBuilder, - IntVector.Builder docsBuilder + private Page processBulkQueries( + Page inputPage, + IntVector.Builder positionsBuilder, + IntVector.Builder segmentsBuilder, + IntVector.Builder docsBuilder ) throws IOException { queryPosition++; BulkKeywordLookup bulkKeywordLookup = queryList.getBulkKeywordLookup(); @@ -207,18 +208,18 @@ private Page processBulkQueries(Page inputPage, bulkKeywordLookup.initializeCaches(indexReader); while (queryPosition < queryList.getPositionCount(inputPage)) { int matches = bulkKeywordLookup.processQuery( - inputPage, - queryPosition, - indexReader, - docsBuilder, - segmentsBuilder, - positionsBuilder + inputPage, + queryPosition, + indexReader, + docsBuilder, + segmentsBuilder, + positionsBuilder ); totalMatches += matches; queryPosition++; } final Page result = buildPage(totalMatches, positionsBuilder, segmentsBuilder, docsBuilder); - + return result; } diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/lookup/BulkLookupSingleValuedTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/lookup/BulkLookupSingleValuedTests.java index 656ee558be848..b2391e3947692 100644 --- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/lookup/BulkLookupSingleValuedTests.java +++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/lookup/BulkLookupSingleValuedTests.java @@ -33,22 +33,16 @@ public class BulkLookupSingleValuedTests extends OperatorTestCase { @Override protected SourceOperator simpleInput(BlockFactory blockFactory, int size) { - final Object[] possibilities = { - "single", - List.of("multiple", "values") - }; + final Object[] possibilities = { "single", List.of("multiple", "values") }; // returns pages with two blocks - // in first block even rows have single values, odd rows have multi values + // in first block even rows have single values, odd rows have multi values // in second block even rows have value == true, odd rows have value == false // return new ListRowsBlockSourceOperator( blockFactory, List.of(ElementType.BYTES_REF, ElementType.BOOLEAN), - IntStream - .range(0, size) - .mapToObj(l -> List.of(possibilities[l % 2], (l % 2) == 0)) - .toList() + IntStream.range(0, size).mapToObj(l -> List.of(possibilities[l % 2], (l % 2) == 0)).toList() ); } @@ -58,7 +52,7 @@ protected void assertSimpleOutput(List input, List results) { final BytesRef scratch = new BytesRef(); for (var page : results) { final BytesRefBlock b0 = page.getBlock(0); - final BooleanBlock b1 = page.getBlock(1); + final BooleanBlock b1 = page.getBlock(1); for (int p = 0; p < page.getPositionCount(); p++) { final BytesRef bytesValue = b0.getBytesRef(p, scratch); final Boolean boolValue = b1.getBoolean(p); diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/AbstractLookupService.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/AbstractLookupService.java index aa272b6a297b0..e63b39968cc02 100644 --- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/AbstractLookupService.java +++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/AbstractLookupService.java @@ -515,8 +515,11 @@ private Operator dropDocBlockOperator(List extractFields) { * Returns an operator to remove false-positive multivalue matches from * BulkKeywordLookup or null when that optimization is not used. */ - private static Operator bulkLookupMvFilterOperator(LookupEnrichQueryGenerator queryList, DriverContext driverContext, Warnings warnings) - { + private static Operator bulkLookupMvFilterOperator( + LookupEnrichQueryGenerator queryList, + DriverContext driverContext, + Warnings warnings + ) { final BulkKeywordLookup bulkLookup = queryList.getBulkKeywordLookup(); if (bulkLookup != null) { @@ -524,13 +527,7 @@ private static Operator bulkLookupMvFilterOperator(LookupEnrichQueryGenerator qu // get the channel ignoreing the DocVector and IntBlock // final int channelOffset = 2 + bulkLookup.getExtractChannelOffset(); - return new FilterOperator( - new BulkLookupSingleValued( - driverContext, - channelOffset, - warnings - ) - ); + return new FilterOperator(new BulkLookupSingleValued(driverContext, channelOffset, warnings)); } return null; } diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/ExpressionQueryList.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/ExpressionQueryList.java index bc9ca91aaef29..49521f5a1f318 100644 --- a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/ExpressionQueryList.java +++ b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/ExpressionQueryList.java @@ -208,7 +208,7 @@ private boolean applyAsFastKeywordFilter( // BulkLookupMvFilterOperator needs the extractChannelOffset later // when filtering out false-positive multivalue matches - // + // int extractChannelOffset = -1; for (int i = 0; i < extractFields.size(); i++) { if (extractFields.get(i).name().equals(rightAttribute.name())) {