Address wrap-up items for Iceberg side input table cache by jrmccluskey · Pull Request #40222 · apache/beam

jrmccluskey · 2026-09-22T13:33:32Z

Completes a handful of cleanup tasks for the side input cache:

Binds historical sort orders unchecked via SortOrderParser.fromJson(schema, node, getOrderId()) in SerializableTableSpec.
Exposes Hadoop Configuration via IcebergCatalogConfig.getHadoopConfiguration() and propagates to SerializableTableSpec.getFileIO(conf) through SideInputTable in write transforms.
Adds worker-local transient Guava cache to ExtractTableIdsDoFn to eliminate hot-key shuffle bottlenecks

Fixes #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.

- Bind historical sort orders unchecked via SortOrderParser.fromJson(schema, node, getOrderId()) in SerializableTableSpec. - Expose Hadoop Configuration via IcebergCatalogConfig.getHadoopConfiguration() and propagate to SerializableTableSpec.getFileIO(conf) through SideInputTable in write transforms. - Add worker-local transient Guava cache to ExtractTableIdsDoFn to eliminate hot-key shuffle bottlenecks on streaming writes. - Add comprehensive unit and integration tests.

…tion in tests - Ensure TableMetadataDriver passes its Clock to ExtractTableIdsDoFn so virtual-time advancement in tests (e.g. testUnusedTablesEvictedFromStreamingCache) correctly expires worker-local table ID caches instead of using wall-clock time. - Fixes testUnusedTablesEvictedFromStreamingCache flakiness on CI environments with limited concurrency.

… cache TTL - Advance ControllableTestClock on trigger rows in testMetadataRefreshedAcrossIntervals, testMetadataRefreshedAcrossIntervalsAsSideInput, and testMetadataRefreshedAcrossIntervalsAsSideInputWithMultipleTables so worker-local table ID caches expire as intended across virtual intervals. - In testStreamingSpecEvolutionWithoutPipelineRestart, wait for the worker-local cache TTL (500ms) to elapse after spec evolution before emitting the second element.

- Add library.java.jackson_databind to dependencies in sdks/java/io/iceberg/build.gradle to resolve usedUndeclaredArtifacts dependency analysis warning caused by JsonNode usage in SerializableTableSpec.

ahmedabu98 · 2026-09-22T19:58:33Z


+  @VisibleForTesting
+  public abstract @Nullable Clock getClock();
+


nit: make this and setClock package private

