Avoid side input when gathering unwindowed WriteFiles results

[ntopousis]

Addresses #39370.

What changed

For bounded, unwindowed WriteFiles, replace the global View.asIterable() used to gather temporary FileResults 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;
• use TimestampCombiner.EARLIEST internally to retain the previous minimum element timestamp, then restore the global-default windowing strategy; and
• sort unwindowed runner-determined results by temporary filename immediately before positional shard assignment.

The windowed-write path is unchanged. FinalizeFn retains its defensive copy and invokes the existing single-call rename and cleanup protocol. FileBasedSink.WriteOperation.finalizeDestination now makes runner-determined shard assignment deterministic.

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 were gathered per large write. One materialization was about 18 MiB, while observed processing rates across the two largest finalizers were roughly 200–225 records per minute as workers repeatedly read and retried GCS data.

Related prior art: #18629 reports the same high-cardinality ISM side-input pathology in BigQuery batch-load finalization. It concerns a different transform, but the finely sharded temporary-file metadata and IsmReader/coder-size stack are closely related.

Retry and finalization semantics

GroupByKey stabilizes the set of results but does not guarantee iteration order. Runner-determined shard numbers are assigned by position, so a retry after a partial rename must not remap a temporary file to a different final shard. FileBasedSink.WriteOperation.finalizeDestination therefore sorts unwindowed runner-determined results by their unique temporary filename at the positional shard-assignment boundary.

This deliberately changes the previously arbitrary content-to-shard ordering from materialization order to lexicographic temporary-filename order. Runner-determined shard count and final filename format are unchanged. Fixed-shard assignment and the windowed-write path are unchanged.

The existing rename options and whole-directory cleanup protocol remain in place. This preserves empty and missing fixed shards, dynamic destinations, filename-policy side inputs, retry-safe temp-to-final mappings, output-filename visibility, and unwindowed temporary-directory cleanup.

Timestamp and windowing compatibility

The old side-input path exposed output-filename elements at TIMESTAMP_MIN_VALUE with the global-default (END_OF_WINDOW) windowing strategy. The new gather uses TimestampCombiner.EARLIEST internally so its grouped list retains the minimum element timestamp, then restores WindowingStrategy.globalDefault() before finalization.

Without that reset, EARLIEST would leak through getPerDestinationOutputFilenames(), and a downstream aggregation applied directly to that collection could emit at the minimum timestamp rather than at the end of the global window. The patch preserves both the filename element timestamp and the externally visible windowing strategy.

Compatibility and scope

The changed gather branch is selected only for non-windowed writes. WriteFiles rejects unbounded inputs unless windowed writes are enabled, so the normal unbounded/streaming write path retains its existing two-reshuffle implementation. Dataflow replacement updates apply to streaming jobs; batch pipeline updates are not supported. The downstream List/ListCoder contract established in #26289 also remains unchanged.

A bounded WriteFiles branch embedded in a larger streaming pipeline is a theoretical compatibility case; maintainer guidance on whether an updateCompatibilityVersion gate is warranted is welcome.

Combine.globally is not a drop-in simplification. With defaults, its empty-input implementation uses View.asIterable(), reintroducing the side-input pattern this change removes. withoutDefaults() avoids that view but suppresses the element needed to finalize an empty write unless it is also paired with an explicit marker.

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. A Dataflow-side mitigation for released SDK versions is also still worth investigating, so this PR addresses rather than auto-closes #39370.

Tests

The following local checks passed with JDK 21:

• A pipeline-graph regression test proves that unwindowed GatherTempFileResults does not contain View.CreatePCollectionView.
• A unit test invokes runner-determined finalization over two permutations of the same FileResult set and verifies identical temp-file-to-final-shard mappings.
• A graph-level test verifies that the public output-filename collection retains the global-default windowing strategy.
• A DirectRunner regression test verifies that output-filename elements retain the previous minimum timestamp.
• The complete FileBasedSinkTest class passes.
• All 24 DirectRunner WriteFilesTest cases pass, including both empty-input cases.
• ./gradlew :sdks:java:core:spotlessCheck
• ./gradlew :sdks:java:core:test --tests org.apache.beam.sdk.io.FileBasedSinkTest --tests org.apache.beam.sdk.io.WriteFilesTest.testUnwindowedGatherDoesNotMaterializeResultsAsSideInput --tests org.apache.beam.sdk.io.WriteFilesTest.testUnwindowedOutputFilenamesKeepGlobalDefaultWindowingStrategy
• ./gradlew :runners:direct-java:needsRunnerTests --tests org.apache.beam.sdk.io.WriteFilesTest
• ./gradlew formatChanges

---

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

• Mention the appropriate issue in your description (for example: addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment fixes #<ISSUE NUMBER> instead.
• Update CHANGES.md with noteworthy changes.
• If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

[gemini-code-assist]

Caution

The consumer version of Gemini Code Assist on GitHub has been sunset. All code review activity has officially ceased.

[gemini-code-assist]

Caution

The consumer version of Gemini Code Assist on GitHub has been sunset. All code review activity has officially ceased.

[gemini-code-assist]

Caution

The consumer version of Gemini Code Assist on GitHub has been sunset. All code review activity has officially ceased.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 56.99%. Comparing base (7eef130) to head (e8484c6).
⚠️ Report is 2 commits behind head on master.

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #39372      +/-   ##
============================================
- Coverage     58.03%   56.99%   -1.05%     
+ Complexity    13051     3632    -9419     
============================================
  Files          2515     1186    -1329     
  Lines        263756   190660   -73096     
  Branches      10764     3774    -6990     
============================================
- Hits         153083   108658   -44425     
+ Misses       104919    78518   -26401     
+ Partials       5754     3484    -2270     

Flag	Coverage Δ
java	72.39% <ø> (+8.16%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[github-actions]

Assigning reviewers:

R: @Abacn for label java.

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

[github-actions]

Assigning reviewers:

R: @Abacn for label java.

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

[github-actions]

Assigning reviewers:

R: @Abacn for label java.

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

[github-actions]

Assigning reviewers:

R: @Abacn for label java.

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

[github-actions]

Assigning reviewers:

R: @Abacn for label java.

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

[github-actions]

Assigning reviewers:

R: @Abacn for label java.

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

[github-actions]

Assigning reviewers:

R: @Abacn for label java.

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

[github-actions]

Assigning reviewers:

R: @Abacn for label java.

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

[github-actions]

Assigning reviewers:

R: @Abacn for label java.

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

[github-actions]

Assigning reviewers:

R: @Abacn for label java.

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

[github-actions]

Assigning reviewers:

R: @Abacn for label java.

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

[github-actions]

Assigning reviewers:

R: @Abacn for label java.

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

[github-actions]

Assigning reviewers:

R: @Abacn for label java.

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

[github-actions]

Assigning reviewers:

R: @Abacn for label java.

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

[github-actions]

Assigning reviewers:

R: @Abacn for label java.

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

[github-actions]

Assigning reviewers:

R: @Abacn for label java.

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

[github-actions]

Assigning reviewers:

R: @Abacn for label java.

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).