Simplify iceberg persistence stats. (#2142)

Author: fweilun

src/moonlink/src/observability/iceberg_persistence.rs

@@ -3,8 +3,7 @@ use crate::observability::latency_guard::LatencyGuard;
 use opentelemetry::metrics::Histogram;
 use opentelemetry::{global, KeyValue};
 
-#[derive(Debug, Clone, Copy)]
-pub(crate) enum IcebergPersistenceStage {
+enum PersistenceStage {
     Overall,
     DataFiles,
     FileIndices,
@@ -14,35 +13,71 @@ pub(crate) enum IcebergPersistenceStage {
 
 #[derive(Debug)]
 pub(crate) struct IcebergPersistenceStats {
+    pub(crate) overall: IcebergPersistenceSingleStats,
+    pub(crate) sync_data_files: IcebergPersistenceSingleStats,
+    pub(crate) sync_file_indices: IcebergPersistenceSingleStats,
+    pub(crate) sync_deletion_vectors: IcebergPersistenceSingleStats,
+    pub(crate) transaction_commit: IcebergPersistenceSingleStats,
+}
+
+impl IcebergPersistenceStats {
+    pub(crate) fn new(mooncake_table_id: String) -> Self {
+        Self {
+            overall: IcebergPersistenceSingleStats::new(
+                mooncake_table_id.to_string(),
+                PersistenceStage::Overall,
+            ),
+            sync_data_files: IcebergPersistenceSingleStats::new(
+                mooncake_table_id.to_string(),
+                PersistenceStage::DataFiles,
+            ),
+            sync_file_indices: IcebergPersistenceSingleStats::new(
+                mooncake_table_id.to_string(),
+                PersistenceStage::FileIndices,
+            ),
+            sync_deletion_vectors: IcebergPersistenceSingleStats::new(
+                mooncake_table_id.to_string(),
+                PersistenceStage::DeletionVectors,
+            ),
+            transaction_commit: IcebergPersistenceSingleStats::new(
+                mooncake_table_id.to_string(),
+                PersistenceStage::TransactionCommit,
+            ),
+        }
+    }
+}
+
+#[derive(Debug)]
+pub(crate) struct IcebergPersistenceSingleStats {
     mooncake_table_id: String,
     latency: Histogram<u64>,
 }
 
-impl IcebergPersistenceStats {
-    pub(crate) fn new(mooncake_table_id: String, stats_type: IcebergPersistenceStage) -> Self {
+impl IcebergPersistenceSingleStats {
+    fn new(mooncake_table_id: String, stats_type: PersistenceStage) -> Self {
         let meter = global::meter("iceberg_persistence");
         let latency = match stats_type {
-            IcebergPersistenceStage::Overall => meter
+            PersistenceStage::Overall => meter
                 .u64_histogram("snapshot_synchronization_latency")
                 .with_description("Latency (ms) for snapshot synchronization")
                 .with_boundaries(vec![50.0, 100.0, 200.0, 300.0, 400.0, 500.0])
                 .build(),
-            IcebergPersistenceStage::DataFiles => meter
+            PersistenceStage::DataFiles => meter
                 .u64_histogram("sync_data_files_latency")
                 .with_description("Latency (ms) for data files synchronization")
                 .with_boundaries(vec![50.0, 100.0, 200.0, 300.0, 400.0, 500.0])
                 .build(),
-            IcebergPersistenceStage::FileIndices => meter
+            PersistenceStage::FileIndices => meter
                 .u64_histogram("sync_file_indices_latency")
                 .with_description("Latency (ms) for file indices synchronization")
                 .with_boundaries(vec![50.0, 100.0, 200.0, 300.0, 400.0, 500.0])
                 .build(),
-            IcebergPersistenceStage::DeletionVectors => meter
+            PersistenceStage::DeletionVectors => meter
                 .u64_histogram("sync_deletion_vectors_latency")
                 .with_description("Latency (ms) for deletion vectors synchronization")
                 .with_boundaries(vec![50.0, 100.0, 200.0, 300.0, 400.0, 500.0])
                 .build(),
-            IcebergPersistenceStage::TransactionCommit => meter
+            PersistenceStage::TransactionCommit => meter
                 .u64_histogram("transaction_commit_latency")
                 .with_description("Latency (ms) for transaction commit")
                 .with_boundaries(vec![50.0, 100.0, 200.0, 300.0, 400.0, 500.0])
@@ -56,7 +91,7 @@ impl IcebergPersistenceStats {
     }
 }
 
-impl BaseLatencyExporter for IcebergPersistenceStats {
+impl BaseLatencyExporter for IcebergPersistenceSingleStats {
     fn start<'a>(&'a self) -> LatencyGuard<'a> {
         LatencyGuard::new(self.mooncake_table_id.clone(), self)
     }

src/moonlink/src/storage/iceberg/iceberg_table_manager.rs

@@ -1,4 +1,4 @@
-use crate::observability::iceberg_persistence::{IcebergPersistenceStage, IcebergPersistenceStats};
+use crate::observability::iceberg_persistence::IcebergPersistenceStats;
 use crate::observability::iceberg_table_recovery::IcebergTableRecoveryStats;
 use crate::observability::latency_exporter::BaseLatencyExporter;
 use crate::storage::cache::object_storage::base_cache::CacheTrait;
@@ -74,20 +74,8 @@ pub struct IcebergTableManager {
     /// Maps from remote data file path to its file id.
     pub(crate) remote_data_file_to_file_id: HashMap<String, FileId>,
 
-    /// Iceberg persistence stats for overall snapshot synchronization.
-    pub(crate) persistence_stats_overall: Arc<IcebergPersistenceStats>,
-
-    /// Iceberg persistence stats for data files synchronization.
-    pub(crate) persistence_stats_sync_data_files: Arc<IcebergPersistenceStats>,
-
-    /// Iceberg persistence stats for deletion vectors synchronization.
-    pub(crate) persistence_stats_sync_deletion_vectors: Arc<IcebergPersistenceStats>,
-
-    /// Iceberg persistence stats for file indices synchronization.
-    pub(crate) persistence_stats_sync_file_indices: Arc<IcebergPersistenceStats>,
-
-    /// Iceberg persistence stats for transaction commit.
-    pub(crate) persistence_stats_transaction_commit: Arc<IcebergPersistenceStats>,
+    /// Iceberg persistence stats.
+    pub(crate) persistence_stats: Arc<IcebergPersistenceStats>,
 }
 
 impl IcebergTableManager {
@@ -112,26 +100,7 @@ impl IcebergTableManager {
             persisted_data_files: HashMap::new(),
             persisted_file_indices: HashMap::new(),
             remote_data_file_to_file_id: HashMap::new(),
-            persistence_stats_overall: Arc::new(IcebergPersistenceStats::new(
-                mooncake_table_id.clone(),
-                IcebergPersistenceStage::Overall,
-            )),
-            persistence_stats_sync_data_files: Arc::new(IcebergPersistenceStats::new(
-                mooncake_table_id.clone(),
-                IcebergPersistenceStage::DataFiles,
-            )),
-            persistence_stats_sync_file_indices: Arc::new(IcebergPersistenceStats::new(
-                mooncake_table_id.clone(),
-                IcebergPersistenceStage::FileIndices,
-            )),
-            persistence_stats_sync_deletion_vectors: Arc::new(IcebergPersistenceStats::new(
-                mooncake_table_id.clone(),
-                IcebergPersistenceStage::DeletionVectors,
-            )),
-            persistence_stats_transaction_commit: Arc::new(IcebergPersistenceStats::new(
-                mooncake_table_id,
-                IcebergPersistenceStage::TransactionCommit,
-            )),
+            persistence_stats: Arc::new(IcebergPersistenceStats::new(mooncake_table_id)),
         })
     }
 
@@ -160,26 +129,7 @@ impl IcebergTableManager {
             persisted_data_files: HashMap::new(),
             persisted_file_indices: HashMap::new(),
             remote_data_file_to_file_id: HashMap::new(),
-            persistence_stats_overall: Arc::new(IcebergPersistenceStats::new(
-                mooncake_table_id.clone(),
-                IcebergPersistenceStage::Overall,
-            )),
-            persistence_stats_sync_data_files: Arc::new(IcebergPersistenceStats::new(
-                mooncake_table_id.clone(),
-                IcebergPersistenceStage::DataFiles,
-            )),
-            persistence_stats_sync_file_indices: Arc::new(IcebergPersistenceStats::new(
-                mooncake_table_id.clone(),
-                IcebergPersistenceStage::FileIndices,
-            )),
-            persistence_stats_sync_deletion_vectors: Arc::new(IcebergPersistenceStats::new(
-                mooncake_table_id.clone(),
-                IcebergPersistenceStage::DeletionVectors,
-            )),
-            persistence_stats_transaction_commit: Arc::new(IcebergPersistenceStats::new(
-                mooncake_table_id,
-                IcebergPersistenceStage::TransactionCommit,
-            )),
+            persistence_stats: Arc::new(IcebergPersistenceStats::new(mooncake_table_id)),
         })
     }
 

src/moonlink/src/storage/iceberg/iceberg_table_syncer.rs

@@ -242,7 +242,7 @@ impl IcebergTableManager {
             });
         }
         // Record data files synchronization latency.
-        let _guard = self.persistence_stats_sync_data_files.start();
+        let _guard = self.persistence_stats.sync_data_files.start();
         let mut local_data_files_to_remote = HashMap::with_capacity(new_data_files.len());
         let mut new_remote_data_files = Vec::with_capacity(new_data_files.len());
         let mut new_iceberg_data_files = Vec::with_capacity(new_data_files.len());
@@ -496,7 +496,7 @@ impl IcebergTableManager {
                 evicted_files_to_delete: Vec::new(),
             });
         }
-        let _guard = self.persistence_stats_sync_deletion_vectors.start();
+        let _guard = self.persistence_stats.sync_deletion_vectors.start();
         let mut puffin_deletion_blobs = HashMap::with_capacity(new_deletion_logs.len());
         let mut evicted_files_to_delete = vec![];
         let prepared: Vec<IcebergResult<PreparedDeletionVectorBlob>>;
@@ -618,7 +618,7 @@ impl IcebergTableManager {
             return Ok(HashMap::new());
         }
         // Record file indices synchronization latency.
-        let _guard = self.persistence_stats_sync_file_indices.start();
+        let _guard = self.persistence_stats.sync_file_indices.start();
         let mut local_index_file_to_remote = HashMap::new();
 
         let iceberg_table = self.iceberg_table.as_ref().unwrap();
@@ -726,8 +726,8 @@ impl IcebergTableManager {
         file_params: PersistenceFileParams,
     ) -> Result<PersistenceResult> {
         // Start recording overall snapshot synchronization latency.
-        let persistence_stats_overall = self.persistence_stats_overall.clone();
-        let _guard = persistence_stats_overall.start();
+        let persistence_stats = self.persistence_stats.clone();
+        let _guard = persistence_stats.overall.start();
 
         // Initialize iceberg table on access.
         self.initialize_iceberg_table_for_once().await?;
@@ -819,7 +819,7 @@ impl IcebergTableManager {
 
         let updated_iceberg_table = {
             // Start recording transaction commit latency.
-            let _guard = self.persistence_stats_transaction_commit.start();
+            let _guard = self.persistence_stats.transaction_commit.start();
             // Commit the transaction.
             txn.commit(&*self.catalog).await?
         };