feat(cdc): share a changelog stream for multiple cdc tables

proto/ddl_service.proto

@@ -141,8 +141,10 @@ message DropViewResponse {
 
 enum TableJobType {
   UNSPECIFIED = 0;
-  // table streaming job w/ a shared upstream CDC source job
-  SHARED_CDC_SOURCE = 1;
+  // table streaming jobs excepts the `SHARED_CDC_SOURCE` type
+  GENERAL = 1;
+  // table streaming job sharing a CDC source job
+  SHARED_CDC_SOURCE = 2;
 }
 
 message CreateTableRequest {

proto/plan_common.proto

@@ -83,8 +83,7 @@ message ExternalTableDesc {
   repeated common.ColumnOrder pk = 3;
   string table_name = 4;
   repeated uint32 stream_key = 5;
-  // use string as a work around until https://github.com/madsim-rs/madsim/issues/169 finished
-  string connect_properties = 6;
+  map<string, string> connect_properties = 6;
 }
 
 enum JoinType {

src/common/src/catalog/external_table.rs

@@ -23,7 +23,7 @@ use crate::util::sort_util::ColumnOrder;
 /// Compute node will use this information to connect to the external database and scan the table.
 #[derive(Debug, Clone, Default, PartialEq, Eq, Hash)]
 pub struct CdcTableDesc {
-    /// Id of the table, to find in storage.
+    /// Id of the upstream source in sharing cdc mode
     pub table_id: TableId,
 
     /// The full name of the table in external database, e.g. `database_name.table.name` in MySQL
@@ -62,8 +62,7 @@ impl CdcTableDesc {
             pk: self.pk.iter().map(|v| v.to_protobuf()).collect(),
             table_name: self.external_table_name.clone(),
             stream_key: self.stream_key.iter().map(|k| *k as _).collect(),
-            connect_properties: serde_json::to_string(&self.connect_properties)
-                .expect("failed to serialize connect_properties"),
+            connect_properties: self.connect_properties.clone(),
         }
     }
 

src/common/src/util/column_index_mapping.rs

@@ -335,7 +335,7 @@ impl ColIndexMapping {
             r#type: strategy.r#type,
             dist_key_indices: map(&strategy.dist_key_indices)?,
             output_indices: map(&strategy.output_indices)?,
-            downstream_table_name: None,
+            downstream_table_name: strategy.downstream_table_name.clone(),
         })
     }
 }

src/connector/src/parser/mod.rs

@@ -168,15 +168,22 @@ impl MessageMeta<'_> {
             // Extract the offset from the meta data.
             SourceColumnType::Offset => Datum::Some(self.offset.into()).into(),
             // Extract custom meta data per connector.