ahmedabu98 · 2026-09-22T19:58:40Z

              ValueInSingleWindow.of(element, timestamp, window, paneInfo));
      if (tableIdentifier != null && !tableIdentifier.trim().isEmpty()) {
-        out.output(tableIdentifier.trim());
+        String canonicalId = tableIdentifier.trim();


readability nit: I don't think we need the outer if (tableIdentifier != null && !tableIdentifier.trim().isEmpty()) check.

Also let's work directly with tableIdentifier instead of trimming it into a "canonical" version

ahmedabu98 · 2026-09-22T19:58:45Z

+        String canonicalId = tableIdentifier.trim();
+        if (localTableIdCache == null && refreshInterval != null) {
+          initCache();
+        }


readability nit: Do we need this if we're already calling it in setup()?

removed, was a codegen add-in because a unit test directly called ProcessElement() without calling setup() first and the obvious "fix" was to check that the cache was initialized in ProcessElement()

ahmedabu98 · 2026-09-22T19:58:48Z

          table.updateSpec().addField("city").commit();
+          // Ensure worker-local table ID cache TTL (interval / 2 = 500ms) has elapsed
+          try {
+            Thread.sleep(600);


give this a little extra time to avoid potential flakes?

ahmedabu98 · 2026-09-22T19:58:50Z

+    private final @Nullable Clock clock;
+    private transient @Nullable Ticker ticker;
+    private transient @Nullable Cache<String, Boolean> localTableIdCache;


Lots of complexity here. I feel like we can simplify with just a simple LinkedHashMap<String, Long> LRU that tracks tableIds and the last emitted millis. So only emit when (now - lastEmitted) > refreshInterval and update lastEmitted=now.
Both ways probably work though so will leave it up to you

Let's go with the simpler implementation

github-actions · 2026-09-22T20:14:53Z

Assigning reviewers:

R: @ahmedabu98 for label java.
R: @chamikaramj for label iceberg.

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).

…p LRU - Replace Guava Cache, CacheBuilder, and Ticker in ExtractTableIdsDoFn with a standard LinkedHashMap-based LRU cache capped at 10,000 entries. - Track lastEmitted timestamp per table identifier and emit when (now - lastEmitted) >= refreshInterval / 2. - Remove getClock() and setClock() from TableMetadataDriver and its Builder to keep the AutoValue API clean and free of testing methods. - Provide testing hooks via static globalTestClock and instance setClock on ExtractTableIdsDoFn.

ahmedabu98

LGTM, thank you!

jrmccluskey · 2026-09-23T17:26:25Z

Test failure was a PubSubToKafkaE2E test flake, merging

…#40248) * Expose Public API for Iceberg Side Input Cache (#40140) * Expose Public API for Iceberg Side Input Cache * Update sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergIO.java Co-authored-by: Ahmed Abualsaud <65791736+ahmedabu98@users.noreply.github.com> * Streamline API, remove redundant checks * Check polled table metric * Evolve Iceberg schema mid-execution in streaming side input cache test * Throw if sub-options are enabled without side input cache enabled * test with partition spec evolution --------- Co-authored-by: Ahmed Abualsaud <65791736+ahmedabu98@users.noreply.github.com> * Address wrap-up items for Iceberg side input table cache (#40222) * Address wrap-up items for Iceberg side input table cache - Bind historical sort orders unchecked via SortOrderParser.fromJson(schema, node, getOrderId()) in SerializableTableSpec. - Expose Hadoop Configuration via IcebergCatalogConfig.getHadoopConfiguration() and propagate to SerializableTableSpec.getFileIO(conf) through SideInputTable in write transforms. - Add worker-local transient Guava cache to ExtractTableIdsDoFn to eliminate hot-key shuffle bottlenecks on streaming writes. - Add comprehensive unit and integration tests. * Propagate Clock to ExtractTableIdsDoFn for deterministic cache expiration in tests - Ensure TableMetadataDriver passes its Clock to ExtractTableIdsDoFn so virtual-time advancement in tests (e.g. testUnusedTablesEvictedFromStreamingCache) correctly expires worker-local table ID caches instead of using wall-clock time. - Fixes testUnusedTablesEvictedFromStreamingCache flakiness on CI environments with limited concurrency. * Make streaming refresh and spec evolution tests robust against worker cache TTL - Advance ControllableTestClock on trigger rows in testMetadataRefreshedAcrossIntervals, testMetadataRefreshedAcrossIntervalsAsSideInput, and testMetadataRefreshedAcrossIntervalsAsSideInputWithMultipleTables so worker-local table ID caches expire as intended across virtual intervals. - In testStreamingSpecEvolutionWithoutPipelineRestart, wait for the worker-local cache TTL (500ms) to elapse after spec evolution before emitting the second element. * Declare jackson-databind dependency in Iceberg IO module - Add library.java.jackson_databind to dependencies in sdks/java/io/iceberg/build.gradle to resolve usedUndeclaredArtifacts dependency analysis warning caused by JsonNode usage in SerializableTableSpec. * Avoid potential flakes in EvolveSpec DoFn, address nits * Simplify worker-local cache in TableMetadataDriver using LinkedHashMap LRU - Replace Guava Cache, CacheBuilder, and Ticker in ExtractTableIdsDoFn with a standard LinkedHashMap-based LRU cache capped at 10,000 entries. - Track lastEmitted timestamp per table identifier and emit when (now - lastEmitted) >= refreshInterval / 2. - Remove getClock() and setClock() from TableMetadataDriver and its Builder to keep the AutoValue API clean and free of testing methods. - Provide testing hooks via static globalTestClock and instance setClock on ExtractTableIdsDoFn. * simplify dofn + fix unit test --------- Co-authored-by: Ahmed Abualsaud <65791736+ahmedabu98@users.noreply.github.com>

github-actions Bot added java io iceberg Iceberg IO connector labels Sep 22, 2026

jrmccluskey added 3 commits September 22, 2026 14:39

Declare jackson-databind dependency in Iceberg IO module

d9c424e

- Add library.java.jackson_databind to dependencies in sdks/java/io/iceberg/build.gradle to resolve usedUndeclaredArtifacts dependency analysis warning caused by JsonNode usage in SerializableTableSpec.

jrmccluskey requested a review from ahmedabu98 September 22, 2026 17:38

ahmedabu98 reviewed Sep 22, 2026

View reviewed changes

github-actions Bot added the Next Action: Reviewers label Sep 22, 2026

jrmccluskey added 3 commits September 22, 2026 20:21

Avoid potential flakes in EvolveSpec DoFn, address nits

f4223eb

simplify dofn + fix unit test

0f81311

ahmedabu98 approved these changes Sep 23, 2026

View reviewed changes

jrmccluskey merged commit c462e49 into apache:master Sep 23, 2026
22 of 24 checks passed

jrmccluskey mentioned this pull request Sep 23, 2026

[Cherry-pick][#39723] Iceberg Side Input Cache Public API and Wrap-Up #40248

Merged

3 tasks


		@VisibleForTesting
		public abstract @Nullable Clock getClock();

Conversation

jrmccluskey commented Sep 22, 2026

GitHub Actions Tests Status (on master branch)

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!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

github-actions Bot commented Sep 22, 2026

Uh oh!

ahmedabu98 left a comment

Choose a reason for hiding this comment

Uh oh!

jrmccluskey commented Sep 23, 2026

Uh oh!

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants