Implement Iceberg Table Metadata Driver transform - #39883
Conversation
|
Assigning reviewers: R: @chamikaramj 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). |
| import org.joda.time.Instant; | ||
|
|
||
| /** | ||
| * A driver transform that extracts table identifiers from incoming {@link Row}s, deduplicates them |
There was a problem hiding this comment.
Cloud you clarify in the docs how this helper will be used, specially given that this seem to be providing a sample of tables ?
There was a problem hiding this comment.
As-written this effectively acts as a maximum cache size, "sample" is maybe not the right word unless you get more tables than the configured maximum. My lack of experience with Iceberg kind of becomes a problem here, I'm not sure what a "typical" workload looks like in terms of the number of tables being operated on. We could do away with this parameter or make it uncapped by default if that makes more sense
| @Test | ||
| public void testWindowedDeduplication() { | ||
| Catalog catalog = getCatalog(); | ||
| TableIdentifier table1 = TableIdentifier.of("default", "t1"); |
There was a problem hiding this comment.
Did you expect to test same table ID being returned by multiple windows here ?
There was a problem hiding this comment.
This is what I get for leaning on Gemini to produce some unit tests. The name is misleading, it's really testing that we deduplicate the target table IDs when we build the spec (since there are 50 elements referring to t1 and 50 elements referring to t2, we deduplicate that to single references to the two tables.) I'll rename this, the window doesn't really matter
| public void processElement( | ||
| @Element String tableIdString, OutputReceiver<KV<String, SerializableTableSpec>> out) { | ||
| TableIdentifier tableId = IcebergUtils.parseTableIdentifier(tableIdString); | ||
| Table table = catalogConfig.catalog().loadTable(tableId); |
There was a problem hiding this comment.
Note that a bad table string here will cause the whole bundle to fail. Is this intended or should we skip bad records and try other elements ?
There was a problem hiding this comment.
We could set up some sort of DLQ here. As best I can tell the current iceberg implementation uses the same approach of outright failing on bad table identifiers though
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #39883 +/- ##
============================================
+ Coverage 57.22% 58.34% +1.12%
- Complexity 3661 13472 +9811
============================================
Files 1195 2567 +1372
Lines 193411 268606 +75195
Branches 3831 11030 +7199
============================================
+ Hits 110670 156717 +46047
- Misses 79171 105952 +26781
- Partials 3570 5937 +2367
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
2 participants
Implements the Iceberg table metadata query transform and materializes the table metadata to be used downstream as a side input.
Part of #39723
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.