[GLUTEN-11145][VL] Iceberg test supports Spark3.5 and Spark4.0

.github/workflows/velox_backend_x86.yml

@@ -963,7 +963,7 @@ jobs:
           export JAVA_HOME=/usr/lib/jvm/java-17-openjdk
           export PATH=$JAVA_HOME/bin:$PATH
           java -version
-          $MVN_CMD clean test -Pspark-3.5 -Pjava-17 -Pbackends-velox -Piceberg -Pdelta -Phudi -Ppaimon -Pspark-ut \
+          $MVN_CMD clean test -Pspark-3.5 -Pjava-17 -Pbackends-velox -Piceberg -Piceberg-test -Pdelta -Phudi -Ppaimon -Pspark-ut \
           -DargLine="-Dspark.test.home=/opt/shims/spark35/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
@@ -1401,6 +1401,7 @@ jobs:
           export PATH=$JAVA_HOME/bin:$PATH
           java -version
           $MVN_CMD clean test -Pspark-4.0 -Pscala-2.13 -Pjava-17 -Pbackends-velox \
+          -Piceberg -Piceberg-test \
           -Pspark-ut -DargLine="-Dspark.test.home=/opt/shims/spark40/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

backends-velox/src-iceberg-spark34/test/java/org/apache/gluten/IcebergTestShims.java

@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.gluten;
+
+import org.apache.spark.rdd.RDD;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.SparkSession;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.types.StructType;
+
+public class IcebergTestShims {
+
+  public static Dataset<Row> internalCreateDataFrame(
+      SparkSession spark, RDD<InternalRow> rdd, StructType schema, boolean isStreaming) {
+    return spark.internalCreateDataFrame(rdd, schema, isStreaming);
+  }
+}

backends-velox/src-iceberg-spark35/test/java/org/apache/gluten/IcebergTestShims.java

@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.gluten;
+
+import org.apache.spark.rdd.RDD;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.SparkSession;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.types.StructType;
+
+public class IcebergTestShims {
+
+  public static Dataset<Row> internalCreateDataFrame(
+      SparkSession spark, RDD<InternalRow> rdd, StructType schema, boolean isStreaming) {
+    return spark.internalCreateDataFrame(rdd, schema, isStreaming);
+  }
+}

backends-velox/src-iceberg-spark40/test/java/org/apache/gluten/IcebergTestShims.java

@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.gluten;
+
+import org.apache.spark.rdd.RDD;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.SparkSession;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.types.StructType;
+
+public class IcebergTestShims {
+
+  public static Dataset<Row> internalCreateDataFrame(
+      SparkSession spark, RDD<InternalRow> rdd, StructType schema, boolean isStreaming) {
+    return ((org.apache.spark.sql.classic.SparkSession) spark)
+        .internalCreateDataFrame(rdd, schema, isStreaming);
+  }
+}

backends-velox/src-iceberg/main/java/org/apache/gluten/TestConfUtil.java

@@ -32,5 +32,7 @@ public class TestConfUtil {
           "spark.ui.enabled",
           "false",
           "spark.gluten.ui.enabled",
+          "false",
+          "spark.sql.ansi.enabled",
           "false");
 }

backends-velox/src-iceberg-spark34/test/java/org/apache/gluten/execution/TestStoragePartitionedJoins.java → backends-velox/src-iceberg10/test/java/org/apache/gluten/execution/TestStoragePartitionedJoins.java

@@ -16,6 +16,8 @@
  */
 package org.apache.gluten.execution;
 
+import org.apache.gluten.IcebergTestShims;
+
 import org.apache.commons.lang3.StringUtils;
 import org.apache.iceberg.PlanningMode;
 import org.apache.iceberg.Schema;
