[Flink] Cache batch side input materialization in Flink 2

[pkuzmickas]

Summary

Automatically cache materialized side-input views for classic Java Flink DataStream execution when the side input is bounded, uses the default trigger, and has zero allowed lateness.

The cache is scoped by Flink job ID, task attempt, PCollectionView, and window. It uses soft values, expires entries five minutes after access, and invalidates an entry after a side-input state update. A retried task rematerializes its values, while operator cleanup leaves shared entries available to peer subtasks and later operators.

No pipeline option is required. Portable side-input delivery remains unchanged. Streaming pipelines only cache side inputs that satisfy the bounded, single-pane eligibility checks.

This restores the reuse provided by the removed Flink DataSet runner's broadcast-variable materialization. Beam's Spark runner and Spark structured-streaming runner also cache batch side-input materialization.

Fixes #39866.

Why

Moving production workloads from the Flink 1 DataSet runner to the Flink 2 DataStream runner caused large performance regressions when they repeatedly accessed materialized side inputs.

The DataStream runner currently reads the stored iterable and reapplies the ViewFn for every side-input access. For a map view, repeated access rebuilds the map for every main-input element.

We observed this across multiple production workloads where map side inputs contained tens of thousands of rows and the main inputs contained billions of records. Caching changes repeated reconstruction into one materialization per job, task attempt, view, and window in each TaskManager JVM.

Performance

These anonymized production measurements used matching input partitions for each baseline and candidate.

Workload	Flink 1 baseline	Flink 2 candidate	Runtime change
A	40m13s	20m19s	-49.5%
B	42m43s	19m39s	-54.0%
C	54m35s	43m28s	-20.4%

Implementation

• Select only bounded side inputs with the default trigger and zero allowed lateness.
• Wrap the classic Java side-input reader automatically when at least one eligible view exists.
• Cache nullable materializations per TaskManager JVM by job, task attempt, view, and window.
• Invalidate an entry after its side-input state write.
• Preserve shared entries when individual operators finish so peer subtasks do not rematerialize them.
• Bound retention with five-minute access expiry and soft values.
• Rematerialize entries after a task retry by including the attempt number in the cache key.
• Apply the change through the Flink 2.0, 2.1, and 2.2 source overrides.

Scope

This PR changes the classic Java Flink 2 DataStream runner. Portable side-input delivery, GroupByKey translation, and pre-aggregation remain unchanged.

Streaming execution is not disabled as a mode. Only bounded side inputs with stable single-pane windowing are eligible for caching.

Validation

• FlinkCachedSideInputReaderTest and FlinkPipelineOptionsTest on Flink 2.0 and 2.2
• Production shadow runs with isolated outputs
• Final current-behavior shadow run completed successfully with valid output and no failures, exceptions, or restarts

Unit coverage includes eligibility checks, repeated reads, sharing between reader instances, job/view/window key separation, retry rematerialization, cached nulls, invalidation, uncached views, and exception propagation.

---

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

• Mention the appropriate issue in the description.
• Update CHANGES.md with noteworthy changes.
• Confirm the Apache ICLA before marking this draft ready.

See the Contributor Guide for more tips on making the review process smoother.

[github-actions]

Assigning reviewers:

R: @Abacn for label website.

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

[Abacn]

Thanks, a few questions

• What is the consideration of exclude portable runner? It would be beneficial to have classic/portable runner feature in sync, as the latter one is the only one available for Python and Go SDKs

• Similarly what is the consideration to disable streaming? Since this is an opt-in option, why hard code to make streaming ineffective

[Abacn]

If it's Flink 2.x only, revert the changes on common runners/flink/src/

[pkuzmickas]

Thank you for the review!

If it's Flink 2.x only, revert the changes on common runners/flink/src/

Removed 👍

What is the consideration of exclude portable runner? It would be beneficial to have classic/portable runner feature in sync, as the latter one is the only one available for Python and Go SDKs

We have only thoroughly tested this with Java production workflows and would like to keep it tightly scoped. Portable side inputs also go through the Fn State API and SDK harness instead of SideInputReader, so this cache doesn't apply directly and seems like a bigger piece of work due to the different SDK harnesses.

Similarly what is the consideration to disable streaming?

Due to the nature of side inputs in unbounded streaming pipelines, the same window can be updated later. Since the cache is shared across the task manager, I'm worried some subtasks could use stale data for a bit. In batch processing side inputs should only be written once before they're read, so this works well.

But I am open to removing it if you think opt-in is enough here. Wdyt?

[Abacn]

Please check the following

1. Beam Batch Transforms Permit Updating / Triggered Side Inputs

   • In Apache Beam’s unified model, running in batch mode (--streaming=false or bounded inputs) does not prohibit non-default windowing or triggering on side inputs. A bounded PCollection used as a side input may specify early firings, element-count triggers, or repeated triggers (e.g. Repeatedly.forever(AfterPane.elementCountAtLeast(N))).
   • When multiple panes are produced for a window, downstream transforms expect to receive updated view values as new panes arrive.
   • While this change attempts to handle updates by calling SideInputCache.invalidate(...) in DoFnOperator.addSideInputValue, this invalidation is racy across parallel subtasks in the same TaskManager JVM:
     • SideInputCache is a static process-wide cache shared across all threads/slots in a TaskManager.
     • When an updated side input element is broadcast, Subtask 0 might process processElement2 and invalidate the cache entry.
     • Concurrently, Subtask 1 (on another thread) is processing main input elements in processElement1 and hasn't yet processed processElement2. Subtask 1 calls CachedSideInputReader.get(), encounters a cache miss due to Subtask 0's invalidation, evaluates its delegate SideInputHandler, reads the old value from Subtask 1's local state, and re-inserts the stale value back into the shared JVM cache.
     • Subtask 0 (which already updated its state) will now read the stale value from the cache.

2. Violation of PTransform Encapsulation (Leaky Public API)

   • Beam pipelines are built by composing modular, reusable PTransforms (from Beam core, connectors, or third-party libraries).
   • An end user authoring a pipeline cannot and should not be expected to know whether every internal DoFn inside every nested composite transform uses a single-pane materialized view, an updating view, or mutates elements in the materialized collection.
   • Introducing --cacheSideInputMaterialization as a global FlinkPipelineOptions flag forces users to know the implementation details of all transforms in their pipeline. Enabling it to optimize one transform could silently break semantic correctness in another.
   • An opt-in flag does not resolve the issue: optimizations in Beam runners must preserve Beam model semantics.

3. Cleaner / Safer Alternatives

   • Inspect Triggering Automatically: The runner can inspect view.getWindowingStrategyInternal(). If the side input uses DefaultTrigger and bounded data, it is guaranteed to emit at most once per window, making caching per (view, window) safe without needing any user-facing flag.
   • Scope the Cache Safely: Instead of a process-wide static cache across slots, scope the cache per-subtask instance, per-bundle, or per-partition (similar to Spark Structured Streaming's CachedSideInputReader and FnApi bundle cache tokens).

[pkuzmickas]

Hey @Abacn thanks for the thorough review, very helpful.

Inspect Triggering Automatically

I went with this suggestion, ran a production test and it seems to be working as effectively as the previous version.

Please let me know if it looks good or something is still not quite right 🤞

[Abacn]

Thank you!