risingwave_frontend/handler/create_source/
validate.rs

1// Copyright 2024 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 risingwave_connector::source::{ADBC_SNOWFLAKE_CONNECTOR, BATCH_POSIX_FS_CONNECTOR};
16
17use super::*;
18
19pub static SOURCE_ALLOWED_CONNECTION_CONNECTOR: LazyLock<HashSet<PbConnectionType>> =
20    LazyLock::new(|| {
21        hashset! {
22            PbConnectionType::Unspecified,
23            PbConnectionType::Kafka,
24            PbConnectionType::Iceberg,
25        }
26    });
27
28pub static SOURCE_ALLOWED_CONNECTION_SCHEMA_REGISTRY: LazyLock<HashSet<PbConnectionType>> =
29    LazyLock::new(|| {
30        hashset! {
31            PbConnectionType::Unspecified,
32            PbConnectionType::SchemaRegistry,
33        }
34    });
35
36// TODO: Better design if we want to support ENCODE KEY where we will have 4 dimensional array
37static CONNECTORS_COMPATIBLE_FORMATS: LazyLock<HashMap<String, HashMap<Format, Vec<Encode>>>> =
38    LazyLock::new(|| {
39        convert_args!(hashmap!(
40                KAFKA_CONNECTOR => hashmap!(
41                    Format::Plain => vec![Encode::Json, Encode::Protobuf, Encode::Avro, Encode::Bytes, Encode::Csv],
42                    Format::Upsert => vec![Encode::Json, Encode::Avro, Encode::Protobuf],
43                    Format::Debezium => vec![Encode::Json, Encode::Avro],
44                    Format::DebeziumMongo => vec![Encode::Json],
45                ),
46                PULSAR_CONNECTOR => hashmap!(
47                    Format::Plain => vec![Encode::Json, Encode::Protobuf, Encode::Avro, Encode::Bytes],
48                    Format::Upsert => vec![Encode::Json, Encode::Avro],
49                    Format::Debezium => vec![Encode::Json],
50                ),
51                KINESIS_CONNECTOR => hashmap!(
52                    Format::Plain => vec![Encode::Json, Encode::Protobuf, Encode::Avro, Encode::Bytes, Encode::Csv],
53                    Format::Upsert => vec![Encode::Json, Encode::Avro],
54                    Format::Debezium => vec![Encode::Json],
55                ),
56                GOOGLE_PUBSUB_CONNECTOR => hashmap!(
57                    Format::Plain => vec![Encode::Json, Encode::Protobuf, Encode::Avro, Encode::Bytes],
58                    Format::Debezium => vec![Encode::Json],
59                ),
60                NEXMARK_CONNECTOR => hashmap!(
61                    Format::Native => vec![Encode::Native],
62                    Format::Plain => vec![Encode::Bytes],
63                ),
64                DATAGEN_CONNECTOR => hashmap!(
65                    Format::Native => vec![Encode::Native],
66                    Format::Plain => vec![Encode::Bytes, Encode::Json],
67                ),
68                OPENDAL_S3_CONNECTOR => hashmap!(
69                    Format::Plain => vec![Encode::Csv, Encode::Json, Encode::Parquet],
70                ),
71                GCS_CONNECTOR => hashmap!(
72                    Format::Plain => vec![Encode::Csv, Encode::Json, Encode::Parquet],
73                ),
74                AZBLOB_CONNECTOR => hashmap!(
75                    Format::Plain => vec![Encode::Csv, Encode::Json, Encode::Parquet],
76                ),
77                POSIX_FS_CONNECTOR => hashmap!(
78                    Format::Plain => vec![Encode::Csv, Encode::Json, Encode::Parquet],
79                ),
80                BATCH_POSIX_FS_CONNECTOR => hashmap!(
81                    Format::Plain => vec![Encode::Csv],
82                ),
83                MYSQL_CDC_CONNECTOR => hashmap!(
84                    Format::Debezium => vec![Encode::Json],
85                    // support source stream job
86                    Format::Plain => vec![Encode::Json],
87                ),
88                POSTGRES_CDC_CONNECTOR => hashmap!(
89                    Format::Debezium => vec![Encode::Json],
90                    // support source stream job
91                    Format::Plain => vec![Encode::Json],
92                ),
93                ORACLE_CDC_CONNECTOR => hashmap!(
94                    Format::Debezium => vec![Encode::Json],
95                    // support source stream job
96                    Format::Plain => vec![Encode::Json],
97                ),
98                MONGODB_CDC_CONNECTOR => hashmap!(
99                    Format::DebeziumMongo => vec![Encode::Json],
100                ),
101                NATS_CONNECTOR => hashmap!(
102                    Format::Plain => vec![Encode::Json, Encode::Protobuf, Encode::Bytes],
103                ),
104                MQTT_CONNECTOR => hashmap!(
105                    Format::Plain => vec![Encode::Json, Encode::Protobuf, Encode::Bytes],
106                ),
107                TEST_CONNECTOR => hashmap!(
108                    Format::Plain => vec![Encode::Json],
109                ),
110                ICEBERG_CONNECTOR => hashmap!(
111                    Format::None => vec![Encode::None],
112                ),
113                ADBC_SNOWFLAKE_CONNECTOR => hashmap!(
114                    Format::None => vec![Encode::None],
115                ),
116                SQL_SERVER_CDC_CONNECTOR => hashmap!(
117                    Format::Debezium => vec![Encode::Json],
118                    // support source stream job
119                    Format::Plain => vec![Encode::Json],
120                ),
121        ))
122    });
123
124fn validate_license(connector: &str) -> Result<()> {
125    if connector == SQL_SERVER_CDC_CONNECTOR {
126        Feature::SqlServerCdcSource.check_available()?;
127    }
128    Ok(())
129}
130
131fn validate_decimal_handling_mode(props: &BTreeMap<String, String>) -> Result<()> {
132    if let Some(mode) = props.get("debezium.decimal.handling.mode")
133        && mode != "string"
134    {
135        return Err(RwError::from(ProtocolError(format!(
136            "'debezium.decimal.handling.mode' must be 'string', got: '{mode}'"
137        ))));
138    }
139    Ok(())
140}
141
142/// Requires an explicitly supplied heartbeat interval to be a positive signed 32-bit integer.
143/// Keep this policy in sync with Java's `SourceValidateHandler.validateHeartbeatInterval`.
144/// Validates user-supplied options, not the final Debezium configuration. On CREATE, an omitted
145/// interval uses the connector default; on ALTER, omission leaves the existing interval unchanged.
146/// Unrelated connector heartbeat mechanisms are not checked here.
147pub fn validate_heartbeat_interval(props: &BTreeMap<String, String>) -> Result<()> {
148    let Some(value) = props.get("debezium.heartbeat.interval.ms") else {
149        return Ok(());
150    };
151
152    // Match Java's ASCII integer syntax and Debezium's signed 32-bit range.
153    if !value.parse::<i32>().is_ok_and(|interval| interval > 0) {
154        return Err(ErrorCode::InvalidParameterValue(format!(
155            "'debezium.heartbeat.interval.ms' must be a positive integer, got: '{value}'"
156        ))
157        .into());
158    }
159    Ok(())
160}
161
162/// Returns whether the Pulsar schema is used. The Pulsar schema options themselves are validated by
163/// `PulsarSchemaConfig::from_options` when building the parser config. Only the connector and
164/// format, which that config is unaware of, are checked here.
165fn validate_pulsar_schema_options(
166    format_encode: &FormatEncodeOptions,
167    connector: &str,
168) -> Result<bool> {
169    let options = WithOptions::try_from(format_encode.row_options())?;
170    if !options.contains_key(PULSAR_SCHEMA_URL_KEY)
171        && !options.secret_ref().contains_key(PULSAR_SCHEMA_URL_KEY)
172    {
173        return Ok(false);
174    }
175
176    if connector != PULSAR_CONNECTOR
177        || format_encode.format != Format::Plain
178        || format_encode.row_encode != Encode::Avro
179    {
180        return Err(RwError::from(ProtocolError(format!(
181            "Pulsar schema requires connector = '{PULSAR_CONNECTOR}' with FORMAT PLAIN ENCODE AVRO"
182        ))));
183    }
184
185    Ok(true)
186}
187
188pub fn validate_compatibility(
189    format_encode: &FormatEncodeOptions,
190    props: &mut BTreeMap<String, String>,
191) -> Result<()> {
192    let mut connector = props
193        .get_connector()
194        .ok_or_else(|| RwError::from(ProtocolError("missing field 'connector'".to_owned())))?;
195
196    if connector == OPENDAL_S3_CONNECTOR {
197        // reject s3_v2 creation
198        return Err(RwError::from(Deprecated(
199            OPENDAL_S3_CONNECTOR.to_owned(),
200            LEGACY_S3_CONNECTOR.to_owned(),
201        )));
202    }
203    if connector == LEGACY_S3_CONNECTOR {
204        // S3 connector is deprecated, use OPENDAL_S3_CONNECTOR instead
205        // do s3 -> s3_v2 migration
206        let entry = props.get_mut(UPSTREAM_SOURCE_KEY).unwrap();
207        *entry = OPENDAL_S3_CONNECTOR.to_owned();
208        connector = OPENDAL_S3_CONNECTOR.to_owned();
209    }
210
211    let compatible_formats = CONNECTORS_COMPATIBLE_FORMATS
212        .get(&connector)
213        .ok_or_else(|| {
214            RwError::from(ProtocolError(format!(
215                "connector {:?} is not supported, accept {:?}",
216                connector,
217                CONNECTORS_COMPATIBLE_FORMATS.keys()
218            )))
219        })?;
220
221    // RisingWave consumes schema-less JSON from Debezium and cannot reconstruct the scale of
222    // binary logical decimals emitted by `precise`. `double` can lose precision, so an explicit
223    // override must retain the common `string` default from `debezium.properties`.
224    validate_decimal_handling_mode(props)?;
225
226    validate_license(&connector)?;
227    let uses_pulsar_schema = validate_pulsar_schema_options(format_encode, &connector)?;
228    if connector != KAFKA_CONNECTOR && !uses_pulsar_schema {
229        let res = match (&format_encode.format, &format_encode.row_encode) {
230            (Format::Plain, Encode::Protobuf) | (Format::Plain, Encode::Avro) => {
231                let mut options = WithOptions::try_from(format_encode.row_options())?;
232                let (_, use_schema_registry) = get_schema_location(options.inner_mut())?;
233                use_schema_registry
234            }
235            (Format::Debezium, Encode::Avro) => true,
236            (_, _) => false,
237        };
238        if res {
239            return Err(RwError::from(ProtocolError(format!(
240                "The {} must be kafka when schema registry is used",
241                UPSTREAM_SOURCE_KEY
242            ))));
243        }
244    }
245
246    let compatible_encodes = compatible_formats
247        .get(&format_encode.format)
248        .ok_or_else(|| {
249            RwError::from(ProtocolError(format!(
250                "connector {} does not support format {:?}",
251                connector, format_encode.format
252            )))
253        })?;
254    if !compatible_encodes.contains(&format_encode.row_encode) {
255        return Err(RwError::from(ProtocolError(format!(
256            "connector {} does not support format {:?} with encode {:?}",
257            connector, format_encode.format, format_encode.row_encode
258        ))));
259    }
260
261    if connector == POSTGRES_CDC_CONNECTOR || connector == CITUS_CDC_CONNECTOR {
262        match props.get("slot.name") {
263            None => {
264                // Build a random slot name with UUID
265                // e.g. "rw_cdc_f9a3567e6dd54bf5900444c8b1c03815"
266                let uuid = uuid::Uuid::new_v4();
267                props.insert("slot.name".into(), format!("rw_cdc_{}", uuid.simple()));
268            }
269            Some(slot_name) => {
270                // please refer to
271                // - https://github.com/debezium/debezium/blob/97956ce25b7612e3413d363658661896b7d2e0a2/debezium-connector-postgres/src/main/java/io/debezium/connector/postgresql/PostgresConnectorConfig.java#L1179
272                // - https://doxygen.postgresql.org/slot_8c.html#afac399f07320b9adfd2c599cf822aaa3
273                if !slot_name
274                    .chars()
275                    .all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '_')
276                    || slot_name.len() > 63
277                {
278                    return Err(RwError::from(ProtocolError(format!(
279                        "Invalid replication slot name: {:?}. Valid replication slot name must contain only digits, lowercase characters and underscores with length <= 63",
280                        slot_name
281                    ))));
282                }
283            }
284        }
285
286        if !props.contains_key("schema.name") {
287            // Default schema name is "public"
288            props.insert("schema.name".into(), "public".into());
289        }
290        if !props.contains_key("publication.name") {
291            // Build a random publication name with UUID to avoid conflicts between sources
292            // e.g. "rw_publication_f9a3567e6dd54bf5900444c8b1c03815"
293            let uuid = uuid::Uuid::new_v4();
294            props.insert(
295                "publication.name".into(),
296                format!("rw_publication_{}", uuid.simple()),
297            );
298        }
299        if !props.contains_key("publication.create.enable") {
300            // Default auto create publication if doesn't exist
301            props.insert("publication.create.enable".into(), "true".into());
302        }
303    }
304
305    if connector == SQL_SERVER_CDC_CONNECTOR && !props.contains_key("schema.name") {
306        // Default schema name is "dbo"
307        props.insert("schema.name".into(), "dbo".into());
308    }
309
310    // Validate cdc.source.wait.streaming.start.timeout for all CDC connectors
311    if (connector == MYSQL_CDC_CONNECTOR
312        || connector == POSTGRES_CDC_CONNECTOR
313        || connector == CITUS_CDC_CONNECTOR
314        || connector == MONGODB_CDC_CONNECTOR
315        || connector == SQL_SERVER_CDC_CONNECTOR
316        || connector == ORACLE_CDC_CONNECTOR)
317        && let Some(timeout_value) = props.get("cdc.source.wait.streaming.start.timeout")
318        && timeout_value.parse::<u32>().is_err()
319    {
320        return Err(ErrorCode::InvalidConfigValue {
321            config_entry: "cdc.source.wait.streaming.start.timeout".to_owned(),
322            config_value: timeout_value.to_owned(),
323        }
324        .into());
325    }
326
327    // Validate debezium.max.queue.size for all CDC connectors
328    if (connector == MYSQL_CDC_CONNECTOR
329        || connector == POSTGRES_CDC_CONNECTOR
330        || connector == CITUS_CDC_CONNECTOR
331        || connector == MONGODB_CDC_CONNECTOR
332        || connector == SQL_SERVER_CDC_CONNECTOR
333        || connector == ORACLE_CDC_CONNECTOR)
334        && let Some(queue_size_value) = props.get("debezium.max.queue.size")
335        && queue_size_value.parse::<u32>().is_err()
336    {
337        return Err(ErrorCode::InvalidConfigValue {
338            config_entry: "debezium.max.queue.size".to_owned(),
339            config_value: queue_size_value.to_owned(),
340        }
341        .into());
342    }
343
344    validate_heartbeat_interval(props)
345}
346
347#[cfg(test)]
348mod tests {
349    use risingwave_sqlparser::ast::SqlOption;
350
351    use super::*;
352
353    fn format_encode(
354        format: Format,
355        encode: Encode,
356        options: &[(&str, &str)],
357    ) -> FormatEncodeOptions {
358        let row_options = options
359            .iter()
360            .map(|(name, value)| {
361                let name = name.to_string();
362                let value = value.to_string();
363                SqlOption::try_from((&name, &value)).unwrap()
364            })
365            .collect();
366        FormatEncodeOptions {
367            format,
368            row_encode: encode,
369            row_options,
370            key_encode: None,
371        }
372    }
373
374    fn source_options(connector: &str) -> BTreeMap<String, String> {
375        BTreeMap::from([(UPSTREAM_SOURCE_KEY.to_owned(), connector.to_owned())])
376    }
377
378    #[test]
379    fn pulsar_schema_accepts_minimal_plain_avro_options() {
380        let format_encode = format_encode(
381            Format::Plain,
382            Encode::Avro,
383            &[
384                (PULSAR_SCHEMA_URL_KEY, "https://localhost:8443"),
385                (PULSAR_SCHEMA_AUTH_TOKEN_KEY, "schema-token"),
386            ],
387        );
388        validate_compatibility(&format_encode, &mut source_options(PULSAR_CONNECTOR)).unwrap();
389    }
390
391    #[test]
392    fn pulsar_schema_rejects_other_connectors_and_formats() {
393        for (connector, format, encode) in [
394            (KAFKA_CONNECTOR, Format::Plain, Encode::Avro),
395            (PULSAR_CONNECTOR, Format::Upsert, Encode::Avro),
396            (PULSAR_CONNECTOR, Format::Plain, Encode::Protobuf),
397            (PULSAR_CONNECTOR, Format::Plain, Encode::Json),
398        ] {
399            let format_encode = format_encode(
400                format,
401                encode,
402                &[(PULSAR_SCHEMA_URL_KEY, "http://localhost:8080")],
403            );
404            let error =
405                validate_compatibility(&format_encode, &mut source_options(connector)).unwrap_err();
406            assert!(error.to_string().contains(
407                "Pulsar schema requires connector = 'pulsar' with FORMAT PLAIN ENCODE AVRO"
408            ));
409        }
410    }
411
412    #[test]
413    fn confluent_schema_registry_behavior_is_unchanged() {
414        let format_encode = format_encode(
415            Format::Plain,
416            Encode::Avro,
417            &[("schema.registry", "http://localhost:8081")],
418        );
419        validate_compatibility(&format_encode, &mut source_options(KAFKA_CONNECTOR)).unwrap();
420    }
421}
risingwave_frontend/handler/create_source/validate.rs

risingwave_frontend/handler/create_source/
validate.rs
