pingzh opened a new issue, #5695:
URL: https://github.com/apache/datafusion-comet/issues/5695
### What is the problem the feature request solves?
## What is the problem the feature request solves?
Selective joins can discard most fact-table rows, but filtering is much less
valuable when those rows have already been read, decoded, and converted into
Arrow. We should let eligible native joins pass runtime information back to
native readers so that work can be avoided at the scan.
For example:
```sql
SELECT SUM(s.ss_net_paid)
FROM store_sales s
JOIN item i ON s.ss_item_sk = i.i_item_sk
WHERE i.i_category = 'Books';
```
Suppose the filtered build side contains item IDs `{10, 20, 30}`. After that
build is complete, its key domain can provide a safe predicate for the fact
scan. A reader may then skip row groups whose statistics exclude those IDs, or
reject rows during decoding before materializing other columns. This does not
require `ss_item_sk` to be a table partition column. The actual join remains
responsible for producing the result.
In the successful-query subset of our latest 10 TB comparison,
Spark-generated Bloom filters already executed natively in 27 of 79 queries.
However, the observed path was **Spark scan → Spark-to-Arrow conversion →
native Bloom filter → downstream operators**. Those filters cannot undo the
preceding I/O or conversion. Native scans were deliberately disabled for that
shuffle comparison, so it does not measure the benefit proposed here.
DataFusion already has dynamic-filter producers and consumers, but Comet
directly constructs and executes physical plans rather than running the
physical optimizer pass that normally connects them. This is an integration
gap, not simply a disabled DataFusion option.
### Describe the potential solution
Connect build-side runtime filters to eligible native probe-side scans,
starting with a narrow hash-join path. Reuse DataFusion's existing filter
machinery where appropriate, and preserve Comet's Spark-facing execution and
metrics contracts.
There are two distinct milestones:
1. **Producer and row-filter integration:** populate a shared filter from
the completed join build and consume it on probe batches. This is useful
infrastructure, but a filter immediately before the hash probe can only save
later work.
2. **Reader integration — the goal of this issue:** propagate or safely
translate that filter into the native reader before relevant reads/decoding
occur. Demonstrate real row-group skipping and/or reduced decoding, rather than
treating a post-scan row filter as scan pushdown.
Start with native Parquet and the shared Parquet machinery used by our
internal Spark/Delta-authoritative reader. Keep Spark and Delta authoritative
for snapshots, selected files, splits, deletion vectors, and
logical-to-physical column mapping. Executor-local pruning should only
eliminate work within that authorized input; it must not introduce a second
Delta snapshot planner or change Spark task partitioning.
Do not assume that running every DataFusion optimizer rule is safe: rewrites
can change operator structure, break native-to-Spark metric mapping, or violate
Spark partition/aggregation contracts. Evaluate a targeted pass or explicit
producer/consumer wiring, and document where pushdown must stop.
### Additional context
_No response_
--
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]
