Compile with Spark master snapshot and fix compile errors

Author: allisonport-db

build.sbt

@@ -32,10 +32,11 @@ val all_scala_versions = Seq(scala212, scala213)
 // sbt 'set default_scala_version := 2.13.8' [commands]
 // FIXME Why not use scalaVersion?
 val default_scala_version = settingKey[String]("Default Scala version")
-Global / default_scala_version := scala212
+Global / default_scala_version := scala213
+// TODO set scala only to 2.13 for spark but keep 2.12 for other projects?
 
 // Dependent library versions
-val sparkVersion = "3.5.0"
+val sparkVersion = "4.0.0-SNAPSHOT"
 val flinkVersion = "1.16.1"
 val hadoopVersion = "3.3.4"
 val scalaTestVersion = "3.2.15"
@@ -117,11 +118,11 @@ lazy val spark = (project in file("spark"))
       "org.apache.spark" %% "spark-hive" % sparkVersion % "test" classifier "tests",
     ),
     // For adding staged Spark RC versions, Ex:
-    // resolvers += "Apche Spark 3.5.0 (RC1) Staging" at "https://repository.apache.org/content/repositories/orgapachespark-1444/",
+    resolvers += "Spark master staging" at "https://repository.apache.org/content/groups/snapshots/",
     Compile / packageBin / mappings := (Compile / packageBin / mappings).value ++
         listPythonFiles(baseDirectory.value.getParentFile / "python"),
 
-    Antlr4 / antlr4Version:= "4.9.3",
+    Antlr4 / antlr4Version:= "4.13.1",
     Antlr4 / antlr4PackageName := Some("io.delta.sql.parser"),
     Antlr4 / antlr4GenListener := true,
     Antlr4 / antlr4GenVisitor := true,
@@ -344,6 +345,7 @@ val icebergSparkRuntimeArtifactName = {
  s"iceberg-spark-runtime-$expMaj.$expMin"
 }
 
