Spark: Initial support for 4.1.0

| Cause | Type | Category | Description | Affected Files |
|-------|------|----------|-------------|----------------|
| - | Feat | Feature | Introduce Spark41Shims and update build configuration to support Spark 4.1. | pom.xml<br>shims/pom.xml<br>shims/spark41/pom.xml<br>shims/spark41/.../META-INF/services/org.apache.gluten.sql.shims.SparkShimProvider<br>shims/spark41/.../spark41/Spark41Shims.scala<br>shims/spark41/.../spark41/SparkShimProvider.scala |
| [#51477](apache/spark#51477) | Fix | Compatibility | Use class name instead of class object for streaming call detection to ensure Spark 4.1 compatibility. | gluten-core/.../caller/CallerInfo.scala |
| [#50852](apache/spark#50852) | Fix | Compatibility | Add printOutputColumns parameter to generateTreeString methods | shims/spark41/.../GenerateTreeStringShim.scala |
| [#51775](apache/spark#51775) | Fix | Compatibility | Remove unused MDC import in FileSourceScanExecShim.scala | shims/spark41/.../FileSourceScanExecShim.scala |
| [#51979](apache/spark#51979) | Fix | Compatibility | Add missing StoragePartitionJoinParams import in BatchScanExecShim and AbstractBatchScanExec | shims/spark41/.../v2/AbstractBatchScanExec.scala<br>shims/spark41/.../v2/BatchScanExecShim.scala |
| [#51302](apache/spark#51302) | Fix | Compatibility | Remove TimeAdd from ExpressionConverter and ExpressionMappings for test | gluten-substrait/.../ExpressionConverter.scala<br>gluten-substrait/.../ExpressionMappings.scala |
| [#50598](apache/spark#50598) | Fix | Compatibility | Adapt to QueryExecution.createSparkPlan interface change | gluten-substrait/.../GlutenImplicits.scala<br>shims/spark\*/.../shims/spark\*/Spark*Shims.scala |
| [#52599](apache/spark#52599) | Fix | Compatibility | Adapt to DataSourceV2Relation interface change | backends-velox/.../ArrowConvertorRule.scala |
| [#52384](apache/spark#52384) | Fix | Compatibility | Using new interface of ParquetFooterReader | backends-velox/.../ParquetMetadataUtils.scala<br>gluten-ut/spark40/.../parquet/GlutenParquetRowIndexSuite.scala<br>shims/spark*/.../parquet/ParquetFooterReaderShim.scala |
| [#52509](apache/spark#52509) | Fix | Build | Update Scala version to 2.13.17 in pom.xml to fix `java.lang.NoSuchMethodError: 'java.lang.String scala.util.hashing.MurmurHash3$.caseClassHash$default$2()'` | pom.xml |
| - | Fix | Test | Refactor Spark version checks in VeloxHashJoinSuite to improve readability and maintainability | backends-velox/.../VeloxHashJoinSuite.scala |
| [#50849](apache/spark#50849) | Fix | Test | Fix MiscOperatorSuite to support OneRowRelationExec plan Spark 4.1 | backends-velox/.../MiscOperatorSuite.scala |
| [#52723](apache/spark#52723) | Fix | Compatibility | Add GeographyVal and GeometryVal support in ColumnarArrayShim | shims/spark41/.../vectorized/ColumnarArrayShim.java |
| [#48470](apache/spark#48470) | 4.1.0 | Exclude | Exclude split test in VeloxStringFunctionsSuite | backends-velox/.../VeloxStringFunctionsSuite.scala |
| [#51259](apache/spark#51259) | 4.1.0 | Exclude | Only Run ArrowEvalPythonExecSuite tests up to Spark 4.0， we need update ci python to 3.10 | backends-velox/.../python/ArrowEvalPythonExecSuite.scala |

Author: baibaichen

.github/workflows/util/install-spark-resources.sh

@@ -119,6 +119,11 @@ case "$1" in
     cd ${INSTALL_DIR} && \
     install_spark "4.0.1" "3" "2.12"
     ;;
+4.1)
+    # Spark-4.x, scala 2.12 // using 2.12 as a hack as 4.0 does not have 2.13 suffix
+    cd ${INSTALL_DIR} && \
+    install_spark "4.1.0" "3" "2.12"
+    ;;
 *)
     echo "Spark version is expected to be specified."
     exit 1

.github/workflows/velox_backend_x86.yml

@@ -1482,3 +1482,109 @@ jobs:
             **/target/*.log
             **/gluten-ut/**/hs_err_*.log
             **/gluten-ut/**/core.*
+
+  spark-test-spark41:
+    needs: build-native-lib-centos-7
+    runs-on: ubuntu-22.04
+    env:
+      SPARK_TESTING: true
+    container: apache/gluten:centos-8-jdk17
+    steps:
+      - uses: actions/checkout@v2
+      - name: Download All Artifacts
+        uses: actions/download-artifact@v4
+        with:
+          name: velox-native-lib-centos-7-${{github.sha}}
+          path: ./cpp/build/releases
+      - name: Download Arrow Jars
+        uses: actions/download-artifact@v4
+        with:
+          name: arrow-jars-centos-7-${{github.sha}}
+          path: /root/.m2/repository/org/apache/arrow/
+      - name: Prepare
+        run: |
+          dnf module -y install python39 && \
+          alternatives --set python3 /usr/bin/python3.9 && \
+          pip3 install setuptools==77.0.3 && \
+          pip3 install pyspark==3.5.5 cython && \
+          pip3 install pandas==2.2.3 pyarrow==20.0.0
+      - name: Prepare Spark Resources for Spark 4.1.0 #TODO remove after image update
+        run: |
+          rm -rf /opt/shims/spark41
+          bash .github/workflows/util/install-spark-resources.sh 4.1
+          mv /opt/shims/spark41/spark_home/assembly/target/scala-2.12 /opt/shims/spark41/spark_home/assembly/target/scala-2.13
+      - name: Build and Run unit test for Spark 4.1.0 with scala-2.13 (other tests)
+        run: |
+          cd $GITHUB_WORKSPACE/
+          export SPARK_SCALA_VERSION=2.13
+          yum install -y java-17-openjdk-devel
+          export JAVA_HOME=/usr/lib/jvm/java-17-openjdk
+          export PATH=$JAVA_HOME/bin:$PATH
+          java -version
+          $MVN_CMD clean test -Pspark-4.1 -Pscala-2.13 -Pjava-17 -Pbackends-velox \
+          -Pspark-ut -DargLine="-Dspark.test.home=/opt/shims/spark41/spark_home/" \
+          -DtagsToExclude=org.apache.spark.tags.ExtendedSQLTest,org.apache.gluten.tags.UDFTest,org.apache.gluten.tags.EnhancedFeaturesTest,org.apache.gluten.tags.SkipTest
+      - name: Upload test report
+        if: always()
+        uses: actions/upload-artifact@v4
+        with:
+          name: ${{ github.job }}-report
+          path: '**/surefire-reports/TEST-*.xml'
+      - name: Upload unit tests log files
+        if: ${{ !success() }}
+        uses: actions/upload-artifact@v4
+        with:
+          name: ${{ github.job }}-test-log
+          path: |
+            **/target/*.log
+            **/gluten-ut/**/hs_err_*.log
+            **/gluten-ut/**/core.*
+
+  spark-test-spark41-slow:
+    needs: build-native-lib-centos-7
+    runs-on: ubuntu-22.04
+    env:
+      SPARK_TESTING: true
+    container: apache/gluten:centos-8-jdk17
+    steps:
+      - uses: actions/checkout@v2
+      - name: Download All Artifacts
+        uses: actions/download-artifact@v4
+        with:
+          name: velox-native-lib-centos-7-${{github.sha}}
+          path: ./cpp/build/releases
+      - name: Download Arrow Jars
+        uses: actions/download-artifact@v4
+        with:
+          name: arrow-jars-centos-7-${{github.sha}}
+          path: /root/.m2/repository/org/apache/arrow/
+      - name: Prepare Spark Resources for Spark 4.1.0 #TODO remove after image update
+        run: |
+          rm -rf /opt/shims/spark41
+          bash .github/workflows/util/install-spark-resources.sh 4.1
+          mv /opt/shims/spark41/spark_home/assembly/target/scala-2.12 /opt/shims/spark41/spark_home/assembly/target/scala-2.13
+      - name: Build and Run unit test for Spark 4.0 (slow tests)
+        run: |
+          cd $GITHUB_WORKSPACE/
+          yum install -y java-17-openjdk-devel
+          export JAVA_HOME=/usr/lib/jvm/java-17-openjdk
+          export PATH=$JAVA_HOME/bin:$PATH
+          java -version
+          $MVN_CMD clean test -Pspark-4.1 -Pscala-2.13 -Pjava-17 -Pbackends-velox -Pspark-ut \
+          -DargLine="-Dspark.test.home=/opt/shims/spark41/spark_home/" \
+          -DtagsToInclude=org.apache.spark.tags.ExtendedSQLTest
+      - name: Upload test report
+        if: always()
+        uses: actions/upload-artifact@v4
+        with:
+          name: ${{ github.job }}-report
+          path: '**/surefire-reports/TEST-*.xml'
+      - name: Upload unit tests log files
+        if: ${{ !success() }}
+        uses: actions/upload-artifact@v4
+        with:
+          name: ${{ github.job }}-test-log
+          path: |
+            **/target/*.log
+            **/gluten-ut/**/hs_err_*.log
+            **/gluten-ut/**/core.*

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

@@ -38,6 +38,24 @@ import java.nio.charset.StandardCharsets
 
 import scala.collection.convert.ImplicitConversions.`map AsScala`
 
+/**
+ * Extracts a CSVTable from a DataSourceV2Relation.
+ *
+ * Only the table variable of DataSourceV2Relation is accessed to improve compatibility across
+ * different Spark versions.
+ * @since Spark
+ *   4.1
+ */
+private object CSVTableExtractor {
+  def unapply(relation: DataSourceV2Relation): Option[(DataSourceV2Relation, CSVTable)] = {
+    relation.table match {
+      case t: CSVTable =>
+        Some((relation, t))
+      case _ => None
+    }
+  }
+}
+
 @Experimental
 case class ArrowConvertorRule(session: SparkSession) extends Rule[LogicalPlan] {
   override def apply(plan: LogicalPlan): LogicalPlan = {
@@ -56,25 +74,15 @@ case class ArrowConvertorRule(session: SparkSession) extends Rule[LogicalPlan] {
             l.copy(relation = r.copy(fileFormat = new ArrowCSVFileFormat(csvOptions))(session))
           case _ => l
         }
-      case d @ DataSourceV2Relation(
-            t @ CSVTable(
-              name,
-              sparkSession,
-              options,
-              paths,
-              userSpecifiedSchema,
-              fallbackFileFormat),
-            _,
-            _,
-            _,
-            _) if validate(session, t.dataSchema, options.asCaseSensitiveMap().toMap) =>
+      case CSVTableExtractor(d, t)
+          if validate(session, t.dataSchema, t.options.asCaseSensitiveMap().toMap) =>
         d.copy(table = ArrowCSVTable(
-          "arrow" + name,
-          sparkSession,
-          options,
-          paths,
-          userSpecifiedSchema,
-          fallbackFileFormat))
+          "arrow" + t.name,
+          t.sparkSession,
+          t.options,
+          t.paths,
+          t.userSpecifiedSchema,
+          t.fallbackFileFormat))
       case r =>
         r
     }

backends-velox/src/main/scala/org/apache/gluten/utils/ParquetMetadataUtils.scala

@@ -21,7 +21,7 @@ import org.apache.gluten.sql.shims.SparkShimLoader
 
 import org.apache.spark.internal.Logging
 import org.apache.spark.sql.execution.datasources.DataSourceUtils
-import org.apache.spark.sql.execution.datasources.parquet.{ParquetFooterReader, ParquetOptions}
+import org.apache.spark.sql.execution.datasources.parquet.{ParquetFooterReaderShim, ParquetOptions}
 
 import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.fs.{FileSystem, LocatedFileStatus, Path}
@@ -135,7 +135,7 @@ object ParquetMetadataUtils extends Logging {
       parquetOptions: ParquetOptions): Option[String] = {
     val footer =
       try {
-        ParquetFooterReader.readFooter(conf, fileStatus, ParquetMetadataConverter.NO_FILTER)
+        ParquetFooterReaderShim.readFooter(conf, fileStatus, ParquetMetadataConverter.NO_FILTER)
       } catch {
         case e: Exception if ExceptionUtils.hasCause(e, classOf[ParquetCryptoRuntimeException]) =>
           return Some("Encrypted Parquet footer detected.")

backends-velox/src/test/scala/org/apache/gluten/execution/MiscOperatorSuite.scala

@@ -753,7 +753,11 @@ class MiscOperatorSuite extends VeloxWholeStageTransformerSuite with AdaptiveSpa
     val df = sql("SELECT 1")
     checkAnswer(df, Row(1))
     val plan = df.queryExecution.executedPlan
-    assert(plan.find(_.isInstanceOf[RDDScanExec]).isDefined)
+    if (isSparkVersionGE("4.1")) {
+      assert(plan.find(_.getClass.getSimpleName == "OneRowRelationExec").isDefined)
+    } else {
+      assert(plan.find(_.isInstanceOf[RDDScanExec]).isDefined)
+    }
     assert(plan.find(_.isInstanceOf[ProjectExecTransformer]).isDefined)
     assert(plan.find(_.isInstanceOf[RowToVeloxColumnarExec]).isDefined)
   }

backends-velox/src/test/scala/org/apache/gluten/execution/VeloxHashJoinSuite.scala

@@ -92,12 +92,9 @@ class VeloxHashJoinSuite extends VeloxWholeStageTransformerSuite {
 
       // The computing is combined into one single whole stage transformer.
       val wholeStages = plan.collect { case wst: WholeStageTransformer => wst }
-      if (SparkShimLoader.getSparkVersion.startsWith("3.2.")) {
+      if (isSparkVersionLE("3.2")) {
         assert(wholeStages.length == 1)
-      } else if (
-        SparkShimLoader.getSparkVersion.startsWith("3.5.") ||
-        SparkShimLoader.getSparkVersion.startsWith("4.0.")
-      ) {
+      } else if (isSparkVersionGE("3.5")) {
         assert(wholeStages.length == 5)
       } else {
         assert(wholeStages.length == 3)

backends-velox/src/test/scala/org/apache/gluten/execution/VeloxStringFunctionsSuite.scala

@@ -544,7 +544,8 @@ class VeloxStringFunctionsSuite extends VeloxWholeStageTransformerSuite {
         s"from $LINEITEM_TABLE limit 5") { _ => }
   }
 
-  testWithMinSparkVersion("split", "3.4") {
+  // TODO: fix on spark-4.1
+  testWithSpecifiedSparkVersion("split", "3.4", "3.5") {
     runQueryAndCompare(
       s"select l_orderkey, l_comment, split(l_comment, '') " +
         s"from $LINEITEM_TABLE limit 5") {

backends-velox/src/test/scala/org/apache/gluten/execution/python/ArrowEvalPythonExecSuite.scala

@@ -39,7 +39,8 @@ class ArrowEvalPythonExecSuite extends WholeStageTransformerSuite {
       .set("spark.executor.cores", "1")
   }
 
-  test("arrow_udf test: without projection") {
+  // TODO: fix on spark-4.1
+  testWithMaxSparkVersion("arrow_udf test: without projection", "4.0") {
     lazy val base =
       Seq(("1", 1), ("1", 2), ("2", 1), ("2", 2), ("3", 1), ("3", 2), ("0", 1), ("3", 0))
         .toDF("a", "b")
@@ -59,7 +60,8 @@ class ArrowEvalPythonExecSuite extends WholeStageTransformerSuite {
     checkAnswer(df2, expected)
   }
 
-  test("arrow_udf test: with unrelated projection") {
+  // TODO: fix on spark-4.1
+  testWithMaxSparkVersion("arrow_udf test: with unrelated projection", "4.0") {
     lazy val base =
       Seq(("1", 1), ("1", 2), ("2", 1), ("2", 2), ("3", 1), ("3", 2), ("0", 1), ("3", 0))
         .toDF("a", "b")

gluten-core/src/main/scala/org/apache/gluten/extension/caller/CallerInfo.scala

@@ -18,7 +18,7 @@ package org.apache.gluten.extension.caller
 
 import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec
 import org.apache.spark.sql.execution.columnar.InMemoryRelation
-import org.apache.spark.sql.execution.streaming.StreamExecution
+import org.apache.spark.util.SparkVersionUtil
 
 /**
  * Helper API that stores information about the call site of the columnar rule. Specific columnar
@@ -70,7 +70,12 @@ object CallerInfo {
   }
 
   private def inStreamingCall(stack: Seq[StackTraceElement]): Boolean = {
-    stack.exists(_.getClassName.equals(StreamExecution.getClass.getName.split('$').head))
+    val streamName = if (SparkVersionUtil.gteSpark41) {
+      "org.apache.spark.sql.execution.streaming.runtime.StreamExecution"
+    } else {
+      "org.apache.spark.sql.execution.streaming.StreamExecution"
+    }
+    stack.exists(_.getClassName.equals(streamName))
   }
 
   private def inBloomFilterStatFunctionCall(stack: Seq[StackTraceElement]): Boolean = {

gluten-core/src/main/scala/org/apache/spark/util/SparkVersionUtil.scala

@@ -25,6 +25,7 @@ object SparkVersionUtil {
   val gteSpark33: Boolean = comparedWithSpark33 >= 0
   val gteSpark35: Boolean = comparedWithSpark35 >= 0
   val gteSpark40: Boolean = compareMajorMinorVersion((4, 0)) >= 0
+  val gteSpark41: Boolean = compareMajorMinorVersion((4, 1)) >= 0
 
   // Returns X. X < 0 if one < other, x == 0 if one == other, x > 0 if one > other.
   def compareMajorMinorVersion(other: (Int, Int)): Int = {