Address comments from Nick and Oussama

gotocoding-DB · gotocoding-DB · commit 38ad6a527db3 · 2025-11-20T16:37:28.000Z
diff --git a/kernel/src/engine/default/filesystem.rs b/kernel/src/engine/default/filesystem.rs
@@ -18,6 +18,7 @@ use crate::{DeltaResult, Error, FileMeta, FileSlice, StorageHandler};
 ///
 /// Generic over the inner iterator type and item type.
 /// The `event_fn` receives (duration, num_files, bytes_read) to construct the appropriate MetricEvent.
+/// Metrics are emitted either when the iterator is exhausted or when dropped.
 struct MetricsIterator<I, T>
 where
     I: Iterator<Item = DeltaResult<T>>,
@@ -155,6 +156,7 @@ impl<E: TaskExecutor> StorageHandler for ObjectStoreStorageHandler<E> {
         &self,
         path: &Url,
     ) -> DeltaResult<Box<dyn Iterator<Item = DeltaResult<FileMeta>>>> {
+        let start = Instant::now();
         // The offset is used for list-after; the prefix is used to restrict the listing to a specific directory.
         // Unfortunately, `Path` provides no easy way to check whether a name is directory-like,
         // because it strips trailing /, so we're reduced to manually checking the original URL.
@@ -220,7 +222,6 @@ impl<E: TaskExecutor> StorageHandler for ObjectStoreStorageHandler<E> {
             }
         });
 
