ESQL: Expression based Lookup Join follow up and refactor

x-pack/plugin/esql/qa/security/src/javaRestTest/java/org/elasticsearch/xpack/esql/EsqlSecurityIT.java

@@ -828,12 +828,7 @@ private void testLookupJoinFieldLevelSecurityHelper(boolean useExpressionJoin) t
         ResponseException error = expectThrows(ResponseException.class, () -> runESQLCommand("fls_user4_1", query));
         assertThat(error.getResponse().getStatusLine().getStatusCode(), equalTo(HttpStatus.SC_BAD_REQUEST));
         if (useExpressionJoin) {
-            assertThat(
-                error.getMessage(),
-                containsString(
-                    "Join condition must be between one attribute on the left side and one attribute on the right side of the join"
-                )
-            );
+            assertThat(error.getMessage(), containsString("Unsupported join filter expression:value_left == value"));
         } else {
             assertThat(error.getMessage(), containsString("Unknown column [value] in right side of join"));
         }
@@ -907,12 +902,7 @@ private void testLookupJoinFieldLevelSecurityOnAliasHelper(boolean useExpression
         ResponseException error = expectThrows(ResponseException.class, () -> runESQLCommand("fls_user4_1_alias", query));
         assertThat(error.getResponse().getStatusLine().getStatusCode(), equalTo(HttpStatus.SC_BAD_REQUEST));
         if (useExpressionJoin) {
-            assertThat(
-                error.getMessage(),
-                containsString(
-                    "Join condition must be between one attribute on the left side and one attribute on the right side of the join"
-                )
-            );
+            assertThat(error.getMessage(), containsString("Unsupported join filter expression:value_left == value"));
         } else {
             assertThat(error.getMessage(), containsString("Unknown column [value] in right side of join"));
         }

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/analysis/Analyzer.java

@@ -129,7 +129,6 @@
 import org.elasticsearch.xpack.esql.plan.logical.join.JoinConfig;
 import org.elasticsearch.xpack.esql.plan.logical.join.JoinType;
 import org.elasticsearch.xpack.esql.plan.logical.join.JoinTypes;
-import org.elasticsearch.xpack.esql.plan.logical.join.JoinTypes.UsingJoinType;
 import org.elasticsearch.xpack.esql.plan.logical.join.LookupJoin;
 import org.elasticsearch.xpack.esql.plan.logical.local.EsqlProject;
 import org.elasticsearch.xpack.esql.plan.logical.local.LocalRelation;
@@ -759,9 +758,9 @@ private Expression resolveAndOrientJoinCondition(Expression condition, Attribute
                 if (leftIsFromRight && rightIsFromLeft) {
                     return comp.swapLeftAndRight(); // Swapped orientation
                 }
-
-                // Invalid orientation (e.g., both from left or both from right)
-                throw new IllegalArgumentException(
+                return new UnresolvedAttribute(
+                    condition.source(),
+                    "unsupported",
                     "Join condition must be between one attribute on the left side and "
                         + "one attribute on the right side of the join, but found: "
                         + condition.sourceText()
@@ -776,24 +775,14 @@ private Join resolveLookupJoin(LookupJoin join) {
             JoinType type = config.type();
 
             // rewrite the join into an equi-join between the field with the same name between left and right
-            if (type instanceof UsingJoinType using) {
-                List<Attribute> cols = using.columns();
+            if (type == JoinTypes.LEFT) {
                 // the lookup cannot be resolved, bail out
-                if (Expressions.anyMatch(cols, c -> c instanceof UnresolvedAttribute ua && ua.customMessage())) {
+                if (Expressions.anyMatch(
+                    join.references().stream().toList(),
+                    c -> c instanceof UnresolvedAttribute ua && ua.customMessage()
+                )) {
                     return join;
                 }
-
-                JoinType coreJoin = using.coreJoin();
-                // verify the join type
-                if (coreJoin != JoinTypes.LEFT) {
-                    String name = cols.get(0).name();
-                    UnresolvedAttribute errorAttribute = new UnresolvedAttribute(
-                        join.source(),
-                        name,
-                        "Only LEFT join is supported with USING"
-                    );
-                    return join.withConfig(new JoinConfig(type, singletonList(errorAttribute), emptyList(), null));
-                }
                 List<Attribute> leftKeys = new ArrayList<>();
                 List<Attribute> rightKeys = new ArrayList<>();
                 List<Expression> resolvedFilters = new ArrayList<>();
@@ -812,27 +801,29 @@ private Join resolveLookupJoin(LookupJoin join) {
                             leftKeys.add(leftAttribute);
                             rightKeys.add(rightAttribute);
                         } else {
-                            throw new IllegalArgumentException("Unsupported join filter expression: " + expression);
+                            UnresolvedAttribute errorAttribute = new UnresolvedAttribute(
+                                expression.source(),
+                                "unsupported",
+                                "Unsupported join filter expression:" + expression.sourceText()
+                            );
+                            return join.withConfig(new JoinConfig(type, singletonList(errorAttribute), emptyList(), null));
+
                         }
                     }
                 } else {
                     // resolve the using columns against the left and the right side then assemble the new join config
-                    leftKeys = resolveUsingColumns(cols, join.left().output(), "left");
-                    rightKeys = resolveUsingColumns(cols, join.right().output(), "right");
+                    leftKeys = resolveUsingColumns(join.config().leftFields(), join.left().output(), "left");
+                    rightKeys = resolveUsingColumns(join.config().rightFields(), join.right().output(), "right");
                 }
 
-                config = new JoinConfig(coreJoin, leftKeys, rightKeys, Predicates.combineAnd(resolvedFilters));
+                config = new JoinConfig(type, leftKeys, rightKeys, Predicates.combineAnd(resolvedFilters));
                 return new LookupJoin(join.source(), join.left(), join.right(), config, join.isRemote());
-            } else if (type != JoinTypes.LEFT) {
+            } else {
                 // everything else is unsupported for now
-                // LEFT can only happen by being mapped from a USING above. So we need to exclude this as well because this rule can be run
-                // more than once.
                 UnresolvedAttribute errorAttribute = new UnresolvedAttribute(join.source(), "unsupported", "Unsupported join type");
                 // add error message
                 return join.withConfig(new JoinConfig(type, singletonList(errorAttribute), emptyList(), null));
             }
-
-            return join;
         }
 
         private LogicalPlan resolveFork(Fork fork, AnalyzerContext context) {

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/logical/join/JoinTypes.java

@@ -9,13 +9,9 @@
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.util.Maps;
-import org.elasticsearch.xpack.esql.core.capabilities.Resolvables;
-import org.elasticsearch.xpack.esql.core.expression.Attribute;
 
 import java.io.IOException;
-import java.util.List;
 import java.util.Map;
-import java.util.Objects;
 
 /**
  * Utility class defining the concrete types of joins supported by ESQL.
@@ -69,80 +65,6 @@ public void writeTo(StreamOutput out) throws IOException {
         }
     }
 
-    /**
-     * Join type for the USING clause - shorthand for defining an equi-join (equality join meaning the condition checks if columns across
-     * each side of the join are equal).
-     * One important difference is that the USING clause returns the join column only once, at the beginning of the result set.
-     */
-    public static class UsingJoinType implements JoinType {
-        private final List<Attribute> columns;
-        private final JoinType coreJoin;
-
-        public UsingJoinType(JoinType coreJoin, List<Attribute> columns) {
-            this.columns = columns;
-            this.coreJoin = coreJoin;
-        }
-
-        @Override
-        public String joinName() {
-            return coreJoin.joinName() + " USING " + columns.toString();
-        }
-
-        @Override
-        public void writeTo(StreamOutput out) throws IOException {
-            throw new IllegalArgumentException("USING join type should not be serialized due to being rewritten");
-        }
-
-        public JoinType coreJoin() {
-            return coreJoin;
-        }
-
-        public List<Attribute> columns() {
-            return columns;
-        }
-
-        @Override
-        public boolean resolved() {
-            return Resolvables.resolved(columns);
-        }
-
-        @Override
-        public int hashCode() {
-            return Objects.hash(columns, coreJoin);
-        }
-
-        @Override
-        public boolean equals(Object o) {
-            if (this == o) return true;
-            if (o == null || getClass() != o.getClass()) return false;
-            UsingJoinType that = (UsingJoinType) o;
-            return Objects.equals(columns, that.columns) && coreJoin == that.coreJoin;
-        }
-
-        @Override
-        public String toString() {
-            return joinName();
-        }
-    }
-
-    /**
-     * Private class so it doesn't get used yet it is defined to showcase why the join type was defined as an interface instead of a simpler
-     * enum.
-     */
-    private abstract static class NaturalJoinType implements JoinType {
-
-        private final JoinType joinType;
-
-        private NaturalJoinType(JoinType joinType) {
-            this.joinType = joinType;
-        }
-
-        @Override
-        public String joinName() {
-            return "NATURAL " + joinType.joinName();
-        }
-    }
-
     public static JoinType readFrom(StreamInput in) throws IOException {
         byte id = in.readByte();
         JoinType type = JOIN_TYPES.get(id);

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/logical/join/LookupJoin.java

@@ -18,11 +18,9 @@
 import org.elasticsearch.xpack.esql.plan.logical.Limit;
 import org.elasticsearch.xpack.esql.plan.logical.LogicalPlan;
 import org.elasticsearch.xpack.esql.plan.logical.SurrogateLogicalPlan;
-import org.elasticsearch.xpack.esql.plan.logical.join.JoinTypes.UsingJoinType;
 
 import java.util.List;
 
-import static java.util.Collections.emptyList;
 import static org.elasticsearch.xpack.esql.common.Failure.fail;
 import static org.elasticsearch.xpack.esql.plan.logical.join.JoinTypes.LEFT;
 
@@ -40,7 +38,7 @@ public LookupJoin(
         boolean isRemote,
         @Nullable Expression joinOnConditions
     ) {
-        this(source, left, right, new UsingJoinType(LEFT, joinFields), emptyList(), emptyList(), isRemote, joinOnConditions);
+        this(source, left, right, LEFT, joinFields, joinFields, isRemote, joinOnConditions);
     }
 
     public LookupJoin(

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/session/FieldNameUtils.java

@@ -41,7 +41,6 @@
 import org.elasticsearch.xpack.esql.plan.logical.TopN;
 import org.elasticsearch.xpack.esql.plan.logical.UnresolvedRelation;
 import org.elasticsearch.xpack.esql.plan.logical.inference.Completion;
-import org.elasticsearch.xpack.esql.plan.logical.join.JoinTypes;
 import org.elasticsearch.xpack.esql.plan.logical.join.LookupJoin;
 import org.elasticsearch.xpack.esql.session.EsqlSession.PreAnalysisResult;
 
@@ -147,9 +146,7 @@ public static PreAnalysisResult resolveFieldNames(LogicalPlan parsed, boolean ha
                 enrichRefs.removeIf(attr -> attr instanceof EmptyAttribute);
                 referencesBuilder.get().addAll(enrichRefs);
             } else if (p instanceof LookupJoin join) {
-                if (join.config().type() instanceof JoinTypes.UsingJoinType usingJoinType) {
-                    joinRefs.addAll(usingJoinType.columns());
-                }
+                joinRefs.addAll(join.config().leftFields());
                 if (keepRefs.isEmpty()) {
                     // No KEEP commands after the JOIN, so we need to mark this index for "*" field resolution
                     wildcardJoinIndices.add(((UnresolvedRelation) join.right()).indexPattern().indexPattern());

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/parser/StatementParserTests.java

@@ -72,7 +72,6 @@
 import org.elasticsearch.xpack.esql.plan.logical.fuse.Fuse;
 import org.elasticsearch.xpack.esql.plan.logical.inference.Completion;
 import org.elasticsearch.xpack.esql.plan.logical.inference.Rerank;
-import org.elasticsearch.xpack.esql.plan.logical.join.JoinTypes;
 import org.elasticsearch.xpack.esql.plan.logical.join.LookupJoin;
 
 import java.util.ArrayList;
@@ -3240,12 +3239,11 @@ public void testValidJoinPatternFieldJoin() {
         assertThat(as(join.left(), UnresolvedRelation.class).indexPattern().indexPattern(), equalTo(unquoteIndexPattern(basePattern)));
         assertThat(as(join.right(), UnresolvedRelation.class).indexPattern().indexPattern(), equalTo(unquoteIndexPattern(joinPattern)));
 
-        var joinType = as(join.config().type(), JoinTypes.UsingJoinType.class);
-        assertThat(joinType.columns(), hasSize(numberOfOnFields));
+        assertThat(join.config().leftFields(), hasSize(numberOfOnFields));
         for (int i = 0; i < numberOfOnFields; i++) {
-            assertThat(as(joinType.columns().get(i), UnresolvedAttribute.class).name(), equalTo(existingIdentifiers.get(i)));
+            assertThat(as(join.config().leftFields().get(i), UnresolvedAttribute.class).name(), equalTo(existingIdentifiers.get(i)));
         }
-        assertThat(joinType.coreJoin().joinName(), equalTo("LEFT OUTER"));
+        assertThat(join.config().type().joinName(), equalTo("LEFT OUTER"));
     }
 
     /**
@@ -4595,9 +4593,8 @@ public void testDoubleParamsForIdentifier() {
             LookupJoin join = as(limit.child(), LookupJoin.class);
             UnresolvedRelation ur = as(join.right(), UnresolvedRelation.class);
             assertEquals(ur.indexPattern().indexPattern(), "idx");
-            JoinTypes.UsingJoinType joinType = as(join.config().type(), JoinTypes.UsingJoinType.class);
-            assertEquals(joinType.coreJoin().joinName(), "LEFT OUTER");
-            assertEquals(joinType.columns(), List.of(attribute("f9")));
+            assertEquals(join.config().type().joinName(), "LEFT OUTER");
+            assertEquals(join.config().leftFields(), List.of(attribute("f9")));
             Rename rename = as(join.left(), Rename.class);
             assertEquals(rename.renamings(), List.of(new Alias(EMPTY, "f.8", attribute("f7*."))));
             Grok grok = as(rename.child(), Grok.class);
@@ -4705,9 +4702,8 @@ public void testDoubleParamsForIdentifier() {
             LookupJoin join = as(limit.child(), LookupJoin.class);
             UnresolvedRelation ur = as(join.right(), UnresolvedRelation.class);
             assertEquals(ur.indexPattern().indexPattern(), "idx");
-            JoinTypes.UsingJoinType joinType = as(join.config().type(), JoinTypes.UsingJoinType.class);
-            assertEquals(joinType.coreJoin().joinName(), "LEFT OUTER");
-            assertEquals(joinType.columns(), List.of(attribute("f13.f14")));
+            assertEquals(join.config().type().joinName(), "LEFT OUTER");
+            assertEquals(join.config().leftFields(), List.of(attribute("f13.f14")));
             Rename rename = as(join.left(), Rename.class);
             assertEquals(rename.renamings(), List.of(new Alias(EMPTY, "f11*..f.12", attribute("f.9*.f.10."))));
             Grok grok = as(rename.child(), Grok.class);
@@ -4956,9 +4952,8 @@ public void testMixedSingleDoubleParams() {
         LookupJoin join = as(plan, LookupJoin.class);
         UnresolvedRelation ur = as(join.right(), UnresolvedRelation.class);
         assertEquals(ur.indexPattern().indexPattern(), "idx");
-        JoinTypes.UsingJoinType joinType = as(join.config().type(), JoinTypes.UsingJoinType.class);
-        assertEquals(joinType.coreJoin().joinName(), "LEFT OUTER");
-        assertEquals(joinType.columns(), List.of(attribute("f5")));
+        assertEquals(join.config().type().joinName(), "LEFT OUTER");
+        assertEquals(join.config().leftFields(), List.of(attribute("f5")));
         Drop drop = as(join.left(), Drop.class);
         List<? extends NamedExpression> removals = drop.removals();
         assertEquals(removals.size(), 2);