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:
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:
- 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.
- 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
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:
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 requiress_item_skto 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:
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