Vivek1106-04 opened a new pull request, #58046:
URL: https://github.com/apache/spark/pull/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**:
```scala
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.
### 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.
```sql
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
```
| | before | after |
|---|---|---|
| result | `[10]`, `[20]` | `[20]` |
`SPARK-23274` added a guard that the left output names are distinct, but
that does not catch this case: the collision is between a left alias and a
right *base* column, which is a name the guard never looks at. 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.
```sql
(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. The
results are unchanged, this is a plan/performance change only.
I left the `SPARK-23274` left-output-name-distinctness guard in
`verifyConditions` untouched to keep the change contained, though it is now
redundant under expression-id matching. Happy to remove it if reviewers prefer.
### 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, run with
`spark.sql.optimizer.replaceExceptWithFilter` both 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.*' #
1408 passed
build/sbt 'sql/testOnly org.apache.spark.sql.SQLQuerySuite' #
272 passed
build/sbt 'sql/testOnly *SQLQueryTestSuite -- -z except' #
12 passed
```
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 5)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