-            SourceColumnType::Meta if let SourceMeta::Kafka(kafka_meta) = &self.meta => {
-                assert_eq!(desc.name.as_str(), KAFKA_TIMESTAMP_COLUMN_NAME, "unexpected kafka meta column name");
-                kafka_meta.timestamp.map(|ts| {
-                    risingwave_common::cast::i64_to_timestamptz(ts)
-                        .unwrap()
-                        .to_scalar_value()
-                }).into()
+            SourceColumnType::Meta if let SourceMeta::Kafka(kafka_meta) = self.meta => {
+                assert_eq!(
+                    desc.name.as_str(),
+                    KAFKA_TIMESTAMP_COLUMN_NAME,
+                    "unexpected kafka meta column name"
+                );
+                kafka_meta
+                    .timestamp
+                    .map(|ts| {
+                        risingwave_common::cast::i64_to_timestamptz(ts)
+                            .unwrap()
+                            .to_scalar_value()
+                    })
+                    .into()
             },
-            SourceColumnType::Meta if let SourceMeta::DebeziumCdc(cdc_meta) = &self.meta => {
+            SourceColumnType::Meta if let SourceMeta::DebeziumCdc(cdc_meta) = self.meta => {
                 assert_eq!(desc.name.as_str(), TABLE_NAME_COLUMN_NAME, "unexpected cdc meta column name");
                 Datum::Some(cdc_meta.full_table_name.as_str().into()).into()
             }

src/frontend/src/handler/create_source.rs

@@ -291,7 +291,7 @@ pub(crate) async fn bind_columns_from_source(
     source_schema: &ConnectorSchema,
     with_properties: &HashMap<String, String>,
     create_cdc_source_job: bool,
-) -> Result<(Option<Vec<ColumnCatalog>>, Vec<String>, StreamSourceInfo)> {
+) -> Result<(Option<Vec<ColumnCatalog>>, StreamSourceInfo)> {
     const MESSAGE_NAME_KEY: &str = "message";
     const KEY_MESSAGE_NAME_KEY: &str = "key.message";
     const NAME_STRATEGY_KEY: &str = "schema.registry.name.strategy";
@@ -358,32 +358,19 @@ pub(crate) async fn bind_columns_from_source(
             )
         }
         (Format::Plain, Encode::Json) => {
-            let (resolved_columns, use_schema_registry) = if !create_cdc_source_job {
-                let schema_config = get_json_schema_location(&mut options)?;
-                if schema_config.is_some() && sql_defined_schema {
-                    return Err(RwError::from(ProtocolError(
-                        "User-defined schema is not allowed with schema registry.".to_string(),
-                    )));
-                }
-                if schema_config.is_none() && sql_defined_columns.is_empty() {
-                    return Err(RwError::from(InvalidInputSyntax(
-                        "schema definition is required for ENCODE JSON".to_owned(),
-                    )));
-                }
-                (
-                    extract_json_table_schema(&schema_config, with_properties).await?,
-                    json_schema_infer_use_schema_registry(&schema_config),
-                )
+            let schema_config = get_json_schema_location(&mut options)?;
+            let columns = if create_cdc_source_job {
+                Some(debezium_cdc_source_schema())
             } else {
-                (Some(debezium_cdc_source_schema()), false)
+                extract_json_table_schema(&schema_config, with_properties).await?
             };
+
             (
-                resolved_columns,
-                sql_defined_pk_names,
+                columns,
                 StreamSourceInfo {
                     format: FormatType::Plain as i32,
                     row_encode: EncodeType::Json as i32,
-                    use_schema_registry,
+                    use_schema_registry: json_schema_infer_use_schema_registry(&schema_config),
                     cdc_source_job: create_cdc_source_job,
                     ..Default::default()
                 },
@@ -1132,22 +1119,9 @@ pub async fn handle_create_source(
     // gated the feature with a session variable
     let create_cdc_source_job =
         is_cdc_connector(&with_properties) && session.config().get_cdc_backfill();
-    let (columns_from_resolve_source, pk_names, source_info) = try_bind_columns_from_source(
-        &source_schema,
-        pk_names,
-        &stmt.columns,
-        &with_properties,
-        create_cdc_source_job,
-    )
-    .await?;
-
-    if create_cdc_source_job {
-        // set connector to backfill mode
-        with_properties.insert(CDC_SNAPSHOT_MODE_KEY.into(), CDC_SNAPSHOT_BACKFILL.into());
-        // enable cdc sharing mode, which will capture all tables in the given `database.name`
-        with_properties.insert(CDC_SHARING_MODE_KEY.into(), "true".into());
-    }
 
+    let (columns_from_resolve_source, source_info) =
+        bind_columns_from_source(&source_schema, &with_properties, create_cdc_source_job).await?;
     let columns_from_sql = bind_sql_columns(&stmt.columns)?;
 
     let mut columns = bind_all_columns(
@@ -1165,6 +1139,13 @@ pub async fn handle_create_source(
     )
     .await?;
 
+    if create_cdc_source_job {
+        // set connector to backfill mode
+        with_properties.insert(CDC_SNAPSHOT_MODE_KEY.into(), CDC_SNAPSHOT_BACKFILL.into());
+        // enable cdc sharing mode, which will capture all tables in the given `database.name`
+        with_properties.insert(CDC_SHARING_MODE_KEY.into(), "true".into());
+    }
+
     check_and_add_timestamp_column(&with_properties, &mut columns);
 
     let mut col_id_gen = ColumnIdGenerator::new_initial();
@@ -1238,7 +1219,7 @@ pub async fn handle_create_source(
                 Some(Rc::new(SourceCatalog::from(&source))),
                 columns.clone(),
                 row_id_index,
-                true,
+                false,
                 false,
                 context.into(),
             )?;
@@ -1247,7 +1228,11 @@ pub async fn handle_create_source(
             let stream_plan = source_node.to_stream(&mut ToStreamContext::new(false))?;
             let mut graph = build_graph(stream_plan);
             // set parallelism to 1
-            graph.parallelism = Some(Parallelism { parallelism: 1 });
+            // graph.parallelism = Some(Parallelism { parallelism: 1 });
+            graph.parallelism = session
+                .config()
+                .get_streaming_parallelism()
+                .map(|parallelism| Parallelism { parallelism });
             graph
         };
         catalog_writer

src/frontend/src/handler/create_table.rs

@@ -20,12 +20,13 @@ use fixedbitset::FixedBitSet;
 use itertools::Itertools;
 use pgwire::pg_response::{PgResponse, StatementType};
 use risingwave_common::catalog::{
-    CdcTableDesc, ColumnCatalog, ColumnDesc, TableId, TableVersionId, INITIAL_SOURCE_VERSION_ID,
-    INITIAL_TABLE_VERSION_ID, USER_COLUMN_ID_OFFSET,
+    CdcTableDesc, ColumnCatalog, ColumnDesc, TableId, TableVersionId, DEFAULT_SCHEMA_NAME,
+    INITIAL_SOURCE_VERSION_ID, INITIAL_TABLE_VERSION_ID, USER_COLUMN_ID_OFFSET,
 };
 use risingwave_common::error::{ErrorCode, Result, RwError};
 use risingwave_common::util::sort_util::{ColumnOrder, OrderType};
 use risingwave_common::util::value_encoding::DatumToProtoExt;
+use risingwave_connector::source;
 use risingwave_connector::source::cdc::{CDC_SNAPSHOT_BACKFILL, CDC_SNAPSHOT_MODE_KEY};
 use risingwave_connector::source::external::{
     CdcTableType, DATABASE_NAME_KEY, SCHEMA_NAME_KEY, TABLE_NAME_KEY,
@@ -43,6 +44,7 @@ use risingwave_sqlparser::ast::{
 
 use super::RwPgResponse;
 use crate::binder::{bind_data_type, bind_struct_field, Clause};
+use crate::catalog::root_catalog::SchemaPath;
 use crate::catalog::source_catalog::SourceCatalog;
 use crate::catalog::table_catalog::TableVersion;
 use crate::catalog::{check_valid_column_name, CatalogError, ColumnId};
@@ -463,9 +465,10 @@ pub(crate) async fn gen_create_table_plan_with_source(
 
     ensure_table_constraints_supported(&constraints)?;
 
-    let (columns_from_resolve_source, pk_names, mut source_info) =
-        try_bind_columns_from_source(&source_schema, pk_names, &column_defs, &properties, false)
-            .await?;
+    let sql_pk_names = bind_sql_pk_names(&column_defs, &constraints)?;
+
+    let (columns_from_resolve_source, mut source_info) =
+        bind_columns_from_source(&source_schema, &properties, false).await?;
     let columns_from_sql = bind_sql_columns(&column_defs)?;
 
     let mut columns = bind_all_columns(
@@ -545,7 +548,7 @@ pub(crate) async fn gen_create_table_plan_with_source(
             pk: table_pk,
             columns: columns.iter().map(|c| c.column_desc.clone()).collect(),
             stream_key: pk_column_indices,
-            value_indices: (0..columns.len()).collect_vec(), /* FIXME: maybe we can remove `_rw_offset` from TableDesc */
+            value_indices: (0..columns.len()).collect_vec(),
             connect_properties: Default::default(),
         };
 
@@ -771,7 +774,7 @@ fn gen_create_table_plan_for_cdc_source(
     column_defs: Vec<ColumnDef>,
     constraints: Vec<TableConstraint>,
     mut col_id_gen: ColumnIdGenerator,
-) -> Result<(PlanRef, Option<PbSource>, PbTable)> {
+) -> Result<(PlanRef, PbTable)> {
     let session = context.session_ctx().clone();
     let db_name = session.database();
     let (schema_name, name) = Binder::resolve_schema_qualified_name(db_name, table_name)?;
@@ -781,15 +784,27 @@ fn gen_create_table_plan_for_cdc_source(
     // cdc table cannot be append-only
     let append_only = false;
     let source_name = source_name.real_value();
-    let source = session.get_source_by_name(schema_name, &source_name)?;
+
+    let source = {
+        let catalog_reader = session.env().catalog_reader().read_guard();
+        let schema_name = schema_name
+            .clone()
+            .unwrap_or(DEFAULT_SCHEMA_NAME.to_string());
+        let (source, _) = catalog_reader.get_source_by_name(
+            db_name,
+            SchemaPath::Name(schema_name.as_str()),
+            source_name.as_str(),
+        )?;
+        source.clone()
+    };
 
     let mut columns = bind_sql_columns(&column_defs)?;
 
     for c in &mut columns {
         c.column_desc.column_id = col_id_gen.generate(c.name())
     }
 
-    let pk_names = bind_pk_names(&column_defs, &constraints)?;
+    let pk_names = bind_sql_pk_names(&column_defs, &constraints)?;
     let (columns, pk_column_ids, _) = bind_pk_on_relation(columns, pk_names)?;
 
     let definition = context.normalized_sql().to_owned();
@@ -855,18 +870,19 @@ fn gen_create_table_plan_for_cdc_source(
 
     let mut table = materialize.table().to_prost(schema_id, database_id);
     table.owner = session.user_id();
-    Ok((materialize.into(), None, table))
+    Ok((materialize.into(), table))
 }
 
 fn derive_connect_properties(
     source: &SourceCatalog,
     external_table_name: String,
 ) -> Result<BTreeMap<String, String>> {
+    use source::cdc::{MYSQL_CDC_CONNECTOR, POSTGRES_CDC_CONNECTOR};
     // we should remove the prefix from `full_table_name`
     let mut connect_properties = source.properties.clone();
     if let Some(connector) = source.properties.get(UPSTREAM_SOURCE_KEY) {
         let table_name = match connector.as_str() {
-            "mysql-cdc" => {
+            MYSQL_CDC_CONNECTOR => {
                 let db_name = connect_properties.get(DATABASE_NAME_KEY).ok_or_else(|| {
                     anyhow!("{} not found in source properties", DATABASE_NAME_KEY)
                 })?;
@@ -876,7 +892,7 @@ fn derive_connect_properties(
                     .strip_prefix(prefix.as_str())
                     .ok_or_else(|| anyhow!("external table name must contain database prefix"))?
             }
-            "postgres-cdc" => {
+            POSTGRES_CDC_CONNECTOR => {
                 let schema_name = connect_properties
                     .get(SCHEMA_NAME_KEY)
                     .ok_or_else(|| anyhow!("{} not found in source properties", SCHEMA_NAME_KEY))?;
@@ -935,7 +951,6 @@ pub async fn handle_create_table(
         let context = OptimizerContext::from_handler_args(handler_args);
         let source_schema = check_create_table_with_source(context.with_options(), source_schema)?;
         let col_id_gen = ColumnIdGenerator::new_initial();
-        let properties = context.with_options().inner().clone().into_iter().collect();
 
         let ((plan, source, table), job_type) = match (source_schema, cdc_table_info.as_ref()) {
             (Some(source_schema), None) => (
@@ -950,7 +965,7 @@ pub async fn handle_create_table(
                     append_only,
                 )
                 .await?,
-                TableJobType::Unspecified,
+                TableJobType::General,
             ),
             (None, None) => (
                 gen_create_table_plan(
@@ -962,21 +977,22 @@ pub async fn handle_create_table(
                     source_watermarks,
                     append_only,
                 )?,
-                TableJobType::Unspecified,
+                TableJobType::General,
             ),
 
-            (None, Some(cdc_table)) => (
-                gen_create_table_plan_for_cdc_source(
+            (None, Some(cdc_table)) => {
+                let (plan, table) = gen_create_table_plan_for_cdc_source(
                     context.into(),
                     cdc_table.source_name.clone(),
                     table_name.clone(),
                     cdc_table.external_table_name.clone(),
                     column_defs,
                     constraints,
                     col_id_gen,
-                )?,
-                TableJobType::SharedCdcSource,
-            ),
+                )?;
+
+                ((plan, None, table), TableJobType::SharedCdcSource)
+            }
             (Some(_), Some(_)) => return Err(ErrorCode::NotSupported(
                 "Data format and encoding format doesn't apply to table created from a CDC source"
                     .into(),
@@ -985,16 +1001,10 @@ pub async fn handle_create_table(
             .into()),
         };
         let mut graph = build_graph(plan);
-        graph.parallelism = if cdc_table_info.is_some()
-            || CdcTableType::from_properties(&properties).can_backfill()
-        {
-            Some(Parallelism { parallelism: 1 })
-        } else {
-            session
-                .config()
-                .get_streaming_parallelism()
-                .map(|parallelism| Parallelism { parallelism })
-        };
+        graph.parallelism = session
+            .config()
+            .get_streaming_parallelism()
+            .map(|parallelism| Parallelism { parallelism });
         (graph, source, table, job_type)
     };
 

src/frontend/src/optimizer/plan_node/generic/scan.rs

@@ -432,6 +432,13 @@ impl Scan {
         matches!(self.scan_table_type, ScanTableType::CdcTable)
     }
 
+    pub fn append_only(&self) -> bool {
+        if self.is_cdc_table() {
+            return false;
+        }
+        self.table_desc.append_only
+    }
+
     /// Get the descs of the output columns.
     pub fn column_descs(&self) -> Vec<ColumnDesc> {
         self.output_col_idx

src/frontend/src/optimizer/plan_node/logical_scan.rs

@@ -447,8 +447,7 @@ impl PredicatePushdown for LogicalScan {
             .conjunctions
             .extract_if(|expr| expr.count_nows() > 0 || HasCorrelated {}.visit_expr(expr))
             .collect();
-
-        let predicate = predicate.rewrite_expr(&mut ColIndexMapping::with_target_size(
+        let predicate = predicate.rewrite_expr(&mut ColIndexMapping::new(
             self.output_col_idx().iter().map(|i| Some(*i)).collect(),
             self.table_desc().columns.len(),
         ));