[Data] [5/11] Add the Parquet FooterReader actor pool

[goutamvenkat-anyscale]

Why

Footer IO for the footer-based Parquet chunking path (the rest of the split of #64985).

Reading N Parquet footers serially on the driver doesn't scale — it's the step that decides which row groups a read even touches. FooterReader runs as a pool of Ray actors, each reading footers on its own thread pool and streaming results back in batches, so footer IO spreads across the cluster instead of bottlenecking on one process.

Nothing constructs it yet — it's unused until the footer indexer lands in step 6.

Flow

Per file, it produces the row-group descriptors the bin packer (merged in #65210) consumes:

flowchart LR
  B["(path, size) batch"] --> A["FooterReaderActor<br/>SPREAD across the cluster"]
  A --> R[read footer]
  R --> F["split_by_row_group(filter)<br/>prune by statistics"]
  F --> Q["split_by_row_group(~filter)<br/>classify fully_matched"]
  Q --> C["coalesce_row_groups<br/>(from #65210)"]
  C --> O["stream List[FileChunks]<br/>batched by result_batch_size"]

Loading

Three things it decides, each of which the packer and the limit push-down depend on:

• Predicate pruning. fragment.split_by_row_group(filter) drops row groups whose Parquet statistics contradict the pushed filter, so they never reach the packer at all.
• fully_matched classification. A surviving group is fully matched iff ~filter prunes it entirely — the DataFusion trick. Only fully-matched groups have an exact survivor count, and only exact counts may drive limit push-down. Any failure defaults to not-fully-matched, which is always the safe direction. This is the correctness-critical bit: nulls are the interesting case, since Parquet min/max stats are computed over non-null values only, so a group whose non-null values all satisfy the filter looks fully matched by bounds alone while its null rows don't survive.
• Projection-scoped sizing. Byte sizes are accounted over only the projected leaf columns, so bins get sized for what the reader will actually fetch. Nested columns expand to all their leaves.

What changed

Two new files, 368 insertions, 0 deletions:

File	Change
listing/footer_reader.py	new — FooterReader plus FooterReaderActor = ray.remote(FooterReader). Kept as a plain class with the actor built via the functional form, so callers can type handles as ActorProxy[FooterReader]. read_footers is a streaming @ray.method generator that yields List[FileChunks], batched by result_batch_size — the driver pays one object-store fetch per yielded list, so batching cuts driver-side deserialization roughly proportionally.
tests/test_footer_reader.py	new — 14 tests

Testing

The tests drive the plain class directly against local Parquet fixtures — no actor, no cluster. Coverage is aimed at the decisions above rather than the plumbing: predicate pruning by statistics, the partial-match case (id >= 30 splitting a row group, which must not count as exact), a predicate matching nothing, projection shrinking accounted size while leaving row counts alone, coalescing preserving total rows, nested columns expanding to all leaves, and result batching at several result_batch_size values.

$ python -m pytest -q python/ray/data/_internal/datasource_v2/tests/test_footer_reader.py
14 passed in 4.30s

$ python -m pytest -q python/ray/data/_internal/datasource_v2/tests/ python/ray/data/tests/unit/datasource_v2/
145 passed in 0.90s

Stack

Step 5 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 scanner ✅ merged
• [Data] [4/11] Add Parquet footer types and the online bin packer #65210 [4/11] Parquet footer types + online bin packer ✅ merged
• this PR [5/11] — depends on [Data] [4/11] Add Parquet footer types and the online bin packer #65210 for parquet_footer_types and parquet_row_group_coalescing
• [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 6 wires this and the packer together behind an opt-in flag, and needs both this and #65214.

---

Not a duplicate of any open PR (checked gh pr list --state open plus a file-level overlap check against #65142 / #63158 / #64404 — no overlap; 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 the FooterReader class to concurrently read Parquet footers and chunk files into row-group runs, along with corresponding unit tests. The review feedback identifies a critical correctness bug where row groups containing null values in filtered columns can be incorrectly marked as fully matched, which would lead to incorrect downstream row counts. Additionally, the feedback suggests avoiding over-subscribing Arrow's CPU thread pool by removing the CPU count configuration that matches the high I/O concurrency limit, and recommends adding a test case to verify the handling of null values during filtering.

[gemini-code-assist]

Critical Correctness Bug with Null Values

There is a subtle but critical correctness bug in how fully_matched is determined when the filtered column contains Null values.

The Scenario

Suppose a row group has 4 rows: [35, 40, 45, Null], and the filter is id >= 30.

1. The Parquet statistics for id are min=35, max=45, null_count=1 (since min/max stats exclude nulls).
2. split_by_row_group(id >= 30) keeps the row group because min=35 >= 30.
3. split_by_row_group(~(id >= 30)) (which simplifies to id < 30) evaluates to False for all non-nulls (since min=35 >= 30) and False for the null row (since Null < 30 is Null/False).
4. Therefore, split_by_row_group(~self.filter) prunes the row group, meaning it is not in not_fully.
5. The code then incorrectly sets fully_matched = True for this row group.
6. However, the null row actually did not satisfy the original filter id >= 30! Only 3 out of 4 rows survived, meaning num_rows (4) is not the exact survivor count.

This will cause downstream limit push-down to miscount the survivors as 4 instead of 3, leading to incorrect query results.

Suggested Fix

To fix this, we must ensure that if any column referenced in the filter has null_count > 0 in the row group, we do not mark it as fully_matched (unless the filter is explicitly null-safe). We can extract the referenced columns from filter_expr and check their null_count in the row group's column chunk metadata.

[gemini-code-assist]

Setting pa.set_cpu_count(io_concurrency) to a high value like 128 (the default io_concurrency) is problematic. Arrow's CPU thread pool is used for CPU-bound tasks (like decompression, decoding, and expression evaluation), whereas S3/footer I/O is handled by the I/O thread pool (pa.set_io_thread_count). Over-subscribing the CPU thread pool can lead to severe thread contention and performance degradation, especially on machines with fewer CPU cores or when multiple actors run on the same node. We should remove pa.set_cpu_count(io_concurrency) and let Arrow use its default CPU thread count (which matches the system's CPU capacity).

Suggested change

	# Match Arrow's process-wide pools to the actor's IO concurrency so
	# nested S3/footer work isn't bottlenecked on the default 8 threads.
	pa.set_io_thread_count(io_concurrency)
	pa.set_cpu_count(io_concurrency)
	# Match Arrow's process-wide pools to the actor's IO concurrency so
	# nested S3/footer work isn't bottlenecked on the default 8 threads.
	pa.set_io_thread_count(io_concurrency)

[gemini-code-assist]

Please add a test case to verify the behavior of fully_matched when there are Null values in the filtered column. This will help prevent regressions and ensure correctness.

Example test case:

    def test_fully_matched_with_nulls(self, tmp_path):
        # Create a table with nulls in the filtered column
        table = pa.table({"id": [35, 40, 45, None]})
        path, size = _write(tmp_path / "nulls.parquet", table, row_group_size=4)
        
        # id >= 30 should NOT mark the row group as fully matched because of the null
        chunks = _reader(filter_expr=col("id") >= 30)._read_and_chunk(path, size)
        assert len(chunks.row_groups) == 1
        assert chunks.row_groups[0].fully_matched is False

[cursor]

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

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

[cursor]

Wrong row-group byte size source

Medium Severity

On the no-projection path, uncompressed_size is taken from total_byte_size, which can be the compressed size for some Parquet files (apache/arrow#48138). Existing helpers in parquet_datasource.py and parquet_file_reader.py already avoid that accessor and sum per-column total_uncompressed_size instead. Undersized bins then overfill read tasks.

 

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

[cursor]

Nulls break fully-matched classification

High Severity

fully_matched is inferred only from whether ~filter is pruned by split_by_row_group. Under three-valued logic that is not enough: nulls make ~filter unsatisfiable while still failing the original predicate. DataFusion’s version also conjuncts IS NULL on filter columns before the inverted prune; without that, limit push-down can treat partial survivors as an exact count.

 

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