@@ -655,7 +657,8 @@ private Dataset<Row> randomDataDF(Schema schema, int numRows) {
     Iterable<InternalRow> rows = RandomData.generateSpark(schema, numRows, 0);
     JavaRDD<InternalRow> rowRDD = sparkContext.parallelize(Lists.newArrayList(rows));
     StructType rowSparkType = SparkSchemaUtil.convert(schema);
-    return spark.internalCreateDataFrame(JavaRDD.toRDD(rowRDD), rowSparkType, false);
+    return IcebergTestShims.internalCreateDataFrame(
+        spark, JavaRDD.toRDD(rowRDD), rowSparkType, false);
   }
 
   private void append(String table, Dataset<Row> df) throws NoSuchTableException {

backends-velox/src-iceberg-spark34/test/java/org/apache/gluten/source/TestPartitionPruning.java → backends-velox/src-iceberg10/test/java/org/apache/gluten/source/TestPartitionPruning.java

@@ -16,6 +16,7 @@
  */
 package org.apache.gluten.source;
 
+import org.apache.gluten.IcebergTestShims;
 import org.apache.gluten.TestConfUtil;
 
 import org.apache.hadoop.conf.Configuration;
@@ -331,8 +332,8 @@ private Dataset<Row> createTestDataset() {
 
     JavaRDD<InternalRow> rdd = sparkContext.parallelize(rows);
     Dataset<Row> df =
-        spark.internalCreateDataFrame(
-            JavaRDD.toRDD(rdd), SparkSchemaUtil.convert(LOG_SCHEMA), false);
+        IcebergTestShims.internalCreateDataFrame(
+            spark, JavaRDD.toRDD(rdd), SparkSchemaUtil.convert(LOG_SCHEMA), false);
 
     return df.selectExpr("id", "date", "level", "message", "timestamp")
         .selectExpr(

backends-velox/src-iceberg-spark34/test/java/org/apache/gluten/source/TestSparkDataFile.java → backends-velox/src-iceberg10/test/java/org/apache/gluten/source/TestSparkDataFile.java

@@ -16,6 +16,7 @@
  */
 package org.apache.gluten.source;
 
+import org.apache.gluten.IcebergTestShims;
 import org.apache.gluten.TestConfUtil;
 
 import org.apache.hadoop.conf.Configuration;
@@ -139,8 +140,8 @@ private void checkSparkContentFiles(Table table) throws IOException {
     Iterable<InternalRow> rows = RandomData.generateSpark(table.schema(), 200, 0);
     JavaRDD<InternalRow> rdd = sparkContext.parallelize(Lists.newArrayList(rows));
     Dataset<Row> df =
-        spark.internalCreateDataFrame(
-            JavaRDD.toRDD(rdd), SparkSchemaUtil.convert(table.schema()), false);
+        IcebergTestShims.internalCreateDataFrame(
+            spark, JavaRDD.toRDD(rdd), SparkSchemaUtil.convert(table.schema()), false);
 
     df.write().format("iceberg").mode("append").save(tableLocation);
 

backends-velox/src-iceberg-spark34/test/java/org/apache/gluten/sql/TestGlutenAggregatePushDown.java → backends-velox/src-iceberg10/test/java/org/apache/gluten/sql/TestGlutenAggregatePushDown.java

@@ -36,7 +36,7 @@ public static void startMetastoreAndSpark() {
     TestBase.metastore = new TestHiveMetastore();
     metastore.start();
     TestBase.hiveConf = metastore.hiveConf();
-    TestBase.spark.close();
+    TestBase.spark.stop();
     TestBase.spark =
         SparkSession.builder()
             .master("local[2]")

backends-velox/src-iceberg-spark34/test/java/org/apache/iceberg/spark/CatalogTestBase.java → backends-velox/src-iceberg10/test/java/org/apache/iceberg/spark/CatalogTestBase.java

@@ -36,12 +36,7 @@ protected static Object[][] parameters() {
         SparkCatalogConfig.HADOOP.catalogName(),
         SparkCatalogConfig.HADOOP.implementation(),
         SparkCatalogConfig.HADOOP.properties()
-      },
-      {
-        SparkCatalogConfig.SPARK.catalogName(),
-        SparkCatalogConfig.SPARK.implementation(),
-        SparkCatalogConfig.SPARK.properties()
-      },
+      }
     };
   }
 }

backends-velox/src-iceberg-spark34/test/java/org/apache/iceberg/spark/SparkCatalogTestBase.java → backends-velox/src-iceberg10/test/java/org/apache/iceberg/spark/SparkCatalogTestBase.java

@@ -39,11 +39,6 @@ public static Object[][] parameters() {
         SparkCatalogConfig.HADOOP.catalogName(),
         SparkCatalogConfig.HADOOP.implementation(),
         SparkCatalogConfig.HADOOP.properties()
-      },
-      {
-        SparkCatalogConfig.SPARK.catalogName(),
-        SparkCatalogConfig.SPARK.implementation(),
-        SparkCatalogConfig.SPARK.properties()
       }
     };
   }

backends-velox/src-iceberg-spark34/test/java/org/apache/iceberg/spark/SparkTestBase.java → backends-velox/src-iceberg10/test/java/org/apache/iceberg/spark/SparkTestBase.java

@@ -55,7 +55,7 @@
 import java.util.concurrent.atomic.AtomicReference;
 
 import scala.Option;
-import scala.collection.JavaConversions;
+import scala.collection.JavaConverters;
 
 import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.METASTOREURIS;
 
