feat(storage): support different snapshot for streaming jobs

proto/hummock.proto

@@ -149,6 +149,16 @@ message TableChangeLog {
   repeated EpochNewChangeLog change_logs = 1;
 }
 
+message StateTableInfo {
+  uint64 committed_epoch = 1;
+  uint64 safe_epoch = 2;
+}
+
+message StateTableInfoDelta {
+  uint64 committed_epoch = 1;
+  uint64 safe_epoch = 2;
+}
+
 message HummockVersion {
   message Levels {
     repeated Level levels = 1;
@@ -166,6 +176,7 @@ message HummockVersion {
   uint64 safe_epoch = 4;
   map<uint32, TableWatermarks> table_watermarks = 5;
   map<uint32, TableChangeLog> table_change_logs = 6;
+  map<uint32, StateTableInfo> state_table_info = 7;
 }
 
 message HummockVersionDelta {
@@ -191,6 +202,7 @@ message HummockVersionDelta {
     uint64 truncate_epoch = 2;
   }
   map<uint32, ChangeLogDelta> change_log_delta = 10;
+  map<uint32, StateTableInfoDelta> state_table_info_delta = 11;
 }
 
 message HummockVersionDeltas {

src/frontend/src/catalog/system_catalog/rw_catalog/rw_hummock_version.rs

@@ -204,3 +204,27 @@ async fn read_hummock_table_watermarks(
         })
         .collect())
 }
+
+#[derive(Fields)]
+struct RwHummockSnapshot {
+    #[primary_key]
+    table_id: i32,
+    safe_epoch: i64,
+    committed_epoch: i64,
+}
+
+#[system_catalog(table, "rw_catalog.rw_hummock_snapshot")]
+async fn read_hummock_snapshot_groups(
+    reader: &SysCatalogReaderImpl,
+) -> Result<Vec<RwHummockSnapshot>> {
+    let version = reader.meta_client.get_hummock_current_version().await?;
+    Ok(version
+        .state_table_info
+        .iter()
+        .map(|(table_id, info)| RwHummockSnapshot {
+            table_id: table_id.table_id as _,
+            committed_epoch: info.committed_epoch as _,
+            safe_epoch: info.safe_epoch as _,
+        })
+        .collect())
+}

src/meta/node/src/server.rs

@@ -587,6 +587,11 @@ pub async fn start_service_as_election_leader(
         .unwrap(),
     );
 
+    hummock_manager
+        .may_fill_backward_state_table_info()
+        .await
+        .unwrap();
+
     // Initialize services.
     let backup_manager = BackupManager::new(
         env.clone(),

src/meta/src/hummock/manager/commit_epoch.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::collections::{BTreeMap, HashMap};
+use std::collections::{BTreeMap, HashMap, HashSet};
 
 use itertools::Itertools;
 use risingwave_common::catalog::TableId;
@@ -27,7 +27,9 @@ use risingwave_hummock_sdk::{
 use risingwave_pb::hummock::compact_task::{self};
 use risingwave_pb::hummock::group_delta::DeltaType;
 use risingwave_pb::hummock::hummock_version_delta::ChangeLogDelta;
-use risingwave_pb::hummock::{GroupDelta, GroupMetaChange, HummockSnapshot, IntraLevelDelta};
+use risingwave_pb::hummock::{
+    GroupDelta, GroupMetaChange, HummockSnapshot, IntraLevelDelta, StateTableInfoDelta,
+};
 
 use crate::hummock::error::{Error, Result};
 use crate::hummock::manager::transaction::{
@@ -140,8 +142,11 @@ impl HummockManager {
             .latest_version()
             .build_compaction_group_info();
 
+        let mut new_table_ids = None;
+
         // Add new table
         if let Some(new_fragment_table_info) = new_table_fragment_info {
+            let new_table_ids = new_table_ids.insert(HashSet::new());
             if !new_fragment_table_info.internal_table_ids.is_empty() {
                 if let Some(levels) = new_version_delta
                     .latest_version()
@@ -178,6 +183,7 @@ impl HummockManager {
                 for table_id in &new_fragment_table_info.internal_table_ids {
                     table_compaction_group_mapping
                         .insert(*table_id, StaticCompactionGroupId::StateDefault as u64);
+                    new_table_ids.insert(*table_id);
                 }
             }
 
@@ -208,6 +214,7 @@ impl HummockManager {
                 });
                 let _ = table_compaction_group_mapping
                     .insert(table_id, StaticCompactionGroupId::MaterializedView as u64);
+                new_table_ids.insert(table_id);
             }
         }
 
@@ -329,6 +336,28 @@ impl HummockManager {
             group_deltas.push(group_delta);
         }
 
+        // update state table info
+        new_version_delta.with_latest_version(|version, delta| {
+            for table_id in new_table_ids
+                .into_iter()
+                .flat_map(|ids| ids.into_iter().map(|table_id| table_id.table_id))
+                .chain(
+                    version
+                        .levels
+                        .values()
+                        .flat_map(|group| group.member_table_ids.iter().cloned()),
+                )
+            {
+                delta.state_table_info_delta.insert(
+                    TableId::new(table_id),
+                    StateTableInfoDelta {
+                        committed_epoch: epoch,
+                        safe_epoch: version.safe_epoch,
+                    },
+                );
+            }
+        });
+
         new_version_delta.pre_apply();
 
         // Apply stats changes.

src/meta/src/hummock/manager/compaction.rs

@@ -47,7 +47,7 @@ use risingwave_pb::hummock::subscribe_compaction_event_response::{
 use risingwave_pb::hummock::{
     compact_task, CompactStatus as PbCompactStatus, CompactTask, CompactTaskAssignment,
     CompactionConfig, GroupDelta, InputLevel, IntraLevelDelta, Level, SstableInfo,
-    SubscribeCompactionEventRequest, TableOption, TableSchema,
+    StateTableInfoDelta, SubscribeCompactionEventRequest, TableOption, TableSchema,
 };
 use rw_futures_util::pending_on_none;
 use thiserror_ext::AsReport;
@@ -171,6 +171,22 @@ impl<'a> HummockVersionTransaction<'a> {
             version_delta.latest_version().safe_epoch,
             compact_task.watermark,
         );
+        if version_delta.latest_version().safe_epoch < version_delta.safe_epoch {
+            version_delta.state_table_info_delta = version_delta
+                .latest_version()
+                .state_table_info
+                .iter()
+                .map(|(table_id, info)| {
+                    (
+                        *table_id,
+                        StateTableInfoDelta {
+                            committed_epoch: info.committed_epoch,
+                            safe_epoch: version_delta.safe_epoch,
+                        },
+                    )
+                })
+                .collect();
+        }
         version_delta.pre_apply();
     }
 }

src/meta/src/hummock/manager/compaction_group_manager.rs

@@ -33,9 +33,8 @@ use risingwave_pb::hummock::subscribe_compaction_event_request::ReportTask;
 use risingwave_pb::hummock::write_limits::WriteLimit;
 use risingwave_pb::hummock::{
     compact_task, CompactionConfig, CompactionGroupInfo, CompatibilityVersion, GroupConstruct,
-    GroupDelta, GroupDestroy, GroupMetaChange,
+    GroupDelta, GroupDestroy, GroupMetaChange, StateTableInfoDelta,
 };
-use thiserror_ext::AsReport;
 use tokio::sync::OnceCell;
 
 use crate::hummock::compaction::compaction_config::{
@@ -120,7 +119,7 @@ impl HummockManager {
                 CompactionGroupId::from(StaticCompactionGroupId::StateDefault),
             ));
         }
-        self.register_table_ids(&pairs).await?;
+        self.register_table_ids_for_test(&pairs).await?;
         Ok(pairs.iter().map(|(table_id, ..)| *table_id).collect_vec())
     }
 
