Repository navigation
[Iceberg CDC sink] Write deltas transform - #40031
Conversation
|
Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment |
|
Assigning reviewers: R: @kennknowles for label java. Note: If you would like to opt out of this review, comment Available commands:
The PR bot will only process comments in the main thread (not review comments). |
| "WriteDeltas", | ||
| ParDo.of( | ||
| new WriteDeltasFn( | ||
| new TableSetup(catalogConfig, config, destinations, filePrefix), |
There was a problem hiding this comment.
Specifying a filePrefix as to the runId param can be error prone ?
There was a problem hiding this comment.
Maybe filePrefix isn't the best naming. It's just an ID that is generated once per pipeline launch. I'll rename it to runId if that helps
| dataFormat, | ||
| deleteFormat); | ||
|
|
||
| org.apache.iceberg.Schema tableSchema = table.schema(); |
There was a problem hiding this comment.
There are multiple table.schema() invocations here and within RecordDeltaTaskWriter.create. Probably do the invocation once and pass it for consistency ?
There was a problem hiding this comment.
We're using the same table instance for these calls so it should be consistent throughout? Not sure if I'm understanding the comment correctly
There was a problem hiding this comment.
Can the underlying table from TableCache refresh ? If not yeah, it's mostly a readability issue. Up to you.
There was a problem hiding this comment.
Yeah good catch, we can have different table.schema() results if another thread refreshes the cache in the middle (although it's a very very tiny window).
The issue though is that inside the writer, we're using Iceberg's writerFactory which is initialized using GenericFileWriterFactory.Builder(table) and itself calls table.schema() internally.
I fixed it by creating a SerializableTable copy of the table so that the metadata version is constant throughout the process call
|
Thanks @chamikaramj, can you PTAL ? |
| dataFormat, | ||
| deleteFormat); | ||
|
|
||
| org.apache.iceberg.Schema tableSchema = table.schema(); |
There was a problem hiding this comment.
Can the underlying table from TableCache refresh ? If not yeah, it's mostly a readability issue. Up to you.
Adds the CDC sink's write stage: WriteDeltas runs the delta writer over each sorted (destination, shard, window) group under the worker's pinned partition spec.
Outputs ShardDeltaFiles, carrying the resulting data and equality-delete files, with their sequence range, to the committer.
Part of #39979
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, commentfixes #<ISSUE NUMBER>instead.CHANGES.mdwith noteworthy changes.See the Contributor Guide for more tips on how to make review process smoother.
To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md
GitHub Actions Tests Status (on master branch)
See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.