feat(meta): Add creating_status field for stream jobs

proto/catalog.proto

@@ -191,6 +191,13 @@ message Table {
     INTERNAL = 4;
   }
 
+  enum TableStatus {
+    // Prefixed by `TABLE_STATUS` due to protobuf namespacing rules.
+    TABLE_STATUS_UNSPECIFIED = 0;
+    CREATING = 1;
+    CREATED = 2;
+  }
+
   message TableVersion {
     // The version number, which will be 0 by default and be increased by 1 for
     // each schema change in the frontend.
@@ -250,6 +257,9 @@ message Table {
   // In older versions we can just initialize without it.
   bool cleaned_by_watermark = 30;
 
+  // Used to filter created / creating tables in meta.
+  TableStatus create_status = 31;
+
   // Per-table catalog version, used by schema change. `None` for internal tables and tests.
   // Not to be confused with the global catalog version for notification service.
   TableVersion version = 100;

src/frontend/src/catalog/table_catalog.rs

@@ -23,7 +23,9 @@ use risingwave_common::constants::hummock::TABLE_OPTION_DUMMY_RETENTION_SECOND;
 use risingwave_common::error::{ErrorCode, RwError};
 use risingwave_common::util::epoch::Epoch;
 use risingwave_common::util::sort_util::ColumnOrder;
-use risingwave_pb::catalog::table::{OptionalAssociatedSourceId, PbTableType, PbTableVersion};
+use risingwave_pb::catalog::table::{
+    OptionalAssociatedSourceId, PbTableStatus, PbTableType, PbTableVersion,
+};
 use risingwave_pb::catalog::PbTable;
 use risingwave_pb::plan_common::column_desc::GeneratedOrDefaultColumn;
 use risingwave_pb::plan_common::DefaultColumnDesc;
@@ -401,6 +403,7 @@ impl TableCatalog {
             initialized_at_epoch: self.initialized_at_epoch.map(|epoch| epoch.0),
             created_at_epoch: self.created_at_epoch.map(|epoch| epoch.0),
             cleaned_by_watermark: self.cleaned_by_watermark,
+            create_status: PbTableStatus::Creating.into(),
         }
     }
 
@@ -542,6 +545,7 @@ mod tests {
     use risingwave_common::test_prelude::*;
     use risingwave_common::types::*;
     use risingwave_common::util::sort_util::OrderType;
+    use risingwave_pb::catalog::table::PbTableStatus;
     use risingwave_pb::catalog::PbTable;
     use risingwave_pb::plan_common::{PbColumnCatalog, PbColumnDesc};
 
@@ -605,6 +609,7 @@ mod tests {
             cardinality: None,
             created_at_epoch: None,
             cleaned_by_watermark: false,
+            create_status: PbTableStatus::Creating.into(),
         }
         .into();
 

src/storage/src/filter_key_extractor.rs

@@ -447,7 +447,7 @@ mod tests {
     use risingwave_common::util::row_serde::OrderedRowSerde;
     use risingwave_common::util::sort_util::OrderType;
     use risingwave_hummock_sdk::key::TABLE_PREFIX_LEN;
-    use risingwave_pb::catalog::table::TableType;
+    use risingwave_pb::catalog::table::{PbTableStatus, TableType};
     use risingwave_pb::catalog::PbTable;
     use risingwave_pb::common::{PbColumnOrder, PbDirection, PbNullsAre, PbOrderType};
     use risingwave_pb::plan_common::PbColumnCatalog;
@@ -549,6 +549,7 @@ mod tests {
             cardinality: None,
             created_at_epoch: None,
             cleaned_by_watermark: false,
+            create_status: PbTableStatus::Creating.into(),
         }
     }
 

src/tests/compaction_test/src/delete_range_runner.rs

@@ -36,6 +36,7 @@ use risingwave_meta::hummock::test_utils::setup_compute_env_with_config;
 use risingwave_meta::hummock::MockHummockMetaClient;
 use risingwave_object_store::object::object_metrics::ObjectStoreMetrics;
 use risingwave_object_store::object::parse_remote_object_store;
+use risingwave_pb::catalog::table::PbTableStatus;
 use risingwave_pb::catalog::PbTable;
 use risingwave_pb::hummock::{CompactionConfig, CompactionGroupInfo};
 use risingwave_pb::meta::SystemParams;
@@ -150,6 +151,7 @@ async fn compaction_test(
         cardinality: None,
         created_at_epoch: None,
         cleaned_by_watermark: false,
+        create_status: PbTableStatus::Creating.into(),
     };
     let mut delete_range_table = delete_key_table.clone();
     delete_range_table.id = 2;