[Spark] Pass catalog table to DeltaLog API call sites, part 2 (#3804)

<!--
Thanks for sending a pull request!  Here are some tips for you:
1. If this is your first time, please read our contributor guidelines:
https://github.com/delta-io/delta/blob/master/CONTRIBUTING.md
2. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP]
Your PR title ...'.
  3. Be sure to keep the PR description updated to reflect all changes.
  4. Please write your PR title to summarize what this PR proposes.
5. If possible, provide a concise example to reproduce the issue for a
faster review.
6. If applicable, include the corresponding issue number in the PR title
and link it in the body.
-->

#### Which Delta project/connector is this regarding?
<!--
Please add the component selected below to the beginning of the pull
request title
For example: [Spark] Title of my pull request
-->

- [x] Spark
- [ ] Standalone
- [ ] Flink
- [ ] Kernel
- [ ] Other (fill in here)

## Description

<!--
- Describe what this PR changes.
- Describe why we need the change.
 
If this PR resolves an issue be sure to include "Resolves #XXX" to
correctly link and close the issue upon merge.
-->

Fix a number of code paths where we want to pass table identifier to the
commit coordinator client via the update method of DeltaLog



## How was this patch tested?
unit tests
<!--
If tests were added, say they were added here. Please make sure to test
the changes thoroughly including negative and positive cases if
possible.
If the changes were tested in any way other than unit tests, please
clarify how you tested step by step (ideally copy and paste-able, so
that other reviewers can test and check, and descendants can verify in
the future).
If the changes were not tested, please explain why.
-->

## Does this PR introduce _any_ user-facing changes?
No
<!--
If yes, please clarify the previous behavior and the change this PR
proposes - provide the console output, description and/or an example to
show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change
compared to the released Delta Lake versions or within the unreleased
branches such as master.
If no, write 'No'.
-->

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

@@ -172,7 +172,10 @@ class OptimisticTransaction(
       deltaLog: DeltaLog,
       catalogTable: Option[CatalogTable],
       snapshotOpt: Option[Snapshot] = None) =
