[Kernel][Metrics][PR#2] Add SnapshotReport for reporting snapshot construction (#3903)

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

Adds a `SnapshotReport` for reporting snapshot construction.

We record a `SnapshotReport` after successfully constructing a snapshot
or if an exception is thrown during construction. We use
`SnapshotContext` to propagate and update information about the snapshot
construction. For example, we update the "version" as soon as it's
resolved for time-travel by timestamp or load latest snapshot queries.
This means in the case of the exception we include as much information
as is available.

## How was this patch tested?

Adds a test suite `MetricsReportSuite` with unit tests.

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

No.

Author: allisonport-db

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

@@ -26,10 +26,13 @@
 import io.delta.kernel.internal.actions.Metadata;
 import io.delta.kernel.internal.actions.Protocol;
 import io.delta.kernel.internal.fs.Path;
+import io.delta.kernel.internal.metrics.SnapshotQueryContext;
+import io.delta.kernel.internal.metrics.SnapshotReportImpl;
 import io.delta.kernel.internal.replay.CreateCheckpointIterator;
 import io.delta.kernel.internal.replay.LogReplay;
 import io.delta.kernel.internal.snapshot.LogSegment;
 import io.delta.kernel.internal.util.VectorUtils;
+import io.delta.kernel.metrics.SnapshotReport;
 import io.delta.kernel.types.StructType;
 import java.util.List;
 import java.util.Map;
@@ -45,13 +48,15 @@ public class SnapshotImpl implements Snapshot {
   private final Metadata metadata;
   private final LogSegment logSegment;
   private Optional<Long> inCommitTimestampOpt;
+  private final SnapshotReport snapshotReport;
 
   public SnapshotImpl(
       Path dataPath,
       LogSegment logSegment,
       LogReplay logReplay,
       Protocol protocol,
-      Metadata metadata) {
+      Metadata metadata,
+      SnapshotQueryContext snapshotContext) {
     this.logPath = new Path(dataPath, "_delta_log");
     this.dataPath = dataPath;
     this.version = logSegment.version;
@@ -60,6 +65,7 @@ public SnapshotImpl(
     this.protocol = protocol;
     this.metadata = metadata;
     this.inCommitTimestampOpt = Optional.empty();
+    this.snapshotReport = SnapshotReportImpl.forSuccess(snapshotContext);
   }
 
   /////////////////
@@ -107,6 +113,7 @@ public StructType getSchema(Engine engine) {
 
   @Override
   public ScanBuilder getScanBuilder(Engine engine) {
+    // TODO when we add ScanReport we will pass the SnapshotReport downstream here
     return new ScanBuilderImpl(dataPath, protocol, metadata, getSchema(engine), logReplay, engine);
   }
 
@@ -130,6 +137,10 @@ public List<String> getPartitionColumnNames(Engine engine) {
     return VectorUtils.toJavaList(getMetadata().getPartitionColumns());
   }
 
+  public SnapshotReport getSnapshotReport() {
+    return snapshotReport;
+  }
+
   /**
    * Get the domain metadata map from the log replay, which lazily loads and replays a history of
    * domain metadata actions, resolving them to produce the current state of the domain metadata.

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

@@ -26,8 +26,11 @@
 import io.delta.kernel.exceptions.TableNotFoundException;
 import io.delta.kernel.internal.actions.Protocol;
 import io.delta.kernel.internal.fs.Path;
+import io.delta.kernel.internal.metrics.SnapshotQueryContext;
+import io.delta.kernel.internal.metrics.SnapshotReportImpl;
 import io.delta.kernel.internal.snapshot.SnapshotManager;
 import io.delta.kernel.internal.util.Clock;
+import io.delta.kernel.metrics.SnapshotReport;
 import io.delta.kernel.types.StructField;
 import io.delta.kernel.types.StructType;
 import io.delta.kernel.utils.CloseableIterator;
@@ -90,19 +93,39 @@ public String getPath(Engine engine) {
 
   @Override
   public Snapshot getLatestSnapshot(Engine engine) throws TableNotFoundException {
-    return snapshotManager.buildLatestSnapshot(engine);
+    SnapshotQueryContext snapshotContext = SnapshotQueryContext.forLatestSnapshot(tablePath);
+    try {
+      return snapshotManager.buildLatestSnapshot(engine, snapshotContext);
+    } catch (Exception e) {
+      recordSnapshotErrorReport(engine, snapshotContext, e);
+      throw e;
+    }
   }
 
   @Override
   public Snapshot getSnapshotAsOfVersion(Engine engine, long versionId)
       throws TableNotFoundException {
-    return snapshotManager.getSnapshotAt(engine, versionId);
+    SnapshotQueryContext snapshotContext =
+        SnapshotQueryContext.forVersionSnapshot(tablePath, versionId);
+    try {
+      return snapshotManager.getSnapshotAt(engine, versionId, snapshotContext);
+    } catch (Exception e) {
+      recordSnapshotErrorReport(engine, snapshotContext, e);
+      throw e;
+    }
   }
 
   @Override
   public Snapshot getSnapshotAsOfTimestamp(Engine engine, long millisSinceEpochUTC)
       throws TableNotFoundException {
-    return snapshotManager.getSnapshotForTimestamp(engine, millisSinceEpochUTC);
+    SnapshotQueryContext snapshotContext =
+        SnapshotQueryContext.forTimestampSnapshot(tablePath, millisSinceEpochUTC);
+    try {
+      return snapshotManager.getSnapshotForTimestamp(engine, millisSinceEpochUTC, snapshotContext);
+    } catch (Exception e) {
+      recordSnapshotErrorReport(engine, snapshotContext, e);
+      throw e;
+    }
   }
 
   @Override
@@ -316,4 +339,11 @@ private CloseableIterator<ColumnarBatch> getRawChanges(
     logger.info("{}: Reading the commit files with readSchema {}", tablePath, readSchema);
     return DeltaLogActionUtils.readCommitFiles(engine, commitFiles, readSchema);
   }
+
+  /** Creates a {@link SnapshotReport} and pushes it to any {@link MetricsReporter}s. */
+  private void recordSnapshotErrorReport(
+      Engine engine, SnapshotQueryContext snapshotContext, Exception e) {
+    SnapshotReport snapshotReport = SnapshotReportImpl.forError(snapshotContext, e);
+    engine.getMetricsReporters().forEach(reporter -> reporter.report(snapshotReport));
+  }
 }

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

@@ -31,6 +31,8 @@
 import io.delta.kernel.exceptions.TableNotFoundException;
 import io.delta.kernel.internal.actions.*;
 import io.delta.kernel.internal.fs.Path;
+import io.delta.kernel.internal.metrics.SnapshotMetrics;
+import io.delta.kernel.internal.metrics.SnapshotQueryContext;
 import io.delta.kernel.internal.replay.LogReplay;
 import io.delta.kernel.internal.snapshot.LogSegment;
 import io.delta.kernel.internal.snapshot.SnapshotHint;
@@ -105,8 +107,11 @@ public Transaction build(Engine engine) {
       // Table doesn't exist yet. Create an initial snapshot with the new schema.
       Metadata metadata = getInitialMetadata();
       Protocol protocol = getInitialProtocol();
-      LogReplay logReplay = getEmptyLogReplay(engine, metadata, protocol);
-      snapshot = new InitialSnapshot(table.getDataPath(), logReplay, metadata, protocol);
+      SnapshotQueryContext snapshotContext = SnapshotQueryContext.forVersionSnapshot(tablePath, -1);
+      LogReplay logReplay =
+          getEmptyLogReplay(engine, metadata, protocol, snapshotContext.getSnapshotMetrics());
+      snapshot =
+          new InitialSnapshot(table.getDataPath(), logReplay, metadata, protocol, snapshotContext);
     }
 
     boolean isNewTable = snapshot.getVersion(engine) < 0;
@@ -204,8 +209,19 @@ private void validate(Engine engine, SnapshotImpl snapshot, boolean isNewTable)
   }
 
   private class InitialSnapshot extends SnapshotImpl {
-    InitialSnapshot(Path dataPath, LogReplay logReplay, Metadata metadata, Protocol protocol) {
-      super(dataPath, LogSegment.empty(table.getLogPath()), logReplay, protocol, metadata);
+    InitialSnapshot(
+        Path dataPath,
+        LogReplay logReplay,
+        Metadata metadata,
+        Protocol protocol,
+        SnapshotQueryContext snapshotContext) {
+      super(
+          dataPath,
+          LogSegment.empty(table.getLogPath()),
+          logReplay,
+          protocol,
+          metadata,
+          snapshotContext);
     }
 
     @Override
@@ -214,14 +230,16 @@ public long getTimestamp(Engine engine) {
     }
   }
 
-  private LogReplay getEmptyLogReplay(Engine engine, Metadata metadata, Protocol protocol) {
+  private LogReplay getEmptyLogReplay(
+      Engine engine, Metadata metadata, Protocol protocol, SnapshotMetrics snapshotMetrics) {
     return new LogReplay(
         table.getLogPath(),
         table.getDataPath(),
         -1,
         engine,
         LogSegment.empty(table.getLogPath()),
-        Optional.empty()) {
+        Optional.empty(),
+        snapshotMetrics) {
 
       @Override
       protected Tuple2<Protocol, Metadata> loadTableProtocolAndMetadata(

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

@@ -0,0 +1,62 @@
+/*
+ * Copyright (2024) The Delta Lake Project Authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.delta.kernel.internal.metrics;
+
+import io.delta.kernel.metrics.SnapshotMetricsResult;
+import java.util.Optional;
+
+/**
+ * Stores the metrics for an ongoing snapshot construction. These metrics are updated and recorded
+ * throughout the snapshot query using this class.
+ *
+ * <p>At report time, we create an immutable {@link SnapshotMetricsResult} from an instance of
+ * {@link SnapshotMetrics} to capture the metrics collected during the query. The {@link
+ * SnapshotMetricsResult} interface exposes getters for any metrics collected in this class.
+ */
+public class SnapshotMetrics {
+
+  public final Timer timestampToVersionResolutionTimer = new Timer();
+
+  public final Timer loadInitialDeltaActionsTimer = new Timer();
+
+  public SnapshotMetricsResult captureSnapshotMetricsResult() {
+    return new SnapshotMetricsResult() {
+
+      final Optional<Long> timestampToVersionResolutionDurationResult =
+          timestampToVersionResolutionTimer.totalDurationIfRecorded();
+      final long loadInitialDeltaActionsDurationResult =
+          loadInitialDeltaActionsTimer.totalDurationNs();
+
+      @Override
+      public Optional<Long> timestampToVersionResolutionDurationNs() {
+        return timestampToVersionResolutionDurationResult;
+      }
+
+      @Override
+      public long loadInitialDeltaActionsDurationNs() {
+        return loadInitialDeltaActionsDurationResult;
+      }
+    };
+  }
+
+  @Override
+  public String toString() {
+    return String.format(
+        "SnapshotMetrics(timestampToVersionResolutionTimer=%s, "
+            + "loadInitialDeltaActionsTimer=%s)",
+        timestampToVersionResolutionTimer, loadInitialDeltaActionsTimer);
+  }
+}

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

@@ -0,0 +1,97 @@
+/*
+ * Copyright (2024) The Delta Lake Project Authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.delta.kernel.internal.metrics;
+
+import io.delta.kernel.metrics.SnapshotReport;
+import java.util.Optional;
+
+/**
+ * Stores the context for a given Snapshot query. This includes information about the query
+ * parameters (i.e. table path, time travel parameters), updated state as the snapshot query
+ * progresses (i.e. resolved version), and metrics.
+ *
+ * <p>This is used to generate a {@link SnapshotReport}. It exists from snapshot query initiation
+ * until either successful snapshot construction or failure.
+ */
+public class SnapshotQueryContext {
+
+  /** Creates a {@link SnapshotQueryContext} for a Snapshot created by a latest snapshot query */
+  public static SnapshotQueryContext forLatestSnapshot(String tablePath) {
+    return new SnapshotQueryContext(tablePath, Optional.empty(), Optional.empty());
+  }
+
+  /** Creates a {@link SnapshotQueryContext} for a Snapshot created by a AS OF VERSION query */
+  public static SnapshotQueryContext forVersionSnapshot(String tablePath, long version) {
+    return new SnapshotQueryContext(tablePath, Optional.of(version), Optional.empty());
+  }
+
+  /** Creates a {@link SnapshotQueryContext} for a Snapshot created by a AS OF TIMESTAMP query */
+  public static SnapshotQueryContext forTimestampSnapshot(String tablePath, long timestamp) {
+    return new SnapshotQueryContext(tablePath, Optional.empty(), Optional.of(timestamp));
+  }
+
+  private final String tablePath;
+  private Optional<Long> version;
+  private final Optional<Long> providedTimestamp;
+  private final SnapshotMetrics snapshotMetrics = new SnapshotMetrics();
+
+  /**
+   * @param tablePath the table path for the table being queried
+   * @param providedVersion the provided version for a time-travel-by-version query, empty if this
+   *     is not a time-travel-by-version query
+   * @param providedTimestamp the provided timestamp for a time-travel-by-timestamp query, empty if
+   *     this is not a time-travel-by-timestamp query
+   */
+  private SnapshotQueryContext(
+      String tablePath, Optional<Long> providedVersion, Optional<Long> providedTimestamp) {
+    this.tablePath = tablePath;
+    this.version = providedVersion;
+    this.providedTimestamp = providedTimestamp;
+  }
+
+  public String getTablePath() {
+    return tablePath;
+  }
+
+  public Optional<Long> getVersion() {
+    return version;
+  }
+
+  public Optional<Long> getProvidedTimestamp() {
+    return providedTimestamp;
+  }
+
+  public SnapshotMetrics getSnapshotMetrics() {
+    return snapshotMetrics;
+  }
+
+  /**
+   * Updates the {@code version} stored in this snapshot context. This version should be updated
+   * upon version resolution for non time-travel-by-version queries. For latest snapshot queries
+   * this is after log segment construction. For time-travel by timestamp queries this is after
+   * timestamp to version resolution.
+   */
+  public void setVersion(long updatedVersion) {
+    version = Optional.of(updatedVersion);
+  }
+
+  @Override
+  public String toString() {
+    return String.format(
+        "SnapshotQueryContext(tablePath=%s, version=%s, providedTimestamp=%s, snapshotMetric=%s)",
+        tablePath, version, providedTimestamp, snapshotMetrics);
+  }
+}