diff --git a/.github/workflows/test.yaml b/.github/workflows/test.yaml
index 1bb2e27ae0d..a42516bafe0 100644
--- a/.github/workflows/test.yaml
+++ b/.github/workflows/test.yaml
@@ -17,7 +17,11 @@ jobs:
             ~/.sbt
             ~/.ivy2
             ~/.cache/coursier
-          key: delta-sbt-cache
+          # Change the key if dependencies are changed. For each key, GitHub Actions will cache the
+          # the above directories when we use the key for the first time. After that, each run will
+          # just use the cache. The cache is immutable so we need to use a new key when trying to
+          # cache new stuff.
+          key: delta-sbt-cache-spark3.2
       - name: Install Job dependencies
         shell: bash -l {0}
         run: |
@@ -33,7 +37,7 @@ jobs:
           pyenv install 3.7.4
           pyenv global system 3.7.4
           pipenv --python 3.7 install
-          pipenv run pip install pyspark==3.1.1
+          pipenv run pip install pyspark==3.2.0
           pipenv run pip install flake8==3.5.0 pypandoc==1.3.3
           pipenv run pip install importlib_metadata==3.10.0
       - name: Run Scala/Java and Python tests
diff --git a/Dockerfile b/Dockerfile
index 84cd3239d12..bb1779d0b3c 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -2,7 +2,7 @@ FROM python:3.7.3-stretch
 
 RUN apt-get update && apt-get -y install openjdk-8-jdk
 
-RUN pip install pyspark==3.1.1
+RUN pip install pyspark==3.2.0
 
 COPY .  /usr/src/delta
 
diff --git a/build.sbt b/build.sbt
index 55a5877c04b..e30ac1bc5d5 100644
--- a/build.sbt
+++ b/build.sbt
@@ -16,7 +16,7 @@
 
 import java.nio.file.Files
 
-val sparkVersion = "3.1.1"
+val sparkVersion = "3.2.0"
 scalaVersion := "2.12.14"
 
 lazy val commonSettings = Seq(
diff --git a/core/src/main/scala/io/delta/sql/parser/DeltaSqlParser.scala b/core/src/main/scala/io/delta/sql/parser/DeltaSqlParser.scala
index bf1b97d0c27..4ad752da9fa 100644
--- a/core/src/main/scala/io/delta/sql/parser/DeltaSqlParser.scala
+++ b/core/src/main/scala/io/delta/sql/parser/DeltaSqlParser.scala
@@ -217,7 +217,7 @@ class DeltaSqlAstBuilder extends DeltaSqlBaseBaseVisitor[AnyRef] {
   private def createUnresolvedTable(
       tableName: Seq[String],
       commandName: String): UnresolvedTable = {
-    UnresolvedTable(tableName, commandName)
+    UnresolvedTable(tableName, commandName, relationTypeMismatchHint = None)
   }
 
   // Build the text of the CHECK constraint expression. The user-specified whitespace is in the
diff --git a/core/src/main/scala/io/delta/tables/execution/VacuumTableCommand.scala b/core/src/main/scala/io/delta/tables/execution/VacuumTableCommand.scala
index b6007504f85..6165ee09af2 100644
--- a/core/src/main/scala/io/delta/tables/execution/VacuumTableCommand.scala
+++ b/core/src/main/scala/io/delta/tables/execution/VacuumTableCommand.scala
@@ -22,7 +22,7 @@ import org.apache.spark.sql.catalyst.TableIdentifier
 import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference}
 import org.apache.spark.sql.delta.{DeltaErrors, DeltaLog, DeltaTableIdentifier, DeltaTableUtils}
 import org.apache.spark.sql.delta.commands.VacuumCommand
-import org.apache.spark.sql.execution.command.RunnableCommand
+import org.apache.spark.sql.execution.command.LeafRunnableCommand
 import org.apache.spark.sql.types.StringType
 
 /**
@@ -35,7 +35,7 @@ case class VacuumTableCommand(
     path: Option[String],
     table: Option[TableIdentifier],
     horizonHours: Option[Double],
-    dryRun: Boolean) extends RunnableCommand {
+    dryRun: Boolean) extends LeafRunnableCommand {
 
   override val output: Seq[Attribute] =
     Seq(AttributeReference("path", StringType, nullable = true)())
diff --git a/core/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/DeltaDelete.scala b/core/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/DeltaDelete.scala
index 7a05e0dc8c6..4e54c0a5bf6 100644
--- a/core/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/DeltaDelete.scala
+++ b/core/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/DeltaDelete.scala
@@ -26,5 +26,6 @@ case class DeltaDelete(
   extends UnaryNode {
   override def output: Seq[Attribute] = Seq.empty
 
-  // TODO: remove when the new Spark version is releases that has the withNewChildInternal method
+  override protected def withNewChildInternal(newChild: LogicalPlan): DeltaDelete =
+    copy(child = newChild)
 }
diff --git a/core/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/DeltaUpdateTable.scala b/core/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/DeltaUpdateTable.scala
index 566d5d232ed..36417999c2e 100644
--- a/core/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/DeltaUpdateTable.scala
+++ b/core/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/DeltaUpdateTable.scala
@@ -38,7 +38,8 @@ case class DeltaUpdateTable(
 
   override def output: Seq[Attribute] = Seq.empty
 
-  // TODO: remove when the new Spark version is releases that has the withNewChildInternal method
+  override protected def withNewChildInternal(newChild: LogicalPlan): DeltaUpdateTable =
+    copy(child = newChild)
 }
 
 object DeltaUpdateTable {
diff --git a/core/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/deltaConstraints.scala b/core/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/deltaConstraints.scala
index 50702112cf7..798941d62cc 100644
--- a/core/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/deltaConstraints.scala
+++ b/core/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/deltaConstraints.scala
@@ -16,22 +16,26 @@
 
 package org.apache.spark.sql.catalyst.plans.logical
 
+import org.apache.spark.sql.delta.constraints.{AddConstraint, DropConstraint}
+
+import org.apache.spark.sql.connector.catalog.TableChange
+
 /**
  * The logical plan of the ALTER TABLE ... ADD CONSTRAINT command.
  */
 case class AlterTableAddConstraint(
-    table: LogicalPlan, constraintName: String, expr: String) extends Command {
-  // TODO: extend UnaryCommand when new Spark version released, now fails on OSS Delta build
-  override def children: Seq[LogicalPlan] = Seq(table)
-  // TODO: remove when the new Spark version is releases that has the withNewChildInternal method
+    table: LogicalPlan, constraintName: String, expr: String) extends AlterTableCommand {
+  override def changes: Seq[TableChange] = Seq(AddConstraint(constraintName, expr))
+
+  protected def withNewChildInternal(newChild: LogicalPlan): LogicalPlan = copy(table = newChild)
 }
 
 /**
  * The logical plan of the ALTER TABLE ... DROP CONSTRAINT command.
  */
 case class AlterTableDropConstraint(
-    table: LogicalPlan, constraintName: String) extends Command {
-  // TODO: extend UnaryCommand when new Spark version released, now fails on OSS Delta build
-  override def children: Seq[LogicalPlan] = Seq(table)
-  // TODO: remove when the new Spark version is releases that has the withNewChildInternal method
+    table: LogicalPlan, constraintName: String) extends AlterTableCommand {
+  override def changes: Seq[TableChange] = Seq(DropConstraint(constraintName))
+
+  protected def withNewChildInternal(newChild: LogicalPlan): LogicalPlan = copy(table = newChild)
 }
diff --git a/core/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/deltaMerge.scala b/core/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/deltaMerge.scala
index 5d33724de3a..d0c3cc17de9 100644
--- a/core/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/deltaMerge.scala
+++ b/core/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/deltaMerge.scala
@@ -68,7 +68,8 @@ case class DeltaMergeAction(
   override def toString: String = s"$targetColString = $expr"
   private lazy val targetColString: String = targetColNameParts.mkString("`", "`.`", "`")
 
-  // TODO: remove when the new Spark version is releases that has the withNewChildInternal method
+  override protected def withNewChildInternal(newChild: Expression): DeltaMergeAction =
+    copy(expr = newChild)
 }
 
 
@@ -165,7 +166,14 @@ case class DeltaMergeIntoUpdateClause(condition: Option[Expression], actions: Se
   def this(cond: Option[Expression], cols: Seq[UnresolvedAttribute], exprs: Seq[Expression]) =
     this(cond, DeltaMergeIntoClause.toActions(cols, exprs))
 
-  // TODO: remove when the new Spark version is releases that has the withNewChildInternal method
+  override protected def withNewChildrenInternal(
+      newChildren: IndexedSeq[Expression]): DeltaMergeIntoUpdateClause = {
+    if (condition.isDefined) {
+      copy(condition = Some(newChildren.head), actions = newChildren.tail)
+    } else {
+      copy(condition = None, actions = newChildren)
+    }
+  }
 }
 
 /** Represents the clause WHEN MATCHED THEN DELETE in MERGE. See [[DeltaMergeInto]]. */
@@ -175,7 +183,9 @@ case class DeltaMergeIntoDeleteClause(condition: Option[Expression])
   children
   override def actions: Seq[Expression] = Seq.empty
 
-  // TODO: remove when the new Spark version is releases that has the withNewChildInternal method
+  override protected def withNewChildrenInternal(
+      newChildren: IndexedSeq[Expression]): DeltaMergeIntoDeleteClause =
+    copy(condition = if (condition.isDefined) Some(newChildren.head) else None)
 }
 
 /** Represents the clause WHEN NOT MATCHED THEN INSERT in MERGE. See [[DeltaMergeInto]]. */
