[Spark] [Delta X-Compile] Fix tests for both Spark 3.5 and Spark Master (batch 1) (delta-io#2884)

scottsand-db · web-flow · commit e1063a1bfafd · 2024-04-12T08:38:49.000-07:00
#### Which Delta project/connector is this regarding?
- [X] Spark
- [ ] Standalone
- [ ] Flink
- [ ] Kernel
- [ ] Other (fill in here)

## Description

Fixes 25 tests in 6 suites for Delta OSS on Spark Master using shimming.

## How was this patch tested?

Existing UTs

## Does this PR introduce _any_ user-facing changes?

No
diff --git a/build.sbt b/build.sbt
@@ -138,6 +138,7 @@ def crossSparkSettings(): Seq[Setting[_]] = getSparkVersion() match {
     // For adding staged Spark RC versions, e.g.:
     // resolvers += "Apache Spark 3.5.0 (RC1) Staging" at "https://repository.apache.org/content/repositories/orgapachespark-1444/",
     Compile / unmanagedSourceDirectories += (Compile / baseDirectory).value / "src" / "main" / "scala-spark-3.5",
+    Test / unmanagedSourceDirectories += (Test / baseDirectory).value / "src" / "test" / "scala-spark-3.5",
     Antlr4 / antlr4Version := "4.9.3",
 
     // Java-/Scala-/Uni-Doc Settings
@@ -153,6 +154,7 @@ def crossSparkSettings(): Seq[Setting[_]] = getSparkVersion() match {
     targetJvm := "17",
     resolvers += "Spark master staging" at "https://repository.apache.org/content/groups/snapshots/",
     Compile / unmanagedSourceDirectories += (Compile / baseDirectory).value / "src" / "main" / "scala-spark-master",
+    Test / unmanagedSourceDirectories += (Test / baseDirectory).value / "src" / "test" / "scala-spark-master",
     Antlr4 / antlr4Version := "4.13.1",
     Test / javaOptions ++= Seq(
       // Copied from SparkBuild.scala to support Java 17 for unit tests (see apache/spark#34153)
diff --git a/spark/src/test/scala-spark-3.5/shims/DeltaInsertIntoTableSuiteShims.scala b/spark/src/test/scala-spark-3.5/shims/DeltaInsertIntoTableSuiteShims.scala
@@ -0,0 +1,21 @@
+/*
+ * Copyright (2024) The Delta Lake Project Authors.
+ *
+ * Licensed 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.spark.sql.delta
+
+object DeltaInsertIntoTableSuiteShims {
+  val INSERT_INTO_TMP_VIEW_ERROR_MSG = "Inserting into a view is not allowed"
+}
diff --git a/spark/src/test/scala-spark-3.5/shims/DeltaVacuumSuiteShims.scala b/spark/src/test/scala-spark-3.5/shims/DeltaVacuumSuiteShims.scala
@@ -0,0 +1,22 @@
+/*
+ * Copyright (2024) The Delta Lake Project Authors.
+ *
+ * Licensed 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.spark.sql.delta
+
+object DeltaVacuumSuiteShims {
+  val SQL_COMMAND_ON_TEMP_VIEW_NOT_SUPPORTED_ERROR_MSG =
+    "v is a temp view. 'VACUUM' expects a table."
+}
diff --git a/spark/src/test/scala-spark-3.5/shims/DescribeDeltaHistorySuiteShims.scala b/spark/src/test/scala-spark-3.5/shims/DescribeDeltaHistorySuiteShims.scala
@@ -0,0 +1,25 @@
+/*
+ * Copyright (2024) The Delta Lake Project Authors.
+ *
+ * Licensed 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.spark.sql.delta
+
+object DescribeDeltaHistorySuiteShims {
+  val FAILS_ON_VIEWS_ERROR_MSG =
+    "spark_catalog.default.delta_view is a view. 'DESCRIBE HISTORY' expects a table"
+
+  val FAILS_ON_TEMP_VIEWS_ERROR_MSG =
+    "v is a temp view. 'DESCRIBE HISTORY' expects a table"
+}
diff --git a/spark/src/test/scala-spark-3.5/shims/SnapshotManagementSuiteShims.scala b/spark/src/test/scala-spark-3.5/shims/SnapshotManagementSuiteShims.scala
@@ -0,0 +1,21 @@
+/*
+ * Copyright (2024) The Delta Lake Project Authors.
+ *
+ * Licensed 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.spark.sql.delta
+
+object SnapshotManagementSuiteShims {
+  val SHOULD_NOT_RECOVER_CHECKPOINT_ERROR_MSG = ".parquet is not a Parquet file"
+}
diff --git a/spark/src/test/scala-spark-3.5/shims/StatsCollectionSuiteShims.scala b/spark/src/test/scala-spark-3.5/shims/StatsCollectionSuiteShims.scala
@@ -0,0 +1,37 @@
+/*
+ * Copyright (2024) The Delta Lake Project Authors.
+ *
+ * Licensed 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.spark.sql.delta.stats
+
+import scala.reflect.ClassTag
+
+import org.scalactic.source
+import org.scalatest.Assertions
+
+object StatsCollectionSuiteShims extends Assertions {
+
+  /**
+   * Handles a breaking change between Spark 3.5 and Spark Master (4.0) to improve error messaging
+   * in Spark. Previously, in Spark 3.5, when an executor would throw an exception, the driver would
+   * wrap it in a SparkException. Now, in Spark Master (4.0), the original executor exception is
+   * thrown directly.
+   */
+  def interceptAndUnwrap[T <: Throwable : ClassTag](f: => Any)(implicit pos: source.Position): T = {
+    intercept[Throwable] {
+      f
+    }.getCause.asInstanceOf[T]
+  }
+}
diff --git a/spark/src/test/scala-spark-master/shims/DeltaInsertIntoTableSuiteShims.scala b/spark/src/test/scala-spark-master/shims/DeltaInsertIntoTableSuiteShims.scala
@@ -0,0 +1,21 @@
+/*
+ * Copyright (2021) The Delta Lake Project Authors.
+ *
+ * Licensed 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.spark.sql.delta
+
+object DeltaInsertIntoTableSuiteShims {
+  val INSERT_INTO_TMP_VIEW_ERROR_MSG = "[EXPECT_TABLE_NOT_VIEW.NO_ALTERNATIVE]"
+}
diff --git a/spark/src/test/scala-spark-master/shims/DeltaVacuumSuiteShims.scala b/spark/src/test/scala-spark-master/shims/DeltaVacuumSuiteShims.scala
@@ -0,0 +1,22 @@
+/*
+ * Copyright (2021) The Delta Lake Project Authors.
+ *
+ * Licensed 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.spark.sql.delta
+
+object DeltaVacuumSuiteShims {
+  val SQL_COMMAND_ON_TEMP_VIEW_NOT_SUPPORTED_ERROR_MSG =
+    "'VACUUM' expects a table but `v` is a view"
+}
diff --git a/spark/src/test/scala-spark-master/shims/DescribeDeltaHistorySuiteShims.scala b/spark/src/test/scala-spark-master/shims/DescribeDeltaHistorySuiteShims.scala
@@ -0,0 +1,25 @@
+/*
+ * Copyright (2021) The Delta Lake Project Authors.
+ *
+ * Licensed 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.spark.sql.delta
+
+object DescribeDeltaHistorySuiteShims {
+  val FAILS_ON_VIEWS_ERROR_MSG =
+    "'DESCRIBE HISTORY' expects a table but `spark_catalog`.`default`.`delta_view` is a view."
+
+  val FAILS_ON_TEMP_VIEWS_ERROR_MSG =
+    "'DESCRIBE HISTORY' expects a table but `v` is a view."
+}
diff --git a/spark/src/test/scala-spark-master/shims/SnapshotManagementSuiteShims.scala b/spark/src/test/scala-spark-master/shims/SnapshotManagementSuiteShims.scala
@@ -0,0 +1,21 @@
+/*
+ * Copyright (2021) The Delta Lake Project Authors.
+ *
+ * Licensed 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.spark.sql.delta
+
+object SnapshotManagementSuiteShims {
+  val SHOULD_NOT_RECOVER_CHECKPOINT_ERROR_MSG = "Encountered error while reading file"
+}
diff --git a/spark/src/test/scala-spark-master/shims/StatsCollectionSuiteShims.scala b/spark/src/test/scala-spark-master/shims/StatsCollectionSuiteShims.scala
@@ -0,0 +1,37 @@
+/*
+ * Copyright (2021) The Delta Lake Project Authors.
+ *
+ * Licensed 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.spark.sql.delta.stats
+
+import scala.reflect.ClassTag
+
+import org.scalactic.source
+import org.scalatest.Assertions
+
+object StatsCollectionSuiteShims extends Assertions {
+
+  /**
+   * Handles a breaking change between Spark 3.5 and Spark Master (4.0) to improve error messaging
+   * in Spark. Previously, in Spark 3.5, when an executor would throw an exception, the driver would
+   * wrap it in a SparkException. Now, in Spark Master (4.0), the original executor exception is
+   * thrown directly.
+   */
+  def interceptAndUnwrap[T <: Throwable : ClassTag](f: => Any)(implicit pos: source.Position): T = {
+    intercept[T] {
+      f
+    }
+  }
+}
diff --git a/spark/src/test/scala/io/delta/tables/DeltaTableSuite.scala b/spark/src/test/scala/io/delta/tables/DeltaTableSuite.scala
@@ -188,7 +188,7 @@ class DeltaTableSuite extends QueryTest
     }
 
     // DeltaTable can be passed to executor but method call causes exception.
-    val e = intercept[SparkException] {
+    val e = intercept[Exception] {
       withTempDir { dir =>
         testData.write.format("delta").mode("append").save(dir.getAbsolutePath)
         val dt: DeltaTable = DeltaTable.forPath(dir.getAbsolutePath)
diff --git a/spark/src/test/scala/org/apache/spark/sql/delta/DeltaInsertIntoTableSuite.scala b/spark/src/test/scala/org/apache/spark/sql/delta/DeltaInsertIntoTableSuite.scala
@@ -21,6 +21,7 @@ import java.io.File
 
 import scala.collection.JavaConverters._
 
+import org.apache.spark.sql.delta.DeltaInsertIntoTableSuiteShims._
 import org.apache.spark.sql.delta.schema.InvariantViolationException
 import org.apache.spark.sql.delta.schema.SchemaUtils
 import org.apache.spark.sql.delta.sources.DeltaSQLConf
@@ -435,7 +436,8 @@ abstract class DeltaInsertIntoTestsWithTempViews(
           checkAnswer(spark.table("v"), expectedResult)
         } catch {
           case e: AnalysisException =>
-            assert(e.getMessage.contains("Inserting into a view is not allowed") ||
+            assert(
+              e.getMessage.contains(INSERT_INTO_TMP_VIEW_ERROR_MSG) ||
               e.getMessage.contains("Inserting into an RDD-based table is not allowed") ||
               e.getMessage.contains("Table default.v not found") ||
               e.getMessage.contains("Table or view 'v' not found in database 'default'") ||
diff --git a/spark/src/test/scala/org/apache/spark/sql/delta/DeltaVacuumSuite.scala b/spark/src/test/scala/org/apache/spark/sql/delta/DeltaVacuumSuite.scala
@@ -25,6 +25,7 @@ import scala.language.implicitConversions
 
 import org.apache.spark.sql.delta.DeltaOperations.{Delete, Write}
 import org.apache.spark.sql.delta.DeltaTestUtils.createTestAddFile
+import org.apache.spark.sql.delta.DeltaVacuumSuiteShims._
 import org.apache.spark.sql.delta.actions.{AddCDCFile, AddFile, Metadata, RemoveFile}
 import org.apache.spark.sql.delta.commands.VacuumCommand
 import org.apache.spark.sql.delta.sources.DeltaSQLConf
@@ -512,8 +513,7 @@ class DeltaVacuumSuite
         val e = intercept[AnalysisException] {
           vacuumSQLTest(tablePath, viewName)
         }
-        assert(
-          e.getMessage.contains("v is a temp view. 'VACUUM' expects a table."))
+        assert(e.getMessage.contains(SQL_COMMAND_ON_TEMP_VIEW_NOT_SUPPORTED_ERROR_MSG))
       }
     }
   }
diff --git a/spark/src/test/scala/org/apache/spark/sql/delta/DescribeDeltaHistorySuite.scala b/spark/src/test/scala/org/apache/spark/sql/delta/DescribeDeltaHistorySuite.scala
@@ -19,6 +19,7 @@ package org.apache.spark.sql.delta
 // scalastyle:off import.ordering.noEmptyLine
 import java.io.File
 
+import org.apache.spark.sql.delta.DescribeDeltaHistorySuiteShims._
 import org.apache.spark.sql.delta.actions.{Action, AddCDCFile, AddFile, Metadata, Protocol, RemoveFile}
 import org.apache.spark.sql.delta.sources.DeltaSQLConf
 import org.apache.spark.sql.delta.test.DeltaSQLCommandTest
@@ -224,8 +225,8 @@ trait DescribeDeltaHistorySuiteBase
       val e = intercept[AnalysisException] {
         sql(s"DESCRIBE HISTORY $viewName").collect()
       }
-      assert(e.getMessage.contains("spark_catalog.default.delta_view is a view. " +
-        "'DESCRIBE HISTORY' expects a table"))
+
+      assert(e.getMessage.contains(FAILS_ON_VIEWS_ERROR_MSG))
     }
   }
 
@@ -238,7 +239,8 @@ trait DescribeDeltaHistorySuiteBase
         val e = intercept[AnalysisException] {
           sql(s"DESCRIBE HISTORY $viewName").collect()
         }
-        assert(e.getMessage.contains("v is a temp view. 'DESCRIBE HISTORY' expects a table"))
+
+        assert(e.getMessage.contains(FAILS_ON_TEMP_VIEWS_ERROR_MSG))
       }
   }
 