-    this(deltaLog, catalogTable, snapshotOpt.getOrElse(deltaLog.update()))
+    this(
+      deltaLog,
+      catalogTable,
+      snapshotOpt.getOrElse(deltaLog.update(catalogTableOpt = catalogTable)))
 }
 
 object CommitConflictFailure {
@@ -1727,8 +1730,10 @@ trait OptimisticTransactionImpl extends TransactionalWrite
             // Actions of a commit which went in before ours.
             // Requires updating deltaLog to retrieve these actions, as another writer may have used
             // CommitCoordinatorClient for writing.
+            val fileProvider = DeltaCommitFileProvider(
+              deltaLog.update(catalogTableOpt = catalogTable))
             val logs = deltaLog.store.readAsIterator(
-              DeltaCommitFileProvider(deltaLog.update()).deltaFile(attemptVersion),
+              fileProvider.deltaFile(attemptVersion),
               deltaLog.newDeltaHadoopConf())
             try {
               val winningCommitActions = logs.map(Action.fromJson)

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

@@ -268,7 +268,7 @@ case class CoordinatedCommitsPreDowngradeCommand(table: DeltaTableV2)
     }
     var postDisablementUnbackfilledCommitsPresent = false
     if (exceptionOpt.isEmpty) {
-      val snapshotAfterDisabling = table.deltaLog.update()
+      val snapshotAfterDisabling = table.update()
       assert(snapshotAfterDisabling.getTableCommitCoordinatorForWrites.isEmpty)
       postDisablementUnbackfilledCommitsPresent =
         CoordinatedCommitsUtils.unbackfilledCommitsPresent(snapshotAfterDisabling)

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

@@ -109,6 +109,11 @@ case class DeltaTableV2(
     }
   }
 
+  /**
+   * Updates the delta log for this table and returns a new snapshot
+   */
+  def update(): Snapshot = deltaLog.update(catalogTableOpt = catalogTable)
+
   def getTableIdentifierIfExists: Option[TableIdentifier] = tableIdentifier.map { tableName =>
     spark.sessionState.sqlParser.parseMultipartIdentifier(tableName).asTableIdentifier
   }

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

@@ -235,7 +235,9 @@ case class CreateDeltaTableCommand(
     logInfo(log"Table is path-based table: ${MDC(DeltaLogKeys.IS_PATH_TABLE, tableByPath)}. " +
       log"Update catalog with mode: ${MDC(DeltaLogKeys.OPERATION, operation)}")
     val opStartTs = TimeUnit.NANOSECONDS.toMillis(txnUsedForCommit.txnStartTimeNs)
-    val postCommitSnapshot = deltaLog.update(checkIfUpdatedSinceTs = Some(opStartTs))
+    val postCommitSnapshot = deltaLog.update(
+      checkIfUpdatedSinceTs = Some(opStartTs),
+      catalogTableOpt = Some(tableWithLocation))
     val didNotChangeMetadata = txnUsedForCommit.metadata == txnUsedForCommit.snapshot.metadata
     updateCatalog(sparkSession, tableWithLocation, postCommitSnapshot, didNotChangeMetadata)
 

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

@@ -148,7 +148,7 @@ case class OptimizeTableCommand(
 
   override def run(sparkSession: SparkSession): Seq[Row] = {
     val table = getDeltaTable(child, "OPTIMIZE")
-    val snapshot = table.deltaLog.update()
+    val snapshot = table.update()
     if (snapshot.version == -1) {
       throw DeltaErrors.notADeltaTableException(table.deltaLog.dataPath.toString)
     }

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

@@ -144,7 +144,7 @@ trait ReorgTableForUpgradeUniformHelper extends DeltaLogging {
       sparkSession: SparkSession,
       targetIcebergCompatVersion: Int): Seq[Row] = {
 
-    val snapshot = target.deltaLog.update()
+    val snapshot = target.update()
     val currIcebergCompatVersionOpt = getEnabledVersion(snapshot.metadata)
     val targetVersionDeltaConfig = getIcebergCompatVersionConfigForValidVersion(
       targetIcebergCompatVersion)

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

@@ -46,9 +46,9 @@ case class ShowDeltaTableColumnsCommand(child: LogicalPlan)
   override def run(sparkSession: SparkSession): Seq[Row] = {
     // Return the schema from snapshot if it is an Delta table. Or raise
     // `DeltaErrors.notADeltaTableException` if it is a non-Delta table.
-    val deltaLog = getDeltaTable(child, "SHOW COLUMNS").deltaLog
-    recordDeltaOperation(deltaLog, "delta.ddl.showColumns") {
-      deltaLog.update().schema.fieldNames.map { x => Row(x) }.toSeq
+    val deltaTable = getDeltaTable(child, "SHOW COLUMNS")
+    recordDeltaOperation(deltaTable.deltaLog, "delta.ddl.showColumns") {
+      deltaTable.update().schema.fieldNames.map { x => Row(x) }.toSeq
     }
   }
 }

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

@@ -323,7 +323,9 @@ case class AlterTableDropFeatureDeltaCommand(
    */
   private def createCheckpointWithRetries(snapshotRefreshStartTimeTs: Long): Boolean = {
     val log = table.deltaLog
-    val snapshot = log.update(checkIfUpdatedSinceTs = Some(snapshotRefreshStartTimeTs))
+    val snapshot = log.update(
+      checkIfUpdatedSinceTs = Some(snapshotRefreshStartTimeTs),
+      catalogTableOpt = table.catalogTable)
 
     def checkpointAndVerify(snapshot: Snapshot): Boolean = {
       try {
@@ -352,17 +354,21 @@ case class AlterTableDropFeatureDeltaCommand(
       snapshotRefreshStartTs: Long,
       retryOnFailure: Boolean = false): Boolean = {
     val log = table.deltaLog
-    val snapshot = log.update(checkIfUpdatedSinceTs = Some(snapshotRefreshStartTs))
+    val snapshot = log.update(
+      checkIfUpdatedSinceTs = Some(snapshotRefreshStartTs),
+      catalogTableOpt = table.catalogTable)
     val emptyCommitTS = System.nanoTime()
-    log.startTransaction(table.catalogTable, Some(snapshot))
+    table.startTransaction(Some(snapshot))
       .commit(Nil, DeltaOperations.EmptyCommit)
 
     // retryOnFailure is temporary to avoid affecting the behavior of the legacy Drop Feature
     // command behavior.
     if (retryOnFailure) {
       createCheckpointWithRetries(emptyCommitTS)
     } else {
-      log.checkpoint(log.update(checkIfUpdatedSinceTs = Some(emptyCommitTS)))
+      log.checkpoint(log.update(
+        checkIfUpdatedSinceTs = Some(emptyCommitTS),
+        catalogTableOpt = table.catalogTable))
       true
     }
   }
@@ -1166,15 +1172,16 @@ case class AlterTableSetLocationDeltaCommand(
     val catalogTable = table.catalogTable.get
     val locUri = CatalogUtils.stringToURI(location)
 
-    val oldTable = table.deltaLog.update()
+    val oldTable = table.update()
     if (oldTable.version == -1) {
       throw DeltaErrors.notADeltaTableException(table.name())
     }
     val oldMetadata = oldTable.metadata
 
     var updatedTable = catalogTable.withNewStorage(locationUri = Some(locUri))
 
-    val (_, newTable) = DeltaLog.forTableWithSnapshot(sparkSession, new Path(location))
+    val (_, newTable) =
+      DeltaLog.forTableWithSnapshot(sparkSession, updatedTable, options = Map.empty[String, String])
     if (newTable.version == -1) {
       throw DeltaErrors.notADeltaTableException(DeltaTableIdentifier(path = Some(location)))
     }
@@ -1338,7 +1345,7 @@ case class AlterTableClusterByDeltaCommand(
     ClusteredTableUtils.validateNumClusteringColumns(clusteringColumns, Some(deltaLog))
     // If the target table is not a clustered table and there are no clustering columns being added
     // (CLUSTER BY NONE), do not convert the table into a clustered table.
-    val snapshot = deltaLog.update()
+    val snapshot = table.update()
     if (clusteringColumns.isEmpty &&
       !ClusteredTableUtils.isSupported(snapshot.protocol)) {
       logInfo(log"Skipping ALTER TABLE CLUSTER BY NONE on a non-clustered table: " +

spark/src/main/scala/org/apache/spark/sql/delta/commands/backfill/RowTrackingBackfillCommand.scala

@@ -55,7 +55,8 @@ case class RowTrackingBackfillCommand(
 
   override def filesToBackfill(txn: OptimisticTransaction): Dataset[AddFile] =
     // Get a new snapshot after the protocol upgrade.
-    RowTrackingBackfillExecutor.getCandidateFilesToBackfill(deltaLog.update())
+    RowTrackingBackfillExecutor.getCandidateFilesToBackfill(
+      deltaLog.update(catalogTableOpt = catalogTable))
 
   override def opType: String = "delta.rowTracking.backfill"
 
@@ -67,7 +68,7 @@ case class RowTrackingBackfillCommand(
   * the current protocol cannot support write table feature.
   */
   private def upgradeProtocolIfRequired(): Unit = {
-    val snapshot = deltaLog.update()
+    val snapshot = deltaLog.update(catalogTableOpt = catalogTable)
     if (!snapshot.protocol.isFeatureSupported(RowTrackingFeature)) {
       val minProtocolAllowingWriteTableFeature = Protocol(
         snapshot.protocol.minReaderVersion,

spark/src/main/scala/org/apache/spark/sql/delta/hooks/AutoCompact.scala

@@ -203,8 +203,16 @@ trait AutoCompactBase extends PostCommitHook with DeltaLogging {
         maxFileSizeOpt,
         maxDeletedRowsRatio = maxDeletedRowsRatio
       )
-      val rows = new OptimizeExecutor(spark, deltaLog.update(), catalogTable, partitionPredicates,
-        zOrderByColumns = Seq(), isAutoCompact = true, optimizeContext).optimize()
+      val rows = new OptimizeExecutor(
+        spark,
+        deltaLog.update(catalogTableOpt = catalogTable),
+        catalogTable,
+        partitionPredicates,
+        zOrderByColumns = Seq(),
+        isAutoCompact = true,
+        optimizeContext
+      )
+      .optimize()
       val metrics = rows.map(_.getAs[OptimizeMetrics](1))
       recordDeltaEvent(deltaLog, s"$opType.execute.metrics", data = metrics.head)
       metrics