[GLUTEN-12013][VL] Fix bloom-filter bytes corruption on whole-stage AQE fallback

`BloomFilterMightContainJointRewriteRule` previously rewrote every
`BloomFilterAggregate` it encountered, including standalone usages such
as `DataFrame.stat.bloomFilter()`.  That API collects the aggregate
output bytes and passes them directly to `BloomFilter.readFrom()`, which
expects Spark-native format; receiving Velox-format bytes caused
`java.io.IOException: Unexpected Bloom filter version number` (surfaced
as a CI failure in `GlutenDataFrameStatSuite - Bloom filter`).

Fix: only rewrite `BloomFilterAggregate` when it appears inside the
`ScalarSubquery` of a `BloomFilterMightContain`.  Standalone aggregates
are left untouched so that collected bytes remain in Spark-native format.

Add a regression test (`GlutenBloomFilterFallbackSuite`) to guard
against reintroducing this regression.

Local test results (Spark 4.0, Velox backend):
- GlutenDataFrameStatSuite           : 25/25 passed (was failing)
- GlutenBloomFilterFallbackSuite     :  4/4  passed
- GlutenBloomFilterAggregateQuerySuite: 14/14 passed
- GlutenInjectRuntimeFilterSuite     : 13/13 passed

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
+import org.apache.spark.sql.catalyst.expressions.{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
@@ -35,6 +35,11 @@ import org.apache.spark.sql.catalyst.rules.Rule
  * an individual stage fallback to a whole-stage AQE fallback. This guarantees that both sides of
  * the bloom-filter pair always produce and consume the same byte format, regardless of whether
  * 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.
  */
 case class BloomFilterMightContainJointRewriteRule(spark: SparkSession)
   extends Rule[LogicalPlan] {
@@ -44,13 +49,17 @@ case class BloomFilterMightContainJointRewriteRule(spark: SparkSession)
       return plan
     }
     plan.transformAllExpressions {
-      case aggExpr @ AggregateExpression(b: BloomFilterAggregate, _, _, _, _) =>
-        aggExpr.copy(aggregateFunction = VeloxBloomFilterAggregate(
-          b.child,
-          b.estimatedNumItemsExpression,
-          b.numBitsExpression,
-          b.mutableAggBufferOffset,
-          b.inputAggBufferOffset))
+      case BloomFilterMightContain(subq: ScalarSubquery, v) =>
+        val rewrittenPlan = subq.plan.transformAllExpressions {
+          case ae @ AggregateExpression(b: BloomFilterAggregate, _, _, _, _) =>
+            ae.copy(aggregateFunction = VeloxBloomFilterAggregate(
+              b.child,
+              b.estimatedNumItemsExpression,
+              b.numBitsExpression,
+              b.mutableAggBufferOffset,
+              b.inputAggBufferOffset))
+        }
+        VeloxBloomFilterMightContain(subq.withNewPlan(rewrittenPlan), v)
       case BloomFilterMightContain(bf, v) =>
         VeloxBloomFilterMightContain(bf, v)
     }

gluten-ut/test/src/test/scala/org/apache/gluten/sql/GlutenBloomFilterFallbackSuite.scala

@@ -155,6 +155,23 @@ class GlutenBloomFilterFallbackSuite extends WholeStageTransformerSuite {
     }
   }
 
+  // GLUTEN-12013: DataFrame.stat.bloomFilter() must not be affected by the optimizer rule.
+  // The rule must only rewrite BloomFilterAggregate inside a BloomFilterMightContain subquery.
+  // A standalone BloomFilterAggregate (as used here) must remain vanilla so that the collected
+  // bytes are in Spark-native format and BloomFilter.readFrom() succeeds.
+  test("GLUTEN-12013: DataFrame.stat.bloomFilter() produces Spark-readable bytes") {
+    if (BackendsApiManager.getSettings.requireBloomFilterAggMightContainJointFallback()) {
+      val table = "bloom_filter_stat_test"
+      withTempView(table) {
+        (1L to 1000L).toDF("col").createOrReplaceTempView(table)
+        // Must not throw: java.io.IOException: Unexpected Bloom filter version number
+        val bf = spark.table(table).stat.bloomFilter("col", 1000L, 0.01)
+        // Bloom filters have no false negatives: every inserted value must be present.
+        assert(bf.mightContainLong(500L), "Expected 500 to be in bloom filter")
+      }
+    }
+  }
+
   // GLUTEN-12013: native bloom filter disabled -- early-exit path of the optimizer rule.
   // When spark.gluten.sql.native.bloomFilter=false the rule returns the plan unchanged.
   // BloomFilterAggregate / BloomFilterMightContain remain as vanilla Spark expressions and