[Kernel] Return txn metrics as part of TransactionCommitResult (delta-io#4353)

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

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

## Description

Return txn metrics as part of TransactionCommitResult

## How was this patch tested?

Updates test suite to check these metrics as well

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

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

Author: allisonport-db

kernel/kernel-api/src/main/java/io/delta/kernel/TransactionCommitResult.java

@@ -15,9 +15,12 @@
  */
 package io.delta.kernel;
 
+import static java.util.Objects.requireNonNull;
+
 import io.delta.kernel.annotation.Evolving;
 import io.delta.kernel.engine.Engine;
 import io.delta.kernel.hook.PostCommitHook;
+import io.delta.kernel.metrics.TransactionMetricsResult;
 import io.delta.kernel.utils.CloseableIterable;
 import java.util.List;
 
@@ -31,10 +34,15 @@
 public class TransactionCommitResult {
   private final long version;
   private final List<PostCommitHook> postCommitHooks;
+  private final TransactionMetricsResult transactionMetrics;
 
-  public TransactionCommitResult(long version, List<PostCommitHook> postCommitHooks) {
+  public TransactionCommitResult(
+      long version,
+      List<PostCommitHook> postCommitHooks,
+      TransactionMetricsResult transactionMetrics) {
     this.version = version;
-    this.postCommitHooks = postCommitHooks;
+    this.postCommitHooks = requireNonNull(postCommitHooks);
+    this.transactionMetrics = requireNonNull(transactionMetrics);
   }
 
   /**
@@ -62,4 +70,9 @@ public long getVersion() {
   public List<PostCommitHook> getPostCommitHooks() {
     return postCommitHooks;
   }
+
+  /** @return the metrics for this transaction */
+  public TransactionMetricsResult getTransactionMetrics() {
+    return transactionMetrics;
+  }
 }

kernel/kernel-api/src/main/java/io/delta/kernel/internal/TransactionImpl.java

@@ -252,15 +252,20 @@ public TransactionCommitResult commit(Engine engine, CloseableIterable<Row> data
     checkState(!closed, "Transaction is already attempted to commit. Create a new transaction.");
     TransactionMetrics transactionMetrics = new TransactionMetrics();
     try {
-      TransactionCommitResult result =
+      long committedVersion =
           transactionMetrics.totalCommitTimer.time(
               () -> commitWithRetry(engine, dataActions, transactionMetrics));
       recordTransactionReport(
           engine,
-          Optional.of(result.getVersion()) /* committedVersion */,
+          Optional.of(committedVersion),
           transactionMetrics,
           Optional.empty() /* exception */);
-      return result;
+      TransactionMetricsResult txnMetricsCaptured =
+          transactionMetrics.captureTransactionMetricsResult();
+      return new TransactionCommitResult(
+          committedVersion,
+          generatePostCommitHooks(committedVersion, txnMetricsCaptured),
+          txnMetricsCaptured);
     } catch (Exception e) {
       recordTransactionReport(
           engine,
@@ -271,7 +276,7 @@ public TransactionCommitResult commit(Engine engine, CloseableIterable<Row> data
     }
   }
 
-  private TransactionCommitResult commitWithRetry(
+  private long commitWithRetry(
       Engine engine, CloseableIterable<Row> dataActions, TransactionMetrics transactionMetrics) {
     try {
       long commitAsVersion = readSnapshot.getVersion() + 1;
@@ -391,7 +396,7 @@ private Optional<Long> getUpdatedInCommitTimestampAfterConflict(
     return attemptInCommitTimestamp;
   }
 
-  private TransactionCommitResult doCommit(
+  private long doCommit(
       Engine engine,
       long commitAsVersion,
       CommitInfo attemptCommitInfo,
@@ -462,16 +467,7 @@ private TransactionCommitResult doCommit(
           "Write file actions to JSON log file `%s`",
           FileNames.deltaFile(logPath, commitAsVersion));
 
-      List<PostCommitHook> postCommitHooks = new ArrayList<>();
-      if (isReadyForCheckpoint(commitAsVersion)) {
-        postCommitHooks.add(new CheckpointHook(dataPath, commitAsVersion));
-      }
-
-      buildPostCommitCrcInfoIfCurrentCrcAvailable(
-              commitAsVersion, transactionMetrics.captureTransactionMetricsResult())
-          .ifPresent(crcInfo -> postCommitHooks.add(new ChecksumSimpleHook(crcInfo, logPath)));
-
-      return new TransactionCommitResult(commitAsVersion, postCommitHooks);
+      return commitAsVersion;
     } catch (FileAlreadyExistsException e) {
       throw e;
     } catch (IOException ioe) {
@@ -485,6 +481,19 @@ public boolean isBlindAppend() {
     return true;
   }
 
+  private List<PostCommitHook> generatePostCommitHooks(
+      long committedVersion, TransactionMetricsResult txnMetrics) {
+    List<PostCommitHook> postCommitHooks = new ArrayList<>();
+    if (isReadyForCheckpoint(committedVersion)) {
+      postCommitHooks.add(new CheckpointHook(dataPath, committedVersion));
+    }
+
+    buildPostCommitCrcInfoIfCurrentCrcAvailable(committedVersion, txnMetrics)
+        .ifPresent(crcInfo -> postCommitHooks.add(new ChecksumSimpleHook(crcInfo, logPath)));
+
+    return postCommitHooks;
+  }
+
   /**
    * Generates a timestamp which is greater than the commit timestamp of the readSnapshot. This can
    * result in an additional file read and that this will only happen if ICT is enabled.

kernel/kernel-api/src/main/java/io/delta/kernel/metrics/TransactionMetricsResult.java

@@ -57,4 +57,6 @@ public interface TransactionMetricsResult {
    *     this metric may be incomplete.
    */
   long getTotalAddFilesSizeInBytes();
+
+  // TODO add fileSizeHistogram
 }

kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/metrics/TransactionReportSuite.scala

@@ -27,7 +27,7 @@ import io.delta.kernel.internal.actions.{RemoveFile, SingleAction}
 import io.delta.kernel.internal.data.GenericRow
 import io.delta.kernel.internal.metrics.Timer
 import io.delta.kernel.internal.util.Utils
-import io.delta.kernel.metrics.{SnapshotReport, TransactionReport}
+import io.delta.kernel.metrics.{SnapshotReport, TransactionMetricsResult, TransactionReport}
 import io.delta.kernel.types.{IntegerType, StructType}
 import io.delta.kernel.utils.{CloseableIterable, CloseableIterator, DataFileStatus}
 
@@ -51,15 +51,19 @@ class TransactionReportSuite extends AnyFunSuite with MetricsReportTestUtils {
   def getTransactionAndSnapshotReport(
       createTransaction: Engine => Transaction,
       generateCommitActions: (Transaction, Engine) => CloseableIterable[Row],
-      expectException: Boolean)
+      expectException: Boolean,
+      validateTransactionMetrics: (TransactionMetricsResult, Long) => Unit)
       : (TransactionReport, Long, Option[SnapshotReport], Option[Exception]) = {
     val timer = new Timer()
 
     val (metricsReports, exception) = collectMetricsReports(
       engine => {
         val transaction = createTransaction(engine)
         val actionsToCommit = generateCommitActions(transaction, engine)
-        timer.time(() => transaction.commit(engine, actionsToCommit)) // Time the actual operation
+        val txnCommitResult = timer.time(() =>
+          transaction.commit(engine, actionsToCommit)) // Time the actual operation
+        // Validate the txn metrics returned in txnCommitResult
+        validateTransactionMetrics(txnCommitResult.getTransactionMetrics, timer.totalDurationNs())
       },
       expectException)
 
@@ -109,6 +113,18 @@ class TransactionReportSuite extends AnyFunSuite with MetricsReportTestUtils {
       operation: Operation = Operation.MANUAL_UPDATE): Unit = {
     // scalastyle:on
     assert(expectException == expectedCommitVersion.isEmpty)
+    def validateTransactionMetrics(txnMetrics: TransactionMetricsResult, duration: Long): Unit = {
+      // Since we cannot know the actual duration of commit we sanity check that they are > 0 and
+      // less than the total operation duration
+      assert(txnMetrics.getTotalCommitDurationNs > 0)
+      assert(txnMetrics.getTotalCommitDurationNs < duration)
+
+      assert(txnMetrics.getNumCommitAttempts == expectedNumAttempts)
+      assert(txnMetrics.getNumAddFiles == expectedNumAddFiles)
+      assert(txnMetrics.getTotalAddFilesSizeInBytes == expectedTotalAddFilesSizeInBytes)
+      assert(txnMetrics.getNumRemoveFiles == expectedNumRemoveFiles)
+      assert(txnMetrics.getNumTotalActions == expectedNumTotalActions)
+    }
 
     val (transactionReport, duration, snapshotReportOpt, exception) =
       getTransactionAndSnapshotReport(
@@ -117,7 +133,8 @@ class TransactionReportSuite extends AnyFunSuite with MetricsReportTestUtils {
             Table.forPath(engine, path).createTransactionBuilder(engine, engineInfo, operation),
             engine),
         generateCommitActions,
-        expectException)
+        expectException,
+        validateTransactionMetrics)
 
     // Verify contents
     assert(transactionReport.getTablePath == defaultEngine.getFileSystemClient.resolvePath(path))
@@ -143,18 +160,7 @@ class TransactionReportSuite extends AnyFunSuite with MetricsReportTestUtils {
       })
     }
     assert(transactionReport.getCommittedVersion.toScala == expectedCommitVersion)
-
-    // Since we cannot know the actual duration of commit we sanity check that they are > 0 and
-    // less than the total operation duration
-    assert(transactionReport.getTransactionMetrics.getTotalCommitDurationNs > 0)
-    assert(transactionReport.getTransactionMetrics.getTotalCommitDurationNs < duration)
-
-    assert(transactionReport.getTransactionMetrics.getNumCommitAttempts == expectedNumAttempts)
-    assert(transactionReport.getTransactionMetrics.getNumAddFiles == expectedNumAddFiles)
-    assert(transactionReport.getTransactionMetrics.getTotalAddFilesSizeInBytes
-      == expectedTotalAddFilesSizeInBytes)
-    assert(transactionReport.getTransactionMetrics.getNumRemoveFiles == expectedNumRemoveFiles)
-    assert(transactionReport.getTransactionMetrics.getNumTotalActions == expectedNumTotalActions)
+    validateTransactionMetrics(transactionReport.getTransactionMetrics, duration)
   }
 
   def generateAppendActions(fileStatusIter: CloseableIterator[DataFileStatus])(