risingwave_connector/source/cdc/
postgres.rs

1// Copyright 2026 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::collections::BTreeMap;
16
17use risingwave_common::bail;
18use risingwave_pb::secret::PbSecretRef;
19
20use crate::error::ConnectorResult;
21use crate::source::cdc::{
22    CdcSourceDdlValidator, CdcSourceRuntimeFlags, CdcSourceTypeTrait, Postgres,
23    check_alter_with_options_inner, check_common_with_options,
24};
25
26// ------- Runtime flags -------
27
28impl CdcSourceRuntimeFlags for Postgres {
29    const CAN_BACKFILL: bool = true;
30    const CAN_BACKFILL_IN_PARALLEL: bool = true;
31    const CAN_USE_TRANSACTION_METADATA: bool = true;
32    const NEEDS_OFFSET_COMMIT_ON_CHECKPOINT: bool = true;
33}
34
35// ------- DDL validation -------
36
37impl CdcSourceDdlValidator for Postgres {
38    fn check_and_fill_create_with_options(
39        props: &mut BTreeMap<String, String>,
40    ) -> ConnectorResult<()> {
41        check_common_with_options(props)?;
42        match props.get("slot.name") {
43            None => {
44                // Build a random slot name with UUID
45                // e.g. "rw_cdc_f9a3567e6dd54bf5900444c8b1c03815"
46                let uuid = uuid::Uuid::new_v4();
47                props.insert("slot.name".into(), format!("rw_cdc_{}", uuid.simple()));
48            }
49            Some(slot_name) => {
50                // please refer to
51                // - https://github.com/debezium/debezium/blob/97956ce25b7612e3413d363658661896b7d2e0a2/debezium-connector-postgres/src/main/java/io/debezium/connector/postgresql/PostgresConnectorConfig.java#L1179
52                // - https://doxygen.postgresql.org/slot_8c.html#afac399f07320b9adfd2c599cf822aaa3
53                if !slot_name
54                    .chars()
55                    .all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '_')
56                    || slot_name.len() > 63
57                {
58                    bail!(
59                        "Invalid replication slot name: {:?}. Valid replication slot name must contain only digits, lowercase characters and underscores with length <= 63",
60                        slot_name
61                    );
62                }
63            }
64        }
65
66        if !props.contains_key("schema.name") {
67            // Default schema name is "public"
68            props.insert("schema.name".into(), "public".into());
69        }
70        if !props.contains_key("publication.name") {
71            // Build a random publication name with UUID to avoid conflicts between sources
72            // e.g. "rw_publication_f9a3567e6dd54bf5900444c8b1c03815"
73            let uuid = uuid::Uuid::new_v4();
74            props.insert(
75                "publication.name".into(),
76                format!("rw_publication_{}", uuid.simple()),
77            );
78        }
79        if !props.contains_key("publication.create.enable") {
80            // Default auto create publication if doesn't exist
81            props.insert("publication.create.enable".into(), "true".into());
82        }
83        Ok(())
84    }
85
86    fn check_alter_with_options(
87        props: &BTreeMap<String, String>,
88        secret_refs: &BTreeMap<String, PbSecretRef>,
89    ) -> ConnectorResult<()> {
90        check_common_with_options(props)?;
91        check_alter_with_options_inner(
92            Self::CDC_CONNECTOR_NAME,
93            props,
94            secret_refs,
95            &[
96                "cdc.source.wait.streaming.start.timeout",
97                "debezium.max.queue.size",
98                "debezium.queue.memory.ratio",
99                "debezium.heartbeat.interval.ms",
100                "hostname",
101                "port",
102                "password",
103            ],
104        )
105    }
106}
risingwave_connector/source/cdc/postgres.rs

risingwave_connector/source/cdc/
postgres.rs
