Repository navigation
Description
Is your feature request related to a problem or challenge? Please describe what you are trying to do.
We read Delta tables through DataFusion's Python package. In Delta, the log says which files are live, not the folder: a folder also holds files that a later write replaced, until a vacuum removes them. So we cannot register a Delta table as a folder listing, because the replaced files would be read next to the live ones.
What we do today is register one small table per live file and union them, adding each file's partition values as constant columns. That works, but it loses what a partitioned listing table gives. With the partition column declared on one Parquet scan, DataFusion 55 prunes whole files by partition value, including by a TopK's dynamic filter. Constants added per file cannot be pruned that way.
Measured on a 10.5M-row table partitioned by scenario (20 partitions), for a capped sort ordered by the partition column (ORDER BY scenario, ... LIMIT 250001):
- a listing table with the partition column declared, one partition: 525.6K rows read for 250,001 kept (1 of 20 files' rows)
- our union of per-file tables: all 10.5M rows read
On our production tables the same query reads 312.6M rows to keep 250,001.
Describe the solution you'd like
A way to register a Parquet table from Python over an explicit list of files, each with its partition values, and an optional schema so planning reads no file. Rust DataFusion already supports this: a listing table over given files, each a PartitionedFile carrying its partition_values. Something like:
ctx.register_parquet_files(
"t",
files=[
("az://container/t/scenario=a/part-1.parquet", ["a"]),
("az://container/t/scenario=b/part-2.parquet", ["b"]),
],
table_partition_cols=[("scenario", pa.string())],
schema=schema,
)
The name and shape are only a suggestion. What matters is that the files are exactly the ones given (no listing), and that the partition values reach the scan as partition columns, so file pruning and dynamic filters work on them.
Describe alternatives you've considered
- Registering the table's folder as a listing table: reads files the Delta log no longer lists.
- One table per file, unioned, partition values as constant columns: correct rows, but no pruning by partition value (what we do today).
- A pyarrow dataset built from the file list with its partitioning, registered as a dataset: we saw intermittent hangs reading through it.
- deltalake's own table provider: correct, but its scan takes no TopK dynamic filter today (feat(datafusion): implement try_pushdown_sort on DeltaScanExec delta-io/delta-rs#4969).
- Writing our own table provider in Rust and passing it to Python through the FFI: possible, but it means a Rust toolchain, wheels per platform and matching DataFusion's version exactly at every upgrade, for a feature Rust DataFusion already has.
Additional context
Measured with datafusion (Python) 55.0.0 and deltalake 1.6.6 on Windows 11. The same need would apply to any table format that keeps its live files in a log (Delta, Iceberg) and is read through the Python package.
Activity