[Kernel][Metric]Rename snapshot metric following same naming pattern (delta-io#4859)

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

<!--
- 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.
-->
Rename snapshot metric following same naming pattern --
Load_what_[total/listing/etc]_Duration
## How was this patch tested?

<!--
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.
-->
Existing tests are updated
## 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: huan233usc

kernel/kernel-api/src/main/java/io/delta/kernel/internal/metrics/SnapshotMetrics.java

@@ -30,48 +30,48 @@ public class SnapshotMetrics {
 
   public final Timer loadSnapshotTotalTimer = new Timer();
 
-  public final Timer timestampToVersionResolutionTimer = new Timer();
+  public final Timer computeTimestampToVersionTotalDurationTimer = new Timer();
 
-  public final Timer loadInitialDeltaActionsTimer = new Timer();
+  public final Timer loadProtocolMetadataTotalDurationTimer = new Timer();
 
-  public final Timer timeToBuildLogSegmentForVersionTimer = new Timer();
+  public final Timer loadLogSegmentTotalDurationTimer = new Timer();
 
-  public final Timer durationToGetCrcInfoTimer = new Timer();
+  public final Timer loadCrcTotalDurationTimer = new Timer();
 
   public SnapshotMetricsResult captureSnapshotMetricsResult() {
     return new SnapshotMetricsResult() {
+      final Optional<Long> computeTimestampToVersionTotalDurationResult =
+          computeTimestampToVersionTotalDurationTimer.totalDurationIfRecorded();
       final long loadSnapshotTotalDurationResult = loadSnapshotTotalTimer.totalDurationNs();
-      final Optional<Long> timestampToVersionResolutionDurationResult =
-          timestampToVersionResolutionTimer.totalDurationIfRecorded();
-      final long loadInitialDeltaActionsDurationResult =
-          loadInitialDeltaActionsTimer.totalDurationNs();
-      final long timeToBuildLogSegmentForVersionDurationResult =
-          timeToBuildLogSegmentForVersionTimer.totalDurationNs();
-      final long durationToGetCrcInfoDurationResult = durationToGetCrcInfoTimer.totalDurationNs();
+      final long loadProtocolMetadataTotalDurationResult =
+          loadProtocolMetadataTotalDurationTimer.totalDurationNs();
+      final long loadLogSegmentTotalDurationResult =
+          loadLogSegmentTotalDurationTimer.totalDurationNs();
+      final long loadCrcTotalDurationResult = loadCrcTotalDurationTimer.totalDurationNs();
 
       @Override
-      public long getLoadSnapshotTotalDurationNs() {
-        return loadSnapshotTotalDurationResult;
+      public Optional<Long> getComputeTimestampToVersionTotalDurationNs() {
+        return computeTimestampToVersionTotalDurationResult;
       }
 
       @Override
-      public Optional<Long> getTimestampToVersionResolutionDurationNs() {
-        return timestampToVersionResolutionDurationResult;
+      public long getLoadSnapshotTotalDurationNs() {
+        return loadSnapshotTotalDurationResult;
       }
 
       @Override
-      public long getLoadInitialDeltaActionsDurationNs() {
-        return loadInitialDeltaActionsDurationResult;
+      public long getLoadProtocolMetadataTotalDurationNs() {
+        return loadProtocolMetadataTotalDurationResult;
       }
 
       @Override
-      public long getTimeToBuildLogSegmentForVersionNs() {
-        return timeToBuildLogSegmentForVersionDurationResult;
+      public long getLoadLogSegmentTotalDurationNs() {
+        return loadLogSegmentTotalDurationResult;
       }
 
       @Override
-      public long getDurationToGetCrcInfoNs() {
-        return durationToGetCrcInfoDurationResult;
+      public long getLoadCrcTotalDurationNs() {
+        return loadCrcTotalDurationResult;
       }
     };
   }
@@ -80,15 +80,15 @@ public long getDurationToGetCrcInfoNs() {
   public String toString() {
     return String.format(
         "SnapshotMetrics("
+            + "computeTimestampToVersionTotalDurationTimer=%s, "
             + "loadSnapshotTotalTimer=%s,"
-            + "timestampToVersionResolutionTimer=%s, "
-            + "loadInitialDeltaActionsTimer=%s, "
+            + "loadProtocolMetadataTotalDurationTimer=%s, "
             + "timeToBuildLogSegmentForVersionTimer=%s, "
-            + "durationToGetCrcInfoTimer=%s)",
+            + "loadCrcTotalDurationNsTimer=%s)",
+        computeTimestampToVersionTotalDurationTimer,
         loadSnapshotTotalTimer,
-        timestampToVersionResolutionTimer,
-        loadInitialDeltaActionsTimer,
-        timeToBuildLogSegmentForVersionTimer,
-        durationToGetCrcInfoTimer);
+        loadProtocolMetadataTotalDurationTimer,
+        loadLogSegmentTotalDurationTimer,
+        loadCrcTotalDurationTimer);
   }
 }

kernel/kernel-api/src/main/java/io/delta/kernel/internal/replay/LogReplay.java

@@ -258,7 +258,7 @@ private Lazy<Tuple2<Protocol, Metadata>> createLazyProtocolAndMetadata(
     return new Lazy<>(
         () -> {
           final Tuple2<Protocol, Metadata> result =
-              snapshotMetrics.loadInitialDeltaActionsTimer.time(
+              snapshotMetrics.loadProtocolMetadataTotalDurationTimer.time(
                   () -> {
                     final long targetVersion = getVersion();
 
@@ -285,7 +285,7 @@ private Lazy<Tuple2<Protocol, Metadata>> createLazyProtocolAndMetadata(
           logger.info(
               "[{}] Took {}ms to load Protocol and Metadata at version {}",
               dataPath.toString(),
-              snapshotMetrics.loadInitialDeltaActionsTimer.totalDurationMs(),
+              snapshotMetrics.loadProtocolMetadataTotalDurationTimer.totalDurationMs(),
               getVersion());
 
           return result;
@@ -611,7 +611,7 @@ public Optional<SnapshotHint> maybeGetNewerSnapshotHintAndUpdateCache(
                       FileNames.getFileVersion(new Path(checksum.getPath())) >= crcReadLowerBound)
               .flatMap(
                   checksum ->
-                      snapshotMetrics.durationToGetCrcInfoTimer.time(
+                      snapshotMetrics.loadCrcTotalDurationTimer.time(
                           () -> ChecksumReader.getCRCInfo(engine, checksum)));
 
       if (!crcInfoOpt.isPresent()) {

kernel/kernel-api/src/main/java/io/delta/kernel/internal/snapshot/SnapshotManager.java

@@ -82,7 +82,7 @@ public SnapshotImpl buildLatestSnapshot(Engine engine, SnapshotQueryContext snap
     final LogSegment logSegment =
         snapshotContext
             .getSnapshotMetrics()
-            .timeToBuildLogSegmentForVersionTimer
+            .loadLogSegmentTotalDurationTimer
             .time(() -> getLogSegmentForVersion(engine, Optional.empty() /* versionToLoad */));
     snapshotContext.setVersion(logSegment.getVersion());
     snapshotContext.setCheckpointVersion(logSegment.getCheckpointVersionOpt());
@@ -105,7 +105,7 @@ public SnapshotImpl getSnapshotAt(
     final LogSegment logSegment =
         snapshotContext
             .getSnapshotMetrics()
-            .timeToBuildLogSegmentForVersionTimer
+            .loadLogSegmentTotalDurationTimer
             .time(
                 () -> getLogSegmentForVersion(engine, Optional.of(version) /* versionToLoadOpt */));
 
@@ -134,7 +134,7 @@ public SnapshotImpl getSnapshotForTimestamp(
     long versionToRead =
         snapshotContext
             .getSnapshotMetrics()
-            .timestampToVersionResolutionTimer
+            .computeTimestampToVersionTotalDurationTimer
             .time(
                 () ->
                     DeltaHistoryManager.getActiveCommitAtTimestamp(
@@ -149,7 +149,10 @@ public SnapshotImpl getSnapshotForTimestamp(
     logger.info(
         "{}: Took {} ms to fetch version at timestamp {}",
         tablePath,
-        snapshotContext.getSnapshotMetrics().timestampToVersionResolutionTimer.totalDurationMs(),
+        snapshotContext
+            .getSnapshotMetrics()
+            .computeTimestampToVersionTotalDurationTimer
+            .totalDurationMs(),
         millisSinceEpochUTC);
 
     return getSnapshotAt(engine, versionToRead, snapshotContext);

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

@@ -20,42 +20,42 @@
 
 /** Stores the metrics results for a {@link SnapshotReport} */
 @JsonPropertyOrder({
+  "computeTimestampToVersionTotalDurationNs",
   "loadSnapshotTotalDurationNs",
-  "timestampToVersionResolutionDurationNs",
-  "loadInitialDeltaActionsDurationNs",
-  "timeToBuildLogSegmentForVersionNs",
-  "durationToGetCrcInfoNs"
+  "loadProtocolMetadataTotalDurationNs",
+  "loadLogSegmentTotalDurationNs",
+  "loadCrcTotalDurationNs"
 })
 public interface SnapshotMetricsResult {
 
+  /**
+   * @return the duration (ns) to resolve the provided timestamp to a table version for timestamp
+   *     time-travel queries. Empty for time-travel by version or non-time-travel queries.
+   */
+  Optional<Long> getComputeTimestampToVersionTotalDurationNs();
+
   /**
    * @return the total duration (ns) to load the snapshot, including all steps such as resolving
    *     timestamp to version, LISTing the _delta_log, building the log segment, and determining the
    *     latest protocol and metadata.
    */
   long getLoadSnapshotTotalDurationNs();
 
-  /**
-   * @return the duration (ns) to resolve the provided timestamp to a table version for timestamp
-   *     time-travel queries. Empty for time-travel by version or non-time-travel queries.
-   */
-  Optional<Long> getTimestampToVersionResolutionDurationNs();
-
   /**
    * @return the duration (ns) to load the initial delta actions for the snapshot (such as the table
    *     protocol and metadata). 0 if snapshot construction fails before log replay.
    */
-  long getLoadInitialDeltaActionsDurationNs();
+  long getLoadProtocolMetadataTotalDurationNs();
 
   /**
    * @return the duration (ns) to build the log segment for the specified version during snapshot
    *     construction. 0 if snapshot construction fails before this step.
    */
-  long getTimeToBuildLogSegmentForVersionNs();
+  long getLoadLogSegmentTotalDurationNs();
 
   /**
    * @return the duration (ns) to get CRC information during snapshot construction. 0 if snapshot
    *     construction fails before this step or if CRC is not read in loading snapshot.
    */
-  long getDurationToGetCrcInfoNs();
+  long getLoadCrcTotalDurationNs();
 }

kernel/kernel-api/src/test/scala/io/delta/kernel/internal/metrics/MetricsReportSerializerSuite.scala

@@ -40,16 +40,16 @@ class MetricsReportSerializerSuite extends AnyFunSuite {
   }
 
   private def testSnapshotReport(snapshotReport: SnapshotReport): Unit = {
+    val computeTimestampToVersionTotalDuration = optionToString(
+      snapshotReport.getSnapshotMetrics().getComputeTimestampToVersionTotalDurationNs())
     val loadSnapshotTotalDuration =
       snapshotReport.getSnapshotMetrics().getLoadSnapshotTotalDurationNs()
-    val timestampToVersionResolutionDuration = optionToString(
-      snapshotReport.getSnapshotMetrics().getTimestampToVersionResolutionDurationNs())
     val loadProtocolAndMetadataDuration =
-      snapshotReport.getSnapshotMetrics().getLoadInitialDeltaActionsDurationNs()
+      snapshotReport.getSnapshotMetrics().getLoadProtocolMetadataTotalDurationNs()
     val buildLogSegmentDuration =
-      snapshotReport.getSnapshotMetrics().getTimeToBuildLogSegmentForVersionNs()
-    val durationToGetCrcInfo =
-      snapshotReport.getSnapshotMetrics().getDurationToGetCrcInfoNs()
+      snapshotReport.getSnapshotMetrics().getLoadLogSegmentTotalDurationNs()
+    val loadCrcTotalDuration =
+      snapshotReport.getSnapshotMetrics().getLoadCrcTotalDurationNs()
     val exception: Optional[String] = snapshotReport.getException().map(_.toString)
     val expectedJson =
       s"""
@@ -61,11 +61,11 @@ class MetricsReportSerializerSuite extends AnyFunSuite {
          |"checkpointVersion":${optionToString(snapshotReport.getCheckpointVersion())},
          |"providedTimestamp":${optionToString(snapshotReport.getProvidedTimestamp())},
          |"snapshotMetrics":{
+         |"computeTimestampToVersionTotalDurationNs":${computeTimestampToVersionTotalDuration},
          |"loadSnapshotTotalDurationNs":${loadSnapshotTotalDuration},
-         |"timestampToVersionResolutionDurationNs":${timestampToVersionResolutionDuration},
-         |"loadInitialDeltaActionsDurationNs":${loadProtocolAndMetadataDuration},
-         |"timeToBuildLogSegmentForVersionNs":${buildLogSegmentDuration},
-         |"durationToGetCrcInfoNs":${durationToGetCrcInfo}
+         |"loadProtocolMetadataTotalDurationNs":${loadProtocolAndMetadataDuration},
+         |"loadLogSegmentTotalDurationNs":${buildLogSegmentDuration},
+         |"loadCrcTotalDurationNs":${loadCrcTotalDuration}
          |}
          |}
          |""".stripMargin.replaceAll("\n", "")
@@ -74,11 +74,11 @@ class MetricsReportSerializerSuite extends AnyFunSuite {
 
   test("SnapshotReport serializer") {
     val snapshotContext1 = SnapshotQueryContext.forTimestampSnapshot("/table/path", 0)
+    snapshotContext1.getSnapshotMetrics.computeTimestampToVersionTotalDurationTimer.record(10)
     snapshotContext1.getSnapshotMetrics.loadSnapshotTotalTimer.record(2000)
-    snapshotContext1.getSnapshotMetrics.timestampToVersionResolutionTimer.record(10)
-    snapshotContext1.getSnapshotMetrics.loadInitialDeltaActionsTimer.record(1000)
-    snapshotContext1.getSnapshotMetrics.timeToBuildLogSegmentForVersionTimer.record(500)
-    snapshotContext1.getSnapshotMetrics.durationToGetCrcInfoTimer.record(250)
+    snapshotContext1.getSnapshotMetrics.loadProtocolMetadataTotalDurationTimer.record(1000)
+    snapshotContext1.getSnapshotMetrics.loadLogSegmentTotalDurationTimer.record(500)
+    snapshotContext1.getSnapshotMetrics.loadCrcTotalDurationTimer.record(250)
     snapshotContext1.setVersion(25)
     snapshotContext1.setCheckpointVersion(Optional.of(20))
     val exception = new RuntimeException("something something failed")
@@ -98,11 +98,11 @@ class MetricsReportSerializerSuite extends AnyFunSuite {
         |"checkpointVersion":20,
         |"providedTimestamp":0,
         |"snapshotMetrics":{
+        |"computeTimestampToVersionTotalDurationNs":10,
         |"loadSnapshotTotalDurationNs":2000,
-        |"timestampToVersionResolutionDurationNs":10,
-        |"loadInitialDeltaActionsDurationNs":1000,
-        |"timeToBuildLogSegmentForVersionNs":500,
-        |"durationToGetCrcInfoNs":250
+        |"loadProtocolMetadataTotalDurationNs":1000,
+        |"loadLogSegmentTotalDurationNs":500,
+        |"loadCrcTotalDurationNs":250
         |}
         |}
         |""".stripMargin.replaceAll("\n", "")

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

@@ -129,42 +129,43 @@ class SnapshotReportSuite extends AnyFunSuite with MetricsReportTestUtils {
       assert(metrics.getLoadSnapshotTotalDurationNs >= 0)
     }
 
-    // ===== Metric: getTimestampToVersionResolutionDurationNs =====
+    // ===== Metric: getComputeTimestampToVersionTotalDurationNs =====
     if (expectations.expectNonEmptyTimestampToVersionResolutionDuration) {
-      assert(metrics.getTimestampToVersionResolutionDurationNs.isPresent)
-      assert(metrics.getTimestampToVersionResolutionDurationNs.get > 0)
-      assert(metrics.getTimestampToVersionResolutionDurationNs.get < duration)
-      assert(metrics.getTimestampToVersionResolutionDurationNs.get <=
+      assert(metrics.getComputeTimestampToVersionTotalDurationNs.isPresent)
+      assert(metrics.getComputeTimestampToVersionTotalDurationNs.get > 0)
+      assert(metrics.getComputeTimestampToVersionTotalDurationNs.get < duration)
+      assert(metrics.getComputeTimestampToVersionTotalDurationNs.get <=
         metrics.getLoadSnapshotTotalDurationNs)
     } else {
-      assert(!metrics.getTimestampToVersionResolutionDurationNs.isPresent)
+      assert(!metrics.getComputeTimestampToVersionTotalDurationNs.isPresent)
     }
 
-    // ===== Metric: getLoadInitialDeltaActionsDurationNs =====
+    // ===== Metric: getLoadProtocolMetadataTotalDurationNs  =====
     if (expectations.expectNonZeroLoadProtocolAndMetadataDuration) {
-      assert(metrics.getLoadInitialDeltaActionsDurationNs > 0)
-      assert(metrics.getLoadInitialDeltaActionsDurationNs < duration)
-      assert(metrics.getLoadInitialDeltaActionsDurationNs <= metrics.getLoadSnapshotTotalDurationNs)
+      assert(metrics.getLoadProtocolMetadataTotalDurationNs > 0)
+      assert(metrics.getLoadProtocolMetadataTotalDurationNs < duration)
+      assert(
+        metrics.getLoadProtocolMetadataTotalDurationNs <= metrics.getLoadSnapshotTotalDurationNs)
     } else {
-      assert(metrics.getLoadInitialDeltaActionsDurationNs == 0)
+      assert(metrics.getLoadProtocolMetadataTotalDurationNs == 0)
     }
 
-    // ===== Metric: getTimeToBuildLogSegmentForVersionNs =====
+    // ===== Metric: getLoadLogSegmentTotalDurationNs =====
     if (expectations.expectNonZeroBuildLogSegmentDuration) {
-      assert(metrics.getTimeToBuildLogSegmentForVersionNs > 0)
-      assert(metrics.getTimeToBuildLogSegmentForVersionNs < duration)
-      assert(metrics.getTimeToBuildLogSegmentForVersionNs <= metrics.getLoadSnapshotTotalDurationNs)
+      assert(metrics.getLoadLogSegmentTotalDurationNs > 0)
+      assert(metrics.getLoadLogSegmentTotalDurationNs < duration)
+      assert(metrics.getLoadLogSegmentTotalDurationNs <= metrics.getLoadSnapshotTotalDurationNs)
     } else {
-      assert(metrics.getTimeToBuildLogSegmentForVersionNs == 0)
+      assert(metrics.getLoadLogSegmentTotalDurationNs == 0)
     }
 
-    // ===== Metric: getDurationToGetCrcInfoNs =====
+    // ===== Metric: getLoadCrcTotalDurationNs =====
     if (expectations.expectNonZeroDurationToGetCrcInfo) {
-      assert(metrics.getDurationToGetCrcInfoNs > 0)
-      assert(metrics.getDurationToGetCrcInfoNs < duration)
-      assert(metrics.getDurationToGetCrcInfoNs <= metrics.getLoadSnapshotTotalDurationNs)
+      assert(metrics.getLoadCrcTotalDurationNs > 0)
+      assert(metrics.getLoadCrcTotalDurationNs < duration)
+      assert(metrics.getLoadCrcTotalDurationNs <= metrics.getLoadSnapshotTotalDurationNs)
     } else {
-      assert(metrics.getDurationToGetCrcInfoNs == 0)
+      assert(metrics.getLoadCrcTotalDurationNs == 0)
     }
   }