risingwave_connector/source/cdc/
oracle.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 anyhow::anyhow;
18use risingwave_common::bail;
19use risingwave_pb::secret::PbSecretRef;
20
21use crate::error::ConnectorResult;
22use crate::source::cdc::{
23    CdcSourceDdlValidator, CdcSourceRuntimeFlags, CdcSourceTypeTrait, Oracle,
24    check_alter_with_options_inner, check_common_with_options,
25};
26
27const HEARTBEAT_TABLE_AUTO_INITIALIZE_KEY: &str = "heartbeat.table.auto.initialize";
28
29// ------- Runtime flags -------
30
31impl CdcSourceRuntimeFlags for Oracle {}
32
33// ------- DDL validation -------
34
35impl CdcSourceDdlValidator for Oracle {
36    fn check_and_fill_create_with_options(
37        props: &mut BTreeMap<String, String>,
38    ) -> ConnectorResult<()> {
39        check_common_with_options(props)?;
40        if let Some(value) = props.get(HEARTBEAT_TABLE_AUTO_INITIALIZE_KEY)
41            && value.parse::<bool>().map_err(|_| {
42                anyhow!("'{HEARTBEAT_TABLE_AUTO_INITIALIZE_KEY}' must be 'true' or 'false'")
43            })?
44            && !props
45                .get("heartbeat.table.name")
46                .is_some_and(|value| !value.trim().is_empty())
47        {
48            bail!("'{HEARTBEAT_TABLE_AUTO_INITIALIZE_KEY}' requires 'heartbeat.table.name'");
49        }
50        Ok(())
51    }
52
53    fn check_alter_with_options(
54        props: &BTreeMap<String, String>,
55        secret_refs: &BTreeMap<String, PbSecretRef>,
56    ) -> ConnectorResult<()> {
57        check_common_with_options(props)?;
58        check_alter_with_options_inner(
59            Self::CDC_CONNECTOR_NAME,
60            props,
61            secret_refs,
62            &[
63                "cdc.source.wait.streaming.start.timeout",
64                "debezium.max.queue.size",
65                "debezium.queue.memory.ratio",
66                "debezium.heartbeat.interval.ms",
67                "heartbeat.table.name",
68                "hostname",
69                "port",
70                "password",
71            ],
72        )
73    }
74}
risingwave_connector/source/cdc/oracle.rs

risingwave_connector/source/cdc/
oracle.rs
