MatchConfig serialization refactor

Author: julian-elastic

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

@@ -163,6 +163,100 @@ public class LookupFromIndexIT extends AbstractEsqlIntegTestCase {
         new EsField("rkey3", DataType.INTEGER, Collections.emptyMap(), true, EsField.TimeSeriesFieldType.NONE)
     );
 
+    // Precreate left-side key attributes (from source index) - up to 4 keys as seen in tests
+    private static final FieldAttribute KEY0_KEYWORD_ATTR = new FieldAttribute(
+        Source.EMPTY,
+        null,
+        null,
+        "key0",
+        new EsField("key0", DataType.KEYWORD, Collections.emptyMap(), true, EsField.TimeSeriesFieldType.NONE)
+    );
+    private static final FieldAttribute KEY0_LONG_ATTR = new FieldAttribute(
+        Source.EMPTY,
+        null,
+        null,
+        "key0",
+        new EsField("key0", DataType.LONG, Collections.emptyMap(), true, EsField.TimeSeriesFieldType.NONE)
+    );
+    private static final FieldAttribute KEY1_KEYWORD_ATTR = new FieldAttribute(
+        Source.EMPTY,
+        null,
+        null,
+        "key1",
+        new EsField("key1", DataType.KEYWORD, Collections.emptyMap(), true, EsField.TimeSeriesFieldType.NONE)
+    );
+    private static final FieldAttribute KEY1_LONG_ATTR = new FieldAttribute(
+        Source.EMPTY,
+        null,
+        null,
+        "key1",
+        new EsField("key1", DataType.LONG, Collections.emptyMap(), true, EsField.TimeSeriesFieldType.NONE)
+    );
+    private static final FieldAttribute KEY2_KEYWORD_ATTR = new FieldAttribute(
+        Source.EMPTY,
+        null,
+        null,
+        "key2",
+        new EsField("key2", DataType.KEYWORD, Collections.emptyMap(), true, EsField.TimeSeriesFieldType.NONE)
+    );
+    private static final FieldAttribute KEY2_LONG_ATTR = new FieldAttribute(
+        Source.EMPTY,
+        null,
+        null,
+        "key2",
+        new EsField("key2", DataType.LONG, Collections.emptyMap(), true, EsField.TimeSeriesFieldType.NONE)
+    );
+    private static final FieldAttribute KEY3_KEYWORD_ATTR = new FieldAttribute(
+        Source.EMPTY,
+        null,
+        null,
+        "key3",
+        new EsField("key3", DataType.KEYWORD, Collections.emptyMap(), true, EsField.TimeSeriesFieldType.NONE)
+    );
+    private static final FieldAttribute KEY3_INTEGER_ATTR = new FieldAttribute(
+        Source.EMPTY,
+        null,
+        null,
+        "key3",
+        new EsField("key3", DataType.INTEGER, Collections.emptyMap(), true, EsField.TimeSeriesFieldType.NONE)
+    );
+
+    // Index all attributes by name and type for easy lookup - built dynamically from attribute names
+    // Key format: "name:type" (e.g., "key0:KEYWORD", "key0:LONG", "rkey0:KEYWORD", "l:LONG")
+    private static final Map<String, FieldAttribute> ATTRIBUTES_BY_NAME_AND_TYPE = Map.ofEntries(
+        // Left-side attributes (from source index)
+        Map.entry(KEY0_KEYWORD_ATTR.name() + ":" + KEY0_KEYWORD_ATTR.dataType(), KEY0_KEYWORD_ATTR),
+        Map.entry(KEY0_LONG_ATTR.name() + ":" + KEY0_LONG_ATTR.dataType(), KEY0_LONG_ATTR),
+        Map.entry(KEY1_KEYWORD_ATTR.name() + ":" + KEY1_KEYWORD_ATTR.dataType(), KEY1_KEYWORD_ATTR),
+        Map.entry(KEY1_LONG_ATTR.name() + ":" + KEY1_LONG_ATTR.dataType(), KEY1_LONG_ATTR),
+        Map.entry(KEY2_KEYWORD_ATTR.name() + ":" + KEY2_KEYWORD_ATTR.dataType(), KEY2_KEYWORD_ATTR),
+        Map.entry(KEY2_LONG_ATTR.name() + ":" + KEY2_LONG_ATTR.dataType(), KEY2_LONG_ATTR),
+        Map.entry(KEY3_KEYWORD_ATTR.name() + ":" + KEY3_KEYWORD_ATTR.dataType(), KEY3_KEYWORD_ATTR),
+        Map.entry(KEY3_INTEGER_ATTR.name() + ":" + KEY3_INTEGER_ATTR.dataType(), KEY3_INTEGER_ATTR),
+        // Right-side attributes (from lookup index)
+        Map.entry(RKEY0_KEYWORD_ATTR.name() + ":" + RKEY0_KEYWORD_ATTR.dataType(), RKEY0_KEYWORD_ATTR),
+        Map.entry(RKEY0_LONG_ATTR.name() + ":" + RKEY0_LONG_ATTR.dataType(), RKEY0_LONG_ATTR),
+        Map.entry(RKEY1_KEYWORD_ATTR.name() + ":" + RKEY1_KEYWORD_ATTR.dataType(), RKEY1_KEYWORD_ATTR),
+        Map.entry(RKEY1_LONG_ATTR.name() + ":" + RKEY1_LONG_ATTR.dataType(), RKEY1_LONG_ATTR),
+        Map.entry(RKEY2_KEYWORD_ATTR.name() + ":" + RKEY2_KEYWORD_ATTR.dataType(), RKEY2_KEYWORD_ATTR),
+        Map.entry(RKEY2_LONG_ATTR.name() + ":" + RKEY2_LONG_ATTR.dataType(), RKEY2_LONG_ATTR),
+        Map.entry(RKEY3_KEYWORD_ATTR.name() + ":" + RKEY3_KEYWORD_ATTR.dataType(), RKEY3_KEYWORD_ATTR),
+        Map.entry(RKEY3_INTEGER_ATTR.name() + ":" + RKEY3_INTEGER_ATTR.dataType(), RKEY3_INTEGER_ATTR),
+        Map.entry(R_FIELD_ATTR.name() + ":" + R_FIELD_ATTR.dataType(), R_FIELD_ATTR)
+    );
+
+    /**
+     * Gets a FieldAttribute by name and type. Throws IllegalArgumentException if not found.
+     */
+    private static FieldAttribute getAttribute(String name, DataType type) {
+        String key = name + ":" + type;
+        FieldAttribute attr = ATTRIBUTES_BY_NAME_AND_TYPE.get(key);
+        if (attr == null) {
+            throw new IllegalArgumentException("Attribute not found: " + key);
+        }
+        return attr;
+    }
+
     public void testKeywordKey() throws IOException {
         runLookup(List.of(DataType.KEYWORD), new UsingSingleLookupTable(new Object[][] { new String[] { "aa", "bb", "cc", "dd" } }), null);
     }
