[Proposal] Runtime Filters for DataFusion Comet

[Shekharrajak]

What is the problem the feature request solves?

Runtime Filters is a high-performance optimization that reduces I/O by 90% and improves query performance for join-heavy workloads. The feature filters data at scan time using lightweight data structures constructed during hash join build phases.

Runtime filters are lightweight data structures (IN sets, Min/Max bounds, Bloom filters) built during hash join build phases and pushed down to scan operators to filter data before reading from storage.

Example:

-- User writes standard SQL
SELECT o.*, c.name 
FROM orders o 
JOIN customers c ON o.customer_id = c.id 
WHERE c.country = 'USA';

1. Comet will detect join opportunity
2. Builds filter from customers table during join build
3. Applies filter to orders scan automatically
4. Query runs faster with less I/O

Filter Types

1. IN Filter (Small cardinality <1000)

2. Min/Max Filter (Numeric/date types)

   • Min and max bounds

3. Bloom Filter (Large cardinality, future)

   • Probabilistic data structure

The system should automatically selects the optimal filter type:

• Numeric/date → Min/Max Filter (most efficient)
• Small cardinality → IN Filter
• Large cardinality → Bloom Filter

Users should be able to control runtime filters via Spark configuration:

// Enable/disable runtime filters
spark.conf.set("spark.comet.runtimeFilter.enabled", true)

// Adjust thresholds
spark.conf.set("spark.comet.runtimeFilter.inFilterThreshold", 1000)
spark.conf.set("spark.comet.runtimeFilter.bloomFilterFpp", 0.01)

Describe the potential solution

No response

Additional context

No response

[Shekharrajak]

cc @andygrove @kazuyukitanimura @mbutrovich

What's your thought ? Please assign this to me to explore more and let me have some POC.

[mbutrovich]

We rely on Spark's planning and capabilities. One challenge with dynamic filters (relevant search terms: sideways information passing, LIP) becomes "how do you get plan data from one part of the plan to another within Spark's communication semantics?" Spark supports some of this with bloom filters (albeit above scans). Any dynamism that we introduce to query execution should work within the context of Spark's adaptivity, mostly AQE, DPP, and bloom filter injection (maybe we could push the later into the scan with Comet and that would be a decent improvement).

[Shekharrajak]

Thank you for the feedback. Instead of creating a new runtime filter system, we can push Spark's existing bloom filters down into Comet scans, as you suggested ("maybe we could push the later into the scan with Comet").

Comet's native Parquet reader could apply filters during scan, but currently doesn't receive bloom filters.

[coderfender]

I wonder how would we fit this with Dynamic Partition pruning in Spark ?

[wForget]

Spark already supports runtime filter injection. We should pay attention to whether this causes comet operators to fall back, and how to make the filters generated by Spark take effect in the comet native plan.

Additionally, for bloom filter, we may need to implement a new function to be compatible with scenarios where execution falls back to vanilla spark. See: https://github.com/apache/incubator-gluten/blob/60bf4b74269057e4a800546942b07e1b77411d25/backends-velox/src/main/scala/org/apache/gluten/expression/VeloxBloomFilterMightContain.scala#L36-L39

[Shekharrajak]

Leverage Spark's InjectRuntimeFilter - we can extract BloomFilterMightContain expressions that Spark's InjectRuntimeFilter rule already creates. These are already in the scan's dataFilters; we extract and push them down.

[wForget]

BloomFilterAggregate and BloomFilterMightContain seem to be supported #179 #987. @mbutrovich Does this mean that we already support runtime filters for comet?

[Shekharrajak]

Spark Plan (after InjectRuntimeFilter):
  Filter(BloomFilterMightContain(...))   =>  Comet can execute this 
    CometScanExec(orders)                => Reads ALL rows 
      └─ Native: Reads entire Parquet file
         └─ Then Filter operator filters rows (no I/O benefit)

We would like to support this :

Spark Plan (after InjectRuntimeFilter):
  Filter(BloomFilterMightContain(...))  => Created by Spark
    CometScanExec(orders, bloomFilters) => Extract bloom filters 
      └─ Native: Bloom filters pushed into ParquetSource 
         └─ Filters applied DURING scan → I/O reduction 

Please let me know if I am missing anything or you want me to explore in different direction.

[wForget]

@Shekharrajak I guess you may be confusing the BloomFilter generated by Runtime Filter with the Parquet Bloom Filter. The scenario you mentioned below requires enabling Parquet Bloom Filters: the Bloom filter is built at write time, and at query time the equality predicates in pushedDataFilters are used to filter out some row groups.

[Shekharrajak]

Trying to get some hands on and benchmark it - for deeper understanding of the execution flow : #3060 - please have a look, review & comment.