charliec05 opened a new pull request, #59078:
URL: https://github.com/apache/spark/pull/59078
### What changes were proposed in this pull request?
Preserve non-key attribute nullability in `ReplaceDeduplicateWithAggregate`:
wrap the generated `First` aggregate with `KnownNotNull` only when its input
attribute is non-nullable. Retain the generated-alias cache and leave nullable
attributes and streaming deduplication unchanged.
This is safe specifically for this rewrite: the rule always produces a
grouping aggregate, adding a literal grouping key when the deduplication key
list is empty. Every output group therefore contains an input row. The general
`First` expression is not changed.
Add optimizer, batch aggregation, and Data Source V2 write regressions, plus
a SQL migration note.
### Why are the changes needed?
The rule currently replaces each non-key attribute with `first(attribute)`,
whose result is nullable in general. This widens a non-nullable input column's
schema during optimization. An otherwise-valid write to a Data Source V2 `NOT
NULL` column then becomes unresolved and fails with
`PLAN_VALIDATION_FAILED_RULE_IN_BATCH`.
For example, a non-nullable `value` in `spark.range(3).selectExpr("id", "id
+ 10 AS value").dropDuplicates("id")` becomes nullable in the optimized plan,
and appending it to a table declared as `(id BIGINT NOT NULL, value BIGINT NOT
NULL)` fails.
This follows the scoped approach discussed by Josh Rosen in
[SPARK-44517](https://issues.apache.org/jira/browse/SPARK-44517). Unlike the
earlier, now-closed #42117 approach, it does not change `First.nullable`
globally: global aggregation over empty input and `first(..., true)` over
all-null input must remain nullable.
### Does this PR introduce _any_ user-facing change?
Yes. Batch `dropDuplicates` preserves non-key column nullability through
optimization, allowing writes to compatible `NOT NULL` columns. Empty-input
deduplication still returns no rows. The normal `first` aggregate retains its
existing null behavior.
### How was this patch tested?
- Before the fix, the new optimizer regression failed; both codegen modes
reproduced the schema change; the Data Source V2 append reproduced
`PLAN_VALIDATION_FAILED_RULE_IN_BATCH`.
- New coverage includes keyed and keyless deduplication, empty inputs,
nullable columns, a successful `NOT NULL` table append, and unchanged null
results from ordinary global `first` aggregates.
- `build/sbt "catalyst/testOnly
org.apache.spark.sql.catalyst.optimizer.ReplaceOperatorSuite"`: all 16 tests
passed.
- `build/sbt "sql/testOnly org.apache.spark.sql.DataFrameAggregateSuite"`:
all 177 tests passed, including codegen-on/off and empty-input regressions.
- `build/sbt "sql/testOnly
org.apache.spark.sql.connector.DataSourceV2DataFrameSuite -- -z SPARK-44517"`:
the `NOT NULL` table append regression passed.
- `build/sbt "sql/testOnly org.apache.spark.sql.DataFrameSuite -- -z
dropDuplicates"`: the existing deduplication regression passed.
- `dev/lint-scala`: ScalaStyle and ScalaFmt checks passed. Tests ran on Java
21.
- [Full GitHub Actions
CI](https://github.com/charliec05/spark/actions/runs/36294917010) is running on
`9ee811036dc`; its result is pending.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenAI Codex (GPT-5)
This contribution is submitted under the project's Apache License 2.0.
--
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]
