better

Author: gotocoding-DB

ffi/src/lib.rs

@@ -565,6 +565,7 @@ fn get_default_engine_impl(
         &url,
         options,
         Arc::new(TokioBackgroundExecutor::new()),
+        None,
     );
     Ok(engine_to_handle(Arc::new(engine?), allocate_error))
 }

kernel/examples/common/src/lib.rs

@@ -161,13 +161,14 @@ pub fn get_engine(
         Ok(DefaultEngine::new(
             store,
             Arc::new(TokioBackgroundExecutor::new()),
+            None,
         ))
     } else if !args.option.is_empty() {
         let opts = args.option.iter().map(|option| {
             let parts: Vec<&str> = option.split("=").collect();
             (parts[0].to_ascii_lowercase(), parts[1])
         });
-        DefaultEngine::try_new(url, opts, Arc::new(TokioBackgroundExecutor::new()))
+        DefaultEngine::try_new(url, opts, Arc::new(TokioBackgroundExecutor::new()), None)
     } else {
         let mut options = if let Some(ref region) = args.region {
             HashMap::from([("region", region.clone())])
@@ -177,7 +178,7 @@ pub fn get_engine(
         if args.public {
             options.insert("skip_signature", "true".to_string());
         }
-        DefaultEngine::try_new(url, options, Arc::new(TokioBackgroundExecutor::new()))
+        DefaultEngine::try_new(url, options, Arc::new(TokioBackgroundExecutor::new()), None)
     }
 }
 

kernel/examples/write-table/src/main.rs

@@ -80,6 +80,7 @@ async fn try_main() -> DeltaResult<()> {
         &url,
         HashMap::<String, String>::new(),
         Arc::new(TokioBackgroundExecutor::new()),
+        None,
     )?;
 
     // Create or get the table

kernel/src/snapshot.rs

@@ -264,7 +264,7 @@ impl Snapshot {
         engine: &dyn Engine,
         operation_id: Option<MetricId>,
     ) -> DeltaResult<Self> {
-        let operation_id = operation_id.unwrap_or_else(MetricId::new);
+        let operation_id = operation_id.unwrap_or_default();
         let reporter = engine.get_metrics_reporter();
         let timer = Timer::new();
 

kernel/src/snapshot/builder.rs

@@ -83,17 +83,23 @@ impl SnapshotBuilder {
     /// - `engine`: Implementation of [`Engine`] apis.
     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 operation_id = MetricId::new();
-            let reporter = engine.get_metrics_reporter();
-
-            if let Some(ref r) = reporter {
-                r.report(MetricEvent::SnapshotStarted {
-                    operation_id,
-                    table_path: table_root.to_string(),
-                });
-            }
 
+        let operation_id = MetricId::new();
+        let reporter = engine.get_metrics_reporter();
+
+        let table_path = self.table_root.as_ref()
+            .map(|url| url.to_string())
+            .or_else(|| self.existing_snapshot.as_ref().map(|s| s.table_root().to_string()))
+            .unwrap_or_else(|| "unknown".to_string());
+
+        if let Some(ref r) = reporter {
+            r.report(MetricEvent::SnapshotStarted {
+                operation_id,
+                table_path,
+            });
+        }
+
+        if let Some(table_root) = self.table_root {
             let timer = Timer::new();
             let log_segment_result = LogSegment::for_snapshot(
                 engine.storage_handler().as_ref(),
@@ -137,7 +143,28 @@ impl SnapshotBuilder {
                     "SnapshotBuilder should have either table_root or existing_snapshot",
                 )
             })?;
-            Snapshot::try_new_from(existing_snapshot, log_tail, engine, self.version)
+
+            let timer = Timer::new();
+            let result = Snapshot::try_new_from(existing_snapshot, log_tail, engine, self.version);
+
+            match result {
+                Ok(snapshot) => {
+                    if let Some(ref r) = reporter {
+                        r.report(MetricEvent::SnapshotCompleted {
+                            operation_id,
+                            version: snapshot.version(),
+                            total_duration: timer.elapsed(),
+                        });
+                    }
+                    Ok(snapshot)
+                }
+                Err(e) => {
+                    if let Some(ref r) = reporter {
+                        r.report(MetricEvent::SnapshotFailed { operation_id });
+                    }
+                    Err(e)
+                }
+            }
         }
     }
 }
@@ -165,6 +192,7 @@ mod tests {
         let engine = Arc::new(DefaultEngine::new(
             store.clone(),
             Arc::new(TokioBackgroundExecutor::new()),
+            None,
         ));
         (engine, store, table_root)
     }