-        let start = Instant::now();
         let reporter = self.reporter.clone();
 
         if !has_ordered_listing {
@@ -261,8 +262,8 @@ impl<E: TaskExecutor> StorageHandler for ObjectStoreStorageHandler<E> {
         &self,
         files: Vec<FileSlice>,
     ) -> DeltaResult<Box<dyn Iterator<Item = DeltaResult<Bytes>>>> {
-        let store = self.inner.clone();
         let start = Instant::now();
+        let store = self.inner.clone();
 
         // This channel will become the output iterator.
         // Because there will already be buffering in the stream, we set the
diff --git a/kernel/src/engine/default/mod.rs b/kernel/src/engine/default/mod.rs
@@ -19,6 +19,7 @@ use self::parquet::DefaultParquetHandler;
 use super::arrow_conversion::TryFromArrow as _;
 use super::arrow_data::ArrowEngineData;
 use super::arrow_expression::ArrowEvaluationHandler;
+use crate::metrics::MetricsReporter;
 use crate::schema::Schema;
 use crate::transaction::WriteContext;
 use crate::{
@@ -40,7 +41,7 @@ pub struct DefaultEngine<E: TaskExecutor> {
     json: Arc<DefaultJsonHandler<E>>,
     parquet: Arc<DefaultParquetHandler<E>>,
     evaluation: Arc<ArrowEvaluationHandler>,
-    reporter: Option<Arc<dyn crate::metrics::MetricsReporter>>,
+    reporter: Option<Arc<dyn MetricsReporter>>,
 }
 
 impl DefaultEngine<executor::tokio::TokioBackgroundExecutor> {
@@ -60,18 +61,12 @@ impl DefaultEngine<executor::tokio::TokioBackgroundExecutor> {
     }
 
     /// Create a new [`DefaultEngine`] instance with the default executor and metrics reporting.
-    ///
-    /// Uses `TokioBackgroundExecutor` as the default executor.
-    /// For custom executors, use [`DefaultEngine::new_with_executor_and_metrics_reporter`].
-    ///
-    /// # Parameters
-    ///
-    /// - `object_store`: The object store to use.
-    /// - `reporter`: Metrics reporter for tracking operation metrics.
+    /// TODO: Switch to Builder pattern to avoid explosion of constructor methods.
+    #[allow(dead_code)]
     #[internal_api]
     pub(crate) fn new_with_metrics_reporter(
         object_store: Arc<DynObjectStore>,
-        reporter: Arc<dyn crate::metrics::MetricsReporter>,
+        reporter: Arc<dyn MetricsReporter>,
     ) -> Self {
         Self::new_with_executor_and_metrics_reporter(
             object_store,
@@ -113,24 +108,13 @@ impl<E: TaskExecutor> DefaultEngine<E> {
     }
 
     /// Create a new [`DefaultEngine`] instance with a custom executor and metrics reporting.
-    ///
-    /// Most users should use [`DefaultEngine::new`] instead. This method is only
-    /// needed for specialized testing scenarios (e.g., multi-threaded executors) with metrics.
-    ///
-    /// # Parameters
-    ///
-    /// - `object_store`: The object store to use.
-    /// - `task_executor`: Used to spawn async IO tasks. See [executor::TaskExecutor].
-    /// - `reporter`: Metrics reporter for tracking operation metrics.
-    ///
-    /// # TODO
-    ///
-    /// Switch to Builder pattern to avoid explosion of constructor methods.
+    /// TODO: Switch to Builder pattern to avoid explosion of constructor methods.
+    #[allow(dead_code)]
     #[internal_api]
     pub(crate) fn new_with_executor_and_metrics_reporter(
         object_store: Arc<DynObjectStore>,
         task_executor: Arc<E>,
-        reporter: Arc<dyn crate::metrics::MetricsReporter>,
+        reporter: Arc<dyn MetricsReporter>,
     ) -> Self {
         Self {
             storage: Arc::new(ObjectStoreStorageHandler::new(
@@ -194,7 +178,7 @@ impl<E: TaskExecutor> Engine for DefaultEngine<E> {
         self.parquet.clone()
     }
 
-    fn get_metrics_reporter(&self) -> Option<Arc<dyn crate::metrics::MetricsReporter>> {
+    fn get_metrics_reporter(&self) -> Option<Arc<dyn MetricsReporter>> {
         self.reporter.clone()
     }
 }
diff --git a/kernel/src/lib.rs b/kernel/src/lib.rs
@@ -155,6 +155,7 @@ pub use engine_data::{EngineData, FilteredEngineData, RowVisitor};
 pub use error::{DeltaResult, Error};
 pub use expressions::{Expression, ExpressionRef, Predicate, PredicateRef};
 pub use log_compaction::{should_compact, LogCompactionWriter};
+pub use metrics::MetricsReporter;
 pub use snapshot::Snapshot;
 pub use snapshot::SnapshotRef;
 
@@ -717,11 +718,11 @@ 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.
+    /// Get the connector provided [`MetricsReporter`] for metrics collection.
     ///
     /// Returns an optional reporter that will receive metric events from Delta operations.
     /// The default implementation returns None (no metrics reporting).
-    fn get_metrics_reporter(&self) -> Option<Arc<dyn crate::metrics::MetricsReporter>> {
+    fn get_metrics_reporter(&self) -> Option<Arc<dyn MetricsReporter>> {
         None
     }
 }
diff --git a/kernel/src/log_segment.rs b/kernel/src/log_segment.rs
@@ -3,13 +3,16 @@
 use std::num::NonZero;
 use std::sync::{Arc, LazyLock};
 
+use std::time::Instant;
+
 use crate::actions::visitors::SidecarVisitor;
 use crate::actions::{
     get_commit_schema, schema_contains_file_actions, Metadata, Protocol, Sidecar, METADATA_NAME,
     PROTOCOL_NAME, SIDECAR_NAME,
 };
 use crate::last_checkpoint_hint::LastCheckpointHint;
 use crate::log_replay::ActionsBatch;
+use crate::metrics::{MetricEvent, MetricId, MetricsReporter};
 use crate::path::{LogPathFileType, ParsedLogPath};
 use crate::schema::{SchemaRef, StructField, ToSchema as _};
 use crate::utils::require;
@@ -145,22 +148,46 @@ impl LogSegment {
     /// - `time_travel_version`: The version of the log that the Snapshot will be at.
     ///
     /// [`Snapshot`]: crate::snapshot::Snapshot
+    ///
+    /// Reports metrics: `LogSegmentLoaded`.
     #[internal_api]
     pub(crate) fn for_snapshot(
         storage: &dyn StorageHandler,
         log_root: Url,
         log_tail: Vec<ParsedLogPath>,
         time_travel_version: impl Into<Option<Version>>,
+        reporter: Option<&Arc<dyn MetricsReporter>>,
+        operation_id: Option<MetricId>,
     ) -> DeltaResult<Self> {
+        let operation_id = operation_id.unwrap_or_default();
+        let start = Instant::now();
+
         let time_travel_version = time_travel_version.into();
         let checkpoint_hint = LastCheckpointHint::try_read(storage, &log_root)?;
-        Self::for_snapshot_impl(
+        let result = Self::for_snapshot_impl(
             storage,
             log_root,
             log_tail,
             checkpoint_hint,
             time_travel_version,
-        )
+        );
+        let log_segment_loading_duration = start.elapsed();
+
+        match result {
+            Ok(log_segment) => {
+                reporter.inspect(|r| {
+                    r.report(MetricEvent::LogSegmentLoaded {
+                        operation_id,
+                        duration: log_segment_loading_duration,
+                        num_commit_files: log_segment.ascending_commit_files.len() as u64,
+                        num_checkpoint_files: log_segment.checkpoint_parts.len() as u64,
+                        num_compaction_files: log_segment.ascending_compaction_files.len() as u64,
+                    });
+                });
+                Ok(log_segment)
+            }
+            Err(e) => Err(e),
+        }
     }
 
     // factored out for testing
diff --git a/kernel/src/log_segment/tests.rs b/kernel/src/log_segment/tests.rs
@@ -2110,7 +2110,8 @@ fn test_latest_commit_file_field_is_captured() {
     );
 
     let log_segment =
-        LogSegment::for_snapshot(storage.as_ref(), log_root.clone(), vec![], None).unwrap();
+        LogSegment::for_snapshot(storage.as_ref(), log_root.clone(), vec![], None, None, None)
+            .unwrap();
 
     // The latest commit should be version 5
     assert_eq!(log_segment.latest_commit_file.unwrap().version, 5);
@@ -2136,7 +2137,8 @@ fn test_latest_commit_file_with_checkpoint_filtering() {
     );
 
     let log_segment =
-        LogSegment::for_snapshot(storage.as_ref(), log_root.clone(), vec![], None).unwrap();
+        LogSegment::for_snapshot(storage.as_ref(), log_root.clone(), vec![], None, None, None)
+            .unwrap();
 
     // The latest commit should be version 4
     assert_eq!(log_segment.latest_commit_file.unwrap().version, 4);
@@ -2156,7 +2158,8 @@ fn test_latest_commit_file_with_no_commits() {
     );
 
     let log_segment =
-        LogSegment::for_snapshot(storage.as_ref(), log_root.clone(), vec![], None).unwrap();
+        LogSegment::for_snapshot(storage.as_ref(), log_root.clone(), vec![], None, None, None)
+            .unwrap();
 
     // latest_commit_file should be None when there are no commits
     assert!(log_segment.latest_commit_file.is_none());
@@ -2179,7 +2182,8 @@ fn test_latest_commit_file_with_checkpoint_at_same_version() {
     );
 
     let log_segment =
-        LogSegment::for_snapshot(storage.as_ref(), log_root.clone(), vec![], None).unwrap();
+        LogSegment::for_snapshot(storage.as_ref(), log_root.clone(), vec![], None, None, None)
+            .unwrap();
 
     // The latest commit should be version 1 (saved before filtering)
     assert_eq!(log_segment.latest_commit_file.unwrap().version, 1);
@@ -2204,7 +2208,8 @@ fn test_latest_commit_file_edge_case_commit_before_checkpoint() {
     );
 
     let log_segment =
-        LogSegment::for_snapshot(storage.as_ref(), log_root.clone(), vec![], None).unwrap();
+        LogSegment::for_snapshot(storage.as_ref(), log_root.clone(), vec![], None, None, None)
+            .unwrap();
 
     // latest_commit_file should be None since there's no commit at the checkpoint version
     assert!(log_segment.latest_commit_file.is_none());
diff --git a/kernel/src/snapshot.rs b/kernel/src/snapshot.rs
@@ -267,8 +267,8 @@ impl Snapshot {
     ) -> DeltaResult<Arc<Self>> {
         let operation_id = operation_id.unwrap_or_default();
         let reporter = engine.get_metrics_reporter();
-        let start = Instant::now();
 
+        let start = Instant::now();
         let result =
             Self::try_new_from_impl(existing_snapshot, log_tail, engine, version, operation_id);
         let snapshot_duration = start.elapsed();
@@ -298,15 +298,29 @@ impl Snapshot {
 
     /// Implementation of snapshot creation from log segment.
     ///
-    /// Calls `build_table_configuration` which reports `ProtocolMetadataLoaded`.
+    /// Reports metrics: `ProtocolMetadataLoaded`.
     fn try_new_from_log_segment_impl(
         location: Url,
         log_segment: LogSegment,
         engine: &dyn Engine,
         operation_id: MetricId,
     ) -> DeltaResult<Self> {
+        let reporter = engine.get_metrics_reporter();
+
+        let start = Instant::now();
+        let (metadata, protocol) = log_segment.read_metadata(engine)?;
+        let read_metadata_duration = start.elapsed();
+
+        reporter.as_ref().inspect(|r| {
+            r.report(MetricEvent::ProtocolMetadataLoaded {
+                operation_id,
+                duration: read_metadata_duration,
+            });
+        });
+
         let table_configuration =
-            Self::build_table_configuration(&log_segment, engine, location, operation_id)?;
+            TableConfiguration::try_new(metadata, protocol, location, log_segment.end_version)?;
+
         Ok(Self {
             log_segment,
             table_configuration,
@@ -324,8 +338,8 @@ impl Snapshot {
     ) -> DeltaResult<Self> {
         let operation_id = operation_id.unwrap_or_default();
         let reporter = engine.get_metrics_reporter();
-        let start = Instant::now();
 
+        let start = Instant::now();
         let result =
             Self::try_new_from_log_segment_impl(location, log_segment, engine, operation_id);
         let snapshot_duration = start.elapsed();
@@ -516,33 +530,6 @@ impl Snapshot {
             None => Err(Error::generic("Last commit file not found in log segment")),
         }
     }
-
-    /// Build table configuration from log segment.
-    ///
-    /// Reports metrics: `ProtocolMetadataLoaded`.
-    fn build_table_configuration(
-        log_segment: &LogSegment,
-        engine: &dyn Engine,
-        location: Url,
-        operation_id: MetricId,
-    ) -> DeltaResult<TableConfiguration> {
-        let start = Instant::now();
-        let reporter = engine.get_metrics_reporter();
-        let (metadata, protocol) = log_segment.read_metadata(engine)?;
-        let read_metadata_duration = start.elapsed();
-
-        reporter.as_ref().inspect(|r| {
-            r.report(MetricEvent::ProtocolMetadataLoaded {
-                operation_id,
-                duration: read_metadata_duration,
-            });
-        });
-
-        let table_configuration =
-            TableConfiguration::try_new(metadata, protocol, location, log_segment.end_version)?;
-
-        Ok(table_configuration)
-    }
 }
 
 #[cfg(test)]
diff --git a/kernel/src/snapshot/builder.rs b/kernel/src/snapshot/builder.rs
@@ -1,9 +1,7 @@
 //! Builder for creating [`Snapshot`] instances.
-use std::time::Instant;
-
 use crate::log_path::LogPath;
 use crate::log_segment::LogSegment;
-use crate::metrics::{MetricEvent, MetricId};
+use crate::metrics::MetricId;
 use crate::snapshot::SnapshotRef;
 use crate::{DeltaResult, Engine, Error, Snapshot, Version};
 
@@ -84,44 +82,20 @@ impl SnapshotBuilder {
     ///
     /// - `engine`: Implementation of [`Engine`] apis.
     pub fn build(self, engine: &dyn Engine) -> DeltaResult<SnapshotRef> {
-        let start = Instant::now();
         let log_tail = self.log_tail.into_iter().map(Into::into).collect();
 
         let operation_id = MetricId::new();
         let reporter = engine.get_metrics_reporter();
 
         if let Some(table_root) = self.table_root {
-            let log_segment_result = LogSegment::for_snapshot(
+            let log_segment = LogSegment::for_snapshot(
                 engine.storage_handler().as_ref(),
                 table_root.join("_delta_log/")?,
                 log_tail,
                 self.version,
-            );
-            let log_segment_loading_duration = start.elapsed();
-
-            let log_segment = match log_segment_result {
-                Ok(seg) => {
-                    reporter.as_ref().inspect(|r| {
-                        r.report(MetricEvent::LogSegmentLoaded {
-                            operation_id,
-                            duration: log_segment_loading_duration,
-                            num_commit_files: seg.ascending_commit_files.len() as u64,
-                            num_checkpoint_files: seg.checkpoint_parts.len() as u64,
-                            num_compaction_files: seg.ascending_compaction_files.len() as u64,
-                        });
-                    });
-                    seg
-                }
-                Err(e) => {
-                    reporter.as_ref().inspect(|r| {
-                        r.report(MetricEvent::SnapshotFailed {
-                            operation_id,
-                            duration: log_segment_loading_duration,
-                        });
-                    });
-                    return Err(e);
-                }
-            };
+                reporter.as_ref(),
+                Some(operation_id),
+            )?;
 
             Ok(Snapshot::try_new_from_log_segment(
                 table_root,
