Add metrics framework for Snapshot creation

Author: gotocoding-DB

kernel/src/lib.rs

@@ -93,6 +93,7 @@ pub mod error;
 pub mod expressions;
 mod log_compaction;
 mod log_path;
+pub mod metrics;
 pub mod scan;
 pub mod schema;
 pub mod snapshot;
@@ -709,6 +710,14 @@ pub trait Engine: AsAny {
 
     /// Get the connector provided [`ParquetHandler`].
     fn parquet_handler(&self) -> Arc<dyn ParquetHandler>;
+
+    /// Get the connector provided [`metrics::MetricsReporter`] for metrics collection.
+    ///
+    /// Returns an optional reporter that will be notified of metrics from Delta operations.
+    /// The default implementation returns None (no metrics reporting).
+    fn get_metrics_reporter(&self) -> Option<Arc<dyn crate::metrics::MetricsReporter>> {
+        None
+    }
 }
 
 // we have an 'internal' feature flag: default-engine-base, which is actually just the shared

kernel/src/metrics/mod.rs

@@ -0,0 +1,117 @@
+//! Metrics collection for Delta Kernel operations.
+//!
+//! This module provides metrics tracking for various Delta operations including
+//! snapshot creation, scans, and transactions. Metrics are collected during operations
+//! and can be reported via the `MetricsReporter` trait.
+//!
+//! # Example: Implementing a Custom MetricsReporter
+//!
+//! ```
+//! use std::sync::Arc;
+//! use delta_kernel::metrics::{MetricsReporter, SnapshotMetrics};
+//!
+//! struct LoggingReporter;
+//!
+//! impl MetricsReporter for LoggingReporter {
+//!     fn report_snapshot(&self, metrics: &SnapshotMetrics) {
+//!         println!("Snapshot creation took {:?}", metrics.load_snapshot_total_duration);
+//!         println!("  - Protocol/Metadata load: {:?}", metrics.load_protocol_metadata_duration);
+//!         println!("  - Log segment load: {:?}", metrics.load_log_segment_duration);
+//!         println!("  - Files: {} commits, {} checkpoints, {} compactions",
+//!                  metrics.num_commit_files,
+//!                  metrics.num_checkpoint_files,
+//!                  metrics.num_compaction_files);
+//!     }
+//! }
+//! ```
+//!
+//! # Example: Implementing a Composite Reporter
+//!
+//! If you need to send metrics to multiple destinations, you can create a composite reporter:
+//!
+//! ```
+//! use std::sync::Arc;
+//! use delta_kernel::metrics::{MetricsReporter, SnapshotMetrics};
+//!
+//! struct CompositeReporter {
+//!     reporters: Vec<Arc<dyn MetricsReporter>>,
+//! }
+//!
+//! impl MetricsReporter for CompositeReporter {
+//!     fn report_snapshot(&self, metrics: &SnapshotMetrics) {
+//!         for reporter in &self.reporters {
+//!             reporter.report_snapshot(metrics);
+//!         }
+//!     }
+//! }
+//! ```
+
+use std::time::{Duration, Instant};
+
+/// Trait for reporting metrics from Delta operations.
+///
+/// Implementations of this trait receive metrics after operations complete
+/// and can forward them to monitoring systems like Prometheus, DataDog, etc.
+pub trait MetricsReporter: Send + Sync {
+    /// Report snapshot creation metrics.
+    fn report_snapshot(&self, metrics: &SnapshotMetrics);
+}
+
+/// Metrics collected during snapshot creation.
+///
+/// These metrics track the time spent in various phases of loading a snapshot,
+/// including log segment construction and protocol/metadata loading.
+/// They also track counts of files processed during snapshot creation.
+#[derive(Debug, Clone, Default)]
+pub struct SnapshotMetrics {
+    /// Total time spent loading the snapshot (all operations combined).
+    pub load_snapshot_total_duration: Duration,
+
+    /// Time spent loading protocol and metadata actions.
+    pub load_protocol_metadata_duration: Duration,
+
+    /// Time spent building the log segment (listing and organizing log files).
+    pub load_log_segment_duration: Duration,
+
+    /// Number of commit files processed during snapshot creation.
+    pub num_commit_files: u64,
+
+    /// Number of checkpoint files processed during snapshot creation.
+    pub num_checkpoint_files: u64,
+
+    /// Number of compaction files processed during snapshot creation.
+    pub num_compaction_files: u64,
+}
+
+/// A simple timer for tracking operation durations.
+///
+/// # Example
+/// ```ignore
+/// let timer = Timer::new();
+/// // ... do work ...
+/// let duration = timer.elapsed();
+/// ```
+#[derive(Debug)]
+pub struct Timer {
+    start: Instant,
+}
+
+impl Timer {
+    /// Create a new timer that starts immediately.
+    pub fn new() -> Self {
+        Self {
+            start: Instant::now(),
+        }
+    }
+
+    /// Get the elapsed time as a Duration since this timer was created.
+    pub fn elapsed(&self) -> Duration {
+        self.start.elapsed()
+    }
+}
+
+impl Default for Timer {
+    fn default() -> Self {
+        Self::new()
+    }
+}