@@ -290,7 +290,7 @@ protected boolean checkAnswer(Dataset<Row> df) {
         () -> {
           Option<String> msg =
               QueryTest.getErrorMessageInCheckAnswer(
-                  df, JavaConversions.asScalaBuffer(rows).toSeq(), true);
+                  df, JavaConverters.asScalaBuffer(rows).toSeq(), true);
           if (msg.isDefined()) {
             throw new RuntimeException(msg.get());
           }

backends-velox/src-iceberg-spark34/test/java/org/apache/iceberg/spark/extensions/ExtensionsTestBase.java → backends-velox/src-iceberg10/test/java/org/apache/iceberg/spark/extensions/ExtensionsTestBase.java

@@ -45,7 +45,7 @@ public static void startMetastoreAndSpark() {
     metastore.start();
     TestBase.hiveConf = metastore.hiveConf();
 
-    TestBase.spark.close();
+    TestBase.spark.stop();
 
     TestBase.spark =
         SparkSession.builder()

pom.xml

@@ -73,6 +73,7 @@
     <spark.version>3.5.5</spark.version>
     <sparkshim.artifactId>spark-sql-columnar-shims-spark35</sparkshim.artifactId>
     <iceberg.version>1.10.0</iceberg.version>
+    <iceberg.binary.version>10</iceberg.binary.version>
     <paimon.version>1.2.0</paimon.version>
     <delta.package.name>delta-spark</delta.package.name>
     <delta.version>3.3.2</delta.version>
@@ -1128,6 +1129,7 @@
         <sparkshim.artifactId>spark-sql-columnar-shims-spark34</sparkshim.artifactId>
         <spark.version>3.4.4</spark.version>
         <iceberg.version>1.10.0</iceberg.version>
+        <iceberg.binary.version>10</iceberg.binary.version>
         <delta.package.name>delta-core</delta.package.name>
         <delta.version>2.4.0</delta.version>
         <delta.binary.version>24</delta.binary.version>
@@ -1153,6 +1155,7 @@
         <sparkshim.artifactId>spark-sql-columnar-shims-spark35</sparkshim.artifactId>
         <spark.version>3.5.5</spark.version>
         <iceberg.version>1.10.0</iceberg.version>
+        <iceberg.binary.version>10</iceberg.binary.version>
         <delta.package.name>delta-spark</delta.package.name>
         <delta.version>3.3.2</delta.version>
         <delta.binary.version>33</delta.binary.version>
@@ -1190,7 +1193,9 @@
         <sparkbundle.version>4.0</sparkbundle.version>
         <sparkshim.artifactId>spark-sql-columnar-shims-spark40</sparkshim.artifactId>
         <spark.version>4.0.1</spark.version>
+        <spark.plain.version>40</spark.plain.version>
         <iceberg.version>1.10.0</iceberg.version>
+        <iceberg.binary.version>10</iceberg.binary.version>
         <delta.package.name>delta-spark</delta.package.name>
         <delta.version>4.0.0</delta.version>
         <delta.binary.version>40</delta.binary.version>
@@ -1622,6 +1627,8 @@
                   <sources>
                     <source>${project.basedir}/src-iceberg/main/scala</source>
                     <source>${project.basedir}/src-iceberg/main/java</source>
+                    <source>${project.basedir}/src-iceberg${iceberg.binary.version}/main/scala</source>
+                    <source>${project.basedir}/src-iceberg${iceberg.binary.version}/main/java</source>
                     <source>${project.basedir}/src-iceberg-spark${spark.plain.version}/main/scala</source>
                     <source>${project.basedir}/src-iceberg-spark${spark.plain.version}/main/java</source>
                   </sources>
@@ -1695,6 +1702,8 @@
                 <phase>generate-test-sources</phase>
                 <configuration>
                   <sources>
+                    <source>${project.basedir}/src-iceberg${iceberg.binary.version}/test/scala</source>
+                    <source>${project.basedir}/src-iceberg${iceberg.binary.version}/test/java</source>
                     <source>${project.basedir}/src-iceberg-spark${spark.plain.version}/test/scala</source>
                     <source>${project.basedir}/src-iceberg-spark${spark.plain.version}/test/java</source>
                   </sources>
@@ -1709,6 +1718,7 @@
                 <configuration>
                   <resources>
                     <resource>
+                      <directory>${project.basedir}/src-iceberg${iceberg.binary.version}/test/resources</directory>
                       <directory>${project.basedir}/src-iceberg-spark${spark.plain.version}/test/resources</directory>
                     </resource>
                   </resources>