[Data] Support predicate pushdown at the Delta Lake level

[tobby168]

Summary

read_delta resolves the entire Delta table on the driver before any predicate is known:

DeltaTable(path) -> dt.to_pyarrow_dataset() -> ParquetDatasource.from_pyarrow_dataset()

Every file is enumerated at call time. For tables that rely on Delta's Z-ordering and per-file statistics rather than Hive partitioning, a selective query still touches every file — at minimum for tail metadata.

This routes read_delta through the DataSourceV2 pipeline (already the default) with a Delta-native file indexer that prunes files from the transaction log — by partition values and by per-file min/max statistics — before any file is listed, sized, or scheduled.

On an unpartitioned table of 4 files holding disjoint val ranges (the shape the issue describes):

predicate	rows (V1 / V2)	files listed
val > 250	10 / 10	1 of 4
val < 5	5 / 5	1 of 4
val > 1000	0 / 0	0 of 4
val >= 0	40 / 40	4 of 4
val != 3	39 / 39	4 of 4

Closes #61547.

Design

Log-level pruning is an optimization, never a correctness requirement.

Predicates reach the scanner through the existing ReadFiles.apply_predicate path and are enforced there while reading — data predicates through the PyArrow scanner filter, partition predicates by parsing the file path. PushdownDeltaFilePruning runs after PredicatePushdown and additionally copies them onto the upstream ListFiles indexer, so files that cannot match are never listed. If that rule doesn't fire, the query reads more files but returns the same rows.

Making that true rather than merely intended took two changes, both of which earlier revisions of this PR got wrong:

• Partition predicates are no longer enforced by path parsing. PathPartitionParser.evaluate_predicate_on_partition swallows any evaluation error and keeps the file — sound only while something else applies the predicate, but apply_predicate removes the Filter as soon as a scanner accepts one. A path is a lossy view of the log: a null partition is the literal __HIVE_DEFAULT_PARTITION__ directory, and pc.is_in over the null-typed row the parser builds raises. Scanners can now decline enforcement via enforces_partition_predicate; DeltaScanner does, so the predicate stays in a Filter that binds against the partition columns the reader has already materialized — exactly typed, with real nulls. The rule still forwards it to the log for skipping, so a partition-pruned read still lists 1 of 2 files.
• The rule declares PredicatePushdown in dependencies(). Ruleset only honors declared edges; list position is a tiebreaker, so an ordering assumption left in a comment can be silently inverted by an unrelated rule.

Statistics-based skipping is conservative: a file is dropped only when [min, max] proves no row can match. !=, NOT, UDFs, missing stats, a null bound, an unknown column — every "don't know" keeps the file. Over-retention costs a read; dropping a matching file loses data silently. test_pruning_never_drops_a_matching_file asserts that property over generated inputs (seeded per parameter, so each case is reproducible).

DeltaScanner subclasses ParquetScanner and adds exactly one thing — declining partition-predicate enforcement. Filter pushdown, column pruning and limit pushdown are inherited unchanged, and the reader is ParquetFileReader as-is, because a Delta table's data files are ordinary Parquet laid out Hive-style. Schema and partition columns come from the log, so no Parquet footers are sampled for inference.

The snapshot version is resolved on the driver and pinned into the datasource. Listing is deferred to workers, so leaving it unresolved would let a concurrent commit be read against a schema inferred before it existed, and could make a retried task disagree with its first attempt.

When V2 is used

_delta_table_supports_datasource_v2 is the single gate, and it fails closed — anything V2 would answer differently rather than merely faster stays on V1, so enabling V2 never changes a result:

• Column mapping. The Parquet columns carry generated names while the logical names live only in the log, so opening the files directly with the logical schema finds none of the requested columns. to_pyarrow_dataset currently raises for these, so behavior is unchanged — and if [data] Fix delta column mapping read #64929 lands they pick up its fix without this path needing to know about it.
• A column named path. That is how the reader labels the synthesized file path, so a real column of that name would be shadowed rather than returned.
• **arrow_parquet_args passed. V1 forwards these to iter_batches; V2 has nowhere to put them, so honoring them means staying on V1 rather than silently dropping them.

