andygrove opened a new issue, #2551:
URL: https://github.com/apache/datafusion-ballista/issues/2551
**Describe the bug**
When a scan's predicate filters on a column the data is clustered by, most
scan tasks end up reading nothing and a few tasks do all the work. The scan's
file groups are built at planning time from file sizes and listing order. Row
group and file pruning from Parquet statistics happens later, inside each task.
If the files are sorted by the filtered column, the files that survive pruning
all sit in a few neighbouring file groups, so only the tasks for those groups
have rows to read.
**Example: TPC-H SF1000 Q6**
Setup: `lineitem` is a Hive-style directory layout partitioned by
`l_shipdate` (`lineitem/l_shipdate=1992-01-02/...`). The table is registered
without partition columns, so pruning comes only from Parquet statistics. The
cluster has 32 executors with 8 task slots each. Ballista 55.0.0, DataFusion
55.1.0.
The scan has `file_groups={256 groups}` in listing (date) order, starting at
`l_shipdate=1992-01-02`, with:
```
predicate=l_shipdate@10 >= 1994-01-01 AND l_shipdate@10 < 1995-01-01 AND ...
pruning_predicate=... l_shipdate_max@0 >= 1994-01-01 ... l_shipdate_min@3 <
1995-01-01 ...
```
Stage 0 task metrics from the event log (second iteration):
| task_id | input_rows | exec time (ms) |
|---|---|---|
| 0–7 | 0 | 769–1212 |
| 8 | 14,962,271 | 1,160 |
| 9 | 187,033,424 | 3,502 |
| 10 | 187,017,808 | 3,641 |
| 11 | 187,059,430 | 3,498 |
| 12 | 187,033,804 | 3,415 |
| 13 | 147,137,385 | 3,131 |
| 14–31 | 0 | 667–1077 |
26 of the 32 tasks read 0 rows, and all of the roughly 910M rows go through
tasks 8–13. Those 6 tasks set the stage's wall-clock time, while the other
executors sit idle once their empty tasks finish. Each empty task still takes
0.7–1.9s just to open files and evaluate statistics on data it then skips. The
query took 4.2s at best.
The same pattern should affect any range filter on a column the files are
sorted or clustered by, including the date-filtered TPC-H queries (Q3, Q5, Q7,
Q12, Q14 and others).
**Expected behavior**
The work that remains after pruning should be spread across the available
task slots. Some options:
- Prune files by statistics (and partition values) during planning, before
the files are split into file groups, so the groups are built only from the
files that will be read.
- Assign files to file groups round-robin or interleaved, rather than in
contiguous runs of the listing order, so that a selective range predicate keeps
its surviving files spread across tasks.
**Additional context**
Declaring the partition columns when registering the table would avoid this
for this dataset, because the directories would be pruned while listing.
Statistics-based pruning is still the only option for data that is sorted but
not partitioned, though.
--
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]
