Is your feature request related to a problem or challenge?
Currently, DataFusion's Parquet reader (to my understanding) only applies pruning/data skipping during the execution of the DataSourceExec. While this keeps the effort during planning at a minimum, it also makes it difficult to estimate the cardinality of Parquet scans with fitlers without any external statistics information.
So, for example, given the table movie_info in the imdb benchmark some filters are highly selective. See an example taken from 3b.sql:
> SELECT COUNT(*) FROM movie_info;
+----------+
| count(*) |
+----------+
| 14835720 |
+----------+
1 row(s) fetched.
Elapsed 0.001 seconds.
> SELECT COUNT(*) FROM movie_info WHERE info IN ('Bulgaria');
+----------+
| count(*) |
+----------+
| 2287 |
+----------+
1 row(s) fetched.
Elapsed 0.325 seconds.
Unfortunately, pruning doesn't help much here as the file is not sorted on the info column (scan_efficiency_ratio=69.41% (210.9 M/303.8 M) on the Parquet scan). I think it's not an uncommon scenario that Parquet files are (almost) sorted on particular columns that are regularly used for filters. For these scenarios we could get better statistics. For example, by sorting the movie_info table on info.
> COPY (SELECT * FROM movie_info ORDER BY info) TO './benchmark/data/imdb/movie_info_sorted.parquet' STORED AS parquet;
+----------+
| count |
+----------+
| 14835720 |
+----------+
1 row(s) fetched.
Elapsed 8.571 seconds.
> CREATE EXTERNAL TABLE movie_info_sorted STORED AS PARQUET LOCATION './benchmark/data/imdb/movie_info_sorted.parquet';
> EXPLAIN ANALYZE SELECT COUNT(*) FROM movie_info_sorted WHERE info IN ('Bulgaria');
Now we have: scan_efficiency_ratio=0.27% (800.7 K/298.2 M) and page_index_rows_pruned=1.05 M total → 15.36 K. So, from what I can tell, we could put an Precision::Inexact(15.36K) as a relatively good estimate of the cardinality in the statistics if we do the pruning already during planning.
If we would have these stats available during planning we (I think) could do: i) better join ordering and ii) other optimizations. For example, given the 800.7K scanned bytes in the statistics, it could trigger an eager fetch for this scan as it is considered small (#24922). In an object store environment this could shave-off about 150ms from the query latency (ignoring the effect of better join ordering).
I think there are some possible downsides to consider:
- planning takes longer
- there could be a cheap and highly selective dynamic filter (e.g.,
table_id = 1) that would be cheaper and would make the pruning that we already did unnecessary.
- Something I did not consider apart from making configuration even more involved?
Describe the solution you'd like
Some thing like this:
/// Sets whether the Parquet scanner should do some work during planning to get better statistics
enum EagerParquetPruning {
/// No pruning during planning
Disabled,
/// Do row group pruning during planning
RowGroups,
/// Do row group and page index pruning during planning
PageIndex,
/// Do bloom filter pruning during planning.
BloomFilters,
/// Do full pruning during planning.
Full
}Then, during physical planning of the scan, we can do the pruning and update the statistics for FileGroups etc. Doing that, the remainder of the optimization pipeline should see (hopefully) more accurate statistics. One possible hook would be:
async fn create_physical_plan(
&self,
state: &dyn Session,
conf: FileScanConfig,
) -> Result<Arc<dyn ExecutionPlan>>It's just important that we cannot rely too much on these stats. For scans that filter on columns that are clustered together in the Parquet file, this can become relatively accurate (e.g., down to the data page size). So it's imaginable that the difference between the full file statistics is large (see above) if the data is suitable. This could be an interesting knob for tuning DataFusion to use cases with suitable data files.
Eager pruning results could be stored in a ParquetAccessPlan to try to avoid duplicate work. Maybe we can also add some state to make this more efficient (e.g., avoid re-evaluating the predicates that already have been evaluated).
One option would be to predicate the eager pruning on whether something (e.g., page index) is cached and thus no I/O is needed. But the I/O would be necessary during the scan eventually so I am not sure whether this is necessary.
Describe alternatives you've considered
No response
Additional context
Related to:
Is your feature request related to a problem or challenge?
Currently, DataFusion's Parquet reader (to my understanding) only applies pruning/data skipping during the execution of the
DataSourceExec. While this keeps the effort during planning at a minimum, it also makes it difficult to estimate the cardinality of Parquet scans with fitlers without any external statistics information.So, for example, given the table
movie_infoin the imdb benchmark some filters are highly selective. See an example taken from3b.sql:Unfortunately, pruning doesn't help much here as the file is not sorted on the info column (
scan_efficiency_ratio=69.41% (210.9 M/303.8 M)on the Parquet scan). I think it's not an uncommon scenario that Parquet files are (almost) sorted on particular columns that are regularly used for filters. For these scenarios we could get better statistics. For example, by sorting themovie_infotable oninfo.Now we have:
scan_efficiency_ratio=0.27% (800.7 K/298.2 M)andpage_index_rows_pruned=1.05 M total → 15.36 K. So, from what I can tell, we could put anPrecision::Inexact(15.36K)as a relatively good estimate of the cardinality in the statistics if we do the pruning already during planning.If we would have these stats available during planning we (I think) could do: i) better join ordering and ii) other optimizations. For example, given the
800.7Kscanned bytes in the statistics, it could trigger an eager fetch for this scan as it is considered small (#24922). In an object store environment this could shave-off about 150ms from the query latency (ignoring the effect of better join ordering).I think there are some possible downsides to consider:
table_id = 1) that would be cheaper and would make the pruning that we already did unnecessary.Describe the solution you'd like
Some thing like this:
Then, during physical planning of the scan, we can do the pruning and update the statistics for
FileGroupsetc. Doing that, the remainder of the optimization pipeline should see (hopefully) more accurate statistics. One possible hook would be:It's just important that we cannot rely too much on these stats. For scans that filter on columns that are clustered together in the Parquet file, this can become relatively accurate (e.g., down to the data page size). So it's imaginable that the difference between the full file statistics is large (see above) if the data is suitable. This could be an interesting knob for tuning DataFusion to use cases with suitable data files.
Eager pruning results could be stored in a
ParquetAccessPlanto try to avoid duplicate work. Maybe we can also add some state to make this more efficient (e.g., avoid re-evaluating the predicates that already have been evaluated).One option would be to predicate the eager pruning on whether something (e.g., page index) is cached and thus no I/O is needed. But the I/O would be necessary during the scan eventually so I am not sure whether this is necessary.
Describe alternatives you've considered
No response
Additional context
Related to: