pkuzmickas opened a new pull request, #39867: URL: https://github.com/apache/beam/pull/39867
## 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 behavior provided by the removed Flink DataSet runner's broadcast-variable materialization. Beam's Spark runners use the same general cached-reader pattern. Fixes #39866. ## Why 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. The largest measured workload read an 82,420-row map side input while processing 10,124,403,205 main-input records. Caching changes repeated reconstruction into one materialization per job, view, and window in each TaskManager JVM. ## Performance These anonymized production measurements use the same input partition, Flink 2.2.1, and the default scheduler. | Workload and metric | Stock Flink 2 | Cache candidate | Change | | --- | ---: | ---: | ---: | | Complete batch workflow: runtime | 51m08s | 19m39s | -61.6% | | Complete batch workflow: estimated compute cost | 78.43 | 32.21 | -58.9% | | Side-input stage: 82,420 side rows, 10.12B main records | 2.10% processed after 116m19s; cancelled | completed in 36m44s | about 150x throughput | The complete-workflow candidate used this cache as its only Beam runner performance patch. It also required a Kryo classloading compatibility setting; neither comparison had a task restart. The stage-throughput ratio is approximate because the stock control was cancelled with tasks in flight, so it is not a total-runtime comparison. ## 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. ## Draft follow-ups - [ ] Resolve the preferred job-cache cleanup lifecycle with Flink maintainers. - [ ] Add a `CHANGES.md` entry before marking ready. - [ ] Run the upstream Flink precommit and relevant ValidatesRunner suites. - [ ] Confirm ICLA status before marking ready. ------------------------ Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily: - [x] Mention the appropriate issue in the description. - [ ] Update `CHANGES.md` with noteworthy changes before marking this draft ready. - [ ] Confirm the Apache ICLA before marking this draft ready. See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [making the review process smoother](https://github.com/apache/beam/blob/master/CONTRIBUTING.md#make-the-reviewers-job-easier). -- 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]
