add more log

src/meta/src/barrier/recovery.rs

@@ -87,7 +87,10 @@ impl GlobalBarrierManagerContext {
                     .iter()
                     .map(|t| t.table_id())
                     .collect();
-                debug!("clean dirty table fragments: {:?}", to_drop_streaming_ids);
+                debug!(
+                    ?stream_job_ids,
+                    "clean dirty table fragments: {:?}", to_drop_streaming_ids
+                );
 
                 let _unregister_table_ids = mgr
                     .fragment_manager

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

@@ -34,7 +34,7 @@ use risingwave_pb::hummock::{
     PbGroupDestroy, PbStateTableInfoDelta,
 };
 use tokio::sync::OnceCell;
-use tracing::warn;
+use tracing::error;
 
 use crate::hummock::compaction::compaction_config::{
     validate_compaction_config, CompactionConfigBuilder,
@@ -181,9 +181,11 @@ impl HummockManager {
                     !valid_ids.contains(table_id)
                 })
                 .collect_vec();
-            if !remaining_valid_ids.is_empty() {
-                warn!(?remaining_valid_ids, "unregistered valid table ids");
-            }
+            error!(
+                ?valid_ids,
+                ?remaining_valid_ids,
+                "un-purged valid table ids"
+            );
             to_unregister
         };
         // As we have released versioning lock, the version that `to_unregister` is calculated from

src/meta/src/manager/catalog/fragment.rs

@@ -40,6 +40,7 @@ use risingwave_pb::stream_plan::{
 };
 use risingwave_pb::stream_service::BuildActorInfo;
 use tokio::sync::{RwLock, RwLockReadGuard};
+use tracing::error;
 
 use crate::barrier::Reschedule;
 use crate::manager::cluster::WorkerId;
@@ -351,12 +352,22 @@ impl FragmentManager {
         }
 
         let mut table_fragments = BTreeMapTransaction::new(map);
+        let state_table_ids = table_fragment
+            .fragments()
+            .flat_map(|fragment| fragment.state_table_ids.iter())
+            .cloned()
+            .collect_vec();
         table_fragments.insert(table_id, table_fragment);
         let mut trx = Transaction::default();
 
         let next_revision = current_revision.next();
         next_revision.store(&mut trx);
         commit_meta_with_trx!(self, trx, table_fragments)?;
+        error!(
+            table_id = table_id.table_id,
+            ?state_table_ids,
+            "register new table fragments"
+        );
         guard.table_revision = next_revision;
         Ok(())
     }