pkuzmickas opened a new issue, #39866: URL: https://github.com/apache/beam/issues/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 avoided this cost through broadcast-variable materialization, and Beam's Spark runners use cached side-input readers for batch execution. This behavior can dominate otherwise simple batch stages. In one anonymized production workload, 10,124,403,205 main-input records repeatedly accessed an 82,420-row map side input. Stock Flink 2 processed 2.10% of the stage in 116m19s before cancellation. A cache-only runner patch completed the stage in 36m44s; archived cumulative rates indicate an approximate 150x throughput improvement. ## 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. I would particularly value feedback on cache ownership and whether cleanup should remain operator-driven or move to a job-scoped lifecycle hook. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