Deletion Vectors are rejected up front on both paths. Those data files still physically contain the deleted rows, and the V2 path opens them directly. The previous error text suggested upgrading to deltalake>=0.10.0, which is not the issue under 1.5.0.

Incidental correctness fixes

Cases where the V2 path would otherwise have been silently wrong:

• Path encoding. Add actions record paths URL-encoded while the objects they name are not — a partition value of e=f lives in grp=e%3Df but is recorded as grp=e%253Df. Joining the raw value yields a path that doesn't exist. Decoded once before joining, which reproduces DeltaTable.file_uris() without a second listing.
• Column order. The V2 reader appends path-derived partition columns after the file's own and only reorders to an explicit projection, so partitioned tables would come back with partition columns moved to the end — unlike every previous release. The scanner now pins a projection, with path added back when include_paths is set so a column projection can't drop it.
• Null partition values. Delta writes these as __HIVE_DEFAULT_PARTITION__, which cannot be cast to the column's declared type. Adds opt-in Partitioning.null_fallback, and PathPartitionParser no longer coerces the resulting None. read_parquet behavior is unchanged.
• URI-addressed tables. deltalake resolves its storage backend from the URI while PyArrow rejects a scheme-prefixed path, so the log is opened with the path as given and the reader gets filesystem-native paths. Without this every s3:// / az:// table — read_delta's own documented examples — failed at execution. This was also the remaining half of the catalog= breakage.
• count(). PushdownCountFiles asserted its indexer was a NonSamplingFileIndexer; it only needs one that can be pinned to whole-file listing. It now checks for that and skips the optimization otherwise, rather than failing the query.
• An empty Delta table is answered from the log — honoring columns and include_paths, so unioning it with a populated table doesn't fail on mismatched schemas — instead of tripping the V2 "no files found" guard, which exists to catch empty directories.

Not a duplicate

No open PR addresses #61547 (gh pr list --state open --search "61547 in:body" returns nothing). #62959 was an earlier attempt by the same author on a since-superseded architecture; it was closed by the stale bot without human review, and this is a fresh implementation against the current DataSourceV2 code. richardliaw invited work on the issue in this comment.

The nearest open PR in this area, #64929 (Delta column mapping), touches read_delta in the same function but solves a different problem; the column-mapping fallback above is written so the two compose rather than conflict.

Testing

Run per file, matching how Bazel runs these targets. 270 passed, 0 failed.

pytest -q python/ray/data/tests/unit/test_delta_file_pruning.py                      # 57 passed
pytest -q python/ray/data/_internal/datasource_v2/tests/test_delta_file_indexer.py   # 19 passed
pytest -q python/ray/data/tests/datasource/test_delta.py                             # 15 passed
pytest -q python/ray/data/tests/datasource/test_delta_pushdown.py                    # 64 passed
pytest -q python/ray/data/tests/datasource/test_write_delta.py                       # 58 passed
pytest -q python/ray/data/tests/datasource/test_read_parquet_v2.py                   # 13 passed
pytest -q python/ray/data/tests/test_execution_optimizer_basic.py                    # 18 passed
pytest -q python/ray/data/tests/test_execution_optimizer_limit_pushdown.py           # 26 passed
pytest -q python/ray/data/tests/test_execution_optimizer_advanced.py                 # 45 passed

pre-commit run --files $(git diff --name-only master...HEAD) — all hooks pass.

Tests assert on files skipped, not only on rows: reading every file also produces the right answer, so a pruning optimization that silently stopped pruning would otherwise pass every correctness test. Likewise test_count_matches_v1 counts on a fresh dataset, because reading first populates the cached row count and hides the plan shape count() actually takes.

test_parquet.py, test_partitioning.py and python/ray/data/tests/unit/ show pre-existing failures in the local environment used here (missing torch, S3/moto fixtures, and Ray workers unable to import test modules outside Bazel). Each was verified identical on the base commit — test_parquet.py 7 failed / 34 errors on both, test_partitioning.py 32 failed / 6 errors on both, tests/unit/ 9 failed on both.

