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::error::ConnectorError;
16use risingwave_connector::source::{ADBC_SNOWFLAKE_CONNECTOR, BATCH_POSIX_FS_CONNECTOR};
17
18use super::*;
19
20pub static SOURCE_ALLOWED_CONNECTION_CONNECTOR: LazyLock<HashSet<PbConnectionType>> =
21    LazyLock::new(|| {
22        hashset! {
23            PbConnectionType::Unspecified,
24            PbConnectionType::Kafka,
25            PbConnectionType::Iceberg,
26        }
27    });
28
29pub static SOURCE_ALLOWED_CONNECTION_SCHEMA_REGISTRY: LazyLock<HashSet<PbConnectionType>> =
30    LazyLock::new(|| {
31        hashset! {
32            PbConnectionType::Unspecified,
33            PbConnectionType::SchemaRegistry,
34        }
35    });
36
37// TODO: Better design if we want to support ENCODE KEY where we will have 4 dimensional array
38static CONNECTORS_COMPATIBLE_FORMATS: LazyLock<HashMap<String, HashMap<Format, Vec<Encode>>>> =
39    LazyLock::new(|| {
40        convert_args!(hashmap!(
41                KAFKA_CONNECTOR => hashmap!(
42                    Format::Plain => vec![Encode::Json, Encode::Protobuf, Encode::Avro, Encode::Bytes, Encode::Csv],
43                    Format::Upsert => vec![Encode::Json, Encode::Avro, Encode::Protobuf],
44                    Format::Debezium => vec![Encode::Json, Encode::Avro],
45                    Format::DebeziumMongo => vec![Encode::Json],
46                ),
47                PULSAR_CONNECTOR => hashmap!(
48                    Format::Plain => vec![Encode::Json, Encode::Protobuf, Encode::Avro, Encode::Bytes],
49                    Format::Upsert => vec![Encode::Json, Encode::Avro],
50                    Format::Debezium => vec![Encode::Json],
51                ),
52                KINESIS_CONNECTOR => hashmap!(
53                    Format::Plain => vec![Encode::Json, Encode::Protobuf, Encode::Avro, Encode::Bytes, Encode::Csv],
54                    Format::Upsert => vec![Encode::Json, Encode::Avro],
55                    Format::Debezium => vec![Encode::Json],
56                ),
57                GOOGLE_PUBSUB_CONNECTOR => hashmap!(
58                    Format::Plain => vec![Encode::Json, Encode::Protobuf, Encode::Avro, Encode::Bytes],
59                    Format::Debezium => vec![Encode::Json],
60                ),
61                NEXMARK_CONNECTOR => hashmap!(
62                    Format::Native => vec![Encode::Native],
63                    Format::Plain => vec![Encode::Bytes],
64                ),
65                DATAGEN_CONNECTOR => hashmap!(
66                    Format::Native => vec![Encode::Native],
67                    Format::Plain => vec![Encode::Bytes, Encode::Json],
68                ),
69                OPENDAL_S3_CONNECTOR => hashmap!(
70                    Format::Plain => vec![Encode::Csv, Encode::Json, Encode::Parquet],
71                ),
72                GCS_CONNECTOR => hashmap!(
73                    Format::Plain => vec![Encode::Csv, Encode::Json, Encode::Parquet],
74                ),
75                AZBLOB_CONNECTOR => hashmap!(
76                    Format::Plain => vec![Encode::Csv, Encode::Json, Encode::Parquet],
77                ),
78                POSIX_FS_CONNECTOR => hashmap!(
79                    Format::Plain => vec![Encode::Csv, Encode::Json, Encode::Parquet],
80                ),
81                BATCH_POSIX_FS_CONNECTOR => hashmap!(
82                    Format::Plain => vec![Encode::Csv],
83                ),
84                NATS_CONNECTOR => hashmap!(
85                    Format::Plain => vec![Encode::Json, Encode::Protobuf, Encode::Bytes],
86                ),
87                MQTT_CONNECTOR => hashmap!(
88                    Format::Plain => vec![Encode::Json, Encode::Protobuf, Encode::Bytes],
89                ),
90                TEST_CONNECTOR => hashmap!(
91                    Format::Plain => vec![Encode::Json],
92                ),
93                ICEBERG_CONNECTOR => hashmap!(
94                    Format::None => vec![Encode::None],
95                ),
96                ADBC_SNOWFLAKE_CONNECTOR => hashmap!(
97                    Format::None => vec![Encode::None],
98                ),
99        ))
100    });
101
102/// Returns whether the Pulsar schema is used. The Pulsar schema options themselves are validated by
103/// `PulsarSchemaConfig::from_options` when building the parser config. Only the connector and
104/// format, which that config is unaware of, are checked here.
105fn validate_pulsar_schema_options(
106    format_encode: &FormatEncodeOptions,
107    connector: &str,
108) -> Result<bool> {
109    let options = WithOptions::try_from(format_encode.row_options())?;
110    if !uses_pulsar_schema(&options, options.secret_ref()) {
111        return Ok(false);
112    }
113
114    if connector != PULSAR_CONNECTOR
115        || format_encode.format != Format::Plain
116        || !matches!(format_encode.row_encode, Encode::Avro | Encode::Json)
117    {
118        return Err(RwError::from(ProtocolError(format!(
119            "Pulsar schema requires connector = '{PULSAR_CONNECTOR}' with FORMAT PLAIN ENCODE AVRO or JSON"
120        ))));
121    }
122
123    Ok(true)
124}
125
126pub fn validate_compatibility(
127    format_encode: &FormatEncodeOptions,
128    props: &mut BTreeMap<String, String>,
129) -> Result<()> {
130    let mut connector = props
131        .get_connector()
132        .ok_or_else(|| RwError::from(ProtocolError("missing field 'connector'".to_owned())))?;
133    let uses_pulsar_schema = validate_pulsar_schema_options(format_encode, &connector)?;
134
135    let protocol_error = |e: ConnectorError| RwError::from(ProtocolError(e.to_report_string()));
136    dispatch_cdc_source_type!(
137        props.cdc_source_type(),
138        T,
139        {
140            T::check_license()?;
141            T::check_format_encode(
142                format_to_prost(&format_encode.format),
143                row_encode_to_prost(&format_encode.row_encode),
144            )
145            .map_err(protocol_error)?;
146            T::check_and_fill_create_with_options(props).map_err(protocol_error)?;
147            return Ok(());
148        },
149        {}
150    );
151
152    if connector == OPENDAL_S3_CONNECTOR {
153        // reject s3_v2 creation
154        return Err(RwError::from(Deprecated(
155            OPENDAL_S3_CONNECTOR.to_owned(),
156            LEGACY_S3_CONNECTOR.to_owned(),
157        )));
158    }
159    if connector == LEGACY_S3_CONNECTOR {
160        // S3 connector is deprecated, use OPENDAL_S3_CONNECTOR instead
161        // do s3 -> s3_v2 migration
162        let entry = props.get_mut(UPSTREAM_SOURCE_KEY).unwrap();
163        *entry = OPENDAL_S3_CONNECTOR.to_owned();
164        connector = OPENDAL_S3_CONNECTOR.to_owned();
165    }
166
167    let compatible_formats = CONNECTORS_COMPATIBLE_FORMATS
168        .get(&connector)
169        .ok_or_else(|| {
170            RwError::from(ProtocolError(format!(
171                "connector {:?} is not supported, accept {:?}",
172                connector,
173                CONNECTORS_COMPATIBLE_FORMATS.keys()
174            )))
175        })?;
176    if connector != KAFKA_CONNECTOR && !uses_pulsar_schema {
177        let res = match (&format_encode.format, &format_encode.row_encode) {
178            (Format::Plain, Encode::Protobuf) | (Format::Plain, Encode::Avro) => {
179                let mut options = WithOptions::try_from(format_encode.row_options())?;
180                let (_, use_schema_registry) = get_schema_location(options.inner_mut())?;
181                use_schema_registry
182            }
183            (Format::Debezium, Encode::Avro) => true,
184            (_, _) => false,
185        };
186        if res {
187            return Err(RwError::from(ProtocolError(format!(
188                "The {} must be kafka when schema registry is used",
189                UPSTREAM_SOURCE_KEY
190            ))));
191        }
192    }
193
194    let compatible_encodes = compatible_formats
195        .get(&format_encode.format)
196        .ok_or_else(|| {
197            RwError::from(ProtocolError(format!(
198                "connector {} does not support format {:?}",
199                connector, format_encode.format
200            )))
201        })?;
202    if !compatible_encodes.contains(&format_encode.row_encode) {
203        return Err(RwError::from(ProtocolError(format!(
204            "connector {} does not support format {:?} with encode {:?}",
205            connector, format_encode.format, format_encode.row_encode
206        ))));
207    }
208    Ok(())
209}
210
211#[cfg(test)]
212mod tests {
213    use risingwave_sqlparser::ast::SqlOption;
214
215    use super::*;
216
217    fn format_encode(
218        format: Format,
219        encode: Encode,
220        options: &[(&str, &str)],
221    ) -> FormatEncodeOptions {
222        let row_options = options
223            .iter()
224            .map(|(name, value)| {
225                let name = name.to_string();
226                let value = value.to_string();
227                SqlOption::try_from((&name, &value)).unwrap()
228            })
229            .collect();
230        FormatEncodeOptions {
231            format,
232            row_encode: encode,
233            row_options,
234            key_encode: None,
235        }
236    }
237
238    fn source_options(connector: &str) -> BTreeMap<String, String> {
239        BTreeMap::from([(UPSTREAM_SOURCE_KEY.to_owned(), connector.to_owned())])
240    }
241
242    #[test]
243    fn pulsar_schema_accepts_minimal_plain_avro_and_json_options() {
244        for encode in [Encode::Avro, Encode::Json] {
245            let format_encode = format_encode(
246                Format::Plain,
247                encode,
248                &[
249                    (PULSAR_SCHEMA_URL_KEY, "https://localhost:8443"),
250                    (PULSAR_SCHEMA_AUTH_TOKEN_KEY, "schema-token"),
251                ],
252            );
253            validate_compatibility(&format_encode, &mut source_options(PULSAR_CONNECTOR)).unwrap();
254        }
255    }
256
257    #[test]
258    fn pulsar_schema_rejects_other_connectors_and_formats() {
259        for (connector, format, encode) in [
260            (KAFKA_CONNECTOR, Format::Plain, Encode::Avro),
261            (PULSAR_CONNECTOR, Format::Upsert, Encode::Avro),
262            (KAFKA_CONNECTOR, Format::Plain, Encode::Json),
263            (PULSAR_CONNECTOR, Format::Upsert, Encode::Json),
264            (PULSAR_CONNECTOR, Format::Plain, Encode::Protobuf),
265        ] {
266            let format_encode = format_encode(
267                format,
268                encode,
269                &[(PULSAR_SCHEMA_URL_KEY, "http://localhost:8080")],
270            );
271            let error =
272                validate_compatibility(&format_encode, &mut source_options(connector)).unwrap_err();
273            assert!(error.to_string().contains(
274                "Pulsar schema requires connector = 'pulsar' with FORMAT PLAIN ENCODE AVRO or JSON"
275            ));
276        }
277    }
278
279    #[test]
280    fn confluent_schema_registry_behavior_is_unchanged() {
281        let format_encode = format_encode(
282            Format::Plain,
283            Encode::Avro,
284            &[("schema.registry", "http://localhost:8081")],
285        );
286        validate_compatibility(&format_encode, &mut source_options(KAFKA_CONNECTOR)).unwrap();
287    }
288}
risingwave_frontend/handler/create_source/validate.rs

risingwave_frontend/handler/create_source/
validate.rs
