ESQL: Performance improvements for Lookup Join on Expression (elastic#135036)

Implement two performance improvements for Lookup Join on Expression

Do not send duplicate columns more than once to the lookup node. Example column left_id will be just sent once instead of 3 times, even though it is used 3 times in the expression
Use the termsQueryList that we build for field based join for Expression Based Joins too, if it uses equals operator. This allows us to take advantage of the optimizations made for field based join and apply them for expression based join whenever possible. We end up doing less work per row, so we expect this to be faster.
Covered by existing UTs lookup-join-expression.LookupJoinOnSameLeftFieldTwice and multiple UTs on equals.

Benchmarks show only moderate improvement 50sec to 43sec (14% improvement) in esql_lookup_join_expression_eq_100k_keys_where_few_matching_like_limit_10000.

Author: julian-elastic

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) {