[GLUTEN-12013][VL] Guard optimizer rule against DPP-injected bloom filters

In Spark 4.1, the optimizer processes BloomFilterMightContainJointRewriteRule
after InjectRuntimeFilter, so the rule sees DPP-injected bloom filters
(which use xxhash64(col, seed) as their value expression).  Rewriting those
to VeloxBloomFilterMightContain caused FilterExecTransformer to fail Velox
validation and fall back to JVM, changing the physical plan structure and
breaking 19 TPC-DS plan-stability golden files across v1.4, v2.7, and
modified suites (q2, q10, q16, q24a, q24b, q32, q37, q40, q59, q69, q80,
q82, q85, q92, q94, q95, q10a, q64, q80a).

Fix: guard the first arm of transformAllExpressions with `v: Attribute`.
DPP bloom filters always hash the join key — xxhash64(col, seed) — so
their value is never a bare Attribute and the guard skips them cleanly.
User-facing bloom filters (might_contain(subquery, col)) pass a plain
column reference and continue to be rewritten correctly.

Also guard the catch-all arm against ScalarSubquery to be consistent.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: brijrajk

backends-velox/src/main/scala/org/apache/gluten/extension/BloomFilterMightContainJointRewriteRule.scala

@@ -21,7 +21,7 @@ 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, ScalarSubquery}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, BloomFilterMightContain, ScalarSubquery}
 import org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression, BloomFilterAggregate}
 import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
 import org.apache.spark.sql.catalyst.rules.Rule
@@ -37,9 +37,17 @@ import org.apache.spark.sql.catalyst.rules.Rule
  * stages fall back to JVM execution after AQE re-planning.
  *
  * `BloomFilterAggregate` is only rewritten when it appears inside the [[ScalarSubquery]] of a
- * [[BloomFilterMightContain]]. Standalone usages (in particular `DataFrame.stat.bloomFilter()`,
- * which collects bloom filter bytes and passes them to `BloomFilter.readFrom()`) are intentionally
- * left untouched so that the returned bytes remain in Spark-native format.
+ * [[BloomFilterMightContain]] whose value expression is a plain column reference ([[Attribute]]).
+ * This deliberately excludes:
+ *   - Standalone `BloomFilterAggregate` (e.g., `DataFrame.stat.bloomFilter()`) — so that the
+ *     returned bytes stay in Spark-native format.
+ *   - DPP-injected bloom filters from
+ *     [[org.apache.spark.sql.execution.dynamicpruning.PlanDynamicPruningFilters]] and Spark's
+ *     `InjectRuntimeFilter` — those always hash the join key with `xxhash64(col, seed)` before
+ *     passing it to `BloomFilterMightContain`, so their value expression is never a bare
+ *     [[Attribute]]. Rewriting them would cause `VeloxBloomFilterMightContain` to appear in a
+ *     `FilterExecTransformer`, which fails Velox expression validation and falls back to JVM,
+ *     changing the physical plan structure and breaking plan-stability golden files.
  */
 case class BloomFilterMightContainJointRewriteRule(spark: SparkSession)
   extends Rule[LogicalPlan] {
@@ -49,7 +57,11 @@ case class BloomFilterMightContainJointRewriteRule(spark: SparkSession)
       return plan
     }
     plan.transformAllExpressions {
-      case BloomFilterMightContain(subq: ScalarSubquery, v) =>
+      case BloomFilterMightContain(subq: ScalarSubquery, v: Attribute) =>
+        // Only rewrite when the tested value is a plain column reference (user-facing
+        // bloom filter). DPP-injected bloom filters always hash the join key —
+        // xxhash64(col, seed) — so their value expression is never a bare Attribute;
+        // they are intentionally left as vanilla BloomFilterMightContain.
         val rewrittenPlan = subq.plan.transformAllExpressions {
           case ae @ AggregateExpression(b: BloomFilterAggregate, _, _, _, _) =>
             ae.copy(aggregateFunction = VeloxBloomFilterAggregate(
@@ -60,7 +72,8 @@ case class BloomFilterMightContainJointRewriteRule(spark: SparkSession)
               b.inputAggBufferOffset))
         }
         VeloxBloomFilterMightContain(subq.withNewPlan(rewrittenPlan), v)
-      case BloomFilterMightContain(bf, v) =>
+      case BloomFilterMightContain(bf, v) if !bf.isInstanceOf[ScalarSubquery] =>
+        // Pre-computed literal bloom filter bytes — rewrite to consume Velox-format bytes.
         VeloxBloomFilterMightContain(bf, v)
     }
   }