@@ -340,26 +434,40 @@ private PhysicalPlan buildRightPreJoinPlan(List<DataType> keyTypes, Expression f
         return new FragmentExec(esRelation);
     }
 
+    /**
+     * Gets the left-side attribute for a given index and type.
+     * This ensures consistent NameId usage across MatchConfig and join conditions.
+     */
+    private FieldAttribute getLeftSideAttribute(int index, DataType keyType) {
+        String name = "key" + index;
+        // Handle special case for key3 which can be INTEGER or KEYWORD
+        if (index == 3 && keyType == DataType.INTEGER) {
+            return getAttribute(name, DataType.INTEGER);
+        }
+        // Default to KEYWORD or LONG based on keyType
+        if (keyType == DataType.KEYWORD) {
+            return getAttribute(name, DataType.KEYWORD);
+        } else {
+            return getAttribute(name, DataType.LONG);
+        }
+    }
+
     /**
      * Gets the right-side attribute for a given index and type.
      * This ensures consistent NameId usage across join conditions and rightPreJoinPlan.
      */
     private FieldAttribute getRightSideAttribute(int index, DataType keyType) {
-        return switch (index) {
-            case 0 -> keyType == DataType.KEYWORD ? RKEY0_KEYWORD_ATTR : RKEY0_LONG_ATTR;
-            case 1 -> keyType == DataType.KEYWORD ? RKEY1_KEYWORD_ATTR : RKEY1_LONG_ATTR;
-            case 2 -> keyType == DataType.KEYWORD ? RKEY2_KEYWORD_ATTR : RKEY2_LONG_ATTR;
-            case 3 -> {
-                if (keyType == DataType.INTEGER) {
-                    yield RKEY3_INTEGER_ATTR;
-                } else if (keyType == DataType.KEYWORD) {
-                    yield RKEY3_KEYWORD_ATTR;
-                } else {
-                    throw new IllegalArgumentException("Unsupported key type for rkey3: " + keyType);
-                }
-            }
-            default -> throw new IllegalArgumentException("Unsupported number of keys: " + index);
-        };
+        String name = "rkey" + index;
+        // Handle special case for rkey3 which can be INTEGER or KEYWORD
+        if (index == 3 && keyType == DataType.INTEGER) {
+            return getAttribute(name, DataType.INTEGER);
+        }
+        // Default to KEYWORD or LONG based on keyType
+        if (keyType == DataType.KEYWORD) {
+            return getAttribute(name, DataType.KEYWORD);
+        } else {
+            return getAttribute(name, DataType.LONG);
+        }
     }
 
     private List<Attribute> buildRightSideAttributes(List<DataType> keyTypes) {
@@ -491,22 +599,19 @@ private void runLookup(List<DataType> keyTypes, PopulateIndices populateIndices,
                 TEST_REQUEST_TIMEOUT
             );
             final String finalNodeWithShard = nodeWithShard;
-            boolean expressionJoin = EsqlCapabilities.Cap.LOOKUP_JOIN_ON_BOOLEAN_EXPRESSION.isEnabled() ? randomBoolean() : false;
+            boolean expressionJoin = true;// EsqlCapabilities.Cap.LOOKUP_JOIN_ON_BOOLEAN_EXPRESSION.isEnabled() ? randomBoolean() : false;
             List<MatchConfig> matchFields = new ArrayList<>();
             List<Expression> joinOnConditions = new ArrayList<>();
             if (expressionJoin) {
                 for (int i = 0; i < keyTypes.size(); i++) {
-                    FieldAttribute leftAttr = new FieldAttribute(
-                        Source.EMPTY,
-                        "key" + i,
-                        new EsField("key" + i, keyTypes.get(0), Collections.emptyMap(), true, EsField.TimeSeriesFieldType.NONE)
-                    );
+                    // Use precreated static attributes to ensure NameId consistency
+                    FieldAttribute leftAttr = getLeftSideAttribute(i, keyTypes.get(i));
                     FieldAttribute rightAttr = getRightSideAttribute(i, keyTypes.get(i));
                     joinOnConditions.add(new Equals(Source.EMPTY, leftAttr, rightAttr));
                     // randomly decide to apply the filter as additional join on filter instead of pushed down filter
-                    boolean applyAsJoinOnCondition = EsqlCapabilities.Cap.LOOKUP_JOIN_WITH_FULL_TEXT_FUNCTION.isEnabled()
-                        ? randomBoolean()
-                        : false;
+                    boolean applyAsJoinOnCondition = true;// EsqlCapabilities.Cap.LOOKUP_JOIN_WITH_FULL_TEXT_FUNCTION.isEnabled()
+                    // ? randomBoolean()
+                    // : false;
                     if (applyAsJoinOnCondition
                         && pushedDownFilter instanceof FragmentExec fragmentExec
                         && fragmentExec.fragment() instanceof Filter filter) {
@@ -516,8 +621,10 @@ private void runLookup(List<DataType> keyTypes, PopulateIndices populateIndices,
                 }
             }
             // the matchFields are shared for both types of join
+            // Use precreated static attributes to ensure NameId consistency with join conditions
             for (int i = 0; i < keyTypes.size(); i++) {
-                matchFields.add(new MatchConfig("key" + i, i + 1, keyTypes.get(i)));
+                FieldAttribute keyAttr = getLeftSideAttribute(i, keyTypes.get(i));
+                matchFields.add(new MatchConfig(keyAttr, i + 1, keyTypes.get(i)));
             }
             PhysicalPlan rightPreJoinPlan = pushedDownFilter != null ? pushedDownFilter : buildRightPreJoinPlan(keyTypes, null);
             LookupFromIndexOperator.Factory lookup = new LookupFromIndexOperator.Factory(

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

@@ -52,7 +52,7 @@ protected Collection<Class<? extends Plugin>> nodePlugins() {
     private void ensureIndicesAndData() {
         assumeTrue(
             "LOOKUP JOIN ON general expressions requires capability",
-            EsqlCapabilities.Cap.LOOKUP_JOIN_ON_BOOLEAN_EXPRESSION.isEnabled()
+            EsqlCapabilities.Cap.LOOKUP_JOIN_WITH_GENERAL_EXPRESSION.isEnabled()
         );
         if (indexExists(MAIN_INDEX) == false) {
             createIndices();
@@ -63,7 +63,7 @@ private void ensureIndicesAndData() {
     private void ensureMultiColumnIndicesAndData() {
         assumeTrue(
             "LOOKUP JOIN ON general expressions requires capability",
-            EsqlCapabilities.Cap.LOOKUP_JOIN_ON_BOOLEAN_EXPRESSION.isEnabled()
+            EsqlCapabilities.Cap.LOOKUP_JOIN_WITH_GENERAL_EXPRESSION.isEnabled()
         );
         if (indexExists(MULTI_COL_MAIN_INDEX) == false) {
             loadMultiColumnDataFromCsv();
@@ -1004,4 +1004,44 @@ public void testComplexExpressionRelatingLeftAndRightMultiColumn() {
             );
         }
     }
+
+    public void testLookupJoinWithLikeCondition() {
+        ensureMultiColumnIndicesAndData();
+        // Test LOOKUP JOIN with LIKE condition in the ON clause
+        // Matches rows where other1 ends with "ta" (pattern "*ta")
+        String query = String.format(Locale.ROOT, """
+            FROM %s
+            | RENAME id_int AS id_left, is_active_bool AS is_active_left
+            | LOOKUP JOIN %s ON id_int == id_left AND is_active_left == is_active_bool AND other1 LIKE "*ta"
+            | KEEP id_left, name_str, extra1, other1, other2
+            | SORT id_left, name_str, extra1, other1, other2
+            """, MULTI_COL_MAIN_INDEX, MULTI_COL_LOOKUP_INDEX);
+
+        try (EsqlQueryResponse response = runQuery(query)) {
+            List<List<Object>> values = getValuesList(response);
+            // Expected results based on CSV spec: rows where other1 matches "*ta" pattern
+            // Note: Some rows may have null values when the LIKE condition doesn't match
+            verifyResults(
+                values,
+                new Object[] { 1, "Alice", "foo", "beta", 2000 },  // other1="beta" ends with "ta"
+                new Object[] { List.of(1, 19, 21), null, "zyx", null, null },  // no match for multi-value id
+                new Object[] { 2, null, "bar", null, null },  // no match (other1 doesn't end with "ta")
+                new Object[] { 3, "Charlie", "baz", "delta", 4000 },  // other1="delta" ends with "ta"
+                new Object[] { 4, "David", "qux", "zeta", 6000 },  // other1="zeta" ends with "ta"
+                new Object[] { 5, "Eve", "quux", "eta", 7000 },  // other1="eta" ends with "ta"
+                new Object[] { 5, "Eve", "quux", "theta", 8000 },  // other1="theta" ends with "ta"
+                new Object[] { 6, null, "corge", "iota", 9000 },  // other1="iota" ends with "ta"
+                new Object[] { 7, null, "grault", null, null },  // no match
+                new Object[] { 8, null, "garply", null, null },  // no match
+                new Object[] { 9, null, "waldo", null, null },  // no match
+                new Object[] { 10, null, "fred", null, null },  // no match
+                new Object[] { 12, null, "xyzzy", null, null },  // no match
+                new Object[] { 13, null, "thud", null, null },  // no match
+                new Object[] { 14, null, "foo2", null, null },  // no match
+                new Object[] { 15, null, "bar2", null, null },  // no match
+                new Object[] { List.of(17, 18), null, "xyz", null, null },  // no match for multi-value id
+                new Object[] { null, null, "plugh", null, null }  // no match for null id
+            );
+        }
+    }
 }

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

@@ -696,7 +696,7 @@ private List<MatchConfig> collectLeftSideFieldsToBroadcast(LookupEnrichQueryGene
                         // we do match by just name
                         // we made sure the same attribute is not on the right side with the checks above
                         for (MatchConfig matchField : lookupRequest.getMatchFields()) {
-                            if (attr.name().equals(matchField.fieldName())) {
+                            if (attr.equals(matchField.fieldName())) {
                                 builder.append(attr);
                                 allLeftFieldsToBroadcast.add(matchField);
                                 addedNameIds.add(nameId);

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

@@ -313,10 +313,11 @@ private boolean applyAsLeftRightBinaryComparison(
             // 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
+            // Compare by attribute (equals compares by NameId) to ensure we match the same attribute instance
             Block block = null;
             DataType dataType = null;
             for (int i = 0; i < matchFields.size(); i++) {
-                if (matchFields.get(i).fieldName().equals(leftAttribute.name())) {
+                if (matchFields.get(i).fieldName().equals(leftAttribute)) {
                     block = inputPage.getBlock(i);
                     dataType = matchFields.get(i).type();
                     break;

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

@@ -184,7 +184,7 @@ private List<MatchConfig> uniqueMatchFieldsByName(List<MatchConfig> matchFields)
             return matchFields;
         }
         List<MatchConfig> uniqueFields = new ArrayList<>();
-        Set<String> seenFieldNames = new HashSet<>();
+        Set<NamedExpression> seenFieldNames = new HashSet<>();
         for (MatchConfig matchField : matchFields) {
             if (seenFieldNames.add(matchField.fieldName())) {
                 uniqueFields.add(matchField);

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

@@ -34,9 +34,11 @@
 import org.elasticsearch.xpack.esql.EsqlIllegalArgumentException;
 import org.elasticsearch.xpack.esql.action.EsqlQueryAction;
 import org.elasticsearch.xpack.esql.core.expression.Expression;
+import org.elasticsearch.xpack.esql.core.expression.FieldAttribute;
 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.core.type.EsField;
 import org.elasticsearch.xpack.esql.io.stream.PlanStreamInput;
 import org.elasticsearch.xpack.esql.io.stream.PlanStreamOutput;
 import org.elasticsearch.xpack.esql.plan.physical.FilterExec;
@@ -47,6 +49,7 @@
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 import java.util.stream.Collectors;
 
@@ -118,7 +121,7 @@ protected LookupEnrichQueryGenerator queryList(
             for (int i = 0; i < request.matchFields.size(); i++) {
                 MatchConfig matchField = request.matchFields.get(i);
                 QueryList q = termQueryList(
-                    context.getFieldType(matchField.fieldName()),
+                    context.getFieldType(matchField.fieldName().name()),
                     context,
                     aliasFilter,
                     request.inputPage.getBlock(matchField.channel()),
@@ -244,9 +247,17 @@ static TransportRequest readFrom(StreamInput in, BlockFactory blockFactory) thro
                 matchFields = planIn.readCollectionAsList(MatchConfig::new);
             } else {
                 String matchField = in.readString();
+                FieldAttribute fieldName = new FieldAttribute(
+                    Source.EMPTY,
+                    null,
+                    null,
+                    matchField,
+                    new EsField(matchField, inputDataType, Map.of(), true, EsField.TimeSeriesFieldType.NONE)
+                );
+                ;
                 // For older versions, we only support a single match field.
                 matchFields = new ArrayList<>(1);
-                matchFields.add(new MatchConfig(matchField, 0, inputDataType));
+                matchFields.add(new MatchConfig(fieldName, 0, inputDataType));
             }
             var source = Source.readFrom(planIn);
             // Source.readFrom() requires the query from the Configuration passed to PlanStreamInput.
@@ -319,7 +330,7 @@ public void writeTo(StreamOutput out) throws IOException {
             } else {
                 // older versions only support a single match field, we already checked this above when writing the datatype
                 // send the field name of the first and only match field here
-                out.writeString(matchFields.get(0).fieldName());
+                out.writeString(matchFields.get(0).fieldName().name());
             }
             source.writeTo(planOut);
             if (out.getTransportVersion().supports(ESQL_LOOKUP_JOIN_SOURCE_TEXT)) {
@@ -340,7 +351,7 @@ public void writeTo(StreamOutput out) throws IOException {
         @Override
         protected String extraDescription() {
             return " ,match_fields="
-                + matchFields.stream().map(MatchConfig::fieldName).collect(Collectors.joining(", "))
+                + matchFields.stream().map(x -> x.fieldName().name()).collect(Collectors.joining(", "))
                 + ", right_pre_join_plan="
                 + (rightPreJoinPlan == null ? "null" : rightPreJoinPlan.toString());
         }