[Flink] Cache batch side input materialization in Flink 2

[pkuzmickas]

Summary

Cache materialized side-input views for classic Flink DataStream batch execution when --cacheSideInputMaterialization=true.

The option is disabled by default and bypassed in streaming mode. The cache is scoped by Flink job ID, PCollectionView, and window; it uses soft values, expires entries five minutes after access, invalidates an entry after a side-input state update, and clears job entries during operator cleanup.

This restores the reuse provided by the removed Flink DataSet runner's broadcast-variable materialization. Beam's Spark runner and Spark structured-streaming runner use cached side-input readers for batch execution.

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, view, and window in each TaskManager JVM.

Performance

These anonymized production measurements used the same input partition 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

• Wrap the classic batch SideInputReader only when the new option is enabled.
• Cache nullable materializations by job, view, and window.
• Invalidate after the side-input state write so a concurrent reader cannot retain the previous value.
• Bound retention with five-minute access expiry, soft values, and job cleanup.
• Apply the option to the Flink 2.0, 2.1, and 2.2 source overrides and generated option documentation.

Scope

This PR intentionally excludes streaming execution, portable side-input delivery, GroupByKey translation, and pre-aggregation changes.

Validation

• Flink, Flink 2.0, Flink 2.1, and Flink 2.2 spotlessJavaCheck
• FlinkCachedSideInputReaderTest on Flink 2.0, 2.1, and 2.2
• FlinkPipelineOptionsTest on Flink 2.0, 2.1, and 2.2
• Production shadow runs with isolated outputs and exact-input comparisons

Unit coverage includes repeated reads, job/view/window key separation, cached nulls, invalidation, exception propagation, the disabled default, and the streaming guard.

---

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