[Spark] Replaces startTransaction(Snapshot) calls with catalogTable overload

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

@@ -266,6 +266,7 @@ class DeltaLog private(
    * the new protocol version is not a superset of the original one used by the snapshot.
    */
   def upgradeProtocol(
+      catalogTable: Option[CatalogTable],
       snapshot: Snapshot,
       newVersion: Protocol): Unit = {
     val currentVersion = snapshot.protocol
@@ -277,7 +278,7 @@ class DeltaLog private(
       throw new ProtocolDowngradeException(currentVersion, newVersion)
     }
 
-    val txn = startTransaction(Some(snapshot))
+    val txn = startTransaction(catalogTable, Some(snapshot))
     try {
       SchemaMergingUtils.checkColumnNameDuplication(txn.metadata.schema, "in the table schema")
     } catch {
@@ -288,11 +289,6 @@ class DeltaLog private(
     logConsole(s"Upgraded table at $dataPath to $newVersion.")
   }
 
-  // Test-only!!
-  private[delta] def upgradeProtocol(newVersion: Protocol): Unit = {
-    upgradeProtocol(unsafeVolatileSnapshot, newVersion)
-  }
-
   /**
    * Get all actions starting from "startVersion" (inclusive). If `startVersion` doesn't exist,
    * return an empty Iterator.

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

@@ -612,7 +612,7 @@ case class CreateDeltaTableCommand(
       deltaLog: DeltaLog,
       tableWithLocation: CatalogTable,
       snapshotOpt: Option[Snapshot] = None): OptimisticTransaction = {
-    val txn = deltaLog.startTransaction(snapshotOpt)
+    val txn = deltaLog.startTransaction(None, snapshotOpt)
 
     // During CREATE/REPLACE, we synchronously run conversion (if Uniform is enabled) so
     // we always remove the post commit hook here.

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

@@ -113,7 +113,7 @@ case class RestoreTableCommand(
       require(versionToRestore < latestVersion, s"Version to restore ($versionToRestore)" +
         s"should be less then last available version ($latestVersion)")
 
-      deltaLog.withNewTransaction { txn =>
+      deltaLog.withNewTransaction(sourceTable.catalogTable) { txn =>
         val latestSnapshot = txn.snapshot
         val snapshotToRestore = deltaLog.getSnapshotAt(versionToRestore)
         val latestSnapshotFiles = latestSnapshot.allFiles

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

@@ -231,7 +231,8 @@ case class AlterTableDropFeatureDeltaCommand(
     val log = table.deltaLog
     val snapshot = log.update(checkIfUpdatedSinceTs = Some(snapshotRefreshStartTime))
     val emptyCommitTS = System.nanoTime()
-    log.startTransaction(Some(snapshot)).commit(Nil, DeltaOperations.EmptyCommit)
+    log.startTransaction(table.catalogTable, Some(snapshot))
+      .commit(Nil, DeltaOperations.EmptyCommit)
     log.checkpoint(log.update(checkIfUpdatedSinceTs = Some(emptyCommitTS)))
   }
 

spark/src/test/scala/org/apache/spark/sql/delta/DeltaAlterTableTests.scala

@@ -1549,7 +1549,8 @@ trait DeltaAlterTableByPathTests extends DeltaAlterTableTestBase {
   override protected def createTable(schema: String, tblProperties: Map[String, String]): String = {
       val tmpDir = Utils.createTempDir().getCanonicalPath
       val (deltaLog, snapshot) = getDeltaLogWithSnapshot(tmpDir)
-      val txn = deltaLog.startTransaction(Some(snapshot))
+      // This is a path-based table so we don't need to pass the catalogTable here
+      val txn = deltaLog.startTransaction(None, Some(snapshot))
       val metadata = Metadata(
         schemaString = StructType.fromDDL(schema).json,
         configuration = tblProperties)

spark/src/test/scala/org/apache/spark/sql/delta/rowtracking/RowTrackingConflictResolutionSuite.scala

@@ -19,6 +19,7 @@ package org.apache.spark.sql.delta.rowtracking
 import org.apache.spark.sql.delta.{DeltaLog, DeltaOperations, RowId, RowTrackingFeature}
 import org.apache.spark.sql.delta.actions.{Action, AddFile}
 import org.apache.spark.sql.delta.rowid.RowIdTestUtils
+import org.apache.spark.sql.delta.test.DeltaTestImplicits._
 
 import org.apache.spark.sql.QueryTest
 import org.apache.spark.sql.catalyst.TableIdentifier

spark/src/test/scala/org/apache/spark/sql/delta/test/DeltaTestImplicits.scala

@@ -104,6 +104,14 @@ object DeltaTestImplicits {
     def enableExpiredLogCleanup(): Boolean = {
       deltaLog.enableExpiredLogCleanup(snapshot.metadata)
     }
+
+    def upgradeProtocol(newVersion: Protocol): Unit = {
+      upgradeProtocol(deltaLog.unsafeVolatileSnapshot, newVersion)
+    }
+
+    def upgradeProtocol(snapshot: Snapshot, newVersion: Protocol): Unit = {
+      deltaLog.upgradeProtocol(None, snapshot, newVersion)
+    }
   }
 
   implicit class DeltaTableV2ObjectTestHelper(dt: DeltaTableV2.type) {