ESQL: Performance improvements for Lookup Join on Expression

docs/changelog/135036.yaml

@@ -0,0 +1,5 @@
+pr: 135036
+summary: Performance improvements for Lookup Join on Expression
+area: ES|QL
+type: enhancement
+issues: []

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

@@ -24,6 +24,7 @@
 import org.elasticsearch.xpack.esql.core.expression.Attribute;
 import org.elasticsearch.xpack.esql.core.expression.Expression;
 import org.elasticsearch.xpack.esql.expression.predicate.Predicates;
+import org.elasticsearch.xpack.esql.expression.predicate.operator.comparison.Equals;
 import org.elasticsearch.xpack.esql.expression.predicate.operator.comparison.EsqlBinaryComparison;
 import org.elasticsearch.xpack.esql.optimizer.rules.physical.local.LucenePushdownPredicates;
 import org.elasticsearch.xpack.esql.plan.physical.EsSourceExec;
@@ -38,6 +39,7 @@
 import java.util.List;
 
 import static org.elasticsearch.xpack.esql.action.EsqlCapabilities.Cap.LOOKUP_JOIN_ON_BOOLEAN_EXPRESSION;
+import static org.elasticsearch.xpack.esql.enrich.AbstractLookupService.termQueryList;
 import static org.elasticsearch.xpack.esql.planner.TranslatorHandler.TRANSLATOR_HANDLER;
 
 /**
@@ -154,17 +156,31 @@ private void buildJoinOnForExpressionJoin(
                             if (right instanceof Attribute rightAttribute) {
                                 MappedFieldType fieldType = context.getFieldType(rightAttribute.name());
                                 if (fieldType != null) {
-                                    queryLists.add(
-                                        new BinaryComparisonQueryList(
+                                    // special handle Equals operator
+                                    // TermQuery is faster than BinaryComparisonQueryList, as it does less work per row
+                                    // so here we reuse the existing logic from field based join to build a termQueryList for Equals
+                                    if (binaryComparison instanceof Equals) {
+                                        QueryList termQueryForEquals = termQueryList(
                                             fieldType,
                                             context,
-                                            block,
-                                            binaryComparison,
-                                            clusterService,
                                             aliasFilter,
-                                            warnings
-                                        )
-                                    );
+                                            inputPage.getBlock(matchFields.get(i).channel()),
+                                            matchFields.get(i).type()
+                                        ).onlySingleValues(warnings, "LOOKUP JOIN encountered multi-value");
+                                        queryLists.add(termQueryForEquals);
+                                    } else {
+                                        queryLists.add(
+                                            new BinaryComparisonQueryList(
+                                                fieldType,
+                                                context,
+                                                block,
+                                                binaryComparison,
+                                                clusterService,
+                                                aliasFilter,
+                                                warnings
+                                            )
+                                        );
+                                    }
                                     matched = true;
                                     break;
                                 } else {

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

@@ -31,10 +31,12 @@
 
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.Set;
 import java.util.function.Function;
 
 // TODO rename package
@@ -143,10 +145,11 @@ public LookupFromIndexOperator(
 
     @Override
     protected void performAsync(Page inputPage, ActionListener<OngoingJoin> listener) {
-        Block[] inputBlockArray = new Block[matchFields.size()];
         List<MatchConfig> newMatchFields = new ArrayList<>();
-        for (int i = 0; i < matchFields.size(); i++) {
-            MatchConfig matchField = matchFields.get(i);
+        List<MatchConfig> uniqueMatchFields = uniqueMatchFieldsByName(matchFields);
+        Block[] inputBlockArray = new Block[uniqueMatchFields.size()];
+        for (int i = 0; i < uniqueMatchFields.size(); i++) {
+            MatchConfig matchField = uniqueMatchFields.get(i);
             int inputChannel = matchField.channel();
             final Block inputBlock = inputPage.getBlock(inputChannel);
             inputBlockArray[i] = inputBlock;
@@ -176,6 +179,20 @@ protected void performAsync(Page inputPage, ActionListener<OngoingJoin> listener
         );
     }
 
+    private List<MatchConfig> uniqueMatchFieldsByName(List<MatchConfig> matchFields) {
+        if (joinOnConditions == null) {
+            return matchFields;
+        }
+        List<MatchConfig> uniqueFields = new ArrayList<>();
+        Set<String> seenFieldNames = new HashSet<>();
+        for (MatchConfig matchField : matchFields) {
+            if (seenFieldNames.add(matchField.fieldName())) {
+                uniqueFields.add(matchField);
+            }
+        }
+        return uniqueFields;
+    }
+
     @Override
     public Page getOutput() {
         if (ongoing == null) {