ericm-db opened a new pull request, #56707: URL: https://github.com/apache/spark/pull/56707
### What changes were proposed in this pull request? Backport of [SPARK-56972] ([apache/spark#56020](https://github.com/apache/spark/pull/56020)) to `branch-4.2`. Wire the sink name through `MicroBatchExecution` so that, when sink evolution is enabled, each committed batch writes a `CommitMetadataV3` whose `sinkMetadataMap` records the current sink as the active entry alongside any sinks that were active in earlier batches: - Add a per-execution `sinkMetadataMap` that is hydrated from the latest `CommitMetadataV3` in `populateStartOffsets`. - When `spark.sql.streaming.queryEvolution.enableSinkEvolution` is true, the commit-log write in `runBatch` produces `CommitMetadataV3` with every prior entry marked `isActive = false` and the current `(sinkName, sink.getClass.getName)` entered as `isActive = true`. - When sink evolution is disabled, the existing V1/V2 commit-log path is preserved unchanged. This is the minimal write-then-read parity for the sink evolution feature added in SPARK-56719. Provider-mismatch and sink-reuse validation are intentionally deferred. **Stacked PR.** `branch-4.2` does not yet have the predecessors SPARK-56970 ([apache/spark#56018](https://github.com/apache/spark/pull/56018)) and SPARK-56971 ([apache/spark#56019](https://github.com/apache/spark/pull/56019)), which are still under review for 4.2 in [apache/spark#56548](https://github.com/apache/spark/pull/56548). This PR is built on top of #56548 and currently shows those two predecessor commits in its diff; that will resolve once #56548 merges. Only the final commit (`[SPARK-56972][SS] Persist sink name in V3 commit log via MicroBatchExecution`) is the subject of this PR. The cherry-pick of `cfa759af5b6` produced the same diff as on master (+156/-3); the only conflict was an import-line collision in `MicroBatchExecution.scala`, resolved by keeping the branch's existing import and adding `CommitLog`, `CommitMetadataV3`, and `SinkMetadataInfo`. ### Why are the changes needed? SPARK-56719 introduced the `DataStreamWriter.name()` API and the in-memory `sinkName` plumbing inside `MicroBatchExecution`, but the sink name was not yet persisted to the checkpoint. Without persistence, restarts cannot observe historical sink identity and the feature is not durable. ### Does this PR introduce _any_ user-facing change? Behavior change only when `enableSinkEvolution` is true (off by default): the commit log directory now contains V3 commit log files instead of V1/V2 files. Wire format compatibility is preserved when the flag is left off. ### How was this patch tested? - Cherry-picked `cfa759af5b6` on top of the SPARK-56971 4.2 backport branch (#56548); resolved the single import-line conflict in `MicroBatchExecution.scala`. - `StreamingSinkEvolutionSuite` passes on this branch (12 tests, including the four new V3 commit-log cases: named-sink active entry, historical-sink retention across rename, V1/V2 preserved when disabled, and mid-checkpoint upgrade to V3). - `sql/core` main and test sources compile cleanly (`build/sbt sql/Test/compile`). ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (claude-opus-4-8) This pull request and its description were written by Isaac. -- 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]
