Repository navigation
[Iceberg CDC sink] Assign CDC keys - #40030
Conversation
|
Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment |
|
Assigning reviewers: R: @Abacn 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). |
| throw new IllegalArgumentException( | ||
| "sequence_number_column '" | ||
| + seqColumn | ||
| + "' must be INT64 (was: " |
There was a problem hiding this comment.
Should we also support other types here ? For example, int32 and may be an string that could be casted to an int (and fail if it's an invalid string) ? It might be hard to change type of a dataset with an existing sequence number column.
There was a problem hiding this comment.
Sequence numbers are usually long values. If needed, users can just do a translation prior to the sink to conver their seqColumn to INT64. We can relax it in the future if we see it becoming a dominant pattern, but generally I'd rather keep it strict for now.
| * across shards breaks same-commit dedup. | ||
| */ | ||
| private int shardFor(TableSetup.Dest dest, Row data, byte[] pkBytes) { | ||
| @Nullable PartitionShardPlan partitionShardPlan = dest.partitionShardPlan(); |
There was a problem hiding this comment.
Seems like we are taking this form the destination table but in practice, the primary key for an end-to-end CDC pipeline might be a business decision that is based on the source table (for example, Delta Lake).
There was a problem hiding this comment.
We have withEqualityColumns in the sink config, which lets users specify the primary key columns. We only infer them from the destination if no override is given.
| @Nullable Map<String, String> changeTypeMap = config.getChangeTypeMap(); | ||
| String name = changeTypeMap != null ? changeTypeMap.getOrDefault(raw, raw) : raw; | ||
| try { | ||
| return ValueKind.valueOf(name); |
There was a problem hiding this comment.
So in this case the output records inherit the native ValueKind metadata, which can contradict the resolved kind ? Should we update the outgoing ValueKind for elements for consistency ?
There was a problem hiding this comment.
I don't think there's much value in updating it. The ValueKind metadata matters when the user is providing it to the sink. After this stage (when we're already inside the sink), we only look at the resolved kind and don't really look at the ValueKind metadata anymore.
The resolved kind gets persisted (see CdcRecord.of(data, kind, seq) above) so that it can survive the downstream GBK. And downstream we also use the same resolved kind for sorting and writing.
Adds the CDC sink's first stage: AssignCdcKeys reads each record's change kind and sequence number and keys it by (destination, shard) with a sort key of [primary key][sequence number][kind].
Invalid records diverted to the failed output.
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.