[Data] [4/11] Add Parquet footer types and the online bin packer

[goutamvenkat-anyscale]

Why

Groundwork for footer-based Parquet chunking (the rest of the split of #64985). Everything here is pure logic with no Ray dependency, landed on its own so it can be reviewed as an algorithm rather than as part of a read-path rewrite.

Today ParquetFileChunker splits each file by ceil(file_size / 1 GiB) without reading the footer, and the partitioner bin-packs those guesses using a fixed encoding-ratio estimate. Bin sizes are inaccurate because the compression ratio is unknown until the footer is read. Replacing that needs a packer that works from real row-group metadata, arrives as a stream, and bounds its own memory.

Nothing calls this yet — it is unused until the footer indexer lands.

Flow

The default path: pack the row groups themselves

Out of the box both coalescing and splitting are off (coalesce_bytes = 0,
split_coalesced = False). One bin item is one physical row group, and
OnlineBinPacker First-Fits those whole items into big bins of max_bin_bytes.
One big bin becomes one read task. Nothing is ever cut.

flowchart LR
  RG["physical row groups<br/>from the footer"] --> IT["one bin item<br/>per row group"]
  IT --> P["OnlineBinPacker<br/>First-Fit, items stay whole"]
  P --> SEAL["seal + evict full big bins<br/>(max_bin_bytes)"]
  SEAL --> M["one big bin = one read task<br/>FileManifest + ParquetRowGroupChunkMetadata"]

Loading

Optional: a second, smaller bin size underneath

Setting coalesce_bytes > 0 inserts a first packing pass upstream, in the footer
reader: coalesce_row_groups packs consecutive row groups into contiguous runs of
~coalesce_bytes. Each run is a small bin, and the packer then packs small bins into
big bins instead of packing row groups directly. Fewer, larger items means fewer
descriptors crossing to the driver.

That raises one question the default path never has to answer: a small bin is a
pre-packed bundle, so what happens when it doesn't fit in the space left in a big bin?
split_coalesced is that answer, and it is also opt-in:

flowchart TD
  RG["physical row groups"] -->|"optional level 1: coalesce_bytes"| SB["small bins<br/>contiguous row-group runs"]
  SB --> FIT{"fits in an open big bin?"}
  FIT -->|yes| PUT["drop it in whole"]
  FIT -->|"no — split_coalesced = false (default)"| NEW["keep it intact,<br/>start a new big bin"]
  FIT -->|"no — split_coalesced = true"| CUT["break it back into its row groups:<br/>fill the leftover space,<br/>carry the remainder onward"]
  PUT --> SEAL["level 2: seal + evict<br/>full big bins (max_bin_bytes)"]
  NEW --> SEAL
  CUT --> SEAL
  SEAL --> M["one big bin = one read task"]

Loading

So the three configurations are:

coalesce_bytes	split_coalesced	behaviour
0 (default)	False (default)	pack whole row groups into big bins
> 0	False	pack whole small bins into big bins
> 0	True	pack small bins, breaking one back open when it would otherwise waste the tail of a big bin

Two properties worth checking during review:

• split_coalesced cannot change the default path. Without coalescing there are no
  small bins to break open — a bin item already is one row group — so the flag is inert
  and packing is plain First-Fit either way. Guarded by
  test_split_coalesced_is_noop_without_coalescing.
• A small bin is only ever cut at the seams it was built from. Never mid-row-group, so
  every piece stays a contiguous run with exact sizes and row counts. A run larger than a
  whole big bin is simply cut across several of them.

Orthogonal to all of the above, big bins come from two pools. A file's chunks go to a
shared pool (mixed files) until that file's cumulative bytes exceed the cap, after
which it switches to a single-file pool — so a large file's row groups stay contiguous
instead of scattering across every open bin. Bins seal and evict as they fill, which
bounds memory over an unbounded stream and lets read tasks start before every footer has
landed.

What changed

File	Change
chunkers/parquet_footer_types.py	new — RowGroupInfo, FileChunks, BinItem, Bin. Frozen dataclasses, no imports beyond typing.
chunkers/parquet_row_group_coalescing.py	new — coalesce_row_groups(). A run breaks on a change in fully_matched (never merge across the match-class boundary, or limit push-down miscounts), a gap in rg_idx (filter-pruned groups), or once the accumulator is full. target == 0 disables it.
partitioners/online_bin_packer.py	new — OnlineBinPacker plus the _BinPool / _SharedBinPool / _SingleFileBinPool hierarchy. A file stays in the shared pool until its cumulative bytes exceed the cap, then switches to a dedicated pool so a large file's row groups stay contiguous instead of scattering across every open bin. Bins seal and evict as they fill, which bounds memory over an unbounded stream and lets read tasks start before every footer has landed.
chunkers/file_chunker.py	adds ParquetRowGroupChunkMetadata alongside the existing ParquetFileChunkMetadata. Additive on purpose — the size-based chunker still needs its own metadata type until the read path switches over.
tests/unit/datasource_v2/test_online_bin_packer.py	new — 14 tests, no Ray cluster.

864 insertions, 0 deletions.

Testing

Coverage is aimed at the invariants rather than the happy path: that every row group is covered exactly once across all emitted bins (parametrized over split_coalesced), that splitting is a no-op without coalescing, that an oversize coalesced run is cut at boundaries and can fill residual shared-bin space, and that a full single-file bin is sealed immediately.

$ python -m pytest -q python/ray/data/tests/unit/datasource_v2/test_online_bin_packer.py
14 passed in 0.05s

$ python -m pytest -q python/ray/data/tests/unit/datasource_v2/
74 passed in 0.17s

$ python -m pytest -q python/ray/data/_internal/datasource_v2/tests/
35 passed in 0.75s

Stack

Step 4 of an 11-step split of #64985, which was 43 files / +3234 −719 and not reviewable as one unit.

• [Data] [1/11] Extract FileIndexer.list_file_infos #65167 — [1/11] Extract FileIndexer.list_file_infos ✅ merged
• [Data] [2/11] Push limit into ReadFiles when Limit sits directly on it #65168 — [2/11] Push limit into ReadFiles when Limit sits directly on it ✅ merged
• [Data] [3/11] Derive ListFiles pushdown state from the ReadFiles scanner #65214 — [3/11] Derive ListFiles pushdown state from the ReadFiles scanner (open)
• this PR — [4/11], no merge-order dependency: it touches a file set disjoint from every other step and applies cleanly to master on its own.
• [Data] [9/11] Pin the Parquet footer actor pool to 1 in tests #65169 — [9/11] Pin the Parquet footer actor pool to 1 in tests (open, independent)

Step 5 (FooterReader) imports parquet_footer_types and parquet_row_group_coalescing from this PR, so it follows this one.

---

Not a duplicate of any open PR (checked gh pr list --repo ray-project/ray --state open plus a file-level overlap check; only #64985, which this replaces).
AI assistance was used; every line reviewed by me and tests run locally.

[gemini-code-assist]

Code Review

This pull request introduces a footer-based Parquet chunking and online bin-packing implementation, including row-group coalescing logic, streaming bin-packing algorithms, and corresponding unit tests. Feedback on the changes highlights an issue in the best-fit bin selection logic (_best_open_bin), where the packer should prioritize maximizing the number of units packed before minimizing the leftover gap to align with its intended behavior.

[cursor]

Cursor Bugbot has reviewed your changes using default effort and found 2 potential issues.

^{Reviewed by Cursor Bugbot for commit dc134d7. Configure here.}

[abhishekverma-ray]

why do we add uncompressed size and compressed size for a path?

[goutamvenkat-anyscale]

no it should be adding only uncompressed sizes for a path

[abhishekverma-ray]

do we have a test case covering the last else condition?

[abhishekverma-ray]

I did not fully understand this one

[goutamvenkat-anyscale]

I clarified it in the comments.

[goutamvenkat-anyscale]

And yes there are a few tests that test this else clause.

[abhishekverma-ray]

Left a few comments, please check. Otherwise looks good to me.

[goutamvenkat-anyscale]

/gemini review

[gemini-code-assist]

Code Review

This pull request introduces a streaming, footer-based Parquet chunking and bin-packing mechanism, including row-group coalescing and an online bin packer to group chunks into partitions. The reviewer suggests adding validation checks across several new dataclasses and classes, specifically ensuring that row-group metadata sizes match counts for coalesced runs, validating that configuration parameters like bin bytes and open bin limits are positive, and verifying that the coalescing target size is non-negative.

[gemini-code-assist]

For defensive programming and to prevent malformed metadata from causing silent issues downstream, we should validate that rg_sizes and rg_rows are populated and match rg_count when rg_count > 1.

Suggested change

	rg_sizes: Tuple[int, ...] = ()
	rg_rows: Tuple[int, ...] = ()
	rg_sizes: Tuple[int, ...] = ()
	rg_rows: Tuple[int, ...] = ()
	def __post_init__(self) -> None:
	if self.rg_count > 1:
	if not self.rg_sizes or len(self.rg_sizes) != self.rg_count:
	raise ValueError(f"rg_sizes must have length {self.rg_count} for coalesced runs")
	if not self.rg_rows or len(self.rg_rows) != self.rg_count:
	raise ValueError(f"rg_rows must have length {self.rg_count} for coalesced runs")

[gemini-code-assist]

Similar to RowGroupInfo, we should validate that rg_sizes and rg_rows are populated and match rg_count when rg_count > 1 in BinItem to ensure metadata integrity.

Suggested change

	rg_sizes: Tuple[int, ...] = ()
	rg_rows: Tuple[int, ...] = ()
	rg_sizes: Tuple[int, ...] = ()
	rg_rows: Tuple[int, ...] = ()
	def __post_init__(self) -> None:
	if self.rg_count > 1:
	if not self.rg_sizes or len(self.rg_sizes) != self.rg_count:
	raise ValueError(f"rg_sizes must have length {self.rg_count} for coalesced runs")
	if not self.rg_rows or len(self.rg_rows) != self.rg_count:
	raise ValueError(f"rg_rows must have length {self.rg_count} for coalesced runs")

[gemini-code-assist]

Add validation for max_bin_bytes and max_shared_open_bins to prevent division by zero, infinite loops, or ValueError when calling max() on an empty sequence of bins.

Suggested change

	def __init__(
	self,
	max_bin_bytes: int,
	*,
	max_shared_open_bins: int = 16,
	split_coalesced: bool = False,
	):
	def __init__(
	self,
	max_bin_bytes: int,
	*,
	max_shared_open_bins: int = 16,
	split_coalesced: bool = False,
	):
	if max_bin_bytes <= 0:
	raise ValueError("max_bin_bytes must be positive")
	if max_shared_open_bins <= 0:
	raise ValueError("max_shared_open_bins must be at least 1")

[gemini-code-assist]

Add validation to ensure target is non-negative. A negative target size is invalid and would result in unexpected behavior (no coalescing).

Suggested change

	if not target:
	return tuple(per_rg)
	if target < 0:
	raise ValueError("target must be non-negative")
	if not target:
	return tuple(per_rg)

[goutamvenkat-anyscale]

@abhishekverma-ray in a follow up PR, I'll decouple the indexer from partitioner that way I can reuse the FilePartitioner abstraction and have each datasource implement its own partitioner