dongjoon-hyun opened a new pull request, #59895:
URL: https://github.com/apache/spark/pull/59895
### What changes were proposed in this pull request?
This PR makes `AddMetadataColumns` skip the target table of `DELETE`,
`UPDATE` and `MERGE` when Spark rewrites the command into a row-level
operation, that is, when the table supports row-level operations
(`SupportsRowLevelOperations`). `UPDATE` and `MERGE` on a table that accepts
any schema keep the current behavior, because Spark does not align their
assignments and leaves them to the rules injected for that table.
`RewriteDeleteFromTable`, `RewriteUpdateTable` and `RewriteMergeIntoTable`
read the metadata columns that the operation requires on their own
(`RewriteRowLevelCommand.buildRelationWithAttrs`). They take the output of the
target table as the data columns to write, and
`ResolveRowLevelCommandAssignments` aligns assignments over the same output.
That output has only data columns when the command is resolved in the first
iteration of the `Resolution` batch, because `AddMetadataColumns` runs before
`ResolveReferences`. When the command is resolved only in a later iteration,
for example because its condition needs type coercion, `AddMetadataColumns`
sees a resolved command that references a metadata column. It then calls
`DataSourceV2Relation.withMetadataColumns()` on the target table, which appends
every metadata column to its output. As a result:
- `ResolveRowLevelCommandAssignments` adds identity assignments such as
`index = index` for the metadata columns.
- `RewriteUpdateTable` handles them as data columns. A metadata column with
`PRESERVE_ON_UPDATE = false` keeps its old value instead of `null`.
- Group-based `DELETE` and `UPDATE`, and delta-based `UPDATE`, pass the
metadata columns in every row and report them in `LogicalWriteInfo.schema()`.
- An `UPDATE` that is represented as a delete and an insert fails in
`ExpandExec`, because the delete and reinsert projections have different widths.
- A `MERGE` with a `NOT MATCHED` clause fails with `No assignment for
'index'`, and a `MERGE` with only `MATCHED` clauses fails with an assertion
error.
For example, take an in-memory table `pk INT NOT NULL, id INT, dep STRING`
partitioned by `dep`, with the metadata columns `index` (`PRESERVE_ON_UPDATE =
false`) and `_partition`, and run:
```sql
UPDATE t SET id = -1 WHERE id IN (1, 100) AND _partition = 'hr' AND id > 0.5
```
**BEFORE**
The target table gains `index` and `_partition`, and `index` is copied:
```
ReplaceData RelationV2[pk#165, id#166, dep#167, index#168, _partition#169]
cat.ns1.test_table
+- Project [..., if (...) index#168 else index#168 AS index#174, if (...)
_partition#169 else _partition#169 AS _partition#175]
+- RelationV2[pk#165, id#166, dep#167, index#168, _partition#169]
cat.ns1.test_table
```
The connector receives the updated row with the metadata `(hr, 0)` and the
write schema `struct<pk:int,id:int,dep:string,index:int,_partition:string>`.
**AFTER**
The plan is the same as for the statement without `AND id > 0.5`. The
connector receives the metadata `(hr, null)` and the write schema
`struct<pk:int,id:int,dep:string>`.
### Why are the changes needed?
`MetadataColumn.PRESERVE_ON_UPDATE` says that the metadata value is replaced
with `null` if it is false. This contract is broken silently for `UPDATE` and
`MERGE` statements that are resolved in a later analyzer iteration. Row lineage
in Iceberg and Delta Lake relies on this contract to reset the row version on
update (SPARK-50820). Such statements also pass metadata columns to connectors
as data columns, or fail.
### Does this PR introduce _any_ user-facing change?
Yes. This is a bug fix for DSv2 `DELETE`, `UPDATE` and `MERGE` statements
that reference a metadata column and are resolved in a later analyzer iteration:
- `UPDATE` and `MERGE` replace metadata columns with `PRESERVE_ON_UPDATE =
false` with `null` on update, as documented.
- Connectors no longer receive metadata columns as data columns, in rows or
in `LogicalWriteInfo.schema()`.
- An `UPDATE` represented as a delete and an insert, and `MERGE`, no longer
fail.
These statements are now analyzed in the same way as statements that are
resolved in the first iteration. So a reference to a metadata column that the
row-level operation does not request in `requiredMetadataAttributes()` now
fails analysis, as it already does without type coercion (for example, without
`AND id > 0.5` above). Before, such a statement could succeed only because
every metadata column of the table was written as a data column.
### How was this patch tested?
Pass the CIs with the newly added test cases. Each runs a statement that
references a metadata column and needs type coercion, and checks the result,
the write schema and the metadata that the connector receives.
- `GroupBasedUpdateTableSuite`, `DeltaBasedUpdateTableSuite`,
`DeltaBasedUpdateAsDeleteAndInsertTableSuite`: `SPARK-60667: update with type
coercion handles metadata columns correctly`
- `GroupBasedDeleteFromTableSuite`: `SPARK-60667: delete with type coercion
preserves metadata columns for carried-over records`
- `GroupBasedMergeIntoTableSuite`, `DeltaBasedMergeIntoTableSuite`,
`DeltaBasedMergeIntoTableUpdateAsDeleteAndInsertSuite`: `SPARK-60667: merge
with type coercion handles metadata columns correctly`
All of them fail without this fix.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.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]
