add ddl type

proto/catalog.proto

@@ -39,6 +39,15 @@ enum StreamJobStatus {
   STREAM_JOB_STATUS_CREATED = 2;
 }
 
+enum DdlType {
+  DDL_TYPE_UNSPECIFIED = 0;
+  DDL_TYPE_TABLE = 1;
+  DDL_TYPE_SINK = 2;
+  DDL_TYPE_INDEX = 3;
+  DDL_TYPE_SOURCE = 4;
+  DDL_TYPE_MATERIALIZED_VIEW = 5;
+}
+
 // How the stream job was created will determine
 // whether they are persisted.
 enum CreateType {

src/meta/src/barrier/command.rs

@@ -21,6 +21,7 @@ use risingwave_common::catalog::TableId;
 use risingwave_common::hash::ActorMapping;
 use risingwave_connector::source::SplitImpl;
 use risingwave_hummock_sdk::HummockEpoch;
+use risingwave_pb::catalog::DdlType;
 use risingwave_pb::meta::PausedReason;
 use risingwave_pb::source::{ConnectorSplit, ConnectorSplits};
 use risingwave_pb::stream_plan::barrier::{BarrierKind, Mutation};
@@ -117,6 +118,7 @@ pub enum Command {
         dispatchers: HashMap<ActorId, Vec<Dispatcher>>,
         init_split_assignment: SplitAssignment,
         definition: String,
+        ddl_type: DdlType,
     },
     /// `CancelStreamingJob` command generates a `Stop` barrier including the actors of the given
     /// table fragment.

src/meta/src/manager/streaming_job.rs

@@ -16,7 +16,7 @@ use std::collections::HashMap;
 
 use risingwave_common::catalog::TableVersionId;
 use risingwave_common::util::epoch::Epoch;
-use risingwave_pb::catalog::{CreateType, Index, PbSource, Sink, Table};
+use risingwave_pb::catalog::{CreateType, DdlType, Index, PbSource, Sink, Table};
 use risingwave_pb::ddl_service::TableJobType;
 
 use crate::model::FragmentId;
@@ -32,6 +32,18 @@ pub enum StreamingJob {
     Source(PbSource),
 }
 
+impl From<&StreamingJob> for DdlType {
+    fn from(job: &StreamingJob) -> Self {
+        match job {
+            StreamingJob::MaterializedView(_) => DdlType::MaterializedView,
+            StreamingJob::Sink(_) => DdlType::Sink,
+            StreamingJob::Table(_, _, _) => DdlType::Table,
+            StreamingJob::Index(_, _) => DdlType::Index,
+            StreamingJob::Source(_) => DdlType::Source,
+        }
+    }
+}
+
 impl StreamingJob {
     pub fn mark_created(&mut self) {
         let created_at_epoch = Some(Epoch::now().0);

src/meta/src/rpc/ddl_controller.rs

@@ -747,6 +747,7 @@ impl DdlController {
             definition: stream_job.definition(),
             mv_table_id: stream_job.mv_table(),
             create_type: stream_job.create_type(),
+            ddl_type: stream_job.into(),
         };
 
         // 4. Mark creating tables, including internal tables and the table of the stream job.

src/meta/src/stream/stream_manager.rs

@@ -18,7 +18,7 @@ use std::sync::Arc;
 use futures::future::{join_all, try_join_all, BoxFuture};
 use itertools::Itertools;
 use risingwave_common::catalog::TableId;
-use risingwave_pb::catalog::{CreateType, Table};
+use risingwave_pb::catalog::{CreateType, DdlType, Table};
 use risingwave_pb::stream_plan::update_mutation::MergeUpdate;
 use risingwave_pb::stream_plan::Dispatcher;
 use risingwave_pb::stream_service::{
@@ -69,6 +69,8 @@ pub struct CreateStreamingJobContext {
     pub mv_table_id: Option<u32>,
 
     pub create_type: CreateType,
+
+    pub ddl_type: DdlType,
 }
 
 impl CreateStreamingJobContext {
@@ -440,6 +442,7 @@ impl GlobalStreamManager {
             mv_table_id,
             internal_tables,
             create_type,
+            ddl_type,
         }: CreateStreamingJobContext,
     ) -> MetaResult<()> {
         // Register to compaction group beforehand.
@@ -483,6 +486,7 @@ impl GlobalStreamManager {
                 dispatchers,
                 init_split_assignment,
                 definition: definition.to_string(),
+                ddl_type,
             })
             .await
         {