Skip tests for Spark 4.1

Author: manuzhang

native/spark-expr/src/bloom_filter/spark_bloom_filter.rs

@@ -24,7 +24,6 @@ use crate::bloom_filter::spark_bit_array::SparkBitArray;
 use crate::hash_funcs::murmur3::spark_compatible_murmur3_hash;
 
 const SPARK_BLOOM_FILTER_VERSION_1: i32 = 1;
-const SPARK_BLOOM_FILTER_VERSION_2: i32 = 2;
 
 /// A Bloom filter implementation that simulates the behavior of Spark's BloomFilter.
 /// It's not a complete implementation of Spark's BloomFilter, but just add the minimum
@@ -61,9 +60,9 @@ impl From<&[u8]> for SparkBloomFilter {
         let mut offset = 0;
         let version = read_num_be_bytes!(i32, 4, buf[offset..]);
         offset += 4;
-        assert!(
-            version == SPARK_BLOOM_FILTER_VERSION_1 || version == SPARK_BLOOM_FILTER_VERSION_2,
-            "Unsupported BloomFilter version: {version}, expecting version: {SPARK_BLOOM_FILTER_VERSION_1} or {SPARK_BLOOM_FILTER_VERSION_2}"
+        assert_eq!(
+            version, SPARK_BLOOM_FILTER_VERSION_1,
+            "Unsupported BloomFilter version: {version}, expecting version: {SPARK_BLOOM_FILTER_VERSION_1}"
         );
         let num_hash_functions = read_num_be_bytes!(i32, 4, buf[offset..]);
         offset += 4;

spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala

@@ -132,6 +132,10 @@ object CometSparkSessionExtensions extends Logging {
     org.apache.spark.SPARK_VERSION >= "4.0"
   }
 
+  def isSpark41Plus: Boolean = {
+    org.apache.spark.SPARK_VERSION >= "4.1"
+  }
+
   /**
    * Whether we should override Spark memory configuration for Comet. This only returns true when
    * Comet native execution is enabled and/or Comet shuffle is enabled and Comet doesn't use

spark/src/test/scala/org/apache/comet/exec/CometExec3_4PlusSuite.scala

@@ -33,6 +33,7 @@ import org.apache.spark.sql.functions.{col, lit}
 import org.apache.spark.util.sketch.BloomFilter
 
 import org.apache.comet.CometConf
+import org.apache.comet.CometSparkSessionExtensions.isSpark41Plus
 
 /**
  * This test suite contains tests for only Spark 3.4+.
@@ -156,6 +157,8 @@ class CometExec3_4PlusSuite extends CometTestBase {
   }
 
   test("test BloomFilterMightContain from random input") {
+    // TODO fix for Spark 4.1
+    assume(!isSpark41Plus)
     val (longs, bfBytes) = bloomFilterFromRandomInput(10000, 10000)
     val table = "test"
 

spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala

@@ -48,7 +48,7 @@ import org.apache.spark.sql.internal.SQLConf.SESSION_LOCAL_TIMEZONE
 import org.apache.spark.unsafe.types.UTF8String
 
 import org.apache.comet.{CometConf, CometExecIterator, ExtendedExplainInfo}
-import org.apache.comet.CometSparkSessionExtensions.{isSpark35Plus, isSpark40Plus}
+import org.apache.comet.CometSparkSessionExtensions.{isSpark35Plus, isSpark40Plus, isSpark41Plus}
 import org.apache.comet.serde.Config.ConfigMap
 import org.apache.comet.testing.{DataGenOptions, ParquetGenerator, SchemaGenOptions}
 
@@ -1123,6 +1123,8 @@ class CometExecSuite extends CometTestBase {
   }
 
   test("bloom_filter_agg") {
+    // TODO fix for Spark 4.1
+    assume(!isSpark41Plus)
     val funcId_bloom_filter_agg = new FunctionIdentifier("bloom_filter_agg")
     spark.sessionState.functionRegistry.registerFunction(
       funcId_bloom_filter_agg,