+/*
 lazy val testDeltaIcebergJar = (project in file("testDeltaIcebergJar"))
   // delta-iceberg depends on delta-spark! So, we need to include it during our test.
   .dependsOn(spark % "test")
@@ -359,6 +361,7 @@ lazy val testDeltaIcebergJar = (project in file("testDeltaIcebergJar"))
       "org.apache.spark" %% "spark-core" % sparkVersion % "test"
     )
   )
+ */
 
 val deltaIcebergSparkIncludePrefixes = Seq(
   // We want everything from this package
@@ -371,6 +374,7 @@ val deltaIcebergSparkIncludePrefixes = Seq(
   "org/apache/spark/sql/delta/commands/convert/IcebergTable"
 )
 
+/*
 // Build using: build/sbt clean icebergShaded/compile iceberg/compile
 // It will fail the first time, just re-run it.
 // scalastyle:off println
@@ -445,6 +449,7 @@ lazy val iceberg = (project in file("iceberg"))
     assemblyPackageScala / assembleArtifact := false
   )
 // scalastyle:on println
+ */
 
 lazy val generateIcebergJarsTask = TaskKey[Unit]("generateIcebergJars", "Generate Iceberg JARs")
 
@@ -1114,7 +1119,8 @@ val createTargetClassesDir = taskKey[Unit]("create target classes dir")
 
 // Don't use these groups for any other projects
 lazy val sparkGroup = project
-  .aggregate(spark, contribs, storage, storageS3DynamoDB, iceberg, testDeltaIcebergJar, sharing)
+  .aggregate(spark, contribs, storage, storageS3DynamoDB, sharing)
+  // .aggregate(spark, contribs, storage, storageS3DynamoDB, iceberg, testDeltaIcebergJar, sharing)
   .settings(
     // crossScalaVersions must be set to Nil on the aggregating project
     crossScalaVersions := Nil,

spark/src/main/scala/io/delta/sql/parser/DeltaSqlParser.scala

@@ -522,7 +522,7 @@ class DeltaSqlAstBuilder extends DeltaSqlBaseBaseVisitor[AnyRef] {
   private def createUnresolvedTable(
       tableName: Seq[String],
       commandName: String): UnresolvedTable = {
-    UnresolvedTable(tableName, commandName, relationTypeMismatchHint = None)
+    UnresolvedTable(tableName, commandName)
   }
 
   // Build the text of the CHECK constraint expression. The user-specified whitespace is in the

spark/src/main/scala/io/delta/tables/DeltaTableBuilder.scala

@@ -27,6 +27,7 @@ import org.apache.spark.annotation._
 import org.apache.spark.sql.SparkSession
 import org.apache.spark.sql.catalyst.TableIdentifier
 import org.apache.spark.sql.catalyst.plans.logical.{CreateTable, LogicalPlan, ReplaceTable}
+import org.apache.spark.sql.catalyst.plans.logical.ColumnDefinition
 import org.apache.spark.sql.catalyst.util.CaseInsensitiveMap
 import org.apache.spark.sql.connector.expressions.Transform
 import org.apache.spark.sql.execution.SQLExecution
@@ -343,15 +344,15 @@ class DeltaTableBuilder private[tables](
         val unresolvedTable = org.apache.spark.sql.catalyst.analysis.UnresolvedIdentifier(table)
         CreateTable(
           unresolvedTable,
-          StructType(columns.toSeq),
+          columns.map(ColumnDefinition.fromV1Column(_, spark.sessionState.sqlParser)).toSeq,
           partitioning,
           tableSpec,
           ifNotExists)
       case ReplaceTableOptions(orCreate) =>
         val unresolvedTable = org.apache.spark.sql.catalyst.analysis.UnresolvedIdentifier(table)
         ReplaceTable(
           unresolvedTable,
-          StructType(columns.toSeq),
+          columns.map(ColumnDefinition.fromV1Column(_, spark.sessionState.sqlParser)).toSeq,
           partitioning,
           tableSpec,
           orCreate)

spark/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/DeltaUpdateTable.scala

@@ -16,7 +16,8 @@
 
 package org.apache.spark.sql.catalyst.plans.logical
 
-import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.delta.DeltaAnalysisException
+
 import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, Expression, ExtractValue, GetStructField}
 
 /**
@@ -66,7 +67,7 @@ object DeltaUpdateTable {
 
     def fail(extraMsg: String): Nothing = {
       val msg = Option(errMsg).map(_ + " - ").getOrElse("") + extraMsg
-      throw new AnalysisException(msg)
+      throw new DeltaAnalysisException(msg)
     }
 
     def extractRecursively(expr: Expression): Seq[String] = expr match {

spark/src/main/scala/org/apache/spark/sql/delta/ColumnWithDefaultExprUtils.scala

@@ -220,7 +220,8 @@ object ColumnWithDefaultExprUtils extends DeltaLogging {
       incrementalExecution.currentBatchId,
       incrementalExecution.prevOffsetSeqMetadata,
       incrementalExecution.offsetSeqMetadata,
-      incrementalExecution.watermarkPropagator
+      incrementalExecution.watermarkPropagator,
+      incrementalExecution.isFirstBatch
     )
     newIncrementalExecution.executedPlan // Force the lazy generation of execution plan
 

spark/src/main/scala/org/apache/spark/sql/delta/DeltaErrors.scala

@@ -245,7 +245,14 @@ trait DeltaErrorsBase
       startPosition: Option[Int] = None,
       plan: Option[LogicalPlan] = None,
       cause: Option[Throwable] = None): AnalysisException = {
-    new ExtendedAnalysisException(msg, line, startPosition, plan, cause)
+    if (plan.isEmpty) {
+      new DeltaAnalysisException(msg, line, startPosition, cause)
+    } else {
+      new ExtendedAnalysisException(
+        new DeltaAnalysisException(msg, line, startPosition, cause),
+        plan.get
+      )
+    }
   }
 
   def notNullColumnMissingException(constraint: Constraints.NotNull): Throwable = {
@@ -283,7 +290,7 @@ trait DeltaErrorsBase
   }
 
   def invalidConstraintName(name: String): AnalysisException = {
-    new AnalysisException(s"Cannot use '$name' as the name of a CHECK constraint.")
+    new DeltaAnalysisException(s"Cannot use '$name' as the name of a CHECK constraint.")
   }
 
   def nonexistentConstraint(constraintName: String, tableName: String): AnalysisException = {
@@ -1084,7 +1091,7 @@ trait DeltaErrorsBase
   }
 
   def bloomFilterInvalidParameterValueException(message: String): Throwable = {
-    new AnalysisException(
+    new DeltaAnalysisException(
       s"Cannot create bloom filter index, invalid parameter value: $message")
   }
 
@@ -1209,7 +1216,7 @@ trait DeltaErrorsBase
     def prettyMap(m: Map[String, String]): String = {
       m.map(e => s"${e._1}=${e._2}").mkString("[", ", ", "]")
     }
-    new AnalysisException(
+    new DeltaAnalysisException(
       s"""You are trying to convert a table which already has a delta log where the table
          |properties in the catalog don't match the configuration in the delta log.
          |Table properties in catalog: ${prettyMap(tableProperties)}
@@ -1387,7 +1394,7 @@ trait DeltaErrorsBase
   def restoreTimestampBeforeEarliestException(
       userTimestamp: String,
       earliestTimestamp: String): Throwable = {
-    new AnalysisException(
+    new DeltaAnalysisException(
       s"Cannot restore table to timestamp ($userTimestamp) as it is before the earliest version " +
         s"available. Please use a timestamp after ($earliestTimestamp)"
     )
@@ -1566,7 +1573,7 @@ trait DeltaErrorsBase
   }
 
   def viewNotSupported(operationName: String): Throwable = {
-    new AnalysisException(s"Operation $operationName can not be performed on a view")
+    new DeltaAnalysisException(s"Operation $operationName can not be performed on a view")
   }
 
   def postCommitHookFailedException(
@@ -1720,7 +1727,7 @@ trait DeltaErrorsBase
   }
 
   def generatedColumnsNonDeterministicExpression(expr: Expression): Throwable = {
-    new AnalysisException(
+    new DeltaAnalysisException(
       s"Found ${expr.sql}. A generated column cannot use a non deterministic expression")
   }
 
@@ -2056,7 +2063,7 @@ trait DeltaErrorsBase
       columnName: String,
       constraints: Map[String, String]): Throwable = {
     val plural = if (constraints.size > 1) "s" else ""
-    new AnalysisException(
+    new DeltaAnalysisException(
       s"""
         |Cannot $operation column $columnName because this column is referenced by the following
         | check constraint$plural:\n\t${constraints.mkString("\n\t")}
@@ -2068,7 +2075,7 @@ trait DeltaErrorsBase
       columnName: String,
       fields: Seq[StructField]): Throwable = {
     val plural = if (fields.size > 1) "s" else ""
-    new AnalysisException(
+    new DeltaAnalysisException(
       s"""
          |Cannot $operation column $columnName because this column is referenced by the following
          | generated column$plural:\n\t${fields.map(_.name).mkString("\n\t")}
@@ -2397,7 +2404,7 @@ trait DeltaErrorsBase
       hasStart: Boolean,
       hasStep: Boolean,
       hasInsert: Boolean): Throwable = {
-    new AnalysisException(s"Inconsistent IDENTITY metadata for column $colName " +
+    new DeltaAnalysisException(s"Inconsistent IDENTITY metadata for column $colName " +
       s"detected: $hasStart, $hasStep, $hasInsert")
   }
 
@@ -3417,7 +3424,7 @@ class MetadataMismatchErrorBuilder {
   }
 
   def finalizeAndThrow(conf: SQLConf): Unit = {
-    throw new AnalysisException(bits.mkString("\n"))
+    throw new DeltaAnalysisException(bits.mkString("\n"))
   }
 }
 

spark/src/main/scala/org/apache/spark/sql/delta/DeltaSharedExceptions.scala

@@ -20,28 +20,75 @@ import scala.collection.JavaConverters._
 
 import org.antlr.v4.runtime.ParserRuleContext
 
+import org.apache.spark.QueryContext
 import org.apache.spark.sql.AnalysisException
 import org.apache.spark.sql.catalyst.parser.{ParseException, ParserUtils}
 import org.apache.spark.sql.catalyst.trees.Origin
 
+// Spark's AnalysisException constructor is no longer public. This means all calls
+// with AnalysisException(msg: String) are invalid.
+// For now we adjust DeltaAnalysisException to expose a constructor that allows this. Otherwise,
+// we should consider changing all of those exceptions to use an errorClass with the public
+// AnalysisException constructors (this can just be INTERNAL_ERROR)
 class DeltaAnalysisException(
+  message: String,
+  line: Option[Int] = None,
+  startPosition: Option[Int] = None,
+  cause: Option[Throwable] = None,
+  errorClass: Option[String] = None,
+  messageParameters: Map[String, String] = Map.empty,
+  context: Array[QueryContext] = Array.empty)
+  extends AnalysisException(
+    message,
+    line,
+    startPosition,
+    cause,
+    errorClass,
+    messageParameters,
+    context)
+    with DeltaThrowable {
+
+  def this(
+    errorClass: String,
+    messageParameters: Array[String]) =
+    this(
+      message = DeltaThrowableHelper.getMessage(errorClass, messageParameters),
+      messageParameters = DeltaThrowableHelper
+        .getParameterNames(errorClass, errorSubClass = null)
+        .zip(messageParameters)
+        .toMap,
+      errorClass = Some(errorClass)
+    )
+
+  def this(
     errorClass: String,
     messageParameters: Array[String],
-    cause: Option[Throwable] = None,
-    origin: Option[Origin] = None)
-  extends AnalysisException(
-    message = DeltaThrowableHelper.getMessage(errorClass, messageParameters),
-    messageParameters = DeltaThrowableHelper
+    cause: Option[Throwable]) =
+    this(
+      message = DeltaThrowableHelper.getMessage(errorClass, messageParameters),
+      messageParameters = DeltaThrowableHelper
         .getParameterNames(errorClass, errorSubClass = null)
         .zip(messageParameters)
         .toMap,
-    errorClass = Some(errorClass),
-    line = origin.flatMap(_.line),
-    startPosition = origin.flatMap(_.startPosition),
-    context = origin.map(_.getQueryContext).getOrElse(Array.empty),
-    cause = cause)
-  with DeltaThrowable {
-  def getMessageParametersArray: Array[String] = messageParameters
+      errorClass = Some(errorClass),
+      cause = cause)
+
+  def this(
+    errorClass: String,
+    messageParameters: Array[String],
+    cause: Option[Throwable],
+    origin: Option[Origin]) =
+    this(
+      message = DeltaThrowableHelper.getMessage(errorClass, messageParameters),
+      messageParameters = DeltaThrowableHelper
+        .getParameterNames(errorClass, errorSubClass = null)
+        .zip(messageParameters)
+        .toMap,
+      errorClass = Some(errorClass),
+      line = origin.flatMap(_.line),
+      startPosition = origin.flatMap(_.startPosition),
+      context = origin.map(_.getQueryContext).getOrElse(Array.empty),
+      cause = cause)
 }
 
 class DeltaIllegalArgumentException(

spark/src/main/scala/org/apache/spark/sql/delta/DeltaTable.scala

@@ -617,7 +617,7 @@ object UnresolvedDeltaPathOrIdentifier {
     (path, tableIdentifier) match {
       case (Some(p), None) => UnresolvedPathBasedDeltaTable(p, Map.empty, cmd)
       case (None, Some(t)) =>
-        UnresolvedTable(t.nameParts, cmd, None)
+        UnresolvedTable(t.nameParts, cmd)
       case _ => throw new IllegalArgumentException(
         s"Exactly one of path or tableIdentifier must be provided to $cmd")
     }
@@ -639,7 +639,7 @@ object UnresolvedPathOrIdentifier {
       cmd: String): LogicalPlan = {
     (path, tableIdentifier) match {
       case (_, Some(t)) =>
-        UnresolvedTable(t.nameParts, cmd, None)
+        UnresolvedTable(t.nameParts, cmd)
       case (Some(p), None) => UnresolvedPathBasedTable(p, Map.empty, cmd)
       case _ => throw new IllegalArgumentException(
         s"At least one of path or tableIdentifier must be provided to $cmd")

spark/src/main/scala/org/apache/spark/sql/delta/catalog/DeltaTableV2.scala

@@ -329,7 +329,7 @@ object DeltaTableV2 {
 
   /** Resolves a table identifier into a DeltaTableV2, leveraging standard v2 table resolution. */
   def apply(spark: SparkSession, tableId: TableIdentifier, cmd: String): DeltaTableV2 = {
-    resolve(spark, UnresolvedTable(tableId.nameParts, cmd, None), cmd)
+    resolve(spark, UnresolvedTable(tableId.nameParts, cmd), cmd)
   }
 
   /** Applies standard v2 table resolution to an unresolved Delta table plan node */

spark/src/main/scala/org/apache/spark/sql/delta/commands/MergeIntoCommandBase.scala

@@ -121,7 +121,7 @@ trait MergeIntoCommandBase extends LeafRunnableCommand
       // queries that add void columns.
       val newNullColumn = SchemaUtils.findNullTypeColumn(migratedSchema.get)
       if (newNullColumn.isDefined) {
-        throw new AnalysisException(
+        throw new DeltaAnalysisException(
           s"""Cannot add column '${newNullColumn.get}' with type 'void'. Please explicitly specify a
               |non-void type.""".stripMargin.replaceAll("\n", " ")
         )