brijrajk commented on code in PR #12151: URL: https://github.com/apache/gluten/pull/12151#discussion_r3600614771
########## backends-velox/src/main/scala/org/apache/gluten/extension/RuntimeBloomFilterRewriteRule.scala: ########## @@ -0,0 +1,78 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.gluten.extension + +import org.apache.gluten.config.GlutenConfig +import org.apache.gluten.expression.VeloxBloomFilterMightContain +import org.apache.gluten.expression.aggregate.VeloxBloomFilterAggregate + +import org.apache.spark.sql.SparkSession +import org.apache.spark.sql.catalyst.expressions.{BloomFilterMightContain, XxHash64} +import org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression, BloomFilterAggregate} +import org.apache.spark.sql.catalyst.rules.Rule +import org.apache.spark.sql.execution.SparkPlan + +/** + * Physical pre-transform rule that rewrites runtime-filter bloom filters (the ones injected by + * Spark's `InjectRuntimeFilter` optimizer rule) to their Velox variants so they offload natively. + * + * Runtime bloom filters cannot be handled by [[BloomFilterMightContainJointRewriteRule]]: that rule + * is registered via `injectOptimizerRule`, which lands in Spark's Operator Optimization batch, + * while `InjectRuntimeFilter` runs in a later batch of `SparkOptimizer`. The runtime-filter + * expressions therefore do not exist yet when the logical rule fires, and a physical-level rewrite + * (as was always done before the logical rule was introduced) is required to keep + * `FilterExecTransformer` and the bloom-filter aggregate native. + * + * The rewrite is restricted to `InjectRuntimeFilter`'s exact expression shapes, which always wrap + * the key in [[XxHash64]] on both the producer and the consumer side: + * - producer: `bloom_filter_agg(xxhash64(key), ...)` -> `velox_bloom_filter_agg(...)` + * - consumer: `might_contain(bf, xxhash64(key))` -> `velox_might_contain(...)` + * + * Because each side is identifiable on its own, both are rewritten consistently to the Velox byte + * format (version=1) even when AQE compiles the bloom-filter subquery separately from the consuming + * filter stage. The `XxHash64` fingerprint also guarantees the other bloom-filter populations are + * never touched: + * - `DataFrame.stat.bloomFilter()` builds `bloom_filter_agg(col, ...)` on the raw column (no + * `XxHash64` wrapper) and deserializes the result with Spark's `BloomFilter.readFrom`, so its + * bytes must stay in Spark-native format. + * - User-facing `might_contain(<scalar subquery>, <value>)` pairs are already rewritten at the + * logical level by [[BloomFilterMightContainJointRewriteRule]] (the GLUTEN-12013 fix), making + * this rule a no-op for them. + * - Literal-value pairs (SPARK-54336) contain no `XxHash64` and stay fully vanilla. + */ +case class RuntimeBloomFilterRewriteRule(spark: SparkSession) extends Rule[SparkPlan] { Review Comment: @zhztheplayer @philo-he I prototyped the `injectFinal` idea and probed it at runtime instead of answering from theory. Findings below, including one correction to my own Jul 14 comment. ## injectFinal PoC results (zhztheplayer's Q1) Implemented as keep-preTransform-plus-add-final (a pure *move* would regress offload: validation needs the Velox expression before `HeuristicTransform`). Same `xxhash64` guards. | Probe | Configs | Before PoC | With PoC | | --- | --- | --- | --- | | D | `filter=false` + `wholeStage.fallback.threshold=1` | CRASH (velox, 1 vs. 0) | 10 rows, correct | | E | `wholeStage.fallback.threshold=1` only | CRASH (IOException 16777218) | **6 of 10 rows, silent wrong result** | | Guards (stat / SPARK-54336 / native-off) | | pass | pass | Probe E is deterministic (same missing keys across reruns). A bloom filter cannot produce false negatives, so the filter itself was corrupt. Strictly worse than the crash, so I have not pushed it. ## Root cause: capacity divergence (new finding -- corrected below) Probe E's plan shows a **phase split**: native partial (`FlushableHashAggregateExecTransformer[VeloxBloomFilterAggregate:Partial]`) feeding a reverted JVM final (`ObjectHashAggregateExec[VeloxBloomFilterAggregate:Final]`). Same byte format, but the two engines size the buffer differently. **Edit:** my original explanation here (different capacity *sources*, expression arg vs. session confs) was wrong -- see the correction a few comments down for the verified mechanism (identical args on both sides, two different sizing formulas, confirmed by rerunning in the container). The bottom line is unchanged: the sizes genuinely diverge and Velox's `BloomFilter::merge` guards the size match with `VELOX_DCHECK_EQ` (compiled out in release) then ORs the full range: silent corruption. **Correction to my Jul 14 comment:** probe E's original crash was this same phase split (vanilla final agg reading native partial buffers), not the consumer reverting; the consumer stayed a native `FilterExecTransformer` throughout. ## Why the logical rule stays (zhztheplayer's Q2) 1. The literal-vs-non-literal pair decision (SPARK-54336 vs GLUTEN-12013) needs whole-plan visibility; after `PlanSubqueries`, a producer-side `bloom_filter_agg(col)` alone is ambiguous, and there is no `xxhash64`-style fingerprint to disambiguate. 2. It bakes the rewrite into `originalPlan`, so the user-facing population is reversion-safe with zero patching (proven by the threshold=1/2 tests). ## The three options (incl. philo-he's proposals) | Option | Verdict | Notes | | --- | --- | --- | | `injectFinal` re-rewrite | Safe only after capacity alignment | Fix: JVM buffer sized from the same session confs as native, or adopt incoming size on first merge (`bits_.empty()` branch supports it). Also propose to Velox: promote the `DCHECK` to a user check. | | Fallback-policy block (philo-he) | Most robust short-term | Only option preventing both split kinds (producer/consumer AND partial/final); no capacity prerequisite. Layering solvable: pass a bloom predicate from `VeloxRuleApi` where `ExpandFallbackPolicy` is constructed. Cost: overrides user's fallback intent. | | Upstream Spark extension point (philo-he) | Best long-term | Deletes the physical rule; but does not cover the phase split by itself, so capacity alignment is needed regardless. Happy to draft the SPARK JIRA. | ## Proposed sequencing 1. Merge this PR as-is: fixes GLUTEN-12013 + SPARK-54336, restores native runtime filters, zero golden changes; the residual gap is default-off, fails loudly, and is byte-identical to main (same probes run against main's bloom code). 2. Follow-up 1: capacity alignment + a phase-split regression test (prerequisite). 3. Follow-up 2: close the reversion gap on top, via `injectFinal` or the policy predicate, your preference. 4. Long-term: SPARK JIRA for the extension point. Happy to file the follow-ups with the probe reproductions. Does this sequencing work for you both? -- 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]