kernel/src/snapshot.rs

@@ -11,6 +11,7 @@ use crate::checkpoint::CheckpointWriter;
 use crate::committer::Committer;
 use crate::listed_log_files::ListedLogFiles;
 use crate::log_segment::LogSegment;
+use crate::metrics::{SnapshotMetrics, Timer};
 use crate::path::ParsedLogPath;
 use crate::scan::ScanBuilder;
 use crate::schema::SchemaRef;
@@ -254,9 +255,47 @@ impl Snapshot {
         log_segment: LogSegment,
         engine: &dyn Engine,
     ) -> DeltaResult<Self> {
+        Self::try_new_from_log_segment_with_metrics(
+            location,
+            log_segment,
+            engine,
+            std::time::Duration::default(),
+        )
+    }
+
+    pub(crate) fn try_new_from_log_segment_with_metrics(
+        location: Url,
+        log_segment: LogSegment,
+        engine: &dyn Engine,
+        load_log_segment_duration: std::time::Duration,
+    ) -> DeltaResult<Self> {
+        let timer = Timer::new();
+
         let (metadata, protocol) = log_segment.read_metadata(engine)?;
+        let load_protocol_metadata_duration = timer.elapsed();
+
         let table_configuration =
             TableConfiguration::try_new(metadata, protocol, location, log_segment.end_version)?;
+
+        let num_commit_files = log_segment.ascending_commit_files.len() as u64;
+        let num_checkpoint_files = log_segment.checkpoint_parts.len() as u64;
+        let num_compaction_files = log_segment.ascending_compaction_files.len() as u64;
+
+        let load_snapshot_total_duration = timer.elapsed();
+
+        let metrics = SnapshotMetrics {
+            load_snapshot_total_duration,
+            load_protocol_metadata_duration,
+            load_log_segment_duration,
+            num_commit_files,
+            num_checkpoint_files,
+            num_compaction_files,
+        };
+
+        if let Some(reporter) = engine.get_metrics_reporter() {
+            reporter.report_snapshot(&metrics);
+        }
+
         Ok(Self {
             log_segment,
             table_configuration,

kernel/src/snapshot/builder.rs

@@ -1,6 +1,7 @@
 //! Builder for creating [`Snapshot`] instances.
 use crate::log_path::LogPath;
 use crate::log_segment::LogSegment;
+use crate::metrics::Timer;
 use crate::snapshot::SnapshotRef;
 use crate::{DeltaResult, Engine, Error, Snapshot, Version};
 
@@ -83,13 +84,22 @@ impl SnapshotBuilder {
     pub fn build(self, engine: &dyn Engine) -> DeltaResult<SnapshotRef> {
         let log_tail = self.log_tail.into_iter().map(Into::into).collect();
         if let Some(table_root) = self.table_root {
+            let timer = Timer::new();
             let log_segment = LogSegment::for_snapshot(
                 engine.storage_handler().as_ref(),
                 table_root.join("_delta_log/")?,
                 log_tail,
                 self.version,
             )?;
-            Ok(Snapshot::try_new_from_log_segment(table_root, log_segment, engine)?.into())
+            let load_log_segment_duration = timer.elapsed();
+
+            Ok(Snapshot::try_new_from_log_segment_with_metrics(
+                table_root,
+                log_segment,
+                engine,
+                load_log_segment_duration,
+            )?
+            .into())
         } else {
             let existing_snapshot = self.existing_snapshot.ok_or_else(|| {
                 Error::internal_error(