Flink source split strategy

[jto]

Improves how the Flink runner splits bounded sources in batch pipelines, so that work is evenly distributed over the source readers. The same logic is shared by the DataSet (SourceInputFormat) and DataStream (FlinkSource enumerators) translations, in a new BoundedSourceSplitter.

Motivation

The runner used to call BoundedSource.split(estimatedSize / parallelism) once and use whatever came back. Beam does not require sources to split in one pass, and many don't honour the desired size: for example a BigQuery export returns one split per exported file, whatever size is requested. The result was often too few splits for the readers, or splits of very uneven size. Combined with lazy split assignment, a few readers could end up with much more work than the others.

What changes

After the initial split() call, the splitter refines the splits in a few bounded passes:

1. Re-split oversized splits. Splits much larger than estimatedSize / (2 x parallelism) are split again, so each reader gets at least two splits, and sources returning splits far larger than requested (e.g. BigQuery exports) are cut down.
2. Balance sizes. If split sizes still vary a lot (coefficient of variation above 0.5), the larger splits are re-split towards the mean size, in up to three rounds.
3. Balance the split count. If round-robin assignment would give the busiest reader more than 10% above the average bytes, the largest splits are halved.
4. Order for round-robin. Splits are sorted by size in a "snake" order, so that assigning split i to reader i % parallelism pairs large splits with small ones.

Steps 2 to 4 need per-split size estimates and are skipped when they are not available. Re-splitting never goes below min(64 MiB, estimatedSize / (2 x parallelism)): small splits mostly add fixed per-split overhead, but small inputs, which may feed transforms that generate a lot more data, are still spread over all readers. fileInputSplitMaxSizeMB is still honoured as a cap.

The DataStream enumerators now pass the actual reader count (currentParallelism()) to the splitter, separately from numSplits (which may be maxParallelism), so ordering and balancing match how splits are actually assigned.

Split assignment: static by default in batch

• sourceStaticSplitThresholdMb no longer has a default. When unset, batch pipelines use static assignment and streaming pipelines keep lazy assignment. Explicit values behave as before: 0 = lazy, negative = static, positive = size-based.
• DataSet pipelines get a new StaticInputSplitAssigner (split i goes to task i % parallelism, like the DataStream enumerator), and fall back to Flink's DefaultInputSplitAssigner when lazy assignment is configured.
• The lazy enumerator hands out the largest splits first, in constant time per assignment.

This is a behaviour change for batch pipelines that relied on the previous lazy default. They can restore it with --sourceStaticSplitThresholdMb=0.

Observability

Each step logs its effect at INFO, in human-readable units: split count, size min/mean/max and coefficient of variation, and the busiest reader's load relative to the average. For example:

Re-split 0 of 374596 splits larger than 192.00 MiB towards 128.00 MiB: ...
Split count balancing for 3500 readers: skipped, 374596 splits already give the busiest reader 1.003 times the average bytes
Ordered 374596 splits for round-robin over 3500 readers: busiest reader load / average went from 1.020 to 1.003

Results

On a production-sized batch job (45 TiB of Avro, parallelism 3500, DataStream translation), comparing the source stage before and after this change, per subtask:

	records out (max / mean)	records out CV	duration (max / mean)	duration CV
before	1.38	0.142	1.27	0.133
after	1.00	0.001	1.13	0.036

On a BigQuery export read (714 files for 3750 readers), the export files are now re-split into 7854 splits, so all readers get work, with the busiest reader at 1.017x the average bytes. Before, about 80% of the readers had nothing to read.

Trade-offs

• The refinement calls getEstimatedSizeBytes() on every split, and split() again on some of them. For file-based sources this is in-memory work once the file pattern has been expanded, but it may be slower for sources whose size estimates are remote calls.
• Static assignment cannot adapt to readers that are slower for reasons other than data volume (node, network, storage). Lazy assignment remains available for that.

Testing

• New BoundedSourceSplitterTest: re-splitting of oversized splits, size balancing, split count balancing by bytes per reader, the minimum split size, small inputs, round-robin ordering for the actual reader count, unsplittable sources, assignment mode resolution and size formatting.
• SourceInputFormatTest: static assignment by default and lazy assignment when requested.
• FlinkSourceSplitEnumeratorTest updated for the new default.

---

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.

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI or the workflows README for a list of workflows and how to trigger them.