Implement Iceberg Table Metadata Driver transform by jrmccluskey · Pull Request #39883 · apache/beam

jrmccluskey · 2026-08-25T15:04:29Z

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:

Mention the appropriate issue in your description (for example: 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, comment fixes #<ISSUE NUMBER> instead.
Update CHANGES.md with noteworthy changes.
If this contribution is large, please file an Apache Individual Contributor License Agreement.

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.

github-actions · 2026-08-25T16:47:06Z

Assigning reviewers:

R: @chamikaramj for label java.

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

stop reviewer notifications - opt out of the automated review tooling
remind me after tests pass - tag the comment author after tests pass
waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

chamikaramj

Thanks!

chamikaramj · 2026-08-26T22:25:52Z

+import org.joda.time.Instant;
+
+/**
+ * A driver transform that extracts table identifiers from incoming {@link Row}s, deduplicates them


Cloud you clarify in the docs how this helper will be used, specially given that this seem to be providing a sample of tables ?

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

I was just using the wording from the existing comment :)

I think if we are trying to cache maximum possible for efficiency, but rest will still work (less efficiently), it makes sense. In general, lets's expand more about downstream use-case here to clarify what it's intended for.

chamikaramj · 2026-08-26T22:31:18Z

+  @Test
+  public void testWindowedDeduplication() {
+    Catalog catalog = getCatalog();
+    TableIdentifier table1 = TableIdentifier.of("default", "t1");


Did you expect to test same table ID being returned by multiple windows here ?

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

chamikaramj · 2026-08-26T23:14:50Z

+    public void processElement(
+        @Element String tableIdString, OutputReceiver<KV<String, SerializableTableSpec>> out) {
+      TableIdentifier tableId = IcebergUtils.parseTableIdentifier(tableIdString);
+      Table table = catalogConfig.catalog().loadTable(tableId);


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 ?

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

Going back to the original design doc, Ahmed had suggested catching NoSuchTableExceptions gracefully since the lack of metadata tells the sink to try creating the table. I've added that logic, which should avoid failing whole bundles when we get an expected exception.

codecov · 2026-08-27T15:46:51Z

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 58.34%. Comparing base (66ecc31) to head (11fffa9).
⚠️ Report is 27 commits behind head on master.

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

Flag	Coverage Δ
java	`64.45% <ø> (-7.75%)`	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

bvolpato

LGTM. One nonblocking question inline about streaming input scope.

bvolpato · 2026-08-29T18:50:52Z

+            .apply("ExtractTableIds", ParDo.of(new ExtractTableIdsDoFn(getDynamicDestinations())))
+            .setCoder(StringUtf8Coder.of());
+
+    PCollection<String> distinctTableIds = tableIds.apply("DistinctTableIds", Distinct.create());


nonblocking: is this meant to support default streaming inputs? Distinct rejects an unbounded GlobalWindow with default trigger. Iceberg write accepts that input shape before its own triggering. fine if this is batch/pre-windowed only, but worth making boundary explicit.

Probably worth deduplicating a different way then, don't want to add an extra restriction here

jrmccluskey · 2026-09-01T13:48:40Z

CC: @ahmedabu98

Currently exploring if moving to another method of deduplication makes sense, the original design doc specifically called out using Distinct but also wants unbounded global window streaming support.

jrmccluskey · 2026-09-01T21:00:31Z

@chamikaramj @ahmedabu98 this should be ready for another review pass!

chamikaramj

Thanks!

ahmedabu98 · 2026-09-01T22:04:47Z

+    }
+
+    PCollection<String> distinctTableIds =
+        triggeredTableIds.apply("DistinctTableIds", Distinct.create());


Deduplicate transform is generally a better alternative for streaming mode. it also has a .withDuration(refreshInterval) method that aligns well here

Looking at Deduplicate it's a stateful DoFn, which the initial problem statement and design scope seemed insistent on avoiding (mainly for the potential bottleneck from shuffles.) Is there a particular reason it's preferred for streaming workloads?

Well I feel silly, adding a schema evolution test case actually found that Distinct didn't re-emit the pane in an unbounded streaming context. That explains it.

Yeah Distinct only emits once per window, so more batch-like.
AFAICT Deduplicate emits once and stores the value in state. It won't emit the value again while it's in state. After refreshInterval time passes, it clears the value and allows a new encounter to be emitted once again.

Deduplicate is a stateful DoFn, which the initial problem statement and design scope seemed insistent on avoiding

That's a good point cuz Deduplicate will still have one thread per tableId.
I think we can relieve some of the pressure here by having the upstream (pre-bottleneck) ExtractTableIdsDoFn transform do some local filtering. It can keep track of when tableIds where last emitted (e.g. Map<String, Instant>) and only output a tableId if it's older than refreshInterval / 2. Doesn't have to be exact or durable, if duplicates still get through then stateful Deduplicate will take care of them.

Also I think this bottleneck is a lot more gentle than the one the doc mentions. We're passing table strings through a side input instead of full data rows through the main path.

Yeah the more I thought about it the less I was worried about the bottleneck in a streaming context since we're 1) passing pretty lightweight objects and 2) anticipate relatively small bundle sizes and worker counts for streaming workloads. If it was the batch patch I would be more concerned since that's the use-case where the potential to DDOS during queries is high

We can leave it as a future improvement if we notice it's affecting throughput. I don't think it'll break update compatibility

Yeah that shouldn't impact coders or graph shape, we can play with that later.

ahmedabu98 · 2026-09-01T22:45:44Z

+    return cachedTableIds
+        .apply("PollTableMetadata", ParDo.of(new CatalogPollingDoFn(getCatalogConfig())))
+        .setCoder(KvCoder.of(StringUtf8Coder.of(), SerializableTableSpec.getCoder()));


How are we going to control parallelism? Ideally we'd pass tableIds to only a few tables* so they can do sequential loadTable calls.

*this can be configurable

I'm not sure I'm following the ask here. Would we want to emit batches of table metadata downstream in that context?

Sorry lemme clarify. I think as it stands, each tableId (after deduplication) can go to a separate worker. If we're writing to many tables, we can end up with many concurrent loadTable calls (one for each table).

Was wondering if it makes sense to reshuffle these table string outputs to a fixed N workers so that we only have at most N concurrent loadTable calls. If there's many tables, the workers can call loadTable on each one sequentially

Reshuffle via random key with a fixed number of buckets would do that, I'll see how that looks.

…panes

ahmedabu98 · 2026-09-02T20:57:16Z

+  @Test
+  public void testMetadataRefreshedAcrossIntervals() {


This test looks good, but can we please add another similar one consuming the table specifically as a side input? Should also assert the update happened

This was a good call, adding that test variation found a number of problems with streaming cases and the triggering setup.

Implement Iceberg Table Metadata Driver transform

83ddc7d

github-actions Bot added java io labels Aug 25, 2026

github-actions Bot added the Next Action: Reviewers label Aug 25, 2026

chamikaramj reviewed Aug 26, 2026

View reviewed changes

Rename unit test for clarity

11fffa9

bvolpato approved these changes Aug 29, 2026

View reviewed changes

Uncap cache size by default

57d0db1

jrmccluskey added 2 commits September 1, 2026 13:51

Handle NoSuchTableExceptions in CatalogPollingDoFn

c7b25dc

unbounded global window support

4bef0d8

chamikaramj approved these changes Sep 1, 2026

View reviewed changes

ahmedabu98 reviewed Sep 1, 2026

View reviewed changes

Comment thread sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/TableMetadataDriverTest.java Outdated

ahmedabu98 reviewed Sep 1, 2026

View reviewed changes

Comment thread sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/TableMetadataDriverTest.java

ahmedabu98 reviewed Sep 1, 2026

View reviewed changes

jrmccluskey added 2 commits September 2, 2026 14:40

Streamline test definitions

cfeb3d5

add schema evolution test case, route through Deduplicate to re-emit …

4555fa8

…panes

ahmedabu98 reviewed Sep 2, 2026

View reviewed changes

jrmccluskey added 2 commits September 3, 2026 14:05

Add reshuffle and polling buckets

d9b260a

Side input tests + multiple tables, fix breakages

f0515f1

Uh oh!

Conversation

jrmccluskey commented Aug 25, 2026

GitHub Actions Tests Status (on master branch)

Uh oh!

github-actions Bot commented Aug 25, 2026

Uh oh!

chamikaramj left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

codecov Bot commented Aug 27, 2026

Codecov Report

Uh oh!

bvolpato left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jrmccluskey commented Sep 1, 2026

Uh oh!

jrmccluskey commented Sep 1, 2026

Uh oh!

chamikaramj left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ahmedabu98 Sep 1, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ahmedabu98 Sep 2, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

ahmedabu98 Sep 1, 2026 •

edited

Loading

ahmedabu98 Sep 2, 2026 •

edited

Loading