[Spark] Rename DeltaTransaction to TransactionHelper (delta-io#4741)

<!--
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

Continue from delta-io#4735. This PR
renames `DeltaTransaction` to `TransactionHelper` and some additional
clean-ups.

## How was this patch tested?

Existing UTs

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

No

Author: ctring

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

@@ -44,7 +44,7 @@ import org.apache.spark.sql.delta.redirect.{RedirectFeature, TableRedirectConfig
 import org.apache.spark.sql.delta.schema.{SchemaMergingUtils, SchemaUtils}
 import org.apache.spark.sql.delta.sources.{DeltaSourceUtils, DeltaSQLConf}
 import org.apache.spark.sql.delta.stats._
-import org.apache.spark.sql.delta.util.{DeltaCommitFileProvider, JsonUtils}
+import org.apache.spark.sql.delta.util.{DeltaCommitFileProvider, JsonUtils, TransactionHelper}
 import org.apache.spark.sql.util.ScalaExtensions._
 import io.delta.storage.commit._
 import io.delta.storage.commit.actions.{AbstractMetadata, AbstractProtocol}
@@ -254,7 +254,7 @@ object OptimisticTransaction {
  *
  * This trait is not thread-safe.
  */
-trait OptimisticTransactionImpl extends DeltaTransaction
+trait OptimisticTransactionImpl extends TransactionHelper
   with TransactionalWrite
   with SQLMetricsReporting
   with DeltaScanGenerator
@@ -478,18 +478,6 @@ trait OptimisticTransactionImpl extends DeltaTransaction
   private[delta] var preCommitLogSegment: LogSegment =
     snapshot.logSegment.copy(checkpointProvider = snapshot.checkpointProvider)
 
-  /**
-   * Generates a timestamp which is greater than the commit timestamp
-   * of the last snapshot. Note that this is only needed when the
-   * feature `inCommitTimestamps` is enabled.
-   */
-  protected[delta] def generateInCommitTimestampForFirstCommitAttempt(
-      currentTimestamp: Long): Option[Long] =
-    Option.when(DeltaConfigs.IN_COMMIT_TIMESTAMPS_ENABLED.fromMetaData(metadata)) {
-      val lastCommitTimestamp = snapshot.timestamp
-      math.max(currentTimestamp, lastCommitTimestamp + 1)
-    }
-
   /** The end to end execution time of this transaction. */
   def txnExecutionTimeMs: Option[Long] = if (commitEndNano == -1) {
     None
@@ -2591,14 +2579,6 @@ trait OptimisticTransactionImpl extends DeltaTransaction
   protected def incrementallyDeriveChecksum(
       attemptVersion: Long,
       currentTransactionInfo: CurrentTransactionInfo): Option[VersionChecksum] = {
-    // Don't include [[AddFile]]s in CRC if this commit is modifying the schema of table in some
-    // way. This is to make sure we don't carry any DROPPED column from previous CRC to this CRC
-    // forever and can start fresh from next commit.
-    // If the oldSnapshot itself is missing, we don't incrementally compute the checksum.
-    val allFilesInCrcWritePathEnabled =
-      Snapshot.allFilesInCrcWritePathEnabled(spark, snapshot) &&
-        (snapshot.version == -1 || snapshot.metadata.schema == metadata.schema)
-
     incrementallyDeriveChecksum(
       spark,
       deltaLog,
@@ -2609,7 +2589,7 @@ trait OptimisticTransactionImpl extends DeltaTransaction
       operationName = currentTransactionInfo.op.name,
       txnIdOpt = Some(currentTransactionInfo.txnId),
       previousVersionState = scala.Left(snapshot),
-      includeAddFilesInCrc = allFilesInCrcWritePathEnabled
+      includeAddFilesInCrc = Snapshot.shouldIncludeAddFilesInCrc(spark, snapshot, metadata)
     ).toOption
   }
 

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

@@ -608,22 +608,7 @@ class Snapshot(
 
   /** Return the set of properties of the table. */
   def getProperties: mutable.Map[String, String] = {
-    val base = new mutable.LinkedHashMap[String, String]()
-    metadata.configuration.foreach { case (k, v) =>
-      if (k != "path") {
-        base.put(k, v)
-      }
-    }
-    base.put(Protocol.MIN_READER_VERSION_PROP, protocol.minReaderVersion.toString)
-    base.put(Protocol.MIN_WRITER_VERSION_PROP, protocol.minWriterVersion.toString)
-    if (protocol.supportsReaderFeatures || protocol.supportsWriterFeatures) {
-      val features = protocol.readerAndWriterFeatureNames.map(name =>
-        s"${TableFeatureProtocolUtils.FEATURE_PROP_PREFIX}$name" ->
-          TableFeatureProtocolUtils.FEATURE_PROP_SUPPORTED)
-      base ++ features.toSeq.sorted
-    } else {
-      base
-    }
+    Snapshot.getProperties(metadata, protocol)
   }
 
   /** The [[CheckpointProvider]] for the underlying checkpoint */
@@ -797,6 +782,40 @@ object Snapshot extends DeltaLogging {
     val shouldVerify = verificationConfEnabled || allFilesInCrcVerificationForceEnabled(spark)
     allFilesInCrcWritePathEnabled(spark, snapshot) && shouldVerify
   }
+
+  /**
+   * Don't include [[AddFile]]s in CRC if this commit is modifying the schema of table in some
+   * way. This is to make sure we don't carry any DROPPED column from previous CRC to this CRC
+   * forever and can start fresh from next commit.
+   * If the oldSnapshot itself is missing, we don't incrementally compute the checksum.
+   */
+  private[delta] def shouldIncludeAddFilesInCrc(
+      spark: SparkSession, snapshot: Snapshot, metadata: Metadata): Boolean = {
+    allFilesInCrcWritePathEnabled(spark, snapshot) &&
+      (snapshot.version == -1 || snapshot.metadata.schema == metadata.schema)
+  }
+
+  /**
+   * Return the set of properties for a given metadata and protocol.
+   */
+  def getProperties(metadata: Metadata, protocol: Protocol): mutable.Map[String, String] = {
+    val base = new mutable.LinkedHashMap[String, String]()
+    metadata.configuration.foreach { case (k, v) =>
+      if (k != "path") {
+        base.put(k, v)
+      }
+    }
+    base.put(Protocol.MIN_READER_VERSION_PROP, protocol.minReaderVersion.toString)
+    base.put(Protocol.MIN_WRITER_VERSION_PROP, protocol.minWriterVersion.toString)
+    if (protocol.supportsReaderFeatures || protocol.supportsWriterFeatures) {
+      val features = protocol.readerAndWriterFeatureNames.map(name =>
+        s"${TableFeatureProtocolUtils.FEATURE_PROP_PREFIX}$name" ->
+          TableFeatureProtocolUtils.FEATURE_PROP_SUPPORTED)
+      base ++ features.toSeq.sorted
+    } else {
+      base
+    }
+  }
 }
 
 /**

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

@@ -19,7 +19,7 @@ package org.apache.spark.sql.delta.hooks
 import scala.collection.mutable
 
 // scalastyle:off import.ordering.noEmptyLine
-import org.apache.spark.sql.delta.{DeltaLog, DeltaTransaction, Snapshot}
+import org.apache.spark.sql.delta.{DeltaLog, Snapshot}
 import org.apache.spark.sql.delta.CommittedTransaction
 import org.apache.spark.sql.delta.metering.DeltaLogging
 import org.apache.spark.sql.delta.sources.DeltaSQLConf

spark/src/main/scala/org/apache/spark/sql/delta/DeltaTransaction.scala renamed to spark/src/main/scala/org/apache/spark/sql/delta/util/TransactionHelper.scala

@@ -14,36 +14,34 @@
  * limitations under the License.
  */
 
-package org.apache.spark.sql.delta
+package org.apache.spark.sql.delta.util
 
-import scala.collection.mutable
 import scala.util.control.NonFatal
 
+import org.apache.spark.sql.delta.{CatalogOwnedTableFeature, CommittedTransaction, CoordinatedCommitsStats, CoordinatedCommitType, DeltaConfigs, DeltaLog, IsolationLevel, Snapshot}
 import org.apache.spark.sql.delta.DeltaOperations.Operation
-import org.apache.spark.sql.delta.actions.{AddFile, CommitInfo, Metadata, Protocol}
+import org.apache.spark.sql.delta.actions.{Metadata, Protocol}
 import org.apache.spark.sql.delta.coordinatedcommits.{CatalogOwnedTableUtils, TableCommitCoordinatorClient}
 import org.apache.spark.sql.delta.hooks.PostCommitHook
 import org.apache.spark.sql.delta.logging.DeltaLogKeys
 import org.apache.spark.sql.delta.metering.DeltaLogging
 import org.apache.spark.sql.delta.sources.DeltaSQLConf
+import org.apache.spark.sql.util.ScalaExtensions._
 
 import org.apache.spark.internal.MDC
 import org.apache.spark.sql.SparkSession
 import org.apache.spark.sql.catalyst.catalog.CatalogTable
 
 /**
- * Represents a transaction that maps to a delta table commit.
- *
- * An instance of this trait tracks the reads and writes as well as accumulates additional
- * information such as statistics of a single table throughout the life of a transaction.
+ * Contains helper methods for Delta transactions.
  */
-trait DeltaTransaction extends DeltaLogging {
-  val deltaLog: DeltaLog
-  val catalogTable: Option[CatalogTable]
-  val snapshot: Snapshot
+trait TransactionHelper extends DeltaLogging {
+  def deltaLog: DeltaLog
+  def catalogTable: Option[CatalogTable]
+  def snapshot: Snapshot
 
   /** Unique identifier for the transaction */
-  val txnId: String
+  def txnId: String
 
   /**
    * Returns the metadata for this transaction. The metadata refers to the metadata of the snapshot
@@ -54,9 +52,6 @@ trait DeltaTransaction extends DeltaLogging {
   /** The protocol of the snapshot that this transaction is reading at. */
   def protocol: Protocol
 
-  /** The end to end execution time of this transaction. */
-  def txnExecutionTimeMs: Option[Long]
-
   /**
    * Default [[IsolationLevel]] as set in table metadata.
    */
@@ -168,4 +163,16 @@ trait DeltaTransaction extends DeltaLogging {
         hook.handleError(spark, e, version)
     }
   }
+
+  /**
+   * Generates a timestamp which is greater than the commit timestamp
+   * of the last snapshot. Note that this is only needed when the
+   * feature `inCommitTimestamps` is enabled.
+   */
+  protected[delta] def generateInCommitTimestampForFirstCommitAttempt(
+      currentTimestamp: Long): Option[Long] =
+    Option.when(DeltaConfigs.IN_COMMIT_TIMESTAMPS_ENABLED.fromMetaData(metadata)) {
+      val lastCommitTimestamp = snapshot.timestamp
+      math.max(currentTimestamp, lastCommitTimestamp + 1)
+    }
 }