andygrove opened a new issue, #26130:
URL: https://github.com/apache/datafusion/issues/26130
### Is your feature request related to a problem or challenge?
`ListingTable` builds a scan's file groups before any pruning by file
statistics. `list_files_for_regular_scan` collects each file's statistics
(`datafusion.execution.collect_statistics` is on by default), and then
`FileGroup::split_files` sorts the files by path and cuts them into contiguous
chunks. The statistics are only used for pruning later, when each partition
opens its files.
When the files are clustered by a filtered column, for example Hive-style
date directories or files written in date order, every file that can match sits
in a few neighbouring groups. Only the partitions that own those groups have
rows to read.
A small reproduction in `datafusion-cli`, where each file holds a disjoint
range of `d`:
```sql
COPY (SELECT value AS d FROM generate_series(0, 9) ORDER BY random()) TO
'/tmp/skew/f0.parquet' STORED AS PARQUET;
COPY (SELECT value AS d FROM generate_series(10, 19) ORDER BY random()) TO
'/tmp/skew/f1.parquet' STORED AS PARQUET;
COPY (SELECT value AS d FROM generate_series(20, 29) ORDER BY random()) TO
'/tmp/skew/f2.parquet' STORED AS PARQUET;
COPY (SELECT value AS d FROM generate_series(30, 39) ORDER BY random()) TO
'/tmp/skew/f3.parquet' STORED AS PARQUET;
COPY (SELECT value AS d FROM generate_series(40, 49) ORDER BY random()) TO
'/tmp/skew/f4.parquet' STORED AS PARQUET;
COPY (SELECT value AS d FROM generate_series(50, 59) ORDER BY random()) TO
'/tmp/skew/f5.parquet' STORED AS PARQUET;
COPY (SELECT value AS d FROM generate_series(60, 69) ORDER BY random()) TO
'/tmp/skew/f6.parquet' STORED AS PARQUET;
COPY (SELECT value AS d FROM generate_series(70, 79) ORDER BY random()) TO
'/tmp/skew/f7.parquet' STORED AS PARQUET;
SET datafusion.execution.target_partitions = 4;
CREATE EXTERNAL TABLE t STORED AS PARQUET LOCATION '/tmp/skew/';
EXPLAIN FORMAT INDENT SELECT d FROM t WHERE d >= 20 AND d < 40;
```
```text
FilterExec: d@0 >= 20 AND d@0 < 40
DataSourceExec: file_groups={4 groups: [[tmp/skew/f0.parquet,
tmp/skew/f1.parquet], [tmp/skew/f2.parquet, tmp/skew/f3.parquet],
[tmp/skew/f4.parquet, tmp/skew/f5.parquet], [tmp/skew/f6.parquet,
tmp/skew/f7.parquet]]}, projection=[d], file_type=parquet, predicate=d@0 >= 20
AND d@0 < 40, pruning_predicate=d_null_count@1 != row_count@2 AND d_max@0 >= 20
AND d_null_count@1 != row_count@2 AND d_min@3 < 40, required_guarantees=[]
```
Only `f2` and `f3` can match, and both are in the second group, so the other
three partitions have nothing to read. The scan's statistics also still count
all 8 files (`Rows=Inexact(80)` with `datafusion.explain.show_statistics =
true`), even though at most 20 rows can match.
In a single process, FileStream work stealing (#21351) mostly hides this,
because an idle partition picks up files from its siblings. Engines that run
each partition as a separate task, such as Ballista and datafusion-distributed,
can't rely on that (see
`datafusion.execution.enable_file_stream_work_stealing`, #23294). For them the
skew decides how long the stage takes. In apache/datafusion-ballista#2551,
TPC-H Q6 over a `lineitem` table stored in `l_shipdate=...` directories and
registered without partition columns had 26 of 32 scan tasks read no rows,
while the other 6 read every matching row.
### Describe the solution you'd like
Prune files against the statistics `ListingTable` has already collected,
during listing and before the files are split into groups. This would sit next
to the existing Hive partition pruning. `scan_with_args_inner` already
separates the non-partition filters from the partition filters. Today it only
uses them to decide whether a limit can be applied while listing.
- A file is dropped only when its statistics prove that no row can match.
Files without usable statistics are kept.
- There is no extra I/O, since the statistics have already been collected.
- `split_files`, and `repartition_file_scans` later on, then spread the
remaining files over the partitions, so no new balancing logic is needed.
- The scan's statistics only count the remaining files, which gives the
optimizer better estimates.
- Scans with a declared `output_partitioning` keep their layout, since their
groups line up with partitions. #25437 covers pruning for the Range case.
### Describe alternatives you've considered
- Assign files to groups round-robin instead of in contiguous runs. This
spreads the matching files too, but every partition still has to check and skip
files that could have been dropped at planning time, and the statistics stay
unpruned.
- Prune in a physical optimizer rule once `FilterPushdown` has put the
predicate on the scan. This works for any `FileScanConfig`, and it is what
Ballista is adding in apache/datafusion-ballista#2558. The rule has to regroup
the files itself, though, and rules that run before `FilterPushdown`, such as
`JoinSelection`, still see unpruned statistics.
- Eager pruning at planning time (#24944, #25328) reads Parquet footers and
page indexes to sharpen estimates. It is opt-in and skips scans with more than
`eager_pruning_file_limit` files (256 by default). File-level pruning from
statistics that were already collected is a cheaper first step and would
complement it.
### Additional context
- apache/datafusion-ballista#2551 has the full TPC-H Q6 task metrics.
- datafusion-contrib/datafusion-distributed#759 is another scan skew report
from an engine that runs with work stealing off.
--
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]