@@ -185,7 +195,13 @@ case class DeltaMergeIntoInsertClause(condition: Option[Expression], actions: Se
   def this(cond: Option[Expression], cols: Seq[UnresolvedAttribute], exprs: Seq[Expression]) =
     this(cond, DeltaMergeIntoClause.toActions(cols, exprs))
 
-  // TODO: remove when the new Spark version is releases that has the withNewChildInternal method
+  override protected def withNewChildrenInternal(
+      newChildren: IndexedSeq[Expression]): DeltaMergeIntoInsertClause =
+    if (condition.isDefined) {
+      copy(condition = Some(newChildren.head), actions = newChildren.tail)
+    } else {
+      copy(condition = None, actions = newChildren)
+    }
 }
 
 /**
@@ -237,7 +253,9 @@ case class DeltaMergeInto(
   // TODO: extend BinaryCommand once the new Spark version is released
   override def children: Seq[LogicalPlan] = Seq(target, source)
   override def output: Seq[Attribute] = Seq.empty
-  // TODO: remove when the new Spark version is releases that has the withNewChildInternal method
+  override protected def withNewChildrenInternal(
+      newChildren: IndexedSeq[LogicalPlan]): DeltaMergeInto =
+    copy(target = newChildren(0), source = newChildren(1))
 }
 
 object DeltaMergeInto {
diff --git a/core/src/main/scala/org/apache/spark/sql/delta/DeltaAnalysis.scala b/core/src/main/scala/org/apache/spark/sql/delta/DeltaAnalysis.scala
index beae875f425..c27cf937bdd 100644
--- a/core/src/main/scala/org/apache/spark/sql/delta/DeltaAnalysis.scala
+++ b/core/src/main/scala/org/apache/spark/sql/delta/DeltaAnalysis.scala
@@ -135,17 +135,21 @@ class DeltaAnalysis(session: SparkSession)
           DeltaMergeIntoUpdateClause(
             update.condition,
             DeltaMergeIntoClause.toActions(update.assignments))
+        case update: UpdateStarAction =>
+          DeltaMergeIntoUpdateClause(update.condition, DeltaMergeIntoClause.toActions(Nil))
         case delete: DeleteAction =>
           DeltaMergeIntoDeleteClause(delete.condition)
-        case insert =>
+        case other =>
           throw new AnalysisException(
-            "Insert clauses cannot be part of the WHEN MATCHED clause in MERGE INTO.")
+            s"${other.prettyName} clauses cannot be part of the WHEN MATCHED clause in MERGE INTO.")
       }
       val notMatchedActions = notMatched.map {
         case insert: InsertAction =>
           DeltaMergeIntoInsertClause(
             insert.condition,
             DeltaMergeIntoClause.toActions(insert.assignments))
+        case insert: InsertStarAction =>
+          DeltaMergeIntoInsertClause(insert.condition, DeltaMergeIntoClause.toActions(Nil))
         case other =>
           throw new AnalysisException(s"${other.prettyName} clauses cannot be part of the " +
             s"WHEN NOT MATCHED clause in MERGE INTO.")
@@ -166,23 +170,6 @@ class DeltaAnalysis(session: SparkSession)
       } else deltaMerge
       d.copy(target = stripTempViewForMergeWrapper(d.target))
 
-    // TODO: remove the 2 cases below after OSS 3.2 is released.
-    case AlterTableAddConstraint(t: ResolvedTable, constraintName, expr)
-        if t.table.isInstanceOf[DeltaTableV2] =>
-      CatalogV2Util.createAlterTable(
-        t.catalog.name +: t.identifier.asMultipartIdentifier,
-        t.catalog,
-        t.identifier.asMultipartIdentifier,
-        Seq(AddConstraint(constraintName, expr)))
-
-    case AlterTableDropConstraint(t: ResolvedTable, constraintName)
-        if t.table.isInstanceOf[DeltaTableV2] =>
-      CatalogV2Util.createAlterTable(
-        t.catalog.name +: t.identifier.asMultipartIdentifier,
-        t.catalog,
-        t.identifier.asMultipartIdentifier,
-        Seq(DropConstraint(constraintName)))
-
   }
 
 
diff --git a/core/src/main/scala/org/apache/spark/sql/delta/DeltaTable.scala b/core/src/main/scala/org/apache/spark/sql/delta/DeltaTable.scala
index 255479f66dd..9b6ee07a22b 100644
--- a/core/src/main/scala/org/apache/spark/sql/delta/DeltaTable.scala
+++ b/core/src/main/scala/org/apache/spark/sql/delta/DeltaTable.scala
@@ -85,7 +85,7 @@ object DeltaTableUtils extends PredicateHelper
    */
   def isDeltaTable(spark: SparkSession, tableName: TableIdentifier): Boolean = {
     val catalog = spark.sessionState.catalog
-    val tableIsNotTemporaryTable = !catalog.isTemporaryTable(tableName)
+    val tableIsNotTemporaryTable = !catalog.isTempView(tableName)
     val tableExists =
       (tableName.database.isEmpty || catalog.databaseExists(tableName.database.get)) &&
       catalog.tableExists(tableName)
diff --git a/core/src/main/scala/org/apache/spark/sql/delta/DeltaTableIdentifier.scala b/core/src/main/scala/org/apache/spark/sql/delta/DeltaTableIdentifier.scala
index c83b358b1e5..a690ec919c0 100644
--- a/core/src/main/scala/org/apache/spark/sql/delta/DeltaTableIdentifier.scala
+++ b/core/src/main/scala/org/apache/spark/sql/delta/DeltaTableIdentifier.scala
@@ -77,7 +77,7 @@ object DeltaTableIdentifier extends Logging {
    */
   def isDeltaPath(spark: SparkSession, identifier: TableIdentifier): Boolean = {
     val catalog = spark.sessionState.catalog
-    def tableIsTemporaryTable = catalog.isTemporaryTable(identifier)
+    def tableIsTemporaryTable = catalog.isTempView(identifier)
     def tableExists: Boolean = {
       try {
         catalog.databaseExists(identifier.database.get) && catalog.tableExists(identifier)
diff --git a/core/src/main/scala/org/apache/spark/sql/delta/DeltaUnsupportedOperationsCheck.scala b/core/src/main/scala/org/apache/spark/sql/delta/DeltaUnsupportedOperationsCheck.scala
index d459a9b5f1b..993e632bc1c 100644
--- a/core/src/main/scala/org/apache/spark/sql/delta/DeltaUnsupportedOperationsCheck.scala
+++ b/core/src/main/scala/org/apache/spark/sql/delta/DeltaUnsupportedOperationsCheck.scala
@@ -83,7 +83,7 @@ case class DeltaUnsupportedOperationsCheck(spark: SparkSession)
       recordDeltaEvent(null, "delta.unsupported.dropPartition")
       fail(operation = "ALTER TABLE DROP PARTITION", a.tableName)
 
-    case a: AlterTableRecoverPartitionsCommand =>
+    case a: RepairTableCommand =>
       recordDeltaEvent(null, "delta.unsupported.recoverPartitions")
       fail(operation = "ALTER TABLE RECOVER PARTITIONS", a.tableName)
 
diff --git a/core/src/main/scala/org/apache/spark/sql/delta/DeltaViewHelper.scala b/core/src/main/scala/org/apache/spark/sql/delta/DeltaViewHelper.scala
index b383c5d710f..556817aa336 100644
--- a/core/src/main/scala/org/apache/spark/sql/delta/DeltaViewHelper.scala
+++ b/core/src/main/scala/org/apache/spark/sql/delta/DeltaViewHelper.scala
@@ -1,25 +1,3 @@
-/*
- * 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.
- */
-
-/*
- * This file contains code from the Apache Spark project (original license above).
- * It contains modifications, which are licensed as follows:
- */
-
 /*
  * Copyright (2021) The Delta Lake Project Authors.
  *
@@ -39,58 +17,94 @@
 package org.apache.spark.sql.delta
 
 import org.apache.spark.sql.catalyst.catalog.CatalogTable
-import org.apache.spark.sql.execution.datasources.LogicalRelation
 import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, Cast, NamedExpression}
 import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, Project, SubqueryAlias, View}
+import org.apache.spark.sql.execution.datasources.LogicalRelation
 import org.apache.spark.sql.internal.SQLConf
 
 object DeltaViewHelper {
-  /**
-   * Eliminate the view node from `plan`. Spark 3.1.1 introduces a View node in the plan when a
-   * view is a SQL view. We need to eliminate it manually in Spark 3.1.1 and above. This View node
-   * doesn't exist in Spark 3.0.x and below.
-   */
-  def stripTempView(plan: LogicalPlan, conf: SQLConf): LogicalPlan = plan transformUp {
-    case v @ View(desc, true, output, child) if child.resolved && !v.sameOutput(child) =>
-      val newOutput = makeNewOutput(desc, output, child, conf)
-      Project(newOutput, child)
+  def stripTempViewForMerge(plan: LogicalPlan, conf: SQLConf): LogicalPlan = {
+    // Check that the two expression lists have the same names and types in the same order, and
+    // are either attributes or direct casts of attributes.
+    def attributesMatch(left: Seq[NamedExpression], right: Seq[NamedExpression]): Boolean = {
+      if (left.length != right.length) return false
 
-    case View(_, true, _, child) =>
-      child
-  }
+      val allowedExprs = (left ++ right).forall {
+        case _: Attribute => true
+        case Alias(Cast(_: Attribute, dataType, timeZone, _), name) => true
+        case _ => false
+      }
 
-  private def makeNewOutput(
-      desc: CatalogTable,
-      output: Seq[Attribute],
-      child: LogicalPlan,
-      conf: SQLConf): Seq[NamedExpression] = {
-    val resolver = conf.resolver
-    val queryColumnNames = desc.viewQueryColumnNames
-    val queryOutput = if (queryColumnNames.nonEmpty) {
-      // Find the attribute that has the expected attribute name from an attribute list, the names
-      // are compared using conf.resolver.
-      // `CheckAnalysis` already guarantees the expected attribute can be found for sure.
-      desc.viewQueryColumnNames.map { colName =>
-        child.output.find(attr => resolver(attr.name, colName)).get
+      val exprsMatch = left.zip(right).forall {
+        case (a, b) => a.dataType == b.dataType && conf.resolver(a.name, b.name)
       }
-    } else {
-      // For view created before Spark 2.2.0, the view text is already fully qualified, the plan
-      // output is the same with the view output.
-      child.output
+
+      allowedExprs && exprsMatch
     }
-    // Map the attributes in the query output to the attributes in the view output by index.
-    output.zip(queryOutput).map {
-      case (attr, originAttr) if !attr.semanticEquals(originAttr) =>
-        // `CheckAnalysis` already guarantees that the cast is a up-cast for sure.
-        Alias(Cast(originAttr, attr.dataType), attr.name)(exprId = attr.exprId,
-          qualifier = attr.qualifier, explicitMetadata = Some(attr.metadata))
-      case (_, originAttr) => originAttr
+
+
+    // We have to do a pretty complicated transformation here to support using two specific things
+    // which are not a Delta table as the target of Delta DML commands:
+    // A view defined as `SELECT * FROM underlying_tbl`
+    // A view defined as `SELECT * FROM underlying_tbl as alias`
+    // This requires stripping their intermediate nodes and pulling out just the scan, because
+    // some of our internal attribute fiddling requires the target plan to have the same attribute
+    // IDs as the underlying scan.
+    object ViewPlan {
+      def unapply(
+          plan: LogicalPlan): Option[(CatalogTable, Seq[NamedExpression], LogicalRelation)] = {
+        // A `SELECT * from underlying_table` view will have:
+        // * A View node marking it as a view.
+        // * An outer Project explicitly casting the scanned types to the types defined in the
+        //   metastore for the view. We don't need this cast for Delta DML commands and it will
+        //   end up being eliminated.
+        // * An inner no-op project.
+        // * A SubqueryAlias explicitly aliasing the scan to its own name (plus another if there's
+        //   a user specified alias.
+        // * The actual scan of the Delta table.
+        // We check for these Projects by ensuring that the name lists are an exact match, and
+        // produce a scan with the outer list's attribute IDs aliased to the view's name.
+        plan match {
+          case View(desc, true, // isTempView
+          Project(outerList,
+          Project(innerList,
+          SubqueryAlias(innerAlias, scan: LogicalRelation))))
+            if attributesMatch(outerList, innerList) && attributesMatch(outerList, scan.output) =>
+            Some(desc, outerList, scan)
+          case View(desc, true, // isTempView
+          Project(outerList,
+          Project(innerList,
+          SubqueryAlias(innerAlias, SubqueryAlias(subalias, scan: LogicalRelation)))))
+            if attributesMatch(outerList, innerList) && attributesMatch(outerList, scan.output) =>
+            Some(desc, outerList, scan)
+          case _ => None
+        }
+      }
+    }
+
+    plan.transformUp {
+      case ViewPlan(desc, outerList, scan) =>
+        val newOutput = scan.output.map { oldAttr =>
+          val newId = outerList.collectFirst {
+            case newAttr if conf.resolver(oldAttr.qualifiedName, newAttr.qualifiedName) =>
+              newAttr.exprId
+          }.getOrElse {
+            throw new IllegalStateException(
+              s"Could not find a new attribute ID for column ${oldAttr.qualifiedName}. This " +
+                s"should have been checked earlier.")
+          }
+          oldAttr.withExprId(newId)
+        }
+        SubqueryAlias(desc.qualifiedName, scan.copy(output = newOutput))
+
+      case v: View if v.isTempView =>
+        v.child
     }
   }
 
-  def stripTempViewForMerge(plan: LogicalPlan, conf: SQLConf): LogicalPlan = {
-    // Check that the two expression lists have the same names and types in the same order, and
-    // are either attributes or direct casts of attributes.
-    stripTempView(plan, conf)
+  def stripTempView(plan: LogicalPlan, conf: SQLConf): LogicalPlan = {
+    plan.transformUp {
+      case v: View if v.isTempView => v.child
+    }
   }
 }
diff --git a/core/src/main/scala/org/apache/spark/sql/delta/catalog/DeltaCatalog.scala b/core/src/main/scala/org/apache/spark/sql/delta/catalog/DeltaCatalog.scala
index 4a732bf14d6..ec5ea150479 100644
--- a/core/src/main/scala/org/apache/spark/sql/delta/catalog/DeltaCatalog.scala
+++ b/core/src/main/scala/org/apache/spark/sql/delta/catalog/DeltaCatalog.scala
@@ -35,14 +35,14 @@ import org.apache.hadoop.fs.Path
 import org.apache.spark.internal.Logging
 import org.apache.spark.sql.{AnalysisException, DataFrame, SparkSession}
 import org.apache.spark.sql.catalyst.TableIdentifier
-import org.apache.spark.sql.catalyst.analysis.{NoSuchDatabaseException, NoSuchNamespaceException, NoSuchTableException, UnresolvedAttribute}
+import org.apache.spark.sql.catalyst.analysis.{NoSuchDatabaseException, NoSuchNamespaceException, NoSuchTableException, UnresolvedAttribute, UnresolvedFieldName, UnresolvedFieldPosition}
 import org.apache.spark.sql.catalyst.catalog.{BucketSpec, CatalogTable, CatalogTableType, CatalogUtils, SessionCatalog}
 import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, QualifiedColType}
 import org.apache.spark.sql.connector.catalog.{DelegatingCatalogExtension, Identifier, StagedTable, StagingTableCatalog, SupportsWrite, Table, TableCapability, TableCatalog, TableChange, V1Table}
 import org.apache.spark.sql.connector.catalog.TableCapability._
 import org.apache.spark.sql.connector.catalog.TableChange._
 import org.apache.spark.sql.connector.expressions.{BucketTransform, FieldReference, IdentityTransform, Transform}
-import org.apache.spark.sql.connector.write.{LogicalWriteInfo, V1WriteBuilder, WriteBuilder}
+import org.apache.spark.sql.connector.write.{LogicalWriteInfo, V1Write, WriteBuilder}
 import org.apache.spark.sql.execution.datasources.{DataSource, PartitioningUtils}
 import org.apache.spark.sql.execution.datasources.parquet.ParquetSchemaConverter
 import org.apache.spark.sql.internal.SQLConf
@@ -396,7 +396,7 @@ class DeltaCatalog extends DelegatingCatalogExtension
 
     override def capabilities(): util.Set[TableCapability] = Set(V1_BATCH_WRITE).asJava
 
-    override def newWriteBuilder(info: LogicalWriteInfo): V1WriteBuilder = {
+    override def newWriteBuilder(info: LogicalWriteInfo): WriteBuilder = {
       writeOptions = info.options.asCaseSensitiveMap().asScala.toMap
       new DeltaV1WriteBuilder
     }
@@ -404,11 +404,13 @@ class DeltaCatalog extends DelegatingCatalogExtension
     /*
      * WriteBuilder for creating a Delta table.
      */
-    private class DeltaV1WriteBuilder extends WriteBuilder with V1WriteBuilder {
-      override def buildForV1Write(): InsertableRelation = {
-        new InsertableRelation {
-          override def insert(data: DataFrame, overwrite: Boolean): Unit = {
-            asSelectQuery = Option(data)
+    private class DeltaV1WriteBuilder extends WriteBuilder {
+      override def build(): V1Write = new V1Write {
+        override def toInsertableRelation(): InsertableRelation = {
+          new InsertableRelation {
+            override def insert(data: DataFrame, overwrite: Boolean): Unit = {
+              asSelectQuery = Option(data)
+            }
           }
         }
       }
@@ -437,12 +439,17 @@ class DeltaCatalog extends DelegatingCatalogExtension
         AlterTableAddColumnsDeltaCommand(
           table,
           newColumns.asInstanceOf[Seq[AddColumn]].map { col =>
+            // Convert V2 `AddColumn` to V1 `QualifiedColType` as `AlterTableAddColumnsDeltaCommand`
+            // is a V1 command.
+            val name = col.fieldNames()
+            val path = if (name.length > 1) Some(UnresolvedFieldName(name.init)) else None
             QualifiedColType(
-              col.fieldNames(),
+              path,
+              name.last,
               col.dataType(),
               col.isNullable,
               Option(col.comment()),
-              Option(col.position()))
+              Option(col.position()).map(UnresolvedFieldPosition))
           }).run(spark)
 
       case (t, newProperties) if t == classOf[SetProperty] =>
diff --git a/core/src/main/scala/org/apache/spark/sql/delta/catalog/DeltaTableV2.scala b/core/src/main/scala/org/apache/spark/sql/delta/catalog/DeltaTableV2.scala
index fd685c28ce9..d344f8f982a 100644
--- a/core/src/main/scala/org/apache/spark/sql/delta/catalog/DeltaTableV2.scala
+++ b/core/src/main/scala/org/apache/spark/sql/delta/catalog/DeltaTableV2.scala
@@ -35,7 +35,7 @@ import org.apache.spark.sql.catalyst.catalog.{CatalogTable, CatalogUtils}
 import org.apache.spark.sql.connector.catalog.{SupportsWrite, Table, TableCapability, TableCatalog, V2TableWithV1Fallback}
 import org.apache.spark.sql.connector.catalog.TableCapability._
 import org.apache.spark.sql.connector.expressions._
-import org.apache.spark.sql.connector.write.{LogicalWriteInfo, SupportsOverwrite, SupportsTruncate, V1WriteBuilder, WriteBuilder}
+import org.apache.spark.sql.connector.write.{LogicalWriteInfo, SupportsOverwrite, SupportsTruncate, V1Write, WriteBuilder}
 import org.apache.spark.sql.execution.datasources.LogicalRelation
 import org.apache.spark.sql.sources.{BaseRelation, Filter, InsertableRelation}
 import org.apache.spark.sql.types.StructType
@@ -188,7 +188,7 @@ case class DeltaTableV2(
 private class WriteIntoDeltaBuilder(
     log: DeltaLog,
     writeOptions: CaseInsensitiveStringMap)
-  extends WriteBuilder with V1WriteBuilder with SupportsOverwrite with SupportsTruncate {
+  extends WriteBuilder with SupportsOverwrite with SupportsTruncate {
 
   private var forceOverwrite = false
 
@@ -210,24 +210,26 @@ private class WriteIntoDeltaBuilder(
     this
   }
 
-  override def buildForV1Write(): InsertableRelation = {
-    new InsertableRelation {
-      override def insert(data: DataFrame, overwrite: Boolean): Unit = {
-        val session = data.sparkSession
-
-        WriteIntoDelta(
-          log,
-          if (forceOverwrite) SaveMode.Overwrite else SaveMode.Append,
-          new DeltaOptions(options.toMap, session.sessionState.conf),
-          Nil,
-          log.snapshot.metadata.configuration,
-          data).run(session)
-
-        // TODO: Push this to Apache Spark
-        // Re-cache all cached plans(including this relation itself, if it's cached) that refer
-        // to this data source relation. This is the behavior for InsertInto
-        session.sharedState.cacheManager.recacheByPlan(
-          session, LogicalRelation(log.createRelation()))
+  override def build(): V1Write = new V1Write {
+    override def toInsertableRelation(): InsertableRelation = {
+      new InsertableRelation {
+        override def insert(data: DataFrame, overwrite: Boolean): Unit = {
+          val session = data.sparkSession
+
+          WriteIntoDelta(
+            log,
+            if (forceOverwrite) SaveMode.Overwrite else SaveMode.Append,
+            new DeltaOptions(options.toMap, session.sessionState.conf),
+            Nil,
+            log.snapshot.metadata.configuration,
+            data).run(session)
+
+          // TODO: Push this to Apache Spark
+          // Re-cache all cached plans(including this relation itself, if it's cached) that refer
+          // to this data source relation. This is the behavior for InsertInto
+          session.sharedState.cacheManager.recacheByPlan(
+            session, LogicalRelation(log.createRelation()))
+        }
       }
     }
   }
diff --git a/core/src/main/scala/org/apache/spark/sql/delta/commands/ConvertToDeltaCommand.scala b/core/src/main/scala/org/apache/spark/sql/delta/commands/ConvertToDeltaCommand.scala
index aebdbc584e1..6d364f202b8 100644
--- a/core/src/main/scala/org/apache/spark/sql/delta/commands/ConvertToDeltaCommand.scala
+++ b/core/src/main/scala/org/apache/spark/sql/delta/commands/ConvertToDeltaCommand.scala
@@ -39,7 +39,7 @@ import org.apache.spark.sql.catalyst.catalog.{CatalogTable, CatalogTableType, Se
 import org.apache.spark.sql.catalyst.expressions.Cast
 import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
 import org.apache.spark.sql.connector.catalog.{Identifier, TableCatalog, V1Table}
-import org.apache.spark.sql.execution.command.RunnableCommand
+import org.apache.spark.sql.execution.command.LeafRunnableCommand
 import org.apache.spark.sql.execution.datasources.PartitioningUtils
 import org.apache.spark.sql.execution.datasources.parquet.{ParquetFileFormat, ParquetToSparkSchemaConverter}
 import org.apache.spark.sql.execution.streaming.{FileStreamSink, MetadataLogFileIndex}
@@ -71,7 +71,7 @@ import org.apache.spark.util.SerializableConfiguration
 abstract class ConvertToDeltaCommandBase(
     tableIdentifier: TableIdentifier,
     partitionSchema: Option[StructType],
-    deltaPath: Option[String]) extends RunnableCommand with DeltaCommand {
+    deltaPath: Option[String]) extends LeafRunnableCommand with DeltaCommand {
 
   protected def isSupportedProvider(lowerCaseProvider: String): Boolean = {
     lowerCaseProvider == "parquet"
@@ -407,9 +407,7 @@ case class ConvertToDeltaCommand(
     tableIdentifier: TableIdentifier,
     partitionSchema: Option[StructType],
     deltaPath: Option[String])
-  extends ConvertToDeltaCommandBase(tableIdentifier, partitionSchema, deltaPath) {
-  // TODO: remove when the new Spark version is releases that has the withNewChildInternal method
-}
+  extends ConvertToDeltaCommandBase(tableIdentifier, partitionSchema, deltaPath)
 
 /**
  * An interface for the table to be converted to Delta.
diff --git a/core/src/main/scala/org/apache/spark/sql/delta/commands/CreateDeltaTableCommand.scala b/core/src/main/scala/org/apache/spark/sql/delta/commands/CreateDeltaTableCommand.scala
index b84ac870405..983202f7ca6 100644
--- a/core/src/main/scala/org/apache/spark/sql/delta/commands/CreateDeltaTableCommand.scala
+++ b/core/src/main/scala/org/apache/spark/sql/delta/commands/CreateDeltaTableCommand.scala
@@ -31,7 +31,7 @@ import org.apache.spark.sql.catalyst.catalog.{CatalogTable, CatalogTableType}
 import org.apache.spark.sql.catalyst.expressions.Attribute
 import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
 import org.apache.spark.sql.connector.catalog.Identifier
-import org.apache.spark.sql.execution.command.RunnableCommand
+import org.apache.spark.sql.execution.command.{LeafRunnableCommand, RunnableCommand}
 import org.apache.spark.sql.types.StructType
 
 /**
@@ -54,7 +54,7 @@ case class CreateDeltaTableCommand(
     operation: TableCreationModes.CreationMode = TableCreationModes.Create,
     tableByPath: Boolean = false,
     override val output: Seq[Attribute] = Nil)
-  extends RunnableCommand
+  extends LeafRunnableCommand
   with DeltaLogging {
 
   override def run(sparkSession: SparkSession): Seq[Row] = {
@@ -442,8 +442,6 @@ case class CreateDeltaTableCommand(
     Thread.currentThread().getStackTrace.exists(_.toString.contains(
       classOf[DataFrameWriter[_]].getCanonicalName + "."))
   }
-
-  // TODO: remove when the new Spark version is releases that has the withNewChildInternal method
 }
 
 object TableCreationModes {
diff --git a/core/src/main/scala/org/apache/spark/sql/delta/commands/DeleteCommand.scala b/core/src/main/scala/org/apache/spark/sql/delta/commands/DeleteCommand.scala
index 2b46743c7c0..051390cf7d4 100644
--- a/core/src/main/scala/org/apache/spark/sql/delta/commands/DeleteCommand.scala
+++ b/core/src/main/scala/org/apache/spark/sql/delta/commands/DeleteCommand.scala
@@ -27,7 +27,7 @@ import org.apache.spark.sql.catalyst.expressions.{EqualNullSafe, Expression, Inp
 import org.apache.spark.sql.catalyst.plans.QueryPlan
 import org.apache.spark.sql.catalyst.plans.logical.{DeltaDelete, LogicalPlan}
 import org.apache.spark.sql.execution.SQLExecution
-import org.apache.spark.sql.execution.command.RunnableCommand
+import org.apache.spark.sql.execution.command.LeafRunnableCommand
 import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics}
 import org.apache.spark.sql.execution.metric.SQLMetrics.createMetric
 import org.apache.spark.sql.functions.udf
@@ -46,7 +46,7 @@ case class DeleteCommand(
     deltaLog: DeltaLog,
     target: LogicalPlan,
     condition: Option[Expression])
-  extends RunnableCommand with DeltaCommand {
+  extends LeafRunnableCommand with DeltaCommand {
 
   override def innerChildren: Seq[QueryPlan[_]] = Seq(target)
 
@@ -211,8 +211,6 @@ case class DeleteCommand(
 
     deleteActions
   }
-
-  // TODO: remove when the new Spark version is releases that has the withNewChildInternal method
 }
 
 object DeleteCommand {
diff --git a/core/src/main/scala/org/apache/spark/sql/delta/commands/DeltaGenerateCommand.scala b/core/src/main/scala/org/apache/spark/sql/delta/commands/DeltaGenerateCommand.scala
index 372da7ce6ef..2619e8513e6 100644
--- a/core/src/main/scala/org/apache/spark/sql/delta/commands/DeltaGenerateCommand.scala
+++ b/core/src/main/scala/org/apache/spark/sql/delta/commands/DeltaGenerateCommand.scala
@@ -25,10 +25,10 @@ import org.apache.spark.sql.{Row, SparkSession}
 import org.apache.spark.sql.catalyst.TableIdentifier
 import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
 import org.apache.spark.sql.catalyst.util.CaseInsensitiveMap
-import org.apache.spark.sql.execution.command.RunnableCommand
+import org.apache.spark.sql.execution.command.LeafRunnableCommand
 
 case class DeltaGenerateCommand(modeName: String, tableId: TableIdentifier)
-  extends RunnableCommand {
+  extends LeafRunnableCommand {
 
   import DeltaGenerateCommand._
 
@@ -52,8 +52,6 @@ case class DeltaGenerateCommand(modeName: String, tableId: TableIdentifier)
     generationFunc(sparkSession, deltaLog)
     Seq.empty
   }
-
-  // TODO: remove when the new Spark version is releases that has the withNewChildInternal method
 }
 
 object DeltaGenerateCommand {
diff --git a/core/src/main/scala/org/apache/spark/sql/delta/commands/DescribeDeltaDetailsCommand.scala b/core/src/main/scala/org/apache/spark/sql/delta/commands/DescribeDeltaDetailsCommand.scala
index e6096608356..4a76429b5cd 100644
--- a/core/src/main/scala/org/apache/spark/sql/delta/commands/DescribeDeltaDetailsCommand.scala
+++ b/core/src/main/scala/org/apache/spark/sql/delta/commands/DescribeDeltaDetailsCommand.scala
@@ -32,7 +32,7 @@ import org.apache.spark.sql.catalyst.analysis.NoSuchTableException
 import org.apache.spark.sql.catalyst.catalog.{CatalogTable, CatalogTableType, CatalogUtils}
 import org.apache.spark.sql.catalyst.expressions.Attribute
 import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
-import org.apache.spark.sql.execution.command.RunnableCommand
+import org.apache.spark.sql.execution.command.LeafRunnableCommand
 import org.apache.spark.sql.types.StructType
 
 /** The result returned by the `describe detail` command. */
@@ -68,7 +68,7 @@ object TableDetail {
  */
 case class DescribeDeltaDetailCommand(
     path: Option[String],
-    tableIdentifier: Option[TableIdentifier]) extends RunnableCommand with DeltaLogging {
+    tableIdentifier: Option[TableIdentifier]) extends LeafRunnableCommand with DeltaLogging {
 
   override val output: Seq[Attribute] = TableDetail.schema.toAttributes
 
@@ -199,6 +199,4 @@ case class DescribeDeltaDetailCommand(
         snapshot.protocol.minReaderVersion,
         snapshot.protocol.minWriterVersion))
   }
-
-  // TODO: remove when the new Spark version is releases that has the withNewChildInternal method
 }
diff --git a/core/src/main/scala/org/apache/spark/sql/delta/commands/DescribeDeltaHistoryCommand.scala b/core/src/main/scala/org/apache/spark/sql/delta/commands/DescribeDeltaHistoryCommand.scala
index 21125c74e62..03ead01d092 100644
--- a/core/src/main/scala/org/apache/spark/sql/delta/commands/DescribeDeltaHistoryCommand.scala
+++ b/core/src/main/scala/org/apache/spark/sql/delta/commands/DescribeDeltaHistoryCommand.scala
@@ -28,7 +28,7 @@ import org.apache.spark.sql.catalyst.catalog.CatalogTableType
 import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
 import org.apache.spark.sql.catalyst.expressions.Attribute
 import org.apache.spark.sql.catalyst.plans.logical.{LeafNode, LogicalPlan, Statistics}
-import org.apache.spark.sql.execution.command.RunnableCommand
+import org.apache.spark.sql.execution.command.LeafRunnableCommand
 
 /**
  * A logical placeholder for describing a Delta table's history, so that the history can be
@@ -53,7 +53,7 @@ case class DescribeDeltaHistoryCommand(
     tableIdentifier: Option[TableIdentifier],
     limit: Option[Int],
     override val output: Seq[Attribute] = ExpressionEncoder[DeltaHistory]().schema.toAttributes)
-  extends RunnableCommand with DeltaLogging {
+  extends LeafRunnableCommand with DeltaLogging {
 
   override def run(sparkSession: SparkSession): Seq[Row] = {
     val basePath =
@@ -93,6 +93,4 @@ case class DescribeDeltaHistoryCommand(
       deltaLog.history.getHistory(limit).toDF().collect().toSeq
     }
   }
-
-  // TODO: remove when the new Spark version is releases that has the withNewChildInternal method
 }
diff --git a/core/src/main/scala/org/apache/spark/sql/delta/commands/MergeIntoCommand.scala b/core/src/main/scala/org/apache/spark/sql/delta/commands/MergeIntoCommand.scala
index 2c601ad12cd..438fc258fbe 100644
--- a/core/src/main/scala/org/apache/spark/sql/delta/commands/MergeIntoCommand.scala
+++ b/core/src/main/scala/org/apache/spark/sql/delta/commands/MergeIntoCommand.scala
@@ -39,7 +39,7 @@ import org.apache.spark.sql.catalyst.expressions.NamedExpression
 import org.apache.spark.sql.catalyst.expressions.codegen._
 import org.apache.spark.sql.catalyst.plans.logical._
 import org.apache.spark.sql.execution.SQLExecution
-import org.apache.spark.sql.execution.command.RunnableCommand
+import org.apache.spark.sql.execution.command.LeafRunnableCommand
 import org.apache.spark.sql.execution.datasources.LogicalRelation
 import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics}
 import org.apache.spark.sql.functions._
@@ -205,7 +205,7 @@ case class MergeIntoCommand(
     condition: Expression,
     matchedClauses: Seq[DeltaMergeIntoMatchedClause],
     notMatchedClauses: Seq[DeltaMergeIntoInsertClause],
-    migratedSchema: Option[StructType]) extends RunnableCommand
+    migratedSchema: Option[StructType]) extends LeafRunnableCommand
   with DeltaCommand with PredicateHelper with AnalysisHelper with ImplicitMetadataOperation {
 
   import SQLMetrics._
@@ -658,8 +658,6 @@ case class MergeIntoCommand(
     }
     r
   }
-
-  // TODO: remove when the new Spark version is releases that has the withNewChildInternal method
 }
 
 object MergeIntoCommand {
diff --git a/core/src/main/scala/org/apache/spark/sql/delta/commands/UpdateCommand.scala b/core/src/main/scala/org/apache/spark/sql/delta/commands/UpdateCommand.scala
index 163bd78cfa4..3ae6a085b6a 100644
--- a/core/src/main/scala/org/apache/spark/sql/delta/commands/UpdateCommand.scala
+++ b/core/src/main/scala/org/apache/spark/sql/delta/commands/UpdateCommand.scala
@@ -27,7 +27,7 @@ import org.apache.spark.sql.catalyst.expressions.{Alias, Expression, If, Literal
 import org.apache.spark.sql.catalyst.plans.QueryPlan
 import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
 import org.apache.spark.sql.execution.SQLExecution
-import org.apache.spark.sql.execution.command.RunnableCommand
+import org.apache.spark.sql.execution.command.LeafRunnableCommand
 import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics}
 import org.apache.spark.sql.execution.metric.SQLMetrics.createMetric
 import org.apache.spark.sql.functions.{input_file_name, udf}
@@ -47,7 +47,7 @@ case class UpdateCommand(
     target: LogicalPlan,
     updateExpressions: Seq[Expression],
     condition: Option[Expression])
-  extends RunnableCommand with DeltaCommand {
+  extends LeafRunnableCommand with DeltaCommand {
 
   override def innerChildren: Seq[QueryPlan[_]] = Seq(target)
 
@@ -235,9 +235,6 @@ case class UpdateCommand(
       new Column(Alias(updated, original.name)())
     }
   }
-
-  // TODO: remove when the new Spark version is releases that has the withNewChildInternal method
-
 }
 
 object UpdateCommand {
diff --git a/core/src/main/scala/org/apache/spark/sql/delta/commands/WriteIntoDelta.scala b/core/src/main/scala/org/apache/spark/sql/delta/commands/WriteIntoDelta.scala
index 92eb75a6277..8236413d4d8 100644
--- a/core/src/main/scala/org/apache/spark/sql/delta/commands/WriteIntoDelta.scala
+++ b/core/src/main/scala/org/apache/spark/sql/delta/commands/WriteIntoDelta.scala
@@ -30,7 +30,7 @@ import org.apache.spark.sql.catalyst.analysis.UnresolvedAttribute
 import org.apache.spark.sql.catalyst.expressions.{And, Expression}
 import org.apache.spark.sql.catalyst.plans.logical.{DeleteFromTable, LogicalPlan}
 import org.apache.spark.sql.catalyst.util.CharVarcharUtils
-import org.apache.spark.sql.execution.command.RunnableCommand
+import org.apache.spark.sql.execution.command.LeafRunnableCommand
 import org.apache.spark.sql.execution.datasources.LogicalRelation
 import org.apache.spark.sql.types.StructType
 
@@ -63,7 +63,7 @@ case class WriteIntoDelta(
     configuration: Map[String, String],
     data: DataFrame,
     schemaInCatalog: Option[StructType] = None)
-  extends RunnableCommand
+  extends LeafRunnableCommand
   with ImplicitMetadataOperation
   with DeltaCommand {
 
diff --git a/core/src/main/scala/org/apache/spark/sql/delta/commands/alterDeltaTableCommands.scala b/core/src/main/scala/org/apache/spark/sql/delta/commands/alterDeltaTableCommands.scala
index c3c3529b83a..4f4fee71d5b 100644
--- a/core/src/main/scala/org/apache/spark/sql/delta/commands/alterDeltaTableCommands.scala
+++ b/core/src/main/scala/org/apache/spark/sql/delta/commands/alterDeltaTableCommands.scala
@@ -37,7 +37,7 @@ import org.apache.spark.sql.catalyst.expressions.{Expression, IsNotNull, IsNull,
 import org.apache.spark.sql.catalyst.plans.logical.{IgnoreCachedData, LogicalPlan, QualifiedColType}
 import org.apache.spark.sql.connector.catalog.TableCatalog
 import org.apache.spark.sql.connector.catalog.TableChange.{After, ColumnPosition, First}
-import org.apache.spark.sql.execution.command.RunnableCommand
+import org.apache.spark.sql.execution.command.LeafRunnableCommand
 import org.apache.spark.sql.types._
 
 /**
@@ -67,7 +67,7 @@ trait AlterDeltaTableCommand extends DeltaCommand {
 case class AlterTableSetPropertiesDeltaCommand(
     table: DeltaTableV2,
     configuration: Map[String, String])
-  extends RunnableCommand with AlterDeltaTableCommand with IgnoreCachedData {
+  extends LeafRunnableCommand with AlterDeltaTableCommand with IgnoreCachedData {
 
   override def run(sparkSession: SparkSession): Seq[Row] = {
     val deltaLog = table.deltaLog
@@ -98,8 +98,6 @@ case class AlterTableSetPropertiesDeltaCommand(
       Seq.empty[Row]
     }
   }
-
-  // TODO: remove when the new Spark version is releases that has the withNewChildInternal method
 }
 
 /**
@@ -116,7 +114,7 @@ case class AlterTableUnsetPropertiesDeltaCommand(
     table: DeltaTableV2,
     propKeys: Seq[String],
     ifExists: Boolean)
-  extends RunnableCommand with AlterDeltaTableCommand with IgnoreCachedData {
+  extends LeafRunnableCommand with AlterDeltaTableCommand with IgnoreCachedData {
 
   override def run(sparkSession: SparkSession): Seq[Row] = {
     val deltaLog = table.deltaLog
@@ -149,8 +147,6 @@ case class AlterTableUnsetPropertiesDeltaCommand(
       Seq.empty[Row]
     }
   }
-
-  // TODO: remove when the new Spark version is releases that has the withNewChildInternal method
 }
 
 /**
@@ -164,7 +160,7 @@ case class AlterTableUnsetPropertiesDeltaCommand(
 case class AlterTableAddColumnsDeltaCommand(
     table: DeltaTableV2,
     colsToAddWithPosition: Seq[QualifiedColType])
-  extends RunnableCommand with AlterDeltaTableCommand with IgnoreCachedData {
+  extends LeafRunnableCommand with AlterDeltaTableCommand with IgnoreCachedData {
 
   override def run(sparkSession: SparkSession): Seq[Row] = {
     val deltaLog = table.deltaLog
@@ -225,7 +221,7 @@ case class AlterTableAddColumnsDeltaCommand(
   object QualifiedColTypeWithPosition {
 
     private def toV2Position(input: Any): ColumnPosition = {
-      input.asInstanceOf[ColumnPosition]
+      input.asInstanceOf[org.apache.spark.sql.catalyst.analysis.FieldPosition].position
     }
 
     def unapply(
@@ -238,8 +234,6 @@ case class AlterTableAddColumnsDeltaCommand(
       Some((col.name.init, field, col.position.map(toV2Position)))
     }
   }
-
-  // TODO: remove when the new Spark version is releases that has the withNewChildInternal method
 }
 
 /**
@@ -259,7 +253,7 @@ case class AlterTableChangeColumnDeltaCommand(
     columnName: String,
     newColumn: StructField,
     colPosition: Option[ColumnPosition])
-  extends RunnableCommand with AlterDeltaTableCommand with IgnoreCachedData {
+  extends LeafRunnableCommand with AlterDeltaTableCommand with IgnoreCachedData {
 
   override def run(sparkSession: SparkSession): Seq[Row] = {
     val deltaLog = table.deltaLog
@@ -405,8 +399,6 @@ case class AlterTableChangeColumnDeltaCommand(
           s" (nullable = ${newColumn.nullable})'")
     }
   }
-
-  // TODO: remove when the new Spark version is releases that has the withNewChildInternal method
 }
 
 /**
@@ -421,7 +413,7 @@ case class AlterTableChangeColumnDeltaCommand(
 case class AlterTableReplaceColumnsDeltaCommand(
     table: DeltaTableV2,
     columns: Seq[StructField])
-  extends RunnableCommand with AlterDeltaTableCommand with IgnoreCachedData {
+  extends LeafRunnableCommand with AlterDeltaTableCommand with IgnoreCachedData {
 
   override def run(sparkSession: SparkSession): Seq[Row] = {
     val deltaLog = table.deltaLog
@@ -452,8 +444,6 @@ case class AlterTableReplaceColumnsDeltaCommand(
       Seq.empty[Row]
     }
   }
-
-  // TODO: remove when the new Spark version is releases that has the withNewChildInternal method
 }
 
 /**
@@ -472,7 +462,7 @@ case class AlterTableReplaceColumnsDeltaCommand(
 case class AlterTableSetLocationDeltaCommand(
     table: DeltaTableV2,
     location: String)
-  extends RunnableCommand with AlterDeltaTableCommand with IgnoreCachedData {
+  extends LeafRunnableCommand with AlterDeltaTableCommand with IgnoreCachedData {
 
   override def run(sparkSession: SparkSession): Seq[Row] = {
     val catalog = sparkSession.sessionState.catalog
@@ -511,8 +501,6 @@ case class AlterTableSetLocationDeltaCommand(
       dropColumnMappingMetadata(oldMetadata.partitionSchema) ==
         dropColumnMappingMetadata(newMetadata.partitionSchema)
   }
-
-  // TODO: remove when the new Spark version is releases that has the withNewChildInternal method
 }
 
 /**
@@ -528,7 +516,7 @@ case class AlterTableAddConstraintDeltaCommand(
     table: DeltaTableV2,
     name: String,
     exprText: String)
-  extends RunnableCommand with AlterDeltaTableCommand with IgnoreCachedData {
+  extends LeafRunnableCommand with AlterDeltaTableCommand with IgnoreCachedData {
 
   override def run(sparkSession: SparkSession): Seq[Row] = {
     val deltaLog = table.deltaLog
@@ -568,8 +556,6 @@ case class AlterTableAddConstraintDeltaCommand(
     }
     Seq()
   }
-
-  // TODO: remove when the new Spark version is releases that has the withNewChildInternal method
 }
 
 /**
@@ -584,7 +570,7 @@ case class AlterTableAddConstraintDeltaCommand(
 case class AlterTableDropConstraintDeltaCommand(
     table: DeltaTableV2,
     name: String)
-  extends RunnableCommand with AlterDeltaTableCommand with IgnoreCachedData {
+  extends LeafRunnableCommand with AlterDeltaTableCommand with IgnoreCachedData {
 
   override def run(sparkSession: SparkSession): Seq[Row] = {
     val deltaLog = table.deltaLog
@@ -600,6 +586,4 @@ case class AlterTableDropConstraintDeltaCommand(
 
     Seq()
   }
-
-  // TODO: remove when the new Spark version is releases that has the withNewChildInternal method
 }
diff --git a/core/src/main/scala/org/apache/spark/sql/delta/constraints/CheckDeltaInvariant.scala b/core/src/main/scala/org/apache/spark/sql/delta/constraints/CheckDeltaInvariant.scala
index 1ec8e39342b..649c155029e 100644
--- a/core/src/main/scala/org/apache/spark/sql/delta/constraints/CheckDeltaInvariant.scala
+++ b/core/src/main/scala/org/apache/spark/sql/delta/constraints/CheckDeltaInvariant.scala
@@ -135,5 +135,6 @@ case class CheckDeltaInvariant(
     ev.copy(code = code, isNull = TrueLiteral, value = JavaCode.literal("null", NullType))
   }
 
-  // TODO: remove when the new Spark version is releases that has the withNewChildInternal method
+  override protected def withNewChildInternal(newChild: Expression): CheckDeltaInvariant =
+    copy(child = newChild)
 }
diff --git a/core/src/main/scala/org/apache/spark/sql/delta/constraints/DeltaInvariantCheckerExec.scala b/core/src/main/scala/org/apache/spark/sql/delta/constraints/DeltaInvariantCheckerExec.scala
index 34bc7e37687..bfe3da42ae6 100644
--- a/core/src/main/scala/org/apache/spark/sql/delta/constraints/DeltaInvariantCheckerExec.scala
+++ b/core/src/main/scala/org/apache/spark/sql/delta/constraints/DeltaInvariantCheckerExec.scala
@@ -47,7 +47,8 @@ case class DeltaInvariantChecker(
   extends UnaryNode {
   override def output: Seq[Attribute] = child.output
 
-  // TODO: remove when the new Spark version is releases that has the withNewChildInternal method
+  override protected def withNewChildInternal(newChild: LogicalPlan): DeltaInvariantChecker =
+    copy(child = newChild)
 }
 
 object DeltaInvariantCheckerStrategy extends SparkStrategy {
@@ -72,16 +73,12 @@ case class DeltaInvariantCheckerExec(
     child: SparkPlan,
     constraints: Seq[Constraint]) extends UnaryExecNode {
 
-  // TODO: we can replace `SparkSession.active` with `session` once OSS Delta
-  //       upgrades to Spark 3.2
-  private def spark: SparkSession = SparkSession.active
-
   override def output: Seq[Attribute] = child.output
 
   override protected def doExecute(): RDD[InternalRow] = {
     if (constraints.isEmpty) return child.execute()
     val invariantChecks =
-      DeltaInvariantCheckerExec.buildInvariantChecks(child.output, constraints, spark)
+      DeltaInvariantCheckerExec.buildInvariantChecks(child.output, constraints, session)
     val boundRefs = invariantChecks.map(_.withBoundReferences(child.output))
 
     child.execute().mapPartitionsInternal { rows =>
@@ -97,7 +94,8 @@ case class DeltaInvariantCheckerExec(
 
   override def outputPartitioning: Partitioning = child.outputPartitioning
 
-  // TODO: remove when the new Spark version is releases that has the withNewChildInternal method
+  override protected def withNewChildInternal(newChild: SparkPlan): DeltaInvariantCheckerExec =
+    copy(child = newChild)
 }
 
 object DeltaInvariantCheckerExec {
diff --git a/core/src/main/scala/org/apache/spark/sql/delta/files/TransactionalWrite.scala b/core/src/main/scala/org/apache/spark/sql/delta/files/TransactionalWrite.scala
index f2ccb724b6f..a0107d7ca5c 100644
--- a/core/src/main/scala/org/apache/spark/sql/delta/files/TransactionalWrite.scala
+++ b/core/src/main/scala/org/apache/spark/sql/delta/files/TransactionalWrite.scala
@@ -212,7 +212,7 @@ trait TransactionalWrite extends DeltaLogging { self: OptimisticTransactionImpl
         val basicWriteJobStatsTracker = new BasicWriteJobStatsTracker(
           new SerializableConfiguration(deltaLog.newDeltaHadoopConf()),
           BasicWriteJobStatsTracker.metrics)
-        registerSQLMetrics(spark, basicWriteJobStatsTracker.metrics)
+        registerSQLMetrics(spark, basicWriteJobStatsTracker.driverSideMetrics)
         statsTrackers.append(basicWriteJobStatsTracker)
       }
 
diff --git a/core/src/main/scala/org/apache/spark/sql/delta/schema/SchemaMergingUtils.scala b/core/src/main/scala/org/apache/spark/sql/delta/schema/SchemaMergingUtils.scala
index 68e738258ed..98bbb0a8706 100644
--- a/core/src/main/scala/org/apache/spark/sql/delta/schema/SchemaMergingUtils.scala
+++ b/core/src/main/scala/org/apache/spark/sql/delta/schema/SchemaMergingUtils.scala
@@ -245,8 +245,7 @@ object SchemaMergingUtils {
    * there's no valid cast.
    */
   private def typeForImplicitCast(sourceType: DataType, targetType: DataType): Option[DataType] = {
-    TypeCoercion.ImplicitTypeCasts.implicitCast(Literal.default(sourceType), targetType)
-      .map(_.dataType)
+    TypeCoercion.implicitCast(Literal.default(sourceType), targetType).map(_.dataType)
   }
 
   def toFieldMap(fields: Seq[StructField]): Map[String, StructField] = {
diff --git a/core/src/main/scala/org/apache/spark/sql/delta/schema/SchemaUtils.scala b/core/src/main/scala/org/apache/spark/sql/delta/schema/SchemaUtils.scala
index 895e258a60e..5a8103220c4 100644
--- a/core/src/main/scala/org/apache/spark/sql/delta/schema/SchemaUtils.scala
+++ b/core/src/main/scala/org/apache/spark/sql/delta/schema/SchemaUtils.scala
@@ -877,11 +877,11 @@ object SchemaUtils {
 
   /**
    * Verifies that the column names are acceptable by Parquet and henceforth Delta. Parquet doesn't
-   * accept the characters ' ,;{}()\n\t'. We ensure that neither the data columns nor the partition
+   * accept the characters ' ,;{}()\n\t='. We ensure that neither the data columns nor the partition
    * columns have these characters.
    */
   def checkFieldNames(names: Seq[String]): Unit = {
-    ParquetSchemaConverter.checkFieldNames(names)
+    names.foreach(ParquetSchemaConverter.checkFieldName)
     // The method checkFieldNames doesn't have a valid regex to search for '\n'. That should be
     // fixed in Apache Spark, and we can remove this additional check here.
     names.find(_.contains("\n")).foreach(col => throw DeltaErrors.invalidColumnName(col))
diff --git a/core/src/main/scala/org/apache/spark/sql/delta/util/AnalysisHelper.scala b/core/src/main/scala/org/apache/spark/sql/delta/util/AnalysisHelper.scala
index c9889a8a2c7..a30f8fee23d 100644
--- a/core/src/main/scala/org/apache/spark/sql/delta/util/AnalysisHelper.scala
+++ b/core/src/main/scala/org/apache/spark/sql/delta/util/AnalysisHelper.scala
@@ -116,6 +116,7 @@ object AnalysisHelper {
     extends LogicalPlan {
     override def output: Seq[Attribute] = Nil
 
-    // TODO: remove when the new Spark version is releases that has the withNewChildInternal method
+    override protected def withNewChildrenInternal(
+      newChildren: IndexedSeq[LogicalPlan]): FakeLogicalPlan = copy(children = newChildren)
   }
 }
diff --git a/core/src/test/scala/org/apache/spark/sql/delta/ActionSerializerSuite.scala b/core/src/test/scala/org/apache/spark/sql/delta/ActionSerializerSuite.scala
index d8aa3c65e7e..acd945255ad 100644
--- a/core/src/test/scala/org/apache/spark/sql/delta/ActionSerializerSuite.scala
+++ b/core/src/test/scala/org/apache/spark/sql/delta/ActionSerializerSuite.scala
@@ -94,8 +94,11 @@ class ActionSerializerSuite extends QueryTest with SharedSparkSession {
     assert(action2.json === json2.replaceAll("\\s", ""))
   }
 
-
-  test("removefile") {
+  // This is the same test as "removefile" in OSS, but due to a Jackson library upgrade the behavior
+  // has diverged between Spark 3.1 and Spark 3.2.
+  // We don't believe this is a practical issue because all extant versions of Delta explicitly
+  // write the dataChange field.
+  test("remove file deserialization") {
     val removeJson = RemoveFile("a", Some(2L)).json
     assert(removeJson.contains(""""deletionTimestamp":2"""))
     assert(!removeJson.contains("""delTimestamp"""))
@@ -104,7 +107,7 @@ class ActionSerializerSuite extends QueryTest with SharedSparkSession {
     val json4 = """{"remove":{"path":"a","deletionTimestamp":5}}"""
     assert(Action.fromJson(json1) === RemoveFile("a", Some(2L), dataChange = true))
     assert(Action.fromJson(json2) === RemoveFile("a", None, dataChange = false))
-    assert(Action.fromJson(json4) === RemoveFile("a", Some(5L), dataChange = false))
+    assert(Action.fromJson(json4) === RemoveFile("a", Some(5L), dataChange = true))
   }
 
   roundTripCompare("SetTransaction",
diff --git a/core/src/test/scala/org/apache/spark/sql/delta/DeltaDDLSuite.scala b/core/src/test/scala/org/apache/spark/sql/delta/DeltaDDLSuite.scala
index 50c141b4374..54bad0e36a5 100644
--- a/core/src/test/scala/org/apache/spark/sql/delta/DeltaDDLSuite.scala
+++ b/core/src/test/scala/org/apache/spark/sql/delta/DeltaDDLSuite.scala
@@ -432,9 +432,12 @@ abstract class DeltaDDLTestBase extends QueryTest with SQLTestUtils {
   }
 
   /**
-   * SHOW CREATE TABLE is NOT supported in spark 3.1 for v2 tables.
+   * Although Spark 3.2 adds the support for SHOW CREATE TABLE for v2 tables, it doesn't work
+   * properly for some delta features, such as Delta constraints and generated columns.
+   *
+   * TODO(SC-83986): We should block it for unsupported tables
    */
-  test("SHOW CREATE TABLE should not include OPTIONS except for path - not supported") {
+  ignore("SHOW CREATE TABLE should not include OPTIONS except for path - not supported") {
     withTable("delta_test") {
       sql(
         s"""
diff --git a/core/src/test/scala/org/apache/spark/sql/delta/DeltaDDLUsingPathSuite.scala b/core/src/test/scala/org/apache/spark/sql/delta/DeltaDDLUsingPathSuite.scala
index 1b091cdbdf2..91ffa87c3e0 100644
--- a/core/src/test/scala/org/apache/spark/sql/delta/DeltaDDLUsingPathSuite.scala
+++ b/core/src/test/scala/org/apache/spark/sql/delta/DeltaDDLUsingPathSuite.scala
@@ -96,9 +96,7 @@ trait DeltaDDLUsingPathTests extends QueryTest
   }
 
   private def errorContains(errMsg: String, str: String): Unit = {
-    val actual = errMsg.replaceAll("`", "")
-    val expected = str.replaceAll("`", "")
-    assert(actual.contains(expected))
+    assert(errMsg.contains(str))
   }
 
   testUsingPath("SELECT") { (table, path) =>
diff --git a/core/src/test/scala/org/apache/spark/sql/delta/DeltaInsertIntoTableSuite.scala b/core/src/test/scala/org/apache/spark/sql/delta/DeltaInsertIntoTableSuite.scala
index a98f066e81b..5cd3325600d 100644
--- a/core/src/test/scala/org/apache/spark/sql/delta/DeltaInsertIntoTableSuite.scala
+++ b/core/src/test/scala/org/apache/spark/sql/delta/DeltaInsertIntoTableSuite.scala
@@ -877,6 +877,15 @@ trait InsertIntoSQLOnlyTests
       }
     }
 
+    test("InsertInto: overwrite - dot in column names - static mode") {
+      import testImplicits._
+      val t1 = "tbl"
+      withTable(t1) {
+        sql(s"CREATE TABLE $t1 (`a.b` string, `c.d` string) USING $v2Format PARTITIONED BY (`a.b`)")
+        sql(s"INSERT OVERWRITE $t1 PARTITION (`a.b` = 'a') VALUES('b')")
+        verifyTable(t1, Seq("a" -> "b").toDF("id", "data"))
+      }
+    }
   }
 
   // END Apache Spark tests
diff --git a/core/src/test/scala/org/apache/spark/sql/delta/DeltaNotSupportedDDLSuite.scala b/core/src/test/scala/org/apache/spark/sql/delta/DeltaNotSupportedDDLSuite.scala
index 89bf676a119..3517b6937de 100644
--- a/core/src/test/scala/org/apache/spark/sql/delta/DeltaNotSupportedDDLSuite.scala
+++ b/core/src/test/scala/org/apache/spark/sql/delta/DeltaNotSupportedDDLSuite.scala
@@ -127,22 +127,28 @@ abstract class DeltaNotSupportedDDLBase extends QueryTest
   test("ALTER TABLE ADD PARTITION") {
     assertUnsupported(
       s"ALTER TABLE $partitionedTableName ADD PARTITION (p1=3)",
-      "can not alter partitions")
+      "does not support partition management")
   }
 
   test("ALTER TABLE DROP PARTITION") {
     assertUnsupported(
       s"ALTER TABLE $partitionedTableName DROP PARTITION (p1=2)",
-      "can not alter partitions")
+      "does not support partition management")
   }
 
   test("ALTER TABLE RECOVER PARTITIONS") {
-    assertUnsupported(s"ALTER TABLE $partitionedTableName RECOVER PARTITIONS")
-    assertUnsupported(s"MSCK REPAIR TABLE $partitionedTableName")
+    assertUnsupported(
+      s"ALTER TABLE $partitionedTableName RECOVER PARTITIONS",
+      "alter table ... recover partitions is not supported for v2 tables")
+    assertUnsupported(
+      s"MSCK REPAIR TABLE $partitionedTableName",
+      "msck repair table is not supported for v2 tables")
   }
 
   test("ALTER TABLE SET SERDEPROPERTIES") {
-    assertUnsupported(s"ALTER TABLE $nonPartitionedTableName SET SERDEPROPERTIES (s1=3)")
+    assertUnsupported(
+      s"ALTER TABLE $nonPartitionedTableName SET SERDEPROPERTIES (s1=3)",
+      "alter table ... set [serde|serdeproperties] is not supported for v2 tables")
   }
 
 
diff --git a/core/src/test/scala/org/apache/spark/sql/delta/DeltaSuite.scala b/core/src/test/scala/org/apache/spark/sql/delta/DeltaSuite.scala
index 906de14a96f..5f5b1a71417 100644
--- a/core/src/test/scala/org/apache/spark/sql/delta/DeltaSuite.scala
+++ b/core/src/test/scala/org/apache/spark/sql/delta/DeltaSuite.scala
@@ -774,35 +774,6 @@ class DeltaSuite extends QueryTest
     }
   }
 
-  test("columns with commas as partition columns") {
-    withTempDir { tempDir =>
-      if (tempDir.exists()) {
-        assert(tempDir.delete())
-      }
-
-      val dfw = spark.range(100).select('id, 'id % 4 as "by,4")
-        .write
-        .format("delta")
-        .partitionBy("by,4")
-
-      // if in column mapping mode, we should not expect invalid character errors
-      if (!columnMappingEnabled) {
-        val e = intercept[AnalysisException] {
-          dfw.save(tempDir.toString)
-        }
-        assert(e.getMessage.contains("invalid character(s)"))
-      }
-
-      withSQLConf(DeltaSQLConf.DELTA_PARTITION_COLUMN_CHECK_ENABLED.key -> "false") {
-        dfw.save(tempDir.toString)
-      }
-
-      val files = spark.read.format("delta").load(tempDir.toString).inputFiles
-
-      val deltaLog = loadDeltaLog(tempDir.getAbsolutePath)
-      assertPartitionExists("by,4", deltaLog, files)
-    }
-  }
 
   test("throw exception when users are trying to write in batch with different partitioning") {
     withTempDir { tempDir =>
diff --git a/core/src/test/scala/org/apache/spark/sql/delta/DeltaTestUtils.scala b/core/src/test/scala/org/apache/spark/sql/delta/DeltaTestUtils.scala
index 3340d0d1412..a6f4d4ef125 100644
--- a/core/src/test/scala/org/apache/spark/sql/delta/DeltaTestUtils.scala
+++ b/core/src/test/scala/org/apache/spark/sql/delta/DeltaTestUtils.scala
@@ -168,12 +168,14 @@ trait DeltaTestUtilsForTempViews
         assert(ex.getMessage.contains(expectedErrorMsgForSQLTempView))
       }
       if (expectedErrorClassForSQLTempView != null) {
+        assert(ex.getErrorClass == expectedErrorClassForSQLTempView)
       }
     } else {
       if (expectedErrorMsgForDataSetTempView != null) {
         assert(ex.getMessage.contains(expectedErrorMsgForDataSetTempView))
       }
       if (expectedErrorClassForDataSetTempView != null) {
+        assert(ex.getErrorClass == expectedErrorClassForDataSetTempView, ex.getMessage)
       }
     }
   }
diff --git a/core/src/test/scala/org/apache/spark/sql/delta/DescribeDeltaHistorySuite.scala b/core/src/test/scala/org/apache/spark/sql/delta/DescribeDeltaHistorySuite.scala
index 4c4f773ba88..62e8cfe050e 100644
--- a/core/src/test/scala/org/apache/spark/sql/delta/DescribeDeltaHistorySuite.scala
+++ b/core/src/test/scala/org/apache/spark/sql/delta/DescribeDeltaHistorySuite.scala
@@ -461,6 +461,18 @@ trait DescribeDeltaHistorySuiteBase
       Seq($"operation", $"operationParameters.mode", $"operationParameters.predicate"))
   }
 
+  testWithFlag("operations - delete with predicate") {
+    val tempDir = Utils.createTempDir().toString
+    Seq((1, "a"), (2, "3")).toDF("id", "data").write.format("delta").partitionBy("id").save(tempDir)
+    val deltaLog = DeltaLog.forTable(spark, tempDir)
+    val deltaTable = io.delta.tables.DeltaTable.forPath(spark, deltaLog.dataPath.toString)
+    deltaTable.delete("id = 1")
+
+    checkLastOperation(
+      tempDir,
+      Seq("DELETE", """["(id = 1)"]"""),
+      Seq($"operation", $"operationParameters.predicate"))
+  }
 
   testWithFlag("old and new writers") {
     val tempDir = Utils.createTempDir().toString
diff --git a/core/src/test/scala/org/apache/spark/sql/delta/GeneratedColumnSuite.scala b/core/src/test/scala/org/apache/spark/sql/delta/GeneratedColumnSuite.scala
index bdfcf37dc74..8f01d50cb22 100644
--- a/core/src/test/scala/org/apache/spark/sql/delta/GeneratedColumnSuite.scala
+++ b/core/src/test/scala/org/apache/spark/sql/delta/GeneratedColumnSuite.scala
@@ -38,9 +38,7 @@ trait GeneratedColumnTest extends QueryTest with SharedSparkSession with DeltaSQ
 
 
   protected def sqlDate(date: String): java.sql.Date = {
-    toJavaDate(stringToDate(
-      UTF8String.fromString(date),
-      getZoneId(SQLConf.get.sessionLocalTimeZone)).get)
+    toJavaDate(stringToDate(UTF8String.fromString(date)).get)
   }
 
   protected def sqlTimestamp(timestamp: String): java.sql.Timestamp = {
@@ -194,9 +192,7 @@ trait GeneratedColumnSuiteBase extends GeneratedColumnTest {
   }
 
   private def errorContains(errMsg: String, str: String): Unit = {
-    val actual = errMsg.replaceAll("`", "")
-    val expected = str.replaceAll("`", "")
-    assert(actual.contains(expected))
+     assert(errMsg.contains(str))
   }
 
   testTableUpdate("append_data") { (table, path) =>
diff --git a/core/src/test/scala/org/apache/spark/sql/delta/MergeIntoSQLSuite.scala b/core/src/test/scala/org/apache/spark/sql/delta/MergeIntoSQLSuite.scala
index 2b99ad8da9a..6884bb65bf7 100644
--- a/core/src/test/scala/org/apache/spark/sql/delta/MergeIntoSQLSuite.scala
+++ b/core/src/test/scala/org/apache/spark/sql/delta/MergeIntoSQLSuite.scala
@@ -192,12 +192,8 @@ class MergeIntoSQLSuite extends MergeIntoSuiteBase  with DeltaSQLCommandTest
       Seq((1, 1), (0, 3)).toDF("srcKey", "srcValue").write.saveAsTable("source")
       append(Seq((2, 2), (1, 4)).toDF("trgKey", "trgValue"))
 
-      // TODO: In DBR we throw AnalysisException, but in OSS Delta we throw ParseException.
-      //       The error message is also slightly different. Here we just catch general Exception.
-      //       We should update this test when OSS delta upgrades to Spark 3.1.
-
       // only the last NOT MATCHED clause can omit the condition
-      val e = intercept[Exception](
+      val e = intercept[AnalysisException](
         sql(s"""
           |MERGE INTO delta.`$tempPath`
           |USING source
@@ -207,31 +203,41 @@ class MergeIntoSQLSuite extends MergeIntoSuiteBase  with DeltaSQLCommandTest
           |WHEN NOT MATCHED THEN
           |  INSERT (trgValue, trgKey) VALUES (srcValue, srcKey)
         """.stripMargin))
-      assert(e.getMessage.contains("only the last NOT MATCHED clause can omit the condition") ||
-        e.getMessage.contains("There should be at most 1 'WHEN NOT MATCHED' clause"))
+      assert(e.getMessage.contains("only the last NOT MATCHED clause can omit the condition"))
     }
   }
 
-
-  // This test is to capture the incorrect behavior caused by
-  // https://github.com/delta-io/delta/issues/618 .
-  // If this test fails then the issue has been fixed. Replace this test with a correct test
-  test("merge into a dataset temp views with star gives incorrect results") {
+  test("merge into a dataset temp views with star") {
     withTempView("v") {
-      withTempView("src") {
-        append(Seq((0, 0), (1, 1)).toDF("key", "value"))
-        readDeltaTable(tempPath).createOrReplaceTempView("v")
-        sql("CREATE TEMP VIEW src AS SELECT * FROM VALUES (10, 1) AS t(value, key)")
-        sql(s"""MERGE INTO v USING src
-             |ON src.key = v.key
-             |WHEN MATCHED THEN
-             |  UPDATE SET *
-             |WHEN NOT MATCHED THEN
-             |  INSERT *
-             |""".stripMargin)
-        val result = readDeltaTable(tempPath).as[(Long, Long)].collect().toSet
-        // This is expected to fail until the issue mentioned above is resolved.
-        assert(result != Set((0, 0), (1, 10)))
+      def testMergeWithView(testClue: String): Unit = {
+        withClue(testClue) {
+          withTempView("src") {
+            sql("CREATE TEMP VIEW src AS SELECT * FROM VALUES (10, 1), (20, 2) AS t(value, key)")
+            sql(
+              s"""
+                 |MERGE INTO v
+                 |USING src
+                 |ON src.key = v.key
+                 |WHEN MATCHED THEN
+                 |  UPDATE SET *
+                 |WHEN NOT MATCHED THEN
+                 |  INSERT *
+                 |""".stripMargin)
+            checkAnswer(spark.sql(s"select * from v"), Seq(Row(0, 0), Row(1, 10), Row(2, 20)))
+          }
+        }
+      }
+
+      // View on path-based table
+      append(Seq((0, 0), (1, 1)).toDF("key", "value"))
+      readDeltaTable(tempPath).createOrReplaceTempView("v")
+      testMergeWithView("with path-based table")
+
+      // View on catalog table
+      withTable("tab") {
+        Seq((0, 0), (1, 1)).toDF("key", "value").write.format("delta").saveAsTable("tab")
+        spark.table("tab").as("name").createOrReplaceTempView("v")
+        testMergeWithView(s"delta.`$tempPath`")
       }
     }
   }
diff --git a/core/src/test/scala/org/apache/spark/sql/delta/MergeIntoSuiteBase.scala b/core/src/test/scala/org/apache/spark/sql/delta/MergeIntoSuiteBase.scala
index 1815921f2ad..da0086f51d7 100644
--- a/core/src/test/scala/org/apache/spark/sql/delta/MergeIntoSuiteBase.scala
+++ b/core/src/test/scala/org/apache/spark/sql/delta/MergeIntoSuiteBase.scala
@@ -585,9 +585,7 @@ abstract class MergeIntoSuiteBase
   }
 
   protected def errorContains(errMsg: String, str: String): Unit = {
-    val actual = errMsg.replaceAll("`", "").toLowerCase(Locale.ROOT)
-    val expected = str.replaceAll("`", "").toLowerCase(Locale.ROOT)
-    assert(actual.contains(expected))
+    assert(errMsg.toLowerCase(Locale.ROOT).contains(str.toLowerCase(Locale.ROOT)))
   }
 
   def errorNotContains(errMsg: String, str: String): Unit = {
diff --git a/core/src/test/scala/org/apache/spark/sql/delta/UpdateSuiteBase.scala b/core/src/test/scala/org/apache/spark/sql/delta/UpdateSuiteBase.scala
index 03c3c56d51b..a77feb0d6cb 100644
--- a/core/src/test/scala/org/apache/spark/sql/delta/UpdateSuiteBase.scala
+++ b/core/src/test/scala/org/apache/spark/sql/delta/UpdateSuiteBase.scala
@@ -355,6 +355,7 @@ abstract class UpdateSuiteBase
       var ae = intercept[AnalysisException] {
         executeUpdate("table", set = "column_doesnt_exist = 'San Francisco'", where = "t = 'a'")
       }
+      assert(ae.getErrorClass == "MISSING_COLUMN")
 
       withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") {
         executeUpdate(target = "table", set = "S = 1, T = 'b'", where = "T = 'a'")
@@ -368,15 +369,18 @@ abstract class UpdateSuiteBase
         ae = intercept[AnalysisException] {
           executeUpdate(target = "table", set = "S = 1", where = "t = 'a'")
         }
+        assert(ae.getErrorClass == "MISSING_COLUMN")
 
         ae = intercept[AnalysisException] {
           executeUpdate(target = "table", set = "S = 1, s = 'b'", where = "s = 1")
         }
+        assert(ae.getErrorClass == "MISSING_COLUMN")
 
         // unresolved column in condition
         ae = intercept[AnalysisException] {
           executeUpdate(target = "table", set = "s = 1", where = "T = 'a'")
         }
+        assert(ae.getErrorClass == "MISSING_COLUMN")
       }
     }
   }
diff --git a/core/src/test/scala/org/apache/spark/sql/delta/schema/CaseSensitivitySuite.scala b/core/src/test/scala/org/apache/spark/sql/delta/schema/CaseSensitivitySuite.scala
index dad58397a15..4cc467fb1eb 100644
--- a/core/src/test/scala/org/apache/spark/sql/delta/schema/CaseSensitivitySuite.scala
+++ b/core/src/test/scala/org/apache/spark/sql/delta/schema/CaseSensitivitySuite.scala
@@ -195,6 +195,7 @@ class CaseSensitivitySuite extends QueryTest
             .option("replaceWhere", "key = 2") // note the different case
             .save(path)
         }
+        assert(e.getErrorClass == "MISSING_COLUMN")
       }
 
       checkAnswer(
diff --git a/core/src/test/scala/org/apache/spark/sql/delta/schema/CheckConstraintsSuite.scala b/core/src/test/scala/org/apache/spark/sql/delta/schema/CheckConstraintsSuite.scala
index be1a229963c..805a61db035 100644
--- a/core/src/test/scala/org/apache/spark/sql/delta/schema/CheckConstraintsSuite.scala
+++ b/core/src/test/scala/org/apache/spark/sql/delta/schema/CheckConstraintsSuite.scala
@@ -50,9 +50,7 @@ class CheckConstraintsSuite extends QueryTest
   }
 
   private def errorContains(errMsg: String, str: String): Unit = {
-    val actual = errMsg.replaceAll("`", "")
-    val expected = str.replaceAll("`", "")
-    assert(actual.contains(expected))
+    errMsg.contains(str)
   }
 
   test("can't add unparseable constraint") {
diff --git a/core/src/test/scala/org/apache/spark/sql/delta/schema/InvariantEnforcementSuite.scala b/core/src/test/scala/org/apache/spark/sql/delta/schema/InvariantEnforcementSuite.scala
index 2eb4953e9c6..34bf8202053 100644
--- a/core/src/test/scala/org/apache/spark/sql/delta/schema/InvariantEnforcementSuite.scala
+++ b/core/src/test/scala/org/apache/spark/sql/delta/schema/InvariantEnforcementSuite.scala
@@ -80,8 +80,7 @@ class InvariantEnforcementSuite extends QueryTest
     val error = e.getMessage
     val allExpected = expectedErrors
     allExpected.foreach { expected =>
-      assert(error.replaceAll("`", "").contains(expected.replaceAll("`", "")),
-        s"$error didn't contain $expected")
+      assert(error.contains(expected), s"$error didn't contain $expected")
     }
   }
 
diff --git a/examples/scala/build.sbt b/examples/scala/build.sbt
index 36f992b4bd6..ecb25e6d179 100644
--- a/examples/scala/build.sbt
+++ b/examples/scala/build.sbt
@@ -39,7 +39,7 @@ lazy val root = (project in file("."))
   .settings(
     name := "hello-world",
     libraryDependencies += "io.delta" %% "delta-core" % getDeltaVersion(),
-    libraryDependencies += "org.apache.spark" %% "spark-sql" % "3.1.0",
+    libraryDependencies += "org.apache.spark" %% "spark-sql" % "3.2.0",
     extraMavenRepo
   )
   
diff --git a/setup.py b/setup.py
index 5351ee39183..8a5451f540c 100644
--- a/setup.py
+++ b/setup.py
@@ -61,7 +61,7 @@ def run(self):
     package_dir={'': 'python'},
     packages=['delta'],
     install_requires=[
-        'pyspark>=3.1.0,<3.2.0',
+        'pyspark>=3.2.0,<3.3.0',
         'importlib_metadata>=1.0.0',
     ],
     python_requires='>=3.6',