AI assistance

This change was developed with AI assistance (Claude Code), including two adversarial review passes. The first found the count(), **arrow_parquet_args and catalog= breakage; the second found that partition predicates still had a single, fail-open enforcement point — returning extra rows for is_in over a nullable partition column in the default configuration — and that URI-addressed tables did not work at all. Both are fixed and covered by tests above.

[gemini-code-assist]

Code Review

This pull request introduces a DataSourceV2 implementation for Delta Lake tables, enabling log-driven file listing and pruning (both partition pruning and statistics-based skipping) via DeltaFileIndexer and prune_add_actions. It also adds logical rules (PushdownDeltaFilePruning) to push down predicates settled by PredicatePushdown to the DeltaFileIndexer, and updates read_delta to route through the V2 path when use_datasource_v2 is enabled. The review feedback highlights several instances of redundant wrapping of PyArrow objects (such as pa.schema() and pa.table()) where the underlying methods already return the correct types.

[gemini-code-assist]

The call to pa.schema() is redundant here because self._open_table().schema().to_arrow() already returns a pyarrow.Schema object. Removing the redundant wrapping simplifies the code and avoids unnecessary object instantiation.

Suggested change

	schema = pa.schema(self._open_table().schema().to_arrow())
	schema = self._open_table().schema().to_arrow()

[gemini-code-assist]

The call to pa.table() is redundant here because table.get_add_actions(flatten=True) already returns a pyarrow.Table object. Removing the redundant wrapping simplifies the code and avoids unnecessary object instantiation.

Suggested change

	add_actions = pa.table(table.get_add_actions(flatten=True))
	add_actions = table.get_add_actions(flatten=True)

[gemini-code-assist]

The call to pa.schema() is redundant here because dt.schema().to_arrow() already returns a pyarrow.Schema object. Removing the redundant wrapping simplifies the code and avoids unnecessary object instantiation.

Suggested change

	schema = pa.schema(dt.schema().to_arrow())
	schema = dt.schema().to_arrow()

[gemini-code-assist]

The call to pa.schema() is redundant here because dt.schema().to_arrow() already returns a pyarrow.Schema object. Removing the redundant wrapping simplifies the code and avoids unnecessary object instantiation.

Suggested change

	schema = pa.schema(dt.schema().to_arrow())
	schema = dt.schema().to_arrow()

[gemini-code-assist]

The call to pyarrow.schema() is redundant here because dt.schema().to_arrow() already returns a pyarrow.Schema object. Removing the redundant wrapping simplifies the code and avoids unnecessary object instantiation.

Suggested change

	schema = pyarrow.schema(dt.schema().to_arrow())
	schema = dt.schema().to_arrow()

[cursor]

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

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

[cursor]

Residual filter needs partition columns

Medium Severity

When enforces_partition_predicate is false, the partition predicate is kept as a residual Filter above ReadFiles. That filter needs the partition columns in each block, but a columns projection can omit them so the reader never synthesizes them. The residual filter then fails at execution for patterns like reading a subset of columns and filtering on a partition key.

Additional Locations (1)

• python/ray/data/_internal/datasource_v2/scanners/delta_scanner.py#L35-L38

 

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

[tobby168]

Thanks both — responses below, plus one behavior change worth flagging that came out of digging into the second Bugbot finding.

@gemini-code-assist: the five pa.schema() / pa.table() suggestions

These don't hold — deltalake returns arro3 objects, not PyArrow ones:

>>> import deltalake, pyarrow as pa
>>> deltalake.__version__
'1.5.0'
>>> type(dt.schema().to_arrow())
<class 'arro3.core._core.Schema'>
>>> isinstance(dt.schema().to_arrow(), pa.Schema)
False
>>> type(dt.get_add_actions(flatten=True))
<class 'arro3.core._core.Table'>
>>> isinstance(dt.get_add_actions(flatten=True), pa.Table)
False