@@ -130,13 +129,14 @@ impl HummockManager {
         &self,
         table_fragments: &[crate::model::TableFragments],
     ) {
-        self.unregister_table_ids_fail_fast(
+        self.unregister_table_ids(
             &table_fragments
                 .iter()
                 .flat_map(|t| t.all_table_ids())
                 .collect_vec(),
         )
-        .await;
+        .await
+        .unwrap();
     }
 
     /// Unregisters stale members and groups
@@ -155,7 +155,10 @@ impl HummockManager {
     }
 
     /// The implementation acquires `versioning` lock.
-    pub async fn register_table_ids(
+    ///
+    /// The method name is temporarily added with a `_for_test` prefix to mark
+    /// that it's currently only used in test.
+    pub async fn register_table_ids_for_test(
         &self,
         pairs: &[(StateTableId, CompactionGroupId)],
     ) -> Result<()> {
@@ -185,6 +188,10 @@ impl HummockManager {
             &self.metrics,
         );
         let mut new_version_delta = version.new_delta();
+        let (committed_epoch, safe_epoch) = {
+            let version = new_version_delta.latest_version();
+            (version.max_committed_epoch, version.safe_epoch)
+        };
 
         for (table_id, raw_group_id) in pairs {
             let mut group_id = *raw_group_id;
@@ -235,6 +242,16 @@ impl HummockManager {
                     ..Default::default()
                 })),
             });
+            assert!(new_version_delta
+                .state_table_info_delta
+                .insert(
+                    TableId::new(*table_id),
+                    StateTableInfoDelta {
+                        committed_epoch,
+                        safe_epoch,
+                    }
+                )
+                .is_none());
         }
         new_version_delta.pre_apply();
         commit_multi_var!(self.meta_store_ref(), version)?;
@@ -290,7 +307,7 @@ impl HummockManager {
                 );
             new_version_delta
                 .removed_table_ids
-                .push(TableId::new(*table_id));
+                .insert(TableId::new(*table_id));
         }
 
         let groups_to_remove = modified_groups
@@ -337,15 +354,6 @@ impl HummockManager {
         Ok(())
     }
 
-    /// The implementation acquires `versioning` lock and `compaction_group_manager` lock.
-    pub async fn unregister_table_ids_fail_fast(&self, table_ids: &[StateTableId]) {
-        self.unregister_table_ids(table_ids)
-            .await
-            .unwrap_or_else(|e| {
-                panic!("unregister table ids fail: {table_ids:?} {}", e.as_report())
-            });
-    }
-
     pub async fn update_compaction_config(
         &self,
         compaction_group_ids: &[CompactionGroupId],

src/meta/src/hummock/manager/mod.rs

@@ -521,7 +521,7 @@ impl HummockManager {
                 .init_compaction_config_for_replay(group.id, group_config)
                 .await
                 .unwrap();
-            self.register_table_ids(&pairs).await?;
+            self.register_table_ids_for_test(&pairs).await?;
             tracing::info!("Registered table ids {:?}", pairs);
         }