Clean up, add more UTs

julian-elastic · julian-elastic · commit 2aa2b49faf79 · 2025-08-20T15:33:30.000-04:00
diff --git a/test/external-modules/esql-heap-attack/src/javaRestTest/java/org/elasticsearch/xpack/esql/heap_attack/HeapAttackIT.java b/test/external-modules/esql-heap-attack/src/javaRestTest/java/org/elasticsearch/xpack/esql/heap_attack/HeapAttackIT.java
@@ -726,7 +726,6 @@ public void testLookupExplosionManyMatchesFiltered() throws IOException {
         // This test will only work with the expanding join optimization
         // that pushes the filter to the right side of the lookup.
         // Without the optimization, it will fail with circuit_breaking_exception
-        // lookupEntries % reductionFactor must be 0 to ensure that the number of matches is reduced
         int sensorDataCount = 10000;
         int lookupEntries = 10000;
         int reductionFactor = 1000; // reduce the number of matches by this factor
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
@@ -10,8 +10,6 @@
 import org.apache.lucene.search.Query;
 import org.elasticsearch.core.Nullable;
 
-import java.io.IOException;
-
 /**
  * An interface to generates queries for the lookup and enrich operators.
  * This interface is used to retrieve queries based on a position index.
@@ -22,7 +20,7 @@ public interface LookupEnrichQueryGenerator {
      * Returns the query at the given position.
      */
     @Nullable
-    Query getQuery(int position) throws IOException;
+    Query getQuery(int position);
 
     /**
      * Returns the number of queries in this generator
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
@@ -5151,7 +5151,7 @@ null           | null             | fred           | null           | null
 ;
 
 
-lookupJoinWithPushableFilterOnLeft
+lookupJoinWithPushableFilterOnRight
 required_capability: join_lookup_v12
 required_capability: lookup_join_on_multiple_fields
 
@@ -5178,7 +5178,7 @@ id_int:integer | name_str:keyword | extra1:keyword | other1:keyword | other2:int
 14             | Nina             | foo2           | omicron        | 15000
 ;
 
-lookupJoinWithTwoPushableFiltersOnLeft
+lookupJoinWithTwoPushableFiltersOnRight
 required_capability: join_lookup_v12
 required_capability: lookup_join_on_multiple_fields
 
diff --git a/x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/LookupFromIndexIT.java b/x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/LookupFromIndexIT.java
@@ -59,6 +59,7 @@
 import org.elasticsearch.xpack.esql.core.expression.ReferenceAttribute;
 import org.elasticsearch.xpack.esql.core.tree.Source;
 import org.elasticsearch.xpack.esql.core.type.DataType;
+import org.elasticsearch.xpack.esql.core.type.EsField;
 import org.elasticsearch.xpack.esql.enrich.LookupFromIndexOperator;
 import org.elasticsearch.xpack.esql.enrich.MatchConfig;
 import org.elasticsearch.xpack.esql.expression.predicate.operator.comparison.GreaterThan;
@@ -228,16 +229,11 @@ public void populate(int docCount, List<String> expected, Predicate<Integer> fil
         }
     }
 
-    Expression buildGreaterThanFilter(long value) {
+    private Expression buildGreaterThanFilter(long value) {
         FieldAttribute filterAttribute = new FieldAttribute(
             Source.EMPTY,
             "l",
-            new org.elasticsearch.xpack.esql.core.type.EsField(
-                "l",
-                org.elasticsearch.xpack.esql.core.type.DataType.LONG,
-                java.util.Collections.emptyMap(),
-                true
-            )
+            new EsField("l", DataType.LONG, Collections.emptyMap(), true, EsField.TimeSeriesFieldType.NONE)
         );
         return new GreaterThan(Source.EMPTY, filterAttribute, new Literal(Source.EMPTY, value, DataType.LONG));
     }
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
@@ -38,7 +38,6 @@
 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;
@@ -74,14 +73,10 @@
 import org.elasticsearch.xpack.esql.EsqlIllegalArgumentException;
 import org.elasticsearch.xpack.esql.core.expression.Alias;
 import org.elasticsearch.xpack.esql.core.expression.FieldAttribute;
-import org.elasticsearch.xpack.esql.core.expression.FoldContext;
 import org.elasticsearch.xpack.esql.core.expression.NamedExpression;
 import org.elasticsearch.xpack.esql.core.tree.Source;
 import org.elasticsearch.xpack.esql.core.type.DataType;
-import org.elasticsearch.xpack.esql.evaluator.EvalMapper;
-import org.elasticsearch.xpack.esql.plan.physical.FilterExec;
 import org.elasticsearch.xpack.esql.planner.EsPhysicalOperationProviders;
-import org.elasticsearch.xpack.esql.planner.Layout;
 import org.elasticsearch.xpack.esql.planner.PlannerUtils;
 import org.elasticsearch.xpack.esql.plugin.EsqlPlugin;
 
@@ -351,6 +346,7 @@ private void doLookup(T request, CancellableTask task, ActionListener<List<Page>
                 warnings
             );
             releasables.add(queryOperator);
+
             List<Operator> operators = new ArrayList<>();
             if (request.extractFields.isEmpty() == false) {
                 var extractFieldsOperator = extractFieldsOperator(shardContext.context, driverContext, request.extractFields);
@@ -418,27 +414,6 @@ public void onFailure(Exception e) {
         }
     }
 
-    private Operator filterExecOperator(
-        FilterExec filterExec,
-        Operator inputOperator, // not needed?
-        EsPhysicalOperationProviders.ShardContext shardContext,
-        DriverContext driverContext,
-        Layout.Builder builder
-    ) {
-        if (filterExec == null) {
-            return null;
-        }
-
-        var evaluatorFactory = EvalMapper.toEvaluator(
-            FoldContext.small()/*is this correct*/,
-            filterExec.condition(),
-            builder.build(),
-            List.of(shardContext)
-        );
-        var filterOperatorFactory = new FilterOperator.FilterOperatorFactory(evaluatorFactory);
-        return filterOperatorFactory.get(driverContext);
-    }
-
     private static Operator extractFieldsOperator(
         EsPhysicalOperationProviders.ShardContext shardContext,
         DriverContext driverContext,
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
@@ -54,8 +54,6 @@ public ExpressionQueryList(
     }
 
     private void buildPrePostJoinFilter(Expression optionalFilter, ClusterService clusterService) {
-        // we support a FilterExec as the pre-join filter
-        // if the filter Exec is not translatable to a QueryBuilder, we will apply it after the join
         try {
             // If the pre-join filter is a FilterExec, we can convert it to a QueryBuilder
             // try to convert it to a QueryBuilder, if not possible apply it after the join
@@ -71,9 +69,9 @@ private void buildPrePostJoinFilter(Expression optionalFilter, ClusterService cl
                 }
             }
             // If the filter is not translatable we will not apply it for now
-            // as performance testing showed no significant difference in performance.
-            // We can revisit this in the future if needed.
-            // The filter is optional, so that is OK
+            // as performance testing showed no performance improvement.
+            // We can revisit this in the future if needed, once we have more optimized workflow in place.
+            // The filter is optional, so it is OK to ignore it if it cannot be translated.
         } catch (IOException e) {
             // as the filter is optional an error in its application will be ignored
             logger.error(() -> "Failed to translate optional pre-join filter: [" + optionalFilter + "]", e);
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
@@ -101,7 +101,6 @@ protected LookupEnrichQueryGenerator queryList(
         Block inputBlock,
         Warnings warnings
     ) {
-
         List<QueryList> queryLists = new ArrayList<>();
         for (int i = 0; i < request.matchFields.size(); i++) {
             MatchConfig matchField = request.matchFields.get(i);
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/PushDownAndCombineFilters.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/PushDownAndCombineFilters.java
@@ -156,12 +156,8 @@ private static LogicalPlan pushDownPastJoin(Filter filter, Join join) {
             }
             // push the right scoped filter down to the right child
             if (scoped.rightFilters().isEmpty() == false && (join.right() instanceof Filter == false)) {
-                // push the filter down to the right child
                 List<Expression> rightPushableFilters = buildRightPushableFilters(scoped.rightFilters());
                 if (rightPushableFilters.isEmpty() == false) {
-                    // right = new Filter(right.source(), right, Predicates.combineAnd(rightPushableFilters));
-                    // update the join with the new right child
-                    // join = (Join) join.replaceRight(right);
                     Expression optionalRightHandSideFilters = Predicates.combineAnd(rightPushableFilters);
                     join = join.withOptionalRightHandFilters(optionalRightHandSideFilters);
                     optimizationApplied = true;
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/mapper/LocalMapper.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/mapper/LocalMapper.java
@@ -23,7 +23,6 @@
 import org.elasticsearch.xpack.esql.plan.logical.join.JoinConfig;
 import org.elasticsearch.xpack.esql.plan.logical.join.JoinTypes;
 import org.elasticsearch.xpack.esql.plan.physical.EsSourceExec;
-import org.elasticsearch.xpack.esql.plan.physical.FilterExec;
 import org.elasticsearch.xpack.esql.plan.physical.HashJoinExec;
 import org.elasticsearch.xpack.esql.plan.physical.LimitExec;
 import org.elasticsearch.xpack.esql.plan.physical.LocalSourceExec;
@@ -112,33 +111,19 @@ private PhysicalPlan mapBinary(BinaryPlan binary) {
                     join.rightOutputFields()
                 );
             }
-            if (right instanceof FilterExec filterExec) {
-                LookupJoinExec lookupJoinExec = getLookupJoinExec(join, filterExec.child(), left, config);
-                if (lookupJoinExec != null) {
-                    // build the right child as a FilterExec with the original lookupJoinExec.right() as the child
-                    FilterExec newRightChild = filterExec.replaceChild(lookupJoinExec.right());
-                    return lookupJoinExec.replaceChildren(lookupJoinExec.left(), newRightChild);
-                }
+            if (right instanceof EsSourceExec source && source.indexMode() == IndexMode.LOOKUP) {
+                return new LookupJoinExec(
+                    join.source(),
+                    left,
+                    right,
+                    config.leftFields(),
+                    config.rightFields(),
+                    join.rightOutputFields(),
+                    join.optionalRightHandFilters()
+                );
             }
-            LookupJoinExec lookupJoinExec = getLookupJoinExec(join, right, left, config);
-            if (lookupJoinExec != null) return lookupJoinExec;
         }
 
         return MapperUtils.unsupported(binary);
     }
-
-    private static LookupJoinExec getLookupJoinExec(Join join, PhysicalPlan right, PhysicalPlan left, JoinConfig config) {
-        if (right instanceof EsSourceExec source && source.indexMode() == IndexMode.LOOKUP) {
-            return new LookupJoinExec(
-                join.source(),
-                left,
-                right,
-                config.leftFields(),
-                config.rightFields(),
-                join.rightOutputFields(),
-                join.optionalRightHandFilters()
-            );
-        }
-        return null;
-    }
 }
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/mapper/Mapper.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/mapper/Mapper.java
@@ -16,7 +16,6 @@
 import org.elasticsearch.xpack.esql.plan.logical.BinaryPlan;
 import org.elasticsearch.xpack.esql.plan.logical.Enrich;
 import org.elasticsearch.xpack.esql.plan.logical.EsRelation;
-import org.elasticsearch.xpack.esql.plan.logical.Filter;
 import org.elasticsearch.xpack.esql.plan.logical.Fork;
 import org.elasticsearch.xpack.esql.plan.logical.LeafPlan;
 import org.elasticsearch.xpack.esql.plan.logical.Limit;
@@ -227,23 +226,18 @@ private PhysicalPlan mapBinary(BinaryPlan bp) {
                     join.rightOutputFields()
                 );
             }
-            if (right instanceof FragmentExec fragment) {
-                boolean isIndexModeLookup = fragment.fragment() instanceof EsRelation relation && relation.indexMode() == IndexMode.LOOKUP;
-                isIndexModeLookup = isIndexModeLookup
-                    || fragment.fragment() instanceof Filter filter
-                        && filter.child() instanceof EsRelation relation
-                        && relation.indexMode() == IndexMode.LOOKUP;
-                if (isIndexModeLookup) {
-                    return new LookupJoinExec(
-                        join.source(),
-                        left,
-                        right,
-                        config.leftFields(),
-                        config.rightFields(),
-                        join.rightOutputFields(),
-                        join.optionalRightHandFilters()
-                    );
-                }
+            if (right instanceof FragmentExec fragment
+                && fragment.fragment() instanceof EsRelation relation
+                && relation.indexMode() == IndexMode.LOOKUP) {
+                return new LookupJoinExec(
+                    join.source(),
+                    left,
+                    right,
+                    config.leftFields(),
+                    config.rightFields(),
+                    join.rightOutputFields(),
+                    join.optionalRightHandFilters()
+                );
             }
         }
 
diff --git a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/enrich/LookupFromIndexOperatorTests.java b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/enrich/LookupFromIndexOperatorTests.java
@@ -28,9 +28,7 @@
 import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.compute.data.BlockFactory;
 import org.elasticsearch.compute.data.BytesRefBlock;
-import org.elasticsearch.compute.data.BytesRefVector;
 import org.elasticsearch.compute.data.IntBlock;
-import org.elasticsearch.compute.data.IntVector;
 import org.elasticsearch.compute.data.LongBlock;
 import org.elasticsearch.compute.data.LongVector;
 import org.elasticsearch.compute.data.Page;
@@ -61,12 +59,18 @@
 import org.elasticsearch.threadpool.TestThreadPool;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
+import org.elasticsearch.xpack.esql.core.expression.Expression;
 import org.elasticsearch.xpack.esql.core.expression.FieldAttribute;
+import org.elasticsearch.xpack.esql.core.expression.Literal;
 import org.elasticsearch.xpack.esql.core.expression.NamedExpression;
 import org.elasticsearch.xpack.esql.core.expression.ReferenceAttribute;
+import org.elasticsearch.xpack.esql.core.tree.Location;
 import org.elasticsearch.xpack.esql.core.tree.Source;
 import org.elasticsearch.xpack.esql.core.type.DataType;
+import org.elasticsearch.xpack.esql.core.type.EsField;
+import org.elasticsearch.xpack.esql.expression.predicate.operator.comparison.LessThan;
 import org.elasticsearch.xpack.esql.planner.EsPhysicalOperationProviders;
+import org.elasticsearch.xpack.esql.plugin.EsqlFlags;
 import org.elasticsearch.xpack.esql.plugin.EsqlPlugin;
 import org.hamcrest.Matcher;
 import org.junit.After;
@@ -75,6 +79,8 @@
 import java.io.IOException;
 import java.io.UncheckedIOException;
 import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -86,6 +92,7 @@
 
 public class LookupFromIndexOperatorTests extends OperatorTestCase {
     private static final int LOOKUP_SIZE = 1000;
+    private static final int LESS_THAN_VALUE = -40;
     private final ThreadPool threadPool = threadPool();
     private final Directory lookupIndexDirectory = newDirectory();
     private final List<Releasable> releasables = new ArrayList<>();
@@ -134,12 +141,17 @@ protected void assertSimpleOutput(List<Page> input, List<Page> results) {
         for (Page r : results) {
             assertThat(r.getBlockCount(), equalTo(numberOfJoinColumns + 2));
             LongVector match = r.<LongBlock>getBlock(0).asVector();
-            BytesRefVector lkwd = r.<BytesRefBlock>getBlock(numberOfJoinColumns).asVector();
-            IntVector lint = r.<IntBlock>getBlock(numberOfJoinColumns + 1).asVector();
+            BytesRefBlock lkwdBlock = r.getBlock(numberOfJoinColumns);
+            IntBlock lintBlock = r.getBlock(numberOfJoinColumns + 1);
             for (int p = 0; p < r.getPositionCount(); p++) {
                 long m = match.getLong(p);
-                assertThat(lkwd.getBytesRef(p, new BytesRef()).utf8ToString(), equalTo("l" + m));
-                assertThat(lint.getInt(p), equalTo((int) -m));
+                if (m > Math.abs(LESS_THAN_VALUE)) {
+                    assertThat(lkwdBlock.getBytesRef(lkwdBlock.getFirstValueIndex(p), new BytesRef()).utf8ToString(), equalTo("l" + m));
+                    assertThat(lintBlock.getInt(lintBlock.getFirstValueIndex(p)), equalTo((int) -m));
+                } else {
+                    assertTrue("at " + p, lkwdBlock.isNull(p));
+                    assertTrue("at " + p, lintBlock.isNull(p));
+                }
             }
         }
     }
@@ -161,6 +173,7 @@ protected Operator.OperatorFactory simple(SimpleOptions options) {
             FieldAttribute.FieldName matchField = new FieldAttribute.FieldName("match" + i);
             matchFields.add(new MatchConfig(matchField, i, inputDataType));
         }
+
         return new LookupFromIndexOperator.Factory(
             matchFields,
             sessionId,
@@ -171,7 +184,20 @@ protected Operator.OperatorFactory simple(SimpleOptions options) {
             lookupIndex,
             loadFields,
             Source.EMPTY,
-            null
+            buildLessThanFilter(LESS_THAN_VALUE)
+        );
+    }
+
+    private Expression buildLessThanFilter(int value) {
+        FieldAttribute filterAttribute = new FieldAttribute(
+            Source.EMPTY,
+            "lint",
+            new EsField("lint", DataType.INTEGER, Collections.emptyMap(), true, EsField.TimeSeriesFieldType.NONE)
+        );
+        return new LessThan(
+            new Source(new Location(0, 0), "lint < " + value),
+            filterAttribute,
+            new Literal(Source.EMPTY, value, DataType.INTEGER)
         );
     }
 
@@ -187,22 +213,25 @@ protected Matcher<String> expectedToStringOfSimple() {
         for (int i = 0; i < numberOfJoinColumns; i++) {
             sb.append(" input_type=LONG match_field=match").append(i).append(" inputChannel=").append(i);
         }
-        sb.append(" optional_filter=null]");
+        sb.append(" optional_filter=lint < ").append(LESS_THAN_VALUE).append("]");
         return matchesPattern(sb.toString());
     }
 
     private LookupFromIndexService lookupService(DriverContext mainContext) {
         boolean beCranky = mainContext.bigArrays().breakerService() instanceof CrankyCircuitBreakerService;
         DiscoveryNode localNode = DiscoveryNodeUtils.create("node", "node");
+        var builtInClusterSettings = new HashSet<>(ClusterSettings.BUILT_IN_CLUSTER_SETTINGS);
+        builtInClusterSettings.add(EsqlFlags.ESQL_STRING_LIKE_ON_INDEX);
         ClusterService clusterService = ClusterServiceUtils.createClusterService(
             threadPool,
             localNode,
             Settings.builder()
                 // Reserve 0 bytes in the sub-driver so we are more likely to hit the cranky breaker in it.
                 .put(BlockFactory.LOCAL_BREAKER_OVER_RESERVED_SIZE_SETTING, ByteSizeValue.ofKb(0))
                 .put(BlockFactory.LOCAL_BREAKER_OVER_RESERVED_MAX_SIZE_SETTING, ByteSizeValue.ofKb(0))
+                .put(EsqlFlags.ESQL_STRING_LIKE_ON_INDEX.getKey(), true)
                 .build(),
-            ClusterSettings.createBuiltInClusterSettings()
+            new ClusterSettings(Settings.EMPTY, builtInClusterSettings)
         );
         IndicesService indicesService = mock(IndicesService.class);
         IndexNameExpressionResolver indexNameExpressionResolver = TestIndexNameExpressionResolver.newInstance();

Original file line number	Diff line number	Diff line change
`@@ -5151,7 +5151,7 @@ null \| null \| fred \| null \| null`
`5151`	`5151`	`;`
`5152`	`5152`
`5153`	`5153`
`5154`		`-lookupJoinWithPushableFilterOnLeft`
	`5154`	`+lookupJoinWithPushableFilterOnRight`
`5155`	`5155`	`required_capability: join_lookup_v12`
`5156`	`5156`	`required_capability: lookup_join_on_multiple_fields`
`5157`	`5157`
`@@ -5178,7 +5178,7 @@ id_int:integer \| name_str:keyword \| extra1:keyword \| other1:keyword \| other2:int`
`5178`	`5178`	`14 \| Nina \| foo2 \| omicron \| 15000`
`5179`	`5179`	`;`
`5180`	`5180`
`5181`		`-lookupJoinWithTwoPushableFiltersOnLeft`
	`5181`	`+lookupJoinWithTwoPushableFiltersOnRight`
`5182`	`5182`	`required_capability: join_lookup_v12`
`5183`	`5183`	`required_capability: lookup_join_on_multiple_fields`
`5184`	`5184`