dongjoon-hyun commented on code in PR #55518:
URL: https://github.com/apache/spark/pull/55518#discussion_r4315467526
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2Writes.scala:
##########
@@ -109,22 +111,75 @@ object V2Writes extends Rule[LogicalPlan] with
PredicateHelper {
WriteToDataSourceV2(Some(r), microBatchWrite, newQuery, customMetrics)
case rd @ ReplaceData(r: DataSourceV2Relation, _, query, _, projections,
_, None) =>
- val rowSchema = projections.rowProjection.schema
+ val (rowSchema, columnUpdateSchema) =
+ writeSchemas(projections.rowProjection.schema, hasNarrowRows(rd))
val metadataSchema = projections.metadataProjection.map(_.schema)
val writeOptions = mergeOptions(Map.empty,
r.options.asCaseSensitiveMap.asScala.toMap)
- val writeBuilder = newWriteBuilder(r.table, writeOptions, rowSchema,
metadataSchema)
+ val writeBuilder = newWriteBuilder(r.table, writeOptions, rowSchema,
metadataSchema,
+ columnUpdateSchema)
val write = writeBuilder.build()
+ validateColumnUpdateWrite(rd, write)
val newQuery = DistributionAndOrderingUtils.prepareQuery(write, query,
r.funCatalog)
rd.copy(write = Some(write), query = newQuery)
case wd @ WriteDelta(r: DataSourceV2Relation, _, query, _, projections, _,
None) =>
val writeOptions = mergeOptions(Map.empty,
r.options.asCaseSensitiveMap.asScala.toMap)
- val deltaWriteBuilder = newDeltaWriteBuilder(r.table, writeOptions,
projections)
+ val deltaWriteBuilder =
+ newDeltaWriteBuilder(r.table, writeOptions, projections,
hasNarrowRows(wd))
val deltaWrite = deltaWriteBuilder.build()
+ validateColumnUpdateWrite(wd, deltaWrite)
val newQuery = DistributionAndOrderingUtils.prepareQuery(deltaWrite,
query, r.funCatalog)
wd.copy(write = Some(deltaWrite), query = newQuery)
}
+ /**
+ * Whether a row-level write delivers narrow rows for updated, copied, and
reinserted records.
+ * This must match the writes whose write relation RewriteUpdateTable
narrows, even when the
+ * declared columns cover the whole table.
+ */
+ private[v2] def hasNarrowRows(write: RowLevelWrite): Boolean = {
+ write.operation.isInstanceOf[SupportsColumnUpdates] &&
write.operation.command == UPDATE
Review Comment:
**6. Whether a write delivers narrow rows is decided twice and recorded
nowhere in the plan.** `RewriteUpdateTable.resolveWriteAttrs` narrows the write
relation for any `SupportsColumnUpdates` operation, while `hasNarrowRows`
re-derives the decision later from whatever operation sits in the write
relation at that point, plus `operation.command == UPDATE`, at five call sites
(L115, L128, L151, `GroupBasedRowLevelOperationScanPlanning.scala:70` and
`DataSourceV2Strategy.scala:594`). Nothing re-validates the narrowed write
relation against the current operation either: `metadataAttrsResolved`
re-derives its expected columns from `operation.requiredMetadataAttributes`,
but `rowAttrsResolved` compares the narrow projection with the narrow
`table.output`.
If the two decisions disagree, for example after a post-hoc rule swaps in a
rebuilt operation (Iceberg's `ResolveBranch` does this after
`RewriteUpdateTable` and keeps the narrow output and projections, see item 9),
this rule reports the narrow row schema as `schema()` with no
`columnUpdateSchema()`, and `ReplaceDataExec` calls `write()`. A copy-on-write
connector that follows the documented contract then rewrites groups with only
the declared columns, and nothing fails at analysis or execution.
Could you fail fast here, e.g. with an internal error when
`!hasNarrowRows(w) && w.table.output != w.originalTable.output`? Every write
Spark builds without narrowing keeps `table.output == originalTable.output`, so
existing plans are not affected.
--
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]