diff --git a/spark/src/test/scala/org/apache/spark/sql/delta/SnapshotManagementSuite.scala b/spark/src/test/scala/org/apache/spark/sql/delta/SnapshotManagementSuite.scala
@@ -24,6 +24,7 @@ import scala.collection.mutable
 import com.databricks.spark.util.{Log4jUsageLogger, UsageRecord}
 import org.apache.spark.sql.delta.DeltaConfigs.MANAGED_COMMIT_OWNER_NAME
 import org.apache.spark.sql.delta.DeltaTestUtils.{verifyBackfilled, verifyUnbackfilled, BOOLEAN_DOMAIN}
+import org.apache.spark.sql.delta.SnapshotManagementSuiteShims._
 import org.apache.spark.sql.delta.managedcommit.{Commit, CommitStore, CommitStoreBuilder, CommitStoreProvider, GetCommitsResponse, InMemoryCommitStore}
 import org.apache.spark.sql.delta.sources.DeltaSQLConf
 import org.apache.spark.sql.delta.storage.LocalLogStore
@@ -195,7 +196,7 @@ class SnapshotManagementSuite extends QueryTest with DeltaSQLTestUtils with Shar
           // Guava cache wraps the root cause
           assert(e.isInstanceOf[SparkException] &&
             e.getMessage.contains("0001.checkpoint") &&
-            e.getMessage.contains(".parquet is not a Parquet file"))
+            e.getMessage.contains(SHOULD_NOT_RECOVER_CHECKPOINT_ERROR_MSG))
         }
       }
     }
@@ -252,7 +253,7 @@ class SnapshotManagementSuite extends QueryTest with DeltaSQLTestUtils with Shar
         val e = intercept[SparkException] { staleLog.update() }
         val version = if (testEmptyCheckpoint) 0 else 1
         assert(e.getMessage.contains(f"$version%020d.checkpoint") &&
-          e.getMessage.contains(".parquet is not a Parquet file"))
+          e.getMessage.contains(SHOULD_NOT_RECOVER_CHECKPOINT_ERROR_MSG))
       }
     }
   }
diff --git a/spark/src/test/scala/org/apache/spark/sql/delta/stats/StatsCollectionSuite.scala b/spark/src/test/scala/org/apache/spark/sql/delta/stats/StatsCollectionSuite.scala
