sunchao opened a new issue, #5965: URL: https://github.com/apache/datafusion-comet/issues/5965
## What is the problem the feature request solves? Spark-generated runtime Bloom predicates can reject many rows from a large fact-table scan, but Comet currently evaluates those predicates above the native Parquet reader. Forwarding an eligible predicate into the reader would let it reject rows before decoding and materializing the remaining projected columns. This issue targets **DataSource V1 native Parquet scans** and an existing Spark-generated Bloom predicate directly above the scan. It is an optimization gap; the existing residual filter and join preserve query results. ### Verified current behavior Source inspection used Comet [`a8e8157e`](https://github.com/apache/datafusion-comet/commit/a8e8157ead46e7971a72049e9a4534c9e6f2b5aa), with fresh Spark-side runtime checks on OSS Spark **4.0.4 and 4.1.3**: 1. Spark's `InjectRuntimeFilter` creates a scalar subquery containing `BloomFilterAggregate(XxHash64(key))` and a `BloomFilterMightContain` predicate on the application side. 2. [`FileSourceStrategy`](https://github.com/apache/spark/blob/c7d67e3f5d4c9d88a480367b44fc54d26adf99ab/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala#L191-L203) explicitly excludes that Bloom predicate from the file scan's `dataFilters`, retaining it in a filter above the scan. 3. [`CometNativeScanExec`](https://github.com/apache/datafusion-comet/blob/a8e8157ead46e7971a72049e9a4534c9e6f2b5aa/spark/src/main/scala/org/apache/spark/sql/comet/CometNativeScanExec.scala#L372-L385) copies the scan's existing `dataFilters`. Its [execution-time scalar-subquery resolver](https://github.com/apache/datafusion-comet/blob/a8e8157ead46e7971a72049e9a4534c9e6f2b5aa/spark/src/main/scala/org/apache/spark/sql/comet/CometNativeScanExec.scala#L181-L212) only processes those attached filters. 4. The parent predicate becomes a separate native [`FilterExec`](https://github.com/apache/datafusion-comet/blob/a8e8157ead46e7971a72049e9a4534c9e6f2b5aa/native/core/src/execution/planner.rs#L1299-L1318). Comet [executes the constructed plan directly](https://github.com/apache/datafusion-comet/blob/a8e8157ead46e7971a72049e9a4534c9e6f2b5aa/native/core/src/execution/jni_api.rs#L948-L968), without a later physical optimizer pass that forwards this predicate into the reader. Enabling `spark.comet.parquet.rowFilterPushdown.enabled` therefore does not by itself restore this missing handoff. ### Spark-side reproduction Run this fixture in a fresh OSS PySpark session without the Comet extension. The small-data thresholds intentionally make Spark generate a runtime Bloom predicate; this is not an all-default configuration. ```python import tempfile from pyspark.sql import SparkSession spark = (SparkSession.builder.master("local[2]") .config("spark.sql.shuffle.partitions", "2") .config("spark.sql.adaptive.enabled", "false") .config("spark.sql.autoBroadcastJoinThreshold", "-1") .config("spark.sql.optimizer.dynamicPartitionPruning.enabled", "false") .config("spark.sql.optimizer.runtime.bloomFilter.enabled", "true") .config("spark.sql.optimizer.runtime.bloomFilter.applicationSideScanSizeThreshold", "0") .config("spark.sql.optimizer.runtime.bloomFilter.creationSideThreshold", "104857600") .config("spark.sql.sources.useV1SourceList", "parquet") .config("spark.sql.parquet.filterPushdown", "true") .getOrCreate()) root = tempfile.mkdtemp(prefix="runtime-bloom-parquet-") spark.range(100000).selectExpr("id AS k", "id % 31 AS value").write.parquet(root + "/fact") spark.range(1000).selectExpr("id AS k", "id % 10 AS flag").write.parquet(root + "/dim") spark.read.parquet(root + "/fact").createOrReplaceTempView("fact") spark.read.parquet(root + "/dim").createOrReplaceTempView("dim") query = spark.sql(""" SELECT sum(f.value) AS result FROM fact f JOIN dim d ON f.k = d.k WHERE d.flag = 1 """) query.explain(True) assert query.collect()[0]["result"] == 1488 ``` On both tested Spark versions, the fact-side physical filter contains `might_contain(...)`, while the fact scan's `DataFilters` contains only `isnotnull(k)`. A separate full-result control selecting and ordering all matching `(k, value)` rows returned **100 identical rows with Bloom generation enabled and disabled**. These runtime checks exercised Spark's generated input plan. The current Comet reader-handoff conclusion is source-backed; native Comet execution and reader counters remain to be validated by the implementation. ## Describe the potential solution Add a narrow, initially opt-in Comet forwarding rule: 1. Identify supported Spark-generated Bloom conjuncts directly above a native V1 Parquet scan. Copy eligible predicates into that scan's data filters while preserving the original residual filter and exact join. 2. Preserve the existing Bloom-producing subquery and Spark's preparation/wait/update lifecycle. Resolve its binary result at execution-time scan serialization, rather than evaluating it during planning or independently recomputing it for every scan task. 3. Preserve exact key bindings, supported data types, hash seed, null behavior, and Bloom serialization. Avoid moving the predicate through arbitrary projections, exchanges, or unverified expression boundaries. Copied/reused scans must have correct payload identities and must not reuse stale resolved filters. 4. Reuse the existing Bloom expression serde, native `might_contain` implementation, and [Parquet predicate attachment](https://github.com/apache/datafusion-comet/blob/a8e8157ead46e7971a72049e9a4534c9e6f2b5aa/native/core/src/parquet/parquet_exec.rs#L200-L221). DataFusion's [row-filter path](https://github.com/apache/datafusion/blob/55.1.0/datafusion/datasource-parquet/src/row_filter.rs#L25-L55) evaluates predicate columns and applies row selection to subsequent decoding. The initial implementation should be able to reuse this machinery. 5. Honor Spark's Parquet filter-pushdown setting and Comet's native row-filter setting. Unsupported cases retain the existing execution path. Preserve Spark's file/split planning and error behavior. ### Acceptance criteria - [ ] Use genuine optimizer-generated Bloom predicates to verify attachment to the native reader and execution of reader-level filtering, beyond an above-scan `FilterExec`. - [ ] Compare full results with Spark and forwarding-disabled Comet; cover null/empty inputs, duplicate keys, supported key/hash types, schema evolution, and unsupported fallback. - [ ] Cover AQE on/off, multiple/reused scalar subqueries, copied scans, repeated actions, cancellation, failures, and combined DPP plus runtime-Bloom cases where both are eligible. - [ ] Verify that the same Bloom-producing subquery is reused, that scan preparation cannot deadlock on it, and that resolved payloads do not leak between executions. - [ ] Benchmark forwarding independently of reader pushdown: both off, reader pushdown only, and reader pushdown plus forwarding, using matched artifacts, inputs, resources, and complete output validation. - [ ] For eligible TPC-DS queries, report repeated query timings, reader rows filtered, decode/materialization CPU, actual storage bytes, shuffle bytes, memory, and overhead on low-rejection/narrow-column controls. Report total suite impact separately. ## Additional context The expected benefit is earlier rejection of rows whose non-key columns would otherwise be decoded or materialized. The existing above-scan Bloom already filters before shuffle in the targeted plan shape, so forwarding the same predicate does not inherently reduce shuffle or join input further. Benefits depend on Bloom selectivity, projected-column cost, and the scan's share of query time. Reader row-count reductions alone do not establish reduced storage I/O or a TPC-DS speedup. This complements the broad runtime-filter proposal #3053. Merged #4053 handles scalar subqueries already present in scan filters, and merged #5699 attaches native join-generated filters; neither implements this Spark-generated Bloom handoff. Parquet file-format Bloom indexes and dynamic file pruning are separate mechanisms. Other scan implementations, including native Delta, require their own qualification. -- 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]
