[SPARK-58239][SQL] DML Command Outputs for DSv2 - #57402
Conversation
juliuszsompolski
left a comment
juliuszsompolski
left a comment
There was a problem hiding this comment.
LGTM, looks good and wired up nice. Just some nits / lazy questions.
| metricRow("num_affected_rows", affectedRows), | ||
| metricRow("num_updated_rows", summary.numTargetRowsUpdated()), | ||
| metricRow("num_deleted_rows", summary.numTargetRowsDeleted()), | ||
| metricRow("num_inserted_rows", summary.numTargetRowsInserted())) |
There was a problem hiding this comment.
nit: I noticed that you have num_copied_rows in update and delete, but not here. Is that mirroring what e.g. Delta is doing?
There was a problem hiding this comment.
Removed num_copied_rows. Delta is not doing that, now it's consistent with what Delta is doing.
| * Sink progress information collected after commit. | ||
| */ | ||
| private[sql] case class StreamWriterCommitProgress(numOutputRows: Long) | ||
|
|
There was a problem hiding this comment.
nit: unrelated stray but whatever
There was a problem hiding this comment.
I don't like these double empty lines, so decided to keep this change in 🙂
| |""".stripMargin | ||
| } | ||
|
|
||
|
|
There was a problem hiding this comment.
nit: unrelated stray but whatever
| * @since 4.3.0 | ||
| */ | ||
| default StructType outputSchema() { | ||
| return new StructType(); |
There was a problem hiding this comment.
just checking: was calling a collect() on a command df returning an empty schema next to the 0 rows before?
There was a problem hiding this comment.
I called .show() on a MERGE query and it showed:
++
||
++
++
So yes, the schema was empty.
szehon-ho
left a comment
szehon-ho
left a comment
There was a problem hiding this comment.
A few follow-ups on the command-output contract (leaving the throwaway RowLevelOperation build cost alone for now).
| val project = Project(projectList, joinPlan) | ||
|
|
||
| InsertOnlyMerge(r, project) | ||
| InsertOnlyMerge(r, project, output = operationTable.operationOutput) |
There was a problem hiding this 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?
| * 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. |
There was a problem hiding this 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(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))) |
There was a problem hiding this 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.
| transaction.foreach(TransactionUtils.commit) | ||
| refreshCache() | ||
| writtenRows | ||
| outputRows |
There was a problem hiding this 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?
3 participants
What changes were proposed in this pull request?
Creates 2 new APIs for DSv2 connectors to be able to provide command outputs:
RowLevelOperation.outputSchema()- The schema of the intended output.Write.output()- The output rows.This PR excludes support for metadata-only DELETEs and Truncate. These will come in a follow-up PR.
Why are the changes needed?
Currently, DML commands do not have any output. Some connectors need to provide users with results. For example, Delta outputs 1 row that contains various metrics about the result of the command.
Does this PR introduce any user-facing change?
Yes, DSv2 API change.
How was this patch tested?
Unit tests via the in-memory table.
Was this patch authored or co-authored using generative AI tooling?
Generated-by: GPT 5.6 Sol