szehon-ho commented on code in PR #57402:
URL: https://github.com/apache/spark/pull/57402#discussion_r3660251022
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/write/Write.java:
##########
@@ -48,6 +49,18 @@ default String description() {
return this.getClass().toString();
}
+ /**
+ * Returns the rows produced by this write after it commits successfully.
+ * <p>
+ * The returned rows must match the schema reported by
+ * {@link RowLevelOperation#outputSchema()}. By default, this method returns
no rows.
Review Comment:
This javadoc ties `Write.output()` to `RowLevelOperation#outputSchema()`,
but `Write` is shared by append / overwrite / streaming, and insert-only MERGE
uses the table's regular `Write` (not the row-level one).
Could we spell out when Spark calls `output()` (after successful commit),
which commands use it today, and that for insert-only MERGE the rows must match
the schema from `RowLevelOperation.outputSchema()` even though the `Write`
itself comes from `SupportsWrite`?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala:
##########
@@ -74,7 +76,7 @@ object RewriteMergeIntoTable extends RewriteRowLevelCommand
with PredicateHelper
}
val project = Project(projectList, joinPlan)
- InsertOnlyMerge(r, project)
+ InsertOnlyMerge(r, project, output = operationTable.operationOutput)
Review Comment:
For insert-only MERGE, the command **schema** comes from
`RowLevelOperation.outputSchema()` here (`operationTable.operationOutput`), but
the **rows** come from `Write.output()` on the `Write` built later via
`table.newWriteBuilder()` in `V2Writes` — not from
`RowLevelOperation.newWriteBuilder()`.
A connector that only implements `outputSchema()` + `Write.output()` on the
row-level Write will get a non-empty command schema but empty rows on this
path. The in-memory table hides that because `InMemoryWriterBuilder` also
overrides `Write.output()`.
Could we document that both Write paths must implement matching `output()` /
`outputSchema()`, or adjust the design so insert-only MERGE does not split
schema and rows across two Writes?
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/UpdateTableSuiteBase.scala:
##########
@@ -76,7 +76,9 @@ abstract class UpdateTableSuiteBase extends
RowLevelOperationSuiteBase {
sql(s"ALTER TABLE $tableNameAsString ALTER COLUMN txt SET DEFAULT
'new-text'")
- sql(s"UPDATE $tableNameAsString SET txt = DEFAULT WHERE pk IN (2, 8, 11)")
+ checkAnswer(
+ sql(s"UPDATE $tableNameAsString SET txt = DEFAULT WHERE pk IN (2, 8,
11)"),
+ Seq(Row("num_affected_rows", 1L)))
Review Comment:
Nit: command-output coverage for UPDATE/DELETE is a bit thin — these
assertions are bolted onto unrelated default-value tests (same for DELETE). A
dedicated test (and maybe asserting the default empty schema/rows when the
connector does not override) would lock the contract more clearly.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/WriteToDataSourceV2Exec.scala:
##########
@@ -491,15 +496,17 @@ trait V2ExistingTableWriteExec extends V2TableWriteExec
with TransactionalExec {
createCustomMetrics(write.supportedCustomMetrics())
override protected def run(): Seq[InternalRow] = {
- val writtenRows = try {
+ try {
writeWithV2(write.toBatch)
} finally {
postDriverMetrics(write.reportDriverMetrics())
}
transaction.foreach(TransactionUtils.commit)
refreshCache()
- writtenRows
+ outputRows
Review Comment:
Open question: should Spark validate that `Write.output()` arity/types match
the planned `output` / `RowLevelOperation.outputSchema()`, or is mismatch left
as a connector footgun for this Evolving API?
--
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]
