ntopousis opened a new pull request, #39372: URL: https://github.com/apache/beam/pull/39372
Addresses #39370. ## What changed For bounded, unwindowed `WriteFiles`, replace the global `View.asIterable()` used to gather temporary `FileResult`s with a main-input shuffle: - gather results into bundle-sized lists; - add an empty-list marker so empty writes still invoke finalization; - flatten, key by `Void`, and `GroupByKey`; - flatten the grouped lists back into the existing `PCollection<List<FileResult<...>>>` contract; - avoid copying the completed list again in `FinalizeFn`. The windowed-write path is unchanged. ## Why At high temporary-file cardinality, the existing global iterable side input can become a finalization bottleneck. In one Dataflow production case, approximately 90,000–100,000 `FileResult` records (about 18 MiB for one materialization) were consumed through `IsmSideInputReader` at only a few hundred records per minute while the worker repeatedly read and retried GCS data. This patch removes that side-input access pattern while retaining the current finalization implementation. ## Compatibility and scope The output of `GatherResults` remains a `List` with `ListCoder`, and the existing `FinalizeFn` and `FileBasedSink.WriteOperation` behavior remain in place. This preserves empty and missing fixed shards, runner-determined shard numbering, dynamic destinations, filename policy side inputs, retry-safe renames, output-filename visibility, and unwindowed temporary-directory cleanup. This is deliberately not parallel finalization. It still materializes one O(N) list and runs one logical finalizer. Chunked finalization requires a separate design for shard assignment, exactly-once empty-shard creation, completion barriers, and cleanup. ## Tests All checks passed with JDK 21: - Added a pipeline-graph regression test proving that the unwindowed `GatherTempFileResults` subtree does not contain `View.CreatePCollectionView`. - Existing `WriteFilesTest` coverage exercises empty writes, fixed and runner-determined sharding, missing empty shards, dynamic destinations, output filenames, and temporary-directory cleanup. - `./gradlew :sdks:java:core:spotlessApply` - `./gradlew :sdks:java:core:test --tests org.apache.beam.sdk.io.WriteFilesTest.testUnwindowedGatherDoesNotMaterializeResultsAsSideInput` - `./gradlew :runners:direct-java:needsRunnerTest --tests org.apache.beam.sdk.io.WriteFilesTest` ------------------------ - [x] Mention the appropriate issue in the description. - [ ] Update `CHANGES.md` if maintainers consider this noteworthy for the next release. - [ ] File an Apache ICLA if required for this contribution. -- 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]
