risingwave_connector/parser/
config.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 std::collections::{BTreeMap, HashSet};
16
17use risingwave_common::bail;
18use risingwave_common::secret::LocalSecretManager;
19use risingwave_connector_codec::decoder::avro::MapHandling;
20use risingwave_pb::catalog::{PbSchemaRegistryNameStrategy, StreamSourceInfo};
21
22use super::unified::json::BigintUnsignedHandlingMode;
23use super::utils::{get_kafka_topic, get_pulsar_topic};
24use super::{DebeziumProps, TimeHandling, TimestampHandling, TimestamptzHandling};
25use crate::WithOptionsSecResolved;
26use crate::connector_common::AwsAuthProps;
27use crate::error::ConnectorResult;
28use crate::parser::PROTOBUF_MESSAGES_AS_JSONB;
29use crate::schema::AWS_GLUE_SCHEMA_ARN_KEY;
30use crate::schema::pulsar_schema::PulsarSchemaConfig;
31use crate::schema::schema_registry::SchemaRegistryConfig;
32use crate::source::cdc::CDC_MONGODB_STRONG_SCHEMA_KEY;
33use crate::source::{SourceColumnDesc, SourceEncode, SourceFormat, extract_source_struct};
34
35pub const PARQUET_CASE_INSENSITIVE_KEY: &str = "parquet.case_insensitive";
36
37/// Note: this is created in `SourceReader::build_stream`
38#[derive(Debug, Clone, Default)]
39pub struct ParserConfig {
40    pub common: CommonParserConfig,
41    pub specific: SpecificParserConfig,
42}
43
44fn parse_parquet_case_insensitive_option(
45    options_with_secret: &BTreeMap<String, String>,
46) -> ConnectorResult<bool> {
47    let value = options_with_secret.get(PARQUET_CASE_INSENSITIVE_KEY);
48    let Some(value) = value else {
49        return Ok(false);
50    };
51    let parsed = match value.trim().to_ascii_lowercase().parse::<bool>() {
52        Ok(parsed) => parsed,
53        Err(_) => bail!(
54            "invalid value for {}, expect true or false",
55            PARQUET_CASE_INSENSITIVE_KEY
56        ),
57    };
58    Ok(parsed)
59}
60
61impl ParserConfig {
62    pub fn get_config(self) -> (Vec<SourceColumnDesc>, SpecificParserConfig) {
63        (self.common.rw_columns, self.specific)
64    }
65}
66
67#[derive(Debug, Clone, Default)]
68pub struct CommonParserConfig {
69    /// Note: this is created by `SourceDescBuilder::builder`
70    pub rw_columns: Vec<SourceColumnDesc>,
71}
72
73#[derive(Debug, Clone, Default)]
74pub struct SpecificParserConfig {
75    pub encoding_config: EncodingProperties,
76    pub protocol_config: ProtocolProperties,
77}
78
79#[derive(Debug, Default, Clone)]
80pub enum EncodingProperties {
81    Avro(AvroProperties),
82    Protobuf(ProtobufProperties),
83    Csv(CsvProperties),
84    Json(JsonProperties),
85    MongoJson(MongoProperties),
86    Bytes(BytesProperties),
87    Parquet(ParquetProperties),
88    Native,
89    /// Encoding can't be specified because the source will determines it. Now only used in Iceberg.
90    None,
91    #[default]
92    Unspecified,
93}
94
95#[derive(Debug, Default, Clone)]
96pub enum ProtocolProperties {
97    Debezium(DebeziumProps),
98    DebeziumMongo,
99    Maxwell,
100    Canal,
101    Plain,
102    Upsert,
103    Native,
104    /// Protocol can't be specified because the source will determines it. Now only used in Iceberg.
105    None,
106    #[default]
107    Unspecified,
108}
109
110impl SpecificParserConfig {
111    // for test only
112    pub const DEFAULT_PLAIN_JSON: SpecificParserConfig = SpecificParserConfig {
113        encoding_config: EncodingProperties::Json(JsonProperties {
114            use_schema_registry: false,
115            timestamp_handling: None,
116            timestamptz_handling: None,
117            time_handling: None,
118            bigint_unsigned_handling: None,
119            handle_toast_columns: false,
120        }),
121        protocol_config: ProtocolProperties::Plain,
122    };
123
124    // The validity of (format, encode) is ensured by `extract_format_encode`
125    pub fn new(
126        info: &StreamSourceInfo,
127        with_properties: &WithOptionsSecResolved,
128    ) -> ConnectorResult<Self> {
129        let info = info.clone();
130        let source_struct = extract_source_struct(&info)?;
131        let format_encode_options_with_secret = LocalSecretManager::global()
132            .fill_secrets(info.format_encode_options, info.format_encode_secret_refs)?;
133        let (options, secret_refs) = with_properties.clone().into_parts();
134        // Make sure `with_properties` is no longer used by accident.
135        // All reads shall go to `options_with_secret` instead.
136        #[expect(unused_variables)]
137        let with_properties = ();
138        let options_with_secret =
139            LocalSecretManager::global().fill_secrets(options, secret_refs)?;
140        let format = source_struct.format;
141        let encode = source_struct.encode;
142        // this transformation is needed since there may be config for the protocol
143        // in the future
144        let protocol_config = match format {
145            SourceFormat::Native => ProtocolProperties::Native,
146            SourceFormat::None => ProtocolProperties::None,
147            SourceFormat::Debezium => {
148                let debezium_props = DebeziumProps::from(&format_encode_options_with_secret);
149                ProtocolProperties::Debezium(debezium_props)
150            }
151            SourceFormat::DebeziumMongo => ProtocolProperties::DebeziumMongo,
152            SourceFormat::Maxwell => ProtocolProperties::Maxwell,
153            SourceFormat::Canal => ProtocolProperties::Canal,
154            SourceFormat::Upsert => ProtocolProperties::Upsert,
155            SourceFormat::Plain => ProtocolProperties::Plain,
156            _ => unreachable!(),
157        };
158
159        let encoding_config = match (format, encode) {
160            (SourceFormat::Plain, SourceEncode::Csv) => EncodingProperties::Csv(CsvProperties {
161                delimiter: info.csv_delimiter as u8,
162                has_header: info.csv_has_header,
163            }),
164            (SourceFormat::Plain, SourceEncode::Parquet) => {
165                let case_insensitive = parse_parquet_case_insensitive_option(&options_with_secret)?;
166                EncodingProperties::Parquet(ParquetProperties { case_insensitive })
167            }
168            (SourceFormat::Plain, SourceEncode::Avro)
169            | (SourceFormat::Upsert, SourceEncode::Avro) => {
170                let mut config = AvroProperties {
171                    record_name: if info.proto_message_name.is_empty() {
172                        None
173                    } else {
174                        Some(info.proto_message_name.clone())
175                    },
176                    key_record_name: info.key_message_name.clone(),
177                    map_handling: MapHandling::from_options(&format_encode_options_with_secret)?,
178                    ..Default::default()
179                };
180                config.schema_location = if let Some(client_config) =
181                    PulsarSchemaConfig::from_options(&format_encode_options_with_secret)?
182                {
183                    SchemaLocation::Pulsar {
184                        client_config,
185                        topic: get_pulsar_topic(&options_with_secret)?.clone(),
186                    }
187                } else if let Some(schema_arn) =
188                    format_encode_options_with_secret.get(AWS_GLUE_SCHEMA_ARN_KEY)
189                {
190                    risingwave_common::license::Feature::GlueSchemaRegistry
191                        .check_available()
192                        .map_err(anyhow::Error::from)?;
193                    SchemaLocation::Glue {
194                        schema_arn: schema_arn.clone(),
195                        aws_auth_props: serde_json::from_value::<AwsAuthProps>(
196                            serde_json::to_value(format_encode_options_with_secret.clone())
197                                .unwrap(),
198                        )
199                        .map_err(|e| anyhow::anyhow!(e))?,
200                        // The option `mock_config` is not public and we can break compatibility.
201                        mock_config: format_encode_options_with_secret
202                            .get("aws.glue.mock_config")
203                            .cloned(),
204                    }
205                } else if info.use_schema_registry {
206                    SchemaLocation::Confluent {
207                        urls: info.row_schema_location.clone(),
208                        client_config: SchemaRegistryConfig::from(
209                            &format_encode_options_with_secret,
210                        ),
211                        name_strategy: PbSchemaRegistryNameStrategy::try_from(info.name_strategy)
212                            .unwrap(),
213                        topic: get_kafka_topic(&options_with_secret)?.clone(),
214                    }
215                } else {
216                    SchemaLocation::File {
217                        url: info.row_schema_location,
218                        aws_auth_props: Some(
219                            serde_json::from_value::<AwsAuthProps>(
220                                serde_json::to_value(format_encode_options_with_secret).unwrap(),
221                            )
222                            .map_err(|e| anyhow::anyhow!(e))?,
223                        ),
224                    }
225                };
226                EncodingProperties::Avro(config)
227            }
228            (SourceFormat::Plain, SourceEncode::Protobuf)
229            | (SourceFormat::Upsert, SourceEncode::Protobuf) => {
230                if info.row_schema_location.is_empty() {
231                    bail!("protobuf file location not provided");
232                }
233                let mut messages_as_jsonb = if let Some(messages_as_jsonb) =
234                    format_encode_options_with_secret.get(PROTOBUF_MESSAGES_AS_JSONB)
235                {
236                    messages_as_jsonb.split(',').map(|s| s.to_owned()).collect()
237                } else {
238                    HashSet::new()
239                };
240                messages_as_jsonb.insert("google.protobuf.Any".to_owned());
241
242                let mut config = ProtobufProperties {
243                    message_name: info.proto_message_name.clone(),
244                    key_message_name: info.key_message_name.clone(),
245                    messages_as_jsonb,
246                    ..Default::default()
247                };
248                config.schema_location = if info.use_schema_registry {
249                    SchemaLocation::Confluent {
250                        urls: info.row_schema_location.clone(),
251                        client_config: SchemaRegistryConfig::from(
252                            &format_encode_options_with_secret,
253                        ),
254                        name_strategy: PbSchemaRegistryNameStrategy::try_from(info.name_strategy)
255                            .unwrap(),
256                        topic: get_kafka_topic(&options_with_secret)?.clone(),
257                    }
258                } else {
259                    SchemaLocation::File {
260                        url: info.row_schema_location,
261                        aws_auth_props: Some(
262                            serde_json::from_value::<AwsAuthProps>(
263                                serde_json::to_value(format_encode_options_with_secret).unwrap(),
264                            )
265                            .map_err(|e| anyhow::anyhow!(e))?,
266                        ),
267                    }
268                };
269                EncodingProperties::Protobuf(config)
270            }
271            (SourceFormat::Debezium, SourceEncode::Avro) => {
272                EncodingProperties::Avro(AvroProperties {
273                    record_name: if info.proto_message_name.is_empty() {
274                        None
275                    } else {
276                        Some(info.proto_message_name.clone())
277                    },
278                    key_record_name: info.key_message_name.clone(),
279                    schema_location: SchemaLocation::Confluent {
280                        urls: info.row_schema_location.clone(),
281                        client_config: SchemaRegistryConfig::from(
282                            &format_encode_options_with_secret,
283                        ),
284                        name_strategy: PbSchemaRegistryNameStrategy::try_from(info.name_strategy)
285                            .unwrap(),
286                        topic: get_kafka_topic(&options_with_secret).unwrap().clone(),
287                    },
288                    ..Default::default()
289                })
290            }
291            (
292                SourceFormat::Plain
293                | SourceFormat::Debezium
294                | SourceFormat::Maxwell
295                | SourceFormat::Canal
296                | SourceFormat::Upsert,
297                SourceEncode::Json,
298            ) => EncodingProperties::Json(JsonProperties {
299                use_schema_registry: info.use_schema_registry,
300                timestamp_handling: None,
301                timestamptz_handling: format_encode_options_with_secret
302                    .get(TimestamptzHandling::OPTION_KEY)
303                    .map(|value| TimestamptzHandling::from_options(value))
304                    .transpose()?,
305                time_handling: None,
306                bigint_unsigned_handling: None,
307                handle_toast_columns: false,
308            }),
309            (SourceFormat::DebeziumMongo, SourceEncode::Json) => {
310                let props = MongoProperties::from(&format_encode_options_with_secret);
311                EncodingProperties::MongoJson(props)
312            }
313            (SourceFormat::Plain, SourceEncode::Bytes) => {
314                EncodingProperties::Bytes(BytesProperties { column_name: None })
315            }
316            (SourceFormat::Native, SourceEncode::Native) => EncodingProperties::Native,
317            (SourceFormat::None, SourceEncode::None) => EncodingProperties::None,
318            (format, encode) => {
319                bail!("Unsupported format {:?} encode {:?}", format, encode);
320            }
321        };
322        Ok(Self {
323            encoding_config,
324            protocol_config,
325        })
326    }
327}
328
329#[derive(Debug, Default, Clone)]
330pub struct AvroProperties {
331    pub schema_location: SchemaLocation,
332    pub record_name: Option<String>,
333    pub key_record_name: Option<String>,
334    pub map_handling: Option<MapHandling>,
335}
336
337/// WIP: may cover protobuf and json schema later.
338#[derive(Debug, Clone)]
339pub enum SchemaLocation {
340    /// Avsc from `https://`, `s3://` or `file://`.
341    File {
342        url: String,
343        aws_auth_props: Option<AwsAuthProps>, // for s3
344    },
345    /// <https://docs.confluent.io/platform/current/schema-registry/index.html>
346    Confluent {
347        urls: String,
348        client_config: SchemaRegistryConfig,
349        name_strategy: PbSchemaRegistryNameStrategy,
350        topic: String,
351    },
352    /// <https://docs.aws.amazon.com/glue/latest/dg/schema-registry.html>
353    Glue {
354        schema_arn: String,
355        aws_auth_props: AwsAuthProps,
356        // When `Some(_)`, ignore AWS and load schemas from provided config
357        mock_config: Option<String>,
358    },
359    /// Pulsar Admin API schema endpoint. Pulsar Avro payloads are raw datum bytes, and their writer
360    /// schema version is carried in message metadata rather than a payload header.
361    Pulsar {
362        client_config: PulsarSchemaConfig,
363        topic: String,
364    },
365}
366
367// TODO: `SpecificParserConfig` shall not `impl`/`derive` a `Default`
368impl Default for SchemaLocation {
369    fn default() -> Self {
370        // backward compatible but undesired
371        Self::File {
372            url: Default::default(),
373            aws_auth_props: None,
374        }
375    }
376}
377
378#[derive(Debug, Default, Clone)]
379pub struct ProtobufProperties {
380    pub schema_location: SchemaLocation,
381    pub message_name: String,
382    pub key_message_name: Option<String>,
383    pub messages_as_jsonb: HashSet<String>,
384}
385
386#[derive(Debug, Default, Clone, Copy)]
387pub struct CsvProperties {
388    pub delimiter: u8,
389    pub has_header: bool,
390}
391
392#[derive(Debug, Default, Clone, Copy)]
393pub struct ParquetProperties {
394    pub case_insensitive: bool,
395}
396
397#[derive(Debug, Default, Clone)]
398pub struct JsonProperties {
399    pub use_schema_registry: bool,
400    pub timestamp_handling: Option<TimestampHandling>,
401    pub timestamptz_handling: Option<TimestamptzHandling>,
402    pub time_handling: Option<TimeHandling>,
403    pub bigint_unsigned_handling: Option<BigintUnsignedHandlingMode>,
404    pub handle_toast_columns: bool,
405}
406
407#[derive(Debug, Default, Clone)]
408pub struct BytesProperties {
409    pub column_name: Option<String>,
410}
411
412#[derive(Debug, Default, Clone)]
413pub struct MongoProperties {
414    pub strong_schema: bool,
415}
416
417impl MongoProperties {
418    pub fn new(strong_schema: bool) -> Self {
419        Self { strong_schema }
420    }
421}
422impl From<&BTreeMap<String, String>> for MongoProperties {
423    fn from(config: &BTreeMap<String, String>) -> Self {
424        let strong_schema = config
425            .get(CDC_MONGODB_STRONG_SCHEMA_KEY)
426            .is_some_and(|k| k.eq_ignore_ascii_case("true"));
427        Self { strong_schema }
428    }
429}
430
431#[cfg(test)]
432mod tests {
433    use risingwave_pb::plan_common::{EncodeType, FormatType};
434
435    use super::*;
436    use crate::schema::pulsar_schema::{PULSAR_SCHEMA_AUTH_TOKEN_KEY, PULSAR_SCHEMA_URL_KEY};
437    use crate::source::{PULSAR_CONNECTOR, UPSTREAM_SOURCE_KEY};
438
439    #[test]
440    fn pulsar_schema_config_uses_pulsar_source_topic() {
441        let info = StreamSourceInfo {
442            format: FormatType::Plain as i32,
443            row_encode: EncodeType::Avro as i32,
444            format_encode_options: BTreeMap::from([
445                (
446                    PULSAR_SCHEMA_URL_KEY.to_owned(),
447                    "https://pulsar-admin:8443".to_owned(),
448                ),
449                (
450                    PULSAR_SCHEMA_AUTH_TOKEN_KEY.to_owned(),
451                    "schema-token".to_owned(),
452                ),
453            ]),
454            ..Default::default()
455        };
456        let source_options = WithOptionsSecResolved::without_secrets(BTreeMap::from([
457            (UPSTREAM_SOURCE_KEY.to_owned(), PULSAR_CONNECTOR.to_owned()),
458            (
459                "topic".to_owned(),
460                "persistent://tenant/namespace/events".to_owned(),
461            ),
462            ("auth.token".to_owned(), "broker-token".to_owned()),
463        ]));
464
465        let config = SpecificParserConfig::new(&info, &source_options).unwrap();
466        let EncodingProperties::Avro(AvroProperties {
467            schema_location:
468                SchemaLocation::Pulsar {
469                    client_config: _,
470                    topic,
471                },
472            ..
473        }) = config.encoding_config
474        else {
475            panic!("expected Pulsar Avro parser config");
476        };
477        assert_eq!(topic, "persistent://tenant/namespace/events");
478    }
479}
risingwave_connector/parser/config.rs

risingwave_connector/parser/
config.rs
