tkaymak opened a new pull request, #39971: URL: https://github.com/apache/beam/pull/39971
Third slice of the Spark 4 Structured Streaming work split out of #39576, following the dispatch seam (#39906) and the Kryo registrations (#39939). Addresses #36841. This adds the DataSourceV2 micro-batch source that exposes any Beam UnboundedSource as a Spark 4 streaming table. All 12 files are new, nothing existing changes. Design notes: - Rows have a fixed two column schema, the element encoded with the Beam FullWindowedValueCoder as BINARY plus the event timestamp. No Catalyst encoder is generated for Beam types, payloads stay opaque until a downstream translator decodes them. - Offsets are opaque, strictly increasing epoch counters. latestOffset always advances so Spark keeps scheduling micro-batches, termination belongs to the lifecycle owner, not to the offsets. - Recovery is durable under the query's checkpoint location. The source id derives deterministically from the read transform's full name, the first run pins its split list because Beam sources do not guarantee deterministic splitting, and each split persists its CheckpointMark per epoch (atomic write via temp file and rename, retention two). On restart the epoch counter fast forwards past everything replayed from Spark's offset log and readers resume from the newest durable mark at or before the replayed epoch. - Executors cache live Beam readers between micro-batches, keyed by checkpoint location, source id and split, mirroring MicrobatchSource in the legacy runner. - Semantics are at least once, a mark is written when a batch finishes reading rather than transactionally with Spark's commit, so a crash between the two replays the last micro-batch. - The batch cutoff honors the maxRecordsPerBatch option from #39952, values below 1, including the default, mean no limit and the batch then ends on the maxBatchDurationMillis deadline. Tests cover element delivery, watermark tracking through typed maps, the epoch offset round trip, the unlimited default, the checkpoint file layout with retention, epoch fast forward and the deterministic source id. The end of stream sentinel used by PAssert arrives with the translators slice, its hooks are deliberately absent here. Full :runners:spark:3:test and :runners:spark:4:test suites green locally on JDK 17 plus spotless, checkstyle and a live ErrorProne compile. The unlimited default test was mutation checked, breaking either the cutoff guard or the option plumbing makes it fail. Remaining slices: the state and timer bridge on transformWithState, then the translators with the end to end tests. End to end evidence remains in draft #39576. R: @Abacn -- 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]
