Problem
The Flink DataStream runner materializes a side-input view each time SideInputReader.get(view, window) runs. For a map side input, this can rebuild the full map for every main-input element. The removed Flink DataSet runner reused broadcast-variable materialization, while Beam's Spark runner and Spark structured-streaming runner use cached side-input readers for batch execution.
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.
Proposed change
Add an opt-in Flink pipeline option that caches each materialized side-input value in the TaskManager JVM, keyed by Flink job ID, view, and window.
The proposed cache:
- applies only to bounded DataStream execution;
- remains disabled by default;
- invalidates an entry after a new side-input value is written;
- uses soft values and five-minute access expiry; and
- removes entries for a job during operator cleanup.
Scope
This issue covers classic Flink DataStream batch side inputs. Streaming execution, portable side-input delivery, and GroupByKey translation changes are out of scope.
Problem
The Flink DataStream runner materializes a side-input view each time
SideInputReader.get(view, window)runs. For a map side input, this can rebuild the full map for every main-input element. The removed Flink DataSet runner reused broadcast-variable materialization, while Beam's Spark runner and Spark structured-streaming runner use cached side-input readers for batch execution.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.
Proposed change
Add an opt-in Flink pipeline option that caches each materialized side-input value in the TaskManager JVM, keyed by Flink job ID, view, and window.
The proposed cache:
Scope
This issue covers classic Flink DataStream batch side inputs. Streaming execution, portable side-input delivery, and GroupByKey translation changes are out of scope.