ndscoder opened a new issue, #1816:
URL: https://github.com/apache/datafusion-python/issues/1816
**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 (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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