The pa.schema(...) / pa.table(...) calls are genuine cross-library conversions over the Arrow PyCapsule interface, not redundant wrapping.

They're easy to mistake for no-ops because arro3 mirrors much of the PyArrow API — .names, .get_field_index() and .num_rows all work on the unwrapped objects. But the operations this code depends on don't:

>>> dt.get_add_actions(flatten=True).filter(mask)
AttributeError: 'arro3.core._core.Table' object has no attribute 'filter'

prune_add_actions filters the add-actions table, so dropping that conversion breaks file pruning outright, and arro3.Schema.empty_table() returns an arro3 table that from_arrow won't accept. Leaving all five as-is.

@cursor: "Empty read drops path column"

Correct, and fixed in 44ace4c. read_delta(columns=["a"], include_paths=True) returned ["a", "path"] for a populated table and ["a"] for an empty one, so a union of the two failed.

Both paths now derive their output columns from one helper, and the test covers each combination of columns and include_paths rather than one option at a time — the gap was that both options were tested, just never together.

@cursor: "Residual filter needs partition columns"

The described scenario doesn't reproduce: ProjectionPushdown already accounts for the columns a residual Filter binds to.

read_delta(t).filter(col("region") == lit("US")).select_columns(["val"])
  Project
    Filter  region == 'US'
      ReadFiles  scanner.columns=('region', 'val')   <-- region retained
        ListFiles
  -> [{'val': 1}]

But it pointed at something real. When columns= is passed to read_delta itself and excludes a partition column, filtering on that column afterwards now raises instead of silently working:

read_delta(t, columns=["val"]).filter(expr=col("region") == lit("US"))
# before: [{'val': 1}]
# now:    KeyError: Field "region" does not exist in schema

That is a consequence of this PR moving partition-predicate enforcement into a Filter. Previously the predicate was absorbed by path-based pruning, so nothing ever needed region in the block.

I've left it raising, for three reasons:

1. The query filters on a column the caller projected away — the dataset's schema genuinely doesn't contain it.
2. read_parquet raises on exactly this pattern today (RuntimeError: Filter expression '(region == "US")' failed on parquet file ... with columns: {'val'}), so raising is the established behavior; read_delta accepting it was the outlier.
3. Both ways I tried to preserve the old behavior — adding partition columns to the scanner projection, and re-narrowing after the read — can't work, because the caller's .filter() is attached after read_delta has already returned.

Flagging it explicitly rather than leaving it to be discovered. Happy to reconsider if maintainers would rather keep the old leniency.

[gemini-code-assist]

Thanks for the detailed breakdown, @tobby168. Your analysis of the arro3 vs pyarrow object distinction is spot on, and it makes perfect sense why those conversions are necessary for the operations you're performing.

Regarding the behavior change for partition columns in read_delta when using columns= to filter them out: your reasoning for letting it raise is sound. Aligning with read_parquet's established behavior and treating the schema as the source of truth is the correct approach for maintaining consistency. Flagging this explicitly is the right call, as it prevents silent failures and makes the API contract clearer for users.

[jackylee-ch]

Drive-by note, not a review — this PR touches the same loop I was about to send a separate fix for, and they're different defects, so flagging here rather than opening a competing PR.

Your if value is None: continue handles a null partition value coming from null_fallback. The case I hit is a missing key: partitions[field] itself raises KeyError whenever a field_types key isn't present in the path — e.g. an unpartitioned file directly under base_dir, which Partitioning's own field comment (partitioning.py:93-96) documents as supported. It also breaks the recipe in PathPartitionFilter's docstring (:470, "Unpartitioned files are denoted with an empty input dictionary"):

f = PathPartitionFilter.of(lambda d: True if d else False, style="hive",
                           base_dir="/tmp/part_probe", field_types={"year": int})
f(["/tmp/part_probe/year=2024/f.csv", "/tmp/part_probe/loose.csv"])
# KeyError: 'year'

Either works for me: I can wait for this to land and send the missing-key fix on top, or you could fold it in here since it's the same two lines. Whichever you prefer.