Spark: Initial support for 4.1-preview2

Author: manuzhang

.github/workflows/spark-ci.yml

@@ -72,17 +72,21 @@ jobs:
     strategy:
       matrix:
         jvm: [11, 17, 21]
-        spark: ['3.4', '3.5', '4.0']
+        spark: ['3.4', '3.5', '4.0', '4.1']
         scala: ['2.12', '2.13']
         exclude:
           # Spark 3.5 is the first version not failing on Java 21 (https://issues.apache.org/jira/browse/SPARK-42369)
           # Full Java 21 support is coming in Spark 4 (https://issues.apache.org/jira/browse/SPARK-43831)
           - jvm: 11
             spark: '4.0'
+          - jvm: 11
+            spark: '4.1'
           - jvm: 21
             spark: '3.4'
           - spark: '4.0'
             scala: '2.12'
+          - spark: '4.1'
+            scala: '2.12'
     env:
       SPARK_LOCAL_IP: localhost
     steps:

dev/stage-binaries.sh

@@ -20,7 +20,7 @@
 
 SCALA_VERSION=2.12
 FLINK_VERSIONS=1.20,2.0,2.1
-SPARK_VERSIONS=3.4,3.5,4.0
+SPARK_VERSIONS=3.4,3.5,4.0,4.1
 KAFKA_VERSIONS=3
 
 ./gradlew -Prelease -DscalaVersion=$SCALA_VERSION -DflinkVersions=$FLINK_VERSIONS -DsparkVersions=$SPARK_VERSIONS -DkafkaVersions=$KAFKA_VERSIONS publishApachePublicationToMavenRepository --no-parallel --no-configuration-cache

gradle.properties

@@ -18,8 +18,8 @@ jmhJsonOutputPath=build/reports/jmh/results.json
 jmhIncludeRegex=.*
 systemProp.defaultFlinkVersions=2.1
 systemProp.knownFlinkVersions=1.20,2.0,2.1
-systemProp.defaultSparkVersions=4.0
-systemProp.knownSparkVersions=3.4,3.5,4.0
+systemProp.defaultSparkVersions=4.1
+systemProp.knownSparkVersions=3.4,3.5,4.0,4.1
 systemProp.defaultKafkaVersions=3
 systemProp.knownKafkaVersions=3
 systemProp.defaultScalaVersion=2.12

gradle/libs.versions.toml

@@ -85,6 +85,7 @@ snowflake-jdbc = "3.26.1"
 spark34 = "3.4.4"
 spark35 = "3.5.7"
 spark40 = "4.0.1"
+spark41 = "4.1.0-preview2"
 sqlite-jdbc = "3.50.3.0"
 testcontainers = "1.21.3"
 tez08 = { strictly = "0.8.4"}  # see rich version usage explanation above
@@ -154,6 +155,7 @@ jackson214-bom = { module = "com.fasterxml.jackson:jackson-bom", version.ref =
 jackson215-bom = { module = "com.fasterxml.jackson:jackson-bom", version.ref = "jackson215" }
 jaxb-api = { module = "javax.xml.bind:jaxb-api", version.ref = "jaxb-api" }
 jaxb-runtime = { module = "org.glassfish.jaxb:jaxb-runtime", version.ref = "jaxb-runtime" }
+jetbrain-annotations = { module = "org.jetbrains:annotations", version = "17.0.0" }
 kafka-clients = { module = "org.apache.kafka:kafka-clients", version.ref = "kafka" }
 kafka-connect-api = { module = "org.apache.kafka:connect-api", version.ref = "kafka" }
 kafka-connect-json = { module = "org.apache.kafka:connect-json", version.ref = "kafka" }

jmh.gradle

@@ -53,6 +53,11 @@ if (sparkVersions.contains("4.0")) {
   jmhProjects.add(project(":iceberg-spark:iceberg-spark-extensions-4.0_2.13"))
 }
 
+if (sparkVersions.contains("4.1")) {
+    jmhProjects.add(project(":iceberg-spark:iceberg-spark-4.1_2.13"))
+    jmhProjects.add(project(":iceberg-spark:iceberg-spark-extensions-4.1_2.13"))
+}
+
 configure(jmhProjects) {
   apply plugin: 'me.champeau.jmh'
   apply plugin: 'io.morethan.jmhreport'

settings.gradle

@@ -175,6 +175,18 @@ if (sparkVersions.contains("4.0")) {
   project(":iceberg-spark:spark-runtime-4.0_2.13").name = "iceberg-spark-runtime-4.0_2.13"
 }
 
+if (sparkVersions.contains("4.1")) {
+    include ":iceberg-spark:spark-4.1_2.13"
+    include ":iceberg-spark:spark-extensions-4.1_2.13"
+    include ":iceberg-spark:spark-runtime-4.1_2.13"
+    project(":iceberg-spark:spark-4.1_2.13").projectDir = file('spark/v4.1/spark')
+    project(":iceberg-spark:spark-4.1_2.13").name = "iceberg-spark-4.1_2.13"
+    project(":iceberg-spark:spark-extensions-4.1_2.13").projectDir = file('spark/v4.1/spark-extensions')
+    project(":iceberg-spark:spark-extensions-4.1_2.13").name = "iceberg-spark-extensions-4.1_2.13"
+    project(":iceberg-spark:spark-runtime-4.1_2.13").projectDir = file('spark/v4.1/spark-runtime')
+    project(":iceberg-spark:spark-runtime-4.1_2.13").name = "iceberg-spark-runtime-4.1_2.13"
+}
+
 if (kafkaVersions.contains("3")) {
   include 'kafka-connect'
   project(':kafka-connect').name = 'iceberg-kafka-connect'

spark/build.gradle

@@ -31,3 +31,7 @@ if (sparkVersions.contains("3.5")) {
 if (sparkVersions.contains("4.0")) {
   apply from: file("$projectDir/v4.0/build.gradle")
 }
+
+if (sparkVersions.contains("4.1")) {
+    apply from: file("$projectDir/v4.1/build.gradle")
+}

spark/v4.1/build.gradle

@@ -17,12 +17,12 @@
  * under the License.
  */
 
-String sparkMajorVersion = '4.0'
+String sparkMajorVersion = '4.1'
 String scalaVersion = '2.13'
 
 JavaVersion javaVersion = JavaVersion.current()
 if (javaVersion != JavaVersion.VERSION_17 && javaVersion != JavaVersion.VERSION_21) {
-  throw new GradleException("Spark 4.0 build requires JDK 17 or 21 but was executed with JDK " + javaVersion)
+  throw new GradleException("Spark 4.1 build requires JDK 17 or 21 but was executed with JDK " + javaVersion)
 }
 
 def sparkProjects = [
@@ -47,6 +47,14 @@ project(":iceberg-spark:iceberg-spark-${sparkMajorVersion}_${scalaVersion}") {
   apply plugin: 'scala'
   apply plugin: 'com.github.alisiikh.scalastyle'
 
+  // Set target to JDK17 for Spark 4.1 to fix following error
+  // "spark/v4.1/spark/src/main/scala/org/apache/spark/sql/stats/ThetaSketchAgg.scala:52:12: Class java.lang.Record not found"
+  tasks.withType(ScalaCompile.class) {
+    sourceCompatibility = "17"
+    targetCompatibility = "17"
+    scalaCompileOptions.additionalParameters.add("-release:17")
+  }
+
   sourceSets {
     main {
       scala.srcDirs = ['src/main/scala', 'src/main/java']
@@ -70,7 +78,7 @@ project(":iceberg-spark:iceberg-spark-${sparkMajorVersion}_${scalaVersion}") {
 
     compileOnly libs.errorprone.annotations
     compileOnly libs.avro.avro
-    compileOnly("org.apache.spark:spark-hive_${scalaVersion}:${libs.versions.spark40.get()}") {
+    compileOnly("org.apache.spark:spark-hive_${scalaVersion}:${libs.versions.spark41.get()}") {
       exclude group: 'org.apache.avro', module: 'avro'
       exclude group: 'org.apache.arrow'
       exclude group: 'org.apache.parquet'

spark/v4.1/spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/IcebergSparkSqlExtensionsParser.scala

@@ -28,7 +28,6 @@ import org.antlr.v4.runtime.tree.TerminalNodeImpl
 import org.apache.iceberg.common.DynConstructors
 import org.apache.iceberg.spark.ExtendedParser
 import org.apache.iceberg.spark.ExtendedParser.RawOrderField
-import org.apache.iceberg.spark.procedures.SparkProcedures
 import org.apache.spark.sql.AnalysisException
 import org.apache.spark.sql.SparkSession
 import org.apache.spark.sql.catalyst.FunctionIdentifier

spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/BaseCatalog.java

@@ -60,6 +60,17 @@ public UnboundProcedure loadProcedure(Identifier ident) {
     throw new RuntimeException("Procedure " + ident + " not found");
   }
 
+  @Override
+  public Identifier[] listProcedures(String[] namespace) {
+    if (isSystemNamespace(namespace)) {
+      return SparkProcedures.names().stream()
+          .map(name -> Identifier.of(namespace, name))
+          .toArray(Identifier[]::new);
+    } else {
+      return new Identifier[0];
+    }
+  }
+
   @Override
   public boolean isFunctionNamespace(String[] namespace) {
     // Allow for empty namespace, as Spark's storage partitioned joins look up