[Flink] Cache materialized side inputs in DataStream batch execution #39866

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.