[SPARK-58383][SQL] Bind the right filter by expression id in ReplaceExceptWithFilter - #58046

What changes were proposed in this pull request?

ReplaceExceptWithFilter rewrites left EXCEPT right into Distinct(Filter(NOT coalesce(cond, false), left)) when the right side is the same relation and projection as the left plus an additional filter.

The filter condition references the base columns below the right projection, but transformCondition remapped it onto left.output by attribute name:

val attributeNameMap: Map[String, Attribute] = plan.output.map(x => (x.name, x)).toMap
if (condition.references.forall(r => attributeNameMap.contains(r.name))) {
  val rewrittenCondition = condition.transform {
    case a: AttributeReference => attributeNameMap(a.name)
  }

When a left alias reuses the name of an unrelated base column, the injected predicate binds to the wrong column.

This PR maps the condition through the base relations instead:

1. The left and right base relations are already known to be sameResult (checked by verifyConditions), so their outputs correspond positionally — that gives right-base attribute -> left-base attribute.
2. The left base attributes are then translated into left.output through the left project list by expression id, either directly or through an alias.
3. If any reference cannot be traced, the rewrite is skipped and the plan keeps the anti-join.

This replaces the name-presence check added by SPARK-23274, and also lets the left-output-name-distinctness guard from the original SPARK-22181 commit go away: it existed because the name-keyed map silently collapsed duplicate names, which expression-id matching cannot do.

Why are the changes needed?

It is a correctness bug: the query returns wrong results. Present since Spark 2.3.0, when the rule was introduced.

CREATE TABLE repro_t(id INT, val INT) USING parquet;
INSERT INTO repro_t VALUES (1, 10), (2, 20);

(SELECT val AS id FROM repro_t)
EXCEPT
(SELECT val AS v FROM repro_t WHERE id = 1);

The right side contains no alias named id, so its WHERE id = 1 unambiguously references the base column. But the condition is ported onto the left output by name and lands on the left's alias of val:

== Optimized Logical Plan ==
Aggregate [id#12239], [id#12239]
+- Project [val#12251 AS id#12239]
   +- Filter NOT coalesce((val#12251 = 1), false)
      +- Relation default.repro_t[id#12250,val#12251] parquet

SPARK-23274 added a check that every reference in the condition is present among the left output names, but that does not catch this case: here the name is present, it just belongs to an unrelated column. The collision is between a left alias and a right base column. Note that alias names are erased during canonicalization (QueryPlan#doCanonicalize rewrites Alias(child, name) to Alias(normalizedChild, "")), which is why val AS id and val AS v compare as sameResult and the rule fires in the first place.

The existing workaround is SET spark.sql.optimizer.replaceExceptWithFilter=false.

Does this PR introduce any user-facing change?

Yes, it fixes wrong results for the queries described above. Affected queries now return the correct answer.

There is also a secondary plan change. Because references are now traced by expression id rather than by name, the rule can apply in a case where it previously bailed out — when the right condition references a base column that the left side exposes under a different name, e.g.

(SELECT val AS v FROM t) EXCEPT (SELECT val AS v FROM t WHERE val = 1)

Previously val was not found among the left output names (v) and the plan fell back to the anti-join; it is now rewritten to the filter form.

Dropping the name-distinctness guard widens the rule the same way for a left side with duplicate output names, e.g. SELECT id AS x, val AS x FROM t, which previously always fell back to the anti-join.

In both cases the results are unchanged — these are plan/performance changes only.

How was this patch tested?

New tests, both of which fail without the rule change and pass with it:

• ReplaceOperatorSuite: SPARK-58383: ReplaceExceptWithFilter should not bind the right condition by name — asserts the rewrite is skipped and the anti-join is kept.
• SQLQuerySuite: SPARK-58383: EXCEPT when a left alias reuses the name of a base column — end-to-end version of the reported repro, plus a duplicate-left-output-name query covering the path opened up by removing the distinctness guard. Both run with spark.sql.optimizer.replaceExceptWithFilter enabled and disabled.

Verified the end-to-end test catches the bug by reverting only the rule change and re-running it: it returned 2 rows instead of 1, with DataFilters: [NOT coalesce((val#87 = 1), false)] on the scan.

Existing suites, all passing:

build/sbt 'catalyst/testOnly org.apache.spark.sql.catalyst.optimizer.*'
build/sbt 'sql/testOnly org.apache.spark.sql.SQLQuerySuite'
build/sbt 'sql/testOnly org.apache.spark.sql.DataFrameSetOperationsSuite'
build/sbt 'sql/testOnly *SQLQueryTestSuite -- -z except'

Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude Code (Opus 5)