risingwave_connector/parser/debezium/
simd_json_parser.rs

1// Copyright 2023 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::fmt::Debug;
16
17use anyhow::Context;
18use simd_json::BorrowedValue;
19use simd_json::prelude::MutableObject;
20
21use crate::error::ConnectorResult;
22use crate::parser::unified::AccessImpl;
23use crate::parser::unified::debezium::MongoJsonAccess;
24use crate::parser::unified::json::{
25    BigintUnsignedHandlingMode, JsonAccess, JsonParseOptions, TimeHandling, TimestampHandling,
26    TimestamptzHandling,
27};
28use crate::parser::{AccessBuilder, MongoProperties};
29
30#[derive(Debug)]
31pub struct DebeziumJsonAccessBuilder {
32    value: Option<Vec<u8>>,
33    json_parse_options: JsonParseOptions,
34}
35
36impl DebeziumJsonAccessBuilder {
37    pub fn new(
38        timestamptz_handling: TimestamptzHandling,
39        timestamp_handling: TimestampHandling,
40        time_handling: TimeHandling,
41        bigint_unsigned_handling: BigintUnsignedHandlingMode,
42        handle_toast_columns: bool,
43    ) -> ConnectorResult<Self> {
44        Ok(Self {
45            value: None,
46            json_parse_options: JsonParseOptions::new_for_debezium(
47                timestamptz_handling,
48                timestamp_handling,
49                time_handling,
50                bigint_unsigned_handling,
51                handle_toast_columns,
52            ),
53        })
54    }
55
56    pub fn new_for_schema_event() -> ConnectorResult<Self> {
57        Ok(Self {
58            value: None,
59            json_parse_options: JsonParseOptions::default(),
60        })
61    }
62}
63
64impl AccessBuilder for DebeziumJsonAccessBuilder {
65    async fn generate_accessor(
66        &mut self,
67        payload: Vec<u8>,
68        _: &crate::source::SourceMeta,
69    ) -> ConnectorResult<AccessImpl<'_>> {
70        self.value = Some(payload);
71        let mut event: BorrowedValue<'_> =
72            simd_json::to_borrowed_value(self.value.as_mut().unwrap())
73                .context("failed to parse debezium json payload")?;
74
75        let payload = if let Some(payload) = event.get_mut("payload") {
76            std::mem::take(payload)
77        } else {
78            event
79        };
80
81        Ok(AccessImpl::Json(JsonAccess::new_with_options(
82            payload,
83            &self.json_parse_options,
84        )))
85    }
86}
87
88#[derive(Debug)]
89pub struct DebeziumMongoJsonAccessBuilder {
90    value: Option<Vec<u8>>,
91    json_parse_options: JsonParseOptions,
92    strong_schema: bool,
93}
94
95impl DebeziumMongoJsonAccessBuilder {
96    pub fn new(props: MongoProperties) -> anyhow::Result<Self> {
97        Ok(Self {
98            value: None,
99            json_parse_options: JsonParseOptions::new_for_debezium(
100                TimestamptzHandling::GuessNumberUnit,
101                TimestampHandling::GuessNumberUnit,
102                TimeHandling::Micro,
103                BigintUnsignedHandlingMode::Long,
104                false,
105            ),
106            strong_schema: props.strong_schema,
107        })
108    }
109}
110
111impl AccessBuilder for DebeziumMongoJsonAccessBuilder {
112    async fn generate_accessor(
113        &mut self,
114        payload: Vec<u8>,
115        _: &crate::source::SourceMeta,
116    ) -> ConnectorResult<AccessImpl<'_>> {
117        self.value = Some(payload);
118        let mut event: BorrowedValue<'_> =
119            simd_json::to_borrowed_value(self.value.as_mut().unwrap())
120                .context("failed to parse debezium mongo json payload")?;
121
122        let payload = if let Some(payload) = event.get_mut("payload") {
123            std::mem::take(payload)
124        } else {
125            event
126        };
127
128        Ok(AccessImpl::MongoJson(MongoJsonAccess::new(
129            JsonAccess::new_with_options(payload, &self.json_parse_options),
130            self.strong_schema,
131        )))
132    }
133}
134
135#[cfg(test)]
136mod tests {
137    use chrono::{NaiveDate, NaiveTime};
138    use risingwave_common::array::{Op, StructValue};
139    use risingwave_common::catalog::ColumnId;
140    use risingwave_common::row::{OwnedRow, Row};
141    use risingwave_common::types::{
142        DataType, Date, Interval, Scalar, ScalarImpl, StructType, Time, Timestamp,
143    };
144    use serde_json::Value;
145    use thiserror_ext::AsReport;
146
147    use crate::parser::{
148        DebeziumParser, DebeziumProps, EncodingProperties, JsonProperties, ProtocolProperties,
149        SourceColumnDesc, SourceStreamChunkBuilder, SpecificParserConfig,
150    };
151    use crate::source::{SourceContext, SourceCtrlOpts};
152
153    fn assert_json_eq(parse_result: &Option<ScalarImpl>, json_str: &str) {
154        if let Some(ScalarImpl::Jsonb(json_val)) = parse_result {
155            let mut json_string = String::new();
156            json_val
157                .as_scalar_ref()
158                .force_str(&mut json_string)
159                .unwrap();
160            let val1: Value = serde_json::from_str(json_string.as_str()).unwrap();
161            let val2: Value = serde_json::from_str(json_str).unwrap();
162            assert_eq!(val1, val2);
163        }
164    }
165
166    async fn build_parser(rw_columns: Vec<SourceColumnDesc>) -> DebeziumParser {
167        let props = SpecificParserConfig {
168            encoding_config: EncodingProperties::Json(JsonProperties {
169                use_schema_registry: false,
170                timestamptz_handling: None,
171                timestamp_handling: None,
172                time_handling: None,
173                bigint_unsigned_handling: None,
174                handle_toast_columns: false,
175            }),
176            protocol_config: ProtocolProperties::Debezium(DebeziumProps::default()),
177        };
178        DebeziumParser::new(props, rw_columns, SourceContext::dummy().into())
179            .await
180            .unwrap()
181    }
182
183    async fn parse_one(
184        mut parser: DebeziumParser,
185        columns: Vec<SourceColumnDesc>,
186        payload: Vec<u8>,
187    ) -> Vec<(Op, OwnedRow)> {
188        let mut builder = SourceStreamChunkBuilder::new(columns, SourceCtrlOpts::for_test());
189        parser
190            .parse_inner(None, Some(payload), builder.row_writer())
191            .await
192            .unwrap();
193        builder.finish_current_chunk();
194        let chunk = builder.consume_ready_chunks().next().unwrap();
195        chunk
196            .rows()
197            .map(|(op, row_ref)| (op, row_ref.into_owned_row()))
198            .collect::<Vec<_>>()
199    }
200
201    mod test1_basic {
202        use super::*;
203
204        fn get_test1_columns() -> Vec<SourceColumnDesc> {
205            vec![
206                SourceColumnDesc::simple("id", DataType::Int32, ColumnId::from(0)),
207                SourceColumnDesc::simple("name", DataType::Varchar, ColumnId::from(1)),
208                SourceColumnDesc::simple("description", DataType::Varchar, ColumnId::from(2)),
209                SourceColumnDesc::simple("weight", DataType::Float64, ColumnId::from(3)),
210            ]
211        }
212
213        #[tokio::test]
214        async fn test1_debezium_json_parser_read() {
215            //     "before": null,
216            //     "after": {
217            //       "id": 101,
218            //       "name": "scooter",
219            //       "description": "Small 2-wheel scooter",
220            //       "weight": 1.234
221            //     },
222            let input = vec![
223                // data with payload field
224                br#"{"payload":{"before":null,"after":{"id":101,"name":"scooter","description":"Small 2-wheel scooter","weight":1.234},"source":{"version":"1.7.1.Final","connector":"mysql","name":"dbserver1","ts_ms":1639547113601,"snapshot":"true","db":"inventory","sequence":null,"table":"products","server_id":0,"gtid":null,"file":"mysql-bin.000003","pos":156,"row":0,"thread":null,"query":null},"op":"r","ts_ms":1639547113602,"transaction":null}}"#.to_vec(),
225                // data without payload field
226                br#"{"before":null,"after":{"id":101,"name":"scooter","description":"Small 2-wheel scooter","weight":1.234},"source":{"version":"1.7.1.Final","connector":"mysql","name":"dbserver1","ts_ms":1639547113601,"snapshot":"true","db":"inventory","sequence":null,"table":"products","server_id":0,"gtid":null,"file":"mysql-bin.000003","pos":156,"row":0,"thread":null,"query":null},"op":"r","ts_ms":1639547113602,"transaction":null}"#.to_vec()];
227
228            let columns = get_test1_columns();
229
230            for data in input {
231                let parser = build_parser(columns.clone()).await;
232                let [(_op, row)]: [_; 1] = parse_one(parser, columns.clone(), data)
233                    .await
234                    .try_into()
235                    .unwrap();
236
237                assert!(row[0].eq(&Some(ScalarImpl::Int32(101))));
238                assert!(row[1].eq(&Some(ScalarImpl::Utf8("scooter".into()))));
239                assert!(row[2].eq(&Some(ScalarImpl::Utf8("Small 2-wheel scooter".into()))));
240                assert!(row[3].eq(&Some(ScalarImpl::Float64(1.234.into()))));
241            }
242        }
243
244        #[tokio::test]
245        async fn test1_debezium_json_parser_insert() {
246            //     "before": null,
247            //     "after": {
248            //       "id": 102,
249            //       "name": "car battery",
250            //       "description": "12V car battery",
251            //       "weight": 8.1
252            //     },
253            let input = vec![
254                // data with payload field
255                br#"{"payload":{"before":null,"after":{"id":102,"name":"car battery","description":"12V car battery","weight":8.1},"source":{"version":"1.7.1.Final","connector":"mysql","name":"dbserver1","ts_ms":1639551564000,"snapshot":"false","db":"inventory","sequence":null,"table":"products","server_id":223344,"gtid":null,"file":"mysql-bin.000003","pos":717,"row":0,"thread":null,"query":null},"op":"c","ts_ms":1639551564960,"transaction":null}}"#.to_vec(),
256                // data without payload field
257                br#"{"before":null,"after":{"id":102,"name":"car battery","description":"12V car battery","weight":8.1},"source":{"version":"1.7.1.Final","connector":"mysql","name":"dbserver1","ts_ms":1639551564000,"snapshot":"false","db":"inventory","sequence":null,"table":"products","server_id":223344,"gtid":null,"file":"mysql-bin.000003","pos":717,"row":0,"thread":null,"query":null},"op":"c","ts_ms":1639551564960,"transaction":null}"#.to_vec()];
258
259            let columns = get_test1_columns();
260
261            for data in input {
262                let parser = build_parser(columns.clone()).await;
263                let [(op, row)]: [_; 1] = parse_one(parser, columns.clone(), data)
264                    .await
265                    .try_into()
266                    .unwrap();
267                assert_eq!(op, Op::Insert);
268
269                assert!(row[0].eq(&Some(ScalarImpl::Int32(102))));
270                assert!(row[1].eq(&Some(ScalarImpl::Utf8("car battery".into()))));
271                assert!(row[2].eq(&Some(ScalarImpl::Utf8("12V car battery".into()))));
272                assert!(row[3].eq(&Some(ScalarImpl::Float64(8.1.into()))));
273            }
274        }
275
276        #[tokio::test]
277        async fn test1_debezium_json_parser_delete() {
278            //     "before": {
279            //       "id": 101,
280            //       "name": "scooter",
281            //       "description": "Small 2-wheel scooter",
282            //       "weight": 1.234
283            //     },
284            //     "after": null,
285            let input = vec![
286                // data with payload field
287                br#"{"payload":{"before":{"id":101,"name":"scooter","description":"Small 2-wheel scooter","weight":1.234},"after":null,"source":{"version":"1.7.1.Final","connector":"mysql","name":"dbserver1","ts_ms":1639551767000,"snapshot":"false","db":"inventory","sequence":null,"table":"products","server_id":223344,"gtid":null,"file":"mysql-bin.000003","pos":1045,"row":0,"thread":null,"query":null},"op":"d","ts_ms":1639551767775,"transaction":null}}"#.to_vec(),
288                // data without payload field
289                br#"{"before":{"id":101,"name":"scooter","description":"Small 2-wheel scooter","weight":1.234},"after":null,"source":{"version":"1.7.1.Final","connector":"mysql","name":"dbserver1","ts_ms":1639551767000,"snapshot":"false","db":"inventory","sequence":null,"table":"products","server_id":223344,"gtid":null,"file":"mysql-bin.000003","pos":1045,"row":0,"thread":null,"query":null},"op":"d","ts_ms":1639551767775,"transaction":null}"#.to_vec()];
290
291            for data in input {
292                let columns = get_test1_columns();
293                let parser = build_parser(columns.clone()).await;
294                let [(op, row)]: [_; 1] = parse_one(parser, columns.clone(), data)
295                    .await
296                    .try_into()
297                    .unwrap();
298
299                assert_eq!(op, Op::Delete);
300
301                assert!(row[0].eq(&Some(ScalarImpl::Int32(101))));
302                assert!(row[1].eq(&Some(ScalarImpl::Utf8("scooter".into()))));
303                assert!(row[2].eq(&Some(ScalarImpl::Utf8("Small 2-wheel scooter".into()))));
304                assert!(row[3].eq(&Some(ScalarImpl::Float64(1.234.into()))));
305            }
306        }
307
308        #[tokio::test]
309        async fn test1_debezium_json_parser_update() {
310            //     "before": {
311            //       "id": 102,
312            //       "name": "car battery",
313            //       "description": "12V car battery",
314            //       "weight": 8.1
315            //     },
316            //     "after": {
317            //       "id": 102,
318            //       "name": "car battery",
319            //       "description": "24V car battery",
320            //       "weight": 9.1
321            //     },
322            let input = vec![
323                // data with payload field
324                br#"{"payload":{"before":{"id":102,"name":"car battery","description":"12V car battery","weight":8.1},"after":{"id":102,"name":"car battery","description":"24V car battery","weight":9.1},"source":{"version":"1.7.1.Final","connector":"mysql","name":"dbserver1","ts_ms":1639551901000,"snapshot":"false","db":"inventory","sequence":null,"table":"products","server_id":223344,"gtid":null,"file":"mysql-bin.000003","pos":1382,"row":0,"thread":null,"query":null},"op":"u","ts_ms":1639551901165,"transaction":null}}"#.to_vec(),
325                // data without payload field
326                br#"{"before":{"id":102,"name":"car battery","description":"12V car battery","weight":8.1},"after":{"id":102,"name":"car battery","description":"24V car battery","weight":9.1},"source":{"version":"1.7.1.Final","connector":"mysql","name":"dbserver1","ts_ms":1639551901000,"snapshot":"false","db":"inventory","sequence":null,"table":"products","server_id":223344,"gtid":null,"file":"mysql-bin.000003","pos":1382,"row":0,"thread":null,"query":null},"op":"u","ts_ms":1639551901165,"transaction":null}"#.to_vec()];
327
328            let columns = get_test1_columns();
329
330            for data in input {
331                let parser = build_parser(columns.clone()).await;
332                let [(op, row)]: [_; 1] = parse_one(parser, columns.clone(), data)
333                    .await
334                    .try_into()
335                    .unwrap();
336
337                assert_eq!(op, Op::Insert);
338
339                assert!(row[0].eq(&Some(ScalarImpl::Int32(102))));
340                assert!(row[1].eq(&Some(ScalarImpl::Utf8("car battery".into()))));
341                assert!(row[2].eq(&Some(ScalarImpl::Utf8("24V car battery".into()))));
342                assert!(row[3].eq(&Some(ScalarImpl::Float64(9.1.into()))));
343            }
344        }
345    }
346    // test2 covers read/insert/update/delete event on the following MySQL table for debezium json:
347    // CREATE TABLE IF NOT EXISTS orders (
348    //     O_KEY BIGINT NOT NULL,
349    //     O_BOOL BOOLEAN,
350    //     O_TINY TINYINT,
351    //     O_INT INT,
352    //     O_REAL REAL,
353    //     O_DOUBLE DOUBLE,
354    //     O_DECIMAL DECIMAL(15, 2),
355    //     O_CHAR CHAR(15),
356    //     O_DATE DATE,
357    //     O_TIME TIME,
358    //     O_DATETIME DATETIME,
359    //     O_TIMESTAMP TIMESTAMP,
360    //     O_JSON JSON,
361    //     PRIMARY KEY (O_KEY));
362    // test2 also covers overflow tests on basic types
363    mod test2_mysql {
364        use super::*;
365
366        fn get_test2_columns() -> Vec<SourceColumnDesc> {
367            vec![
368                SourceColumnDesc::simple("O_KEY", DataType::Int64, ColumnId::from(0)),
369                SourceColumnDesc::simple("O_BOOL", DataType::Boolean, ColumnId::from(1)),
370                SourceColumnDesc::simple("O_TINY", DataType::Int16, ColumnId::from(2)),
371                SourceColumnDesc::simple("O_INT", DataType::Int32, ColumnId::from(3)),
372                SourceColumnDesc::simple("O_REAL", DataType::Float32, ColumnId::from(4)),
373                SourceColumnDesc::simple("O_DOUBLE", DataType::Float64, ColumnId::from(5)),
374                SourceColumnDesc::simple("O_DECIMAL", DataType::Decimal, ColumnId::from(6)),
375                SourceColumnDesc::simple("O_CHAR", DataType::Varchar, ColumnId::from(7)),
376                SourceColumnDesc::simple("O_DATE", DataType::Date, ColumnId::from(8)),
377                SourceColumnDesc::simple("O_TIME", DataType::Time, ColumnId::from(9)),
378                SourceColumnDesc::simple("O_DATETIME", DataType::Timestamp, ColumnId::from(10)),
379                SourceColumnDesc::simple("O_TIMESTAMP", DataType::Timestamptz, ColumnId::from(11)),
380                SourceColumnDesc::simple("O_JSON", DataType::Jsonb, ColumnId::from(12)),
381            ]
382        }
383
384        #[tokio::test]
385        async fn test2_debezium_json_parser_read() {
386            let data = br#"{"payload":{"before":null,"after":{"O_KEY":111,"O_BOOL":1,"O_TINY":-1,"O_INT":-1111,"O_REAL":-11.11,"O_DOUBLE":-111.11111,"O_DECIMAL":-111.11,"O_CHAR":"yes please","O_DATE":"1000-01-01","O_TIME":0,"O_DATETIME":0,"O_TIMESTAMP":"1970-01-01T00:00:01Z","O_JSON":"{\"k1\": \"v1\", \"k2\": 11}"},"source":{"version":"1.9.7.Final","connector":"mysql","name":"RW_CDC_test.orders","ts_ms":1678090651000,"snapshot":"last","db":"test","sequence":null,"table":"orders","server_id":0,"gtid":null,"file":"mysql-bin.000003","pos":951,"row":0,"thread":null,"query":null},"op":"r","ts_ms":1678090651640,"transaction":null}}"#;
387
388            let columns = get_test2_columns();
389
390            let parser = build_parser(columns.clone()).await;
391
392            let [(_op, row)]: [_; 1] = parse_one(parser, columns, data.to_vec())
393                .await
394                .try_into()
395                .unwrap();
396
397            assert!(row[0].eq(&Some(ScalarImpl::Int64(111))));
398            assert!(row[1].eq(&Some(ScalarImpl::Bool(true))));
399            assert!(row[2].eq(&Some(ScalarImpl::Int16(-1))));
400            assert!(row[3].eq(&Some(ScalarImpl::Int32(-1111))));
401            assert!(row[4].eq(&Some(ScalarImpl::Float32((-11.11).into()))));
402            assert!(row[5].eq(&Some(ScalarImpl::Float64((-111.11111).into()))));
403            assert!(row[6].eq(&Some(ScalarImpl::Decimal("-111.11".parse().unwrap()))));
404            assert!(row[7].eq(&Some(ScalarImpl::Utf8("yes please".into()))));
405            assert!(row[8].eq(&Some(ScalarImpl::Date(Date::new(
406                NaiveDate::from_ymd_opt(1000, 1, 1).unwrap()
407            )))));
408            assert!(row[9].eq(&Some(ScalarImpl::Time(Time::new(
409                NaiveTime::from_hms_micro_opt(0, 0, 0, 0).unwrap()
410            )))));
411            assert!(row[10].eq(&Some(ScalarImpl::Timestamp(Timestamp::new(
412                "1970-01-01T00:00:00".parse().unwrap()
413            )))));
414            assert!(row[11].eq(&Some(ScalarImpl::Timestamptz(
415                "1970-01-01T00:00:01Z".parse().unwrap()
416            ))));
417            assert_json_eq(&row[12], "{\"k1\": \"v1\", \"k2\": 11}");
418        }
419
420        #[tokio::test]
421        async fn test2_debezium_json_parser_insert() {
422            let data = br#"{"payload":{"before":null,"after":{"O_KEY":111,"O_BOOL":1,"O_TINY":-1,"O_INT":-1111,"O_REAL":-11.11,"O_DOUBLE":-111.11111,"O_DECIMAL":-111.11,"O_CHAR":"yes please","O_DATE":"1000-01-01","O_TIME":0,"O_DATETIME":0,"O_TIMESTAMP":"1970-01-01T00:00:01Z","O_JSON":"{\"k1\": \"v1\", \"k2\": 11}"},"source":{"version":"1.9.7.Final","connector":"mysql","name":"RW_CDC_test.orders","ts_ms":1678088861000,"snapshot":"false","db":"test","sequence":null,"table":"orders","server_id":223344,"gtid":null,"file":"mysql-bin.000003","pos":789,"row":0,"thread":4,"query":null},"op":"c","ts_ms":1678088861249,"transaction":null}}"#;
423
424            let columns = get_test2_columns();
425            let parser = build_parser(columns.clone()).await;
426            let [(op, row)]: [_; 1] = parse_one(parser, columns, data.to_vec())
427                .await
428                .try_into()
429                .unwrap();
430            assert_eq!(op, Op::Insert);
431
432            assert!(row[0].eq(&Some(ScalarImpl::Int64(111))));
433            assert!(row[1].eq(&Some(ScalarImpl::Bool(true))));
434            assert!(row[2].eq(&Some(ScalarImpl::Int16(-1))));
435            assert!(row[3].eq(&Some(ScalarImpl::Int32(-1111))));
436            assert!(row[4].eq(&Some(ScalarImpl::Float32((-11.11).into()))));
437            assert!(row[5].eq(&Some(ScalarImpl::Float64((-111.11111).into()))));
438            assert!(row[6].eq(&Some(ScalarImpl::Decimal("-111.11".parse().unwrap()))));
439            assert!(row[7].eq(&Some(ScalarImpl::Utf8("yes please".into()))));
440            assert!(row[8].eq(&Some(ScalarImpl::Date(Date::new(
441                NaiveDate::from_ymd_opt(1000, 1, 1).unwrap()
442            )))));
443            assert!(row[9].eq(&Some(ScalarImpl::Time(Time::new(
444                NaiveTime::from_hms_micro_opt(0, 0, 0, 0).unwrap()
445            )))));
446            assert!(row[10].eq(&Some(ScalarImpl::Timestamp(Timestamp::new(
447                "1970-01-01T00:00:00".parse().unwrap()
448            )))));
449            assert!(row[11].eq(&Some(ScalarImpl::Timestamptz(
450                "1970-01-01T00:00:01Z".parse().unwrap()
451            ))));
452            assert_json_eq(&row[12], "{\"k1\": \"v1\", \"k2\": 11}");
453        }
454
455        #[tokio::test]
456        async fn test2_debezium_json_parser_delete() {
457            let data = br#"{"payload":{"before":{"O_KEY":111,"O_BOOL":0,"O_TINY":3,"O_INT":3333,"O_REAL":33.33,"O_DOUBLE":333.33333,"O_DECIMAL":333.33,"O_CHAR":"no thanks","O_DATE":"9999-12-31","O_TIME":86399000000,"O_DATETIME":99999999999000,"O_TIMESTAMP":"2038-01-09T03:14:07Z","O_JSON":"{\"k1\":\"v1_updated\",\"k2\":33}"},"after":null,"source":{"version":"1.9.7.Final","connector":"mysql","name":"RW_CDC_test.orders","ts_ms":1678090653000,"snapshot":"false","db":"test","sequence":null,"table":"orders","server_id":223344,"gtid":null,"file":"mysql-bin.000003","pos":1643,"row":0,"thread":4,"query":null},"op":"d","ts_ms":1678090653611,"transaction":null}}"#;
458
459            let columns = get_test2_columns();
460            let parser = build_parser(columns.clone()).await;
461            let [(op, row)]: [_; 1] = parse_one(parser, columns, data.to_vec())
462                .await
463                .try_into()
464                .unwrap();
465
466            assert_eq!(op, Op::Delete);
467
468            assert!(row[0].eq(&Some(ScalarImpl::Int64(111))));
469            assert!(row[1].eq(&Some(ScalarImpl::Bool(false))));
470            assert!(row[2].eq(&Some(ScalarImpl::Int16(3))));
471            assert!(row[3].eq(&Some(ScalarImpl::Int32(3333))));
472            assert!(row[4].eq(&Some(ScalarImpl::Float32((33.33).into()))));
473            assert!(row[5].eq(&Some(ScalarImpl::Float64((333.33333).into()))));
474            assert!(row[6].eq(&Some(ScalarImpl::Decimal("333.33".parse().unwrap()))));
475            assert!(row[7].eq(&Some(ScalarImpl::Utf8("no thanks".into()))));
476            assert!(row[8].eq(&Some(ScalarImpl::Date(Date::new(
477                NaiveDate::from_ymd_opt(9999, 12, 31).unwrap()
478            )))));
479            assert!(row[9].eq(&Some(ScalarImpl::Time(Time::new(
480                NaiveTime::from_hms_micro_opt(23, 59, 59, 0).unwrap()
481            )))));
482            assert!(row[10].eq(&Some(ScalarImpl::Timestamp(Timestamp::new(
483                "5138-11-16T09:46:39".parse().unwrap()
484            )))));
485            assert!(row[11].eq(&Some(ScalarImpl::Timestamptz(
486                "2038-01-09T03:14:07Z".parse().unwrap()
487            ))));
488            assert_json_eq(&row[12], "{\"k1\":\"v1_updated\",\"k2\":33}");
489        }
490
491        #[tokio::test]
492        async fn test2_debezium_json_parser_update() {
493            let data = br#"{"payload":{"before":{"O_KEY":111,"O_BOOL":1,"O_TINY":-1,"O_INT":-1111,"O_REAL":-11.11,"O_DOUBLE":-111.11111,"O_DECIMAL":-111.11,"O_CHAR":"yes please","O_DATE":"1000-01-01","O_TIME":0,"O_DATETIME":0,"O_TIMESTAMP":"1970-01-01T00:00:01Z","O_JSON":"{\"k1\": \"v1\", \"k2\": 11}"},"after":{"O_KEY":111,"O_BOOL":0,"O_TINY":3,"O_INT":3333,"O_REAL":33.33,"O_DOUBLE":333.33333,"O_DECIMAL":333.33,"O_CHAR":"no thanks","O_DATE":"9999-12-31","O_TIME":86399000000,"O_DATETIME":99999999999000,"O_TIMESTAMP":"2038-01-09T03:14:07Z","O_JSON":"{\"k1\": \"v1_updated\", \"k2\": 33}"},"source":{"version":"1.9.7.Final","connector":"mysql","name":"RW_CDC_test.orders","ts_ms":1678089331000,"snapshot":"false","db":"test","sequence":null,"table":"orders","server_id":223344,"gtid":null,"file":"mysql-bin.000003","pos":1168,"row":0,"thread":4,"query":null},"op":"u","ts_ms":1678089331464,"transaction":null}}"#;
494
495            let columns = get_test2_columns();
496
497            let parser = build_parser(columns.clone()).await;
498            let [(op, row)]: [_; 1] = parse_one(parser, columns, data.to_vec())
499                .await
500                .try_into()
501                .unwrap();
502
503            assert_eq!(op, Op::Insert);
504
505            assert!(row[0].eq(&Some(ScalarImpl::Int64(111))));
506            assert!(row[1].eq(&Some(ScalarImpl::Bool(false))));
507            assert!(row[2].eq(&Some(ScalarImpl::Int16(3))));
508            assert!(row[3].eq(&Some(ScalarImpl::Int32(3333))));
509            assert!(row[4].eq(&Some(ScalarImpl::Float32((33.33).into()))));
510            assert!(row[5].eq(&Some(ScalarImpl::Float64((333.33333).into()))));
511            assert!(row[6].eq(&Some(ScalarImpl::Decimal("333.33".parse().unwrap()))));
512            assert!(row[7].eq(&Some(ScalarImpl::Utf8("no thanks".into()))));
513            assert!(row[8].eq(&Some(ScalarImpl::Date(Date::new(
514                NaiveDate::from_ymd_opt(9999, 12, 31).unwrap()
515            )))));
516            assert!(row[9].eq(&Some(ScalarImpl::Time(Time::new(
517                NaiveTime::from_hms_micro_opt(23, 59, 59, 0).unwrap()
518            )))));
519            assert!(row[10].eq(&Some(ScalarImpl::Timestamp(Timestamp::new(
520                "5138-11-16T09:46:39".parse().unwrap()
521            )))));
522            assert!(row[11].eq(&Some(ScalarImpl::Timestamptz(
523                "2038-01-09T03:14:07Z".parse().unwrap()
524            ))));
525            assert_json_eq(&row[12], "{\"k1\": \"v1_updated\", \"k2\": 33}");
526        }
527
528        #[cfg(not(madsim))] // Traced test does not work with madsim
529        #[tokio::test]
530        #[tracing_test::traced_test]
531        async fn test2_debezium_json_parser_overflow() {
532            let columns = vec![
533                SourceColumnDesc::simple("O_KEY", DataType::Int64, ColumnId::from(0)),
534                SourceColumnDesc::simple("O_BOOL", DataType::Boolean, ColumnId::from(1)),
535                SourceColumnDesc::simple("O_TINY", DataType::Int16, ColumnId::from(2)),
536                SourceColumnDesc::simple("O_INT", DataType::Int32, ColumnId::from(3)),
537                SourceColumnDesc::simple("O_REAL", DataType::Float32, ColumnId::from(4)),
538                SourceColumnDesc::simple("O_DOUBLE", DataType::Float64, ColumnId::from(5)),
539            ];
540            let mut parser = build_parser(columns.clone()).await;
541
542            let mut dummy_builder =
543                SourceStreamChunkBuilder::new(columns, SourceCtrlOpts::for_test());
544
545            let normal_values = ["111", "1", "33", "444", "555.0", "666.0"];
546            let overflow_values = [
547                "9223372036854775808",
548                "2",
549                "32768",
550                "2147483648",
551                "3.80282347E38",
552                "1.797695E308",
553            ];
554
555            for i in 0..6 {
556                let mut values = normal_values;
557                values[i] = overflow_values[i];
558                let data = format!(
559                    r#"{{"payload":{{"before":null,"after":{{"O_KEY":{},"O_BOOL":{},"O_TINY":{},"O_INT":{},"O_REAL":{},"O_DOUBLE":{}}},"source":{{"version":"1.9.7.Final","connector":"mysql","name":"RW_CDC_test.orders","ts_ms":1678158055000,"snapshot":"false","db":"test","sequence":null,"table":"orders","server_id":223344,"gtid":null,"file":"mysql-bin.000003","pos":637,"row":0,"thread":4,"query":null}},"op":"c","ts_ms":1678158055464,"transaction":null}}}}"#,
560                    values[0], values[1], values[2], values[3], values[4], values[5]
561                ).as_bytes().to_vec();
562
563                let res = parser
564                    .parse_inner(None, Some(data), dummy_builder.row_writer())
565                    .await;
566                if i < 5 {
567                    // For other overflow, the parsing succeeds but the type conversion fails
568                    // The errors are ignored and logged.
569                    res.unwrap();
570                    assert!(logs_contain("expected type"), "{i}");
571                } else {
572                    // For f64 overflow, the parsing fails
573                    let e = res.unwrap_err();
574                    assert!(e.to_report_string().contains("InvalidNumber"), "{i}: {e}");
575                }
576            }
577        }
578    }
579
580    // postgres-specific data-type mapping tests
581    mod test3_postgres {
582        use risingwave_pb::plan_common::AdditionalColumn;
583
584        use super::*;
585        use crate::source::SourceColumnType;
586
587        // schema for temporal-type test
588        fn get_temporal_test_columns() -> Vec<SourceColumnDesc> {
589            vec![
590                SourceColumnDesc::simple("o_key", DataType::Int32, ColumnId::from(0)),
591                SourceColumnDesc::simple("o_time_0", DataType::Time, ColumnId::from(1)),
592                SourceColumnDesc::simple("o_time_6", DataType::Time, ColumnId::from(2)),
593                SourceColumnDesc::simple("o_timez_0", DataType::Time, ColumnId::from(3)),
594                SourceColumnDesc::simple("o_timez_6", DataType::Time, ColumnId::from(4)),
595                SourceColumnDesc::simple("o_timestamp_0", DataType::Timestamp, ColumnId::from(5)),
596                SourceColumnDesc::simple("o_timestamp_6", DataType::Timestamp, ColumnId::from(6)),
597                SourceColumnDesc::simple(
598                    "o_timestampz_0",
599                    DataType::Timestamptz,
600                    ColumnId::from(7),
601                ),
602                SourceColumnDesc::simple(
603                    "o_timestampz_6",
604                    DataType::Timestamptz,
605                    ColumnId::from(8),
606                ),
607                SourceColumnDesc::simple("o_interval", DataType::Interval, ColumnId::from(9)),
608                SourceColumnDesc::simple("o_date", DataType::Date, ColumnId::from(10)),
609            ]
610        }
611
612        // schema for numeric-type test
613        fn get_numeric_test_columns() -> Vec<SourceColumnDesc> {
614            vec![
615                SourceColumnDesc::simple("o_key", DataType::Int32, ColumnId::from(0)),
616                SourceColumnDesc::simple("o_smallint", DataType::Int16, ColumnId::from(1)),
617                SourceColumnDesc::simple("o_integer", DataType::Int32, ColumnId::from(2)),
618                SourceColumnDesc::simple("o_bigint", DataType::Int64, ColumnId::from(3)),
619                SourceColumnDesc::simple("o_real", DataType::Float32, ColumnId::from(4)),
620                SourceColumnDesc::simple("o_double", DataType::Float64, ColumnId::from(5)),
621                SourceColumnDesc::simple("o_numeric", DataType::Decimal, ColumnId::from(6)),
622                SourceColumnDesc::simple("o_numeric_6_3", DataType::Decimal, ColumnId::from(7)),
623                SourceColumnDesc::simple("o_money", DataType::Decimal, ColumnId::from(8)),
624            ]
625        }
626
627        // schema for the remaining types
628        fn get_other_types_test_columns() -> Vec<SourceColumnDesc> {
629            vec![
630                SourceColumnDesc::simple("o_key", DataType::Int32, ColumnId::from(0)),
631                SourceColumnDesc::simple("o_boolean", DataType::Boolean, ColumnId::from(1)),
632                SourceColumnDesc::simple("o_bit", DataType::Boolean, ColumnId::from(2)),
633                SourceColumnDesc::simple("o_bytea", DataType::Bytea, ColumnId::from(3)),
634                SourceColumnDesc::simple("o_json", DataType::Jsonb, ColumnId::from(4)),
635                SourceColumnDesc::simple("o_xml", DataType::Varchar, ColumnId::from(5)),
636                SourceColumnDesc::simple("o_uuid", DataType::Varchar, ColumnId::from(6)),
637                SourceColumnDesc {
638                    name: "o_point".to_owned(),
639                    data_type: DataType::Struct(StructType::new(vec![
640                        ("x", DataType::Float32),
641                        ("y", DataType::Float32),
642                    ])),
643                    column_id: 7.into(),
644                    column_type: SourceColumnType::Normal,
645                    is_pk: false,
646                    is_hidden_addition_col: false,
647                    additional_column: AdditionalColumn { column_type: None },
648                },
649                SourceColumnDesc::simple("o_enum", DataType::Varchar, ColumnId::from(8)),
650                SourceColumnDesc::simple("o_char", DataType::Varchar, ColumnId::from(9)),
651                SourceColumnDesc::simple("o_varchar", DataType::Varchar, ColumnId::from(10)),
652                SourceColumnDesc::simple("o_character", DataType::Varchar, ColumnId::from(11)),
653                SourceColumnDesc::simple(
654                    "o_character_varying",
655                    DataType::Varchar,
656                    ColumnId::from(12),
657                ),
658            ]
659        }
660
661        #[tokio::test]
662        async fn test_temporal_types() {
663            // this test includes all supported temporal types, with the schema
664            // CREATE TABLE orders (
665            //     o_key integer,
666            //     o_time_0 time(0),
667            //     o_time_6 time(6),
668            //     o_timez_0 time(0) with time zone,
669            //     o_timez_6 time(6) with time zone,
670            //     o_timestamp_0 timestamp(0),
671            //     o_timestamp_6 timestamp(6),
672            //     o_timestampz_0 timestamp(0) with time zone,
673            //     o_timestampz_6 timestamp(6) with time zone,
674            //     o_interval interval,
675            //     o_date date,
676            //     PRIMARY KEY (o_key)
677            // );
678            // this test covers an insert event on the table above
679            let data = br#"{"payload":{"before":null,"after":{"o_key":0,"o_time_0":40271000000,"o_time_6":40271000010,"o_timez_0":"11:11:11Z","o_timez_6":"11:11:11.00001Z","o_timestamp_0":1321009871000,"o_timestamp_6":1321009871123456,"o_timestampz_0":"2011-11-11T03:11:11Z","o_timestampz_6":"2011-11-11T03:11:11.123456Z","o_interval":"P1Y2M3DT4H5M6.78S","o_date":"1999-09-09"},"source":{"version":"1.9.7.Final","connector":"postgresql","name":"RW_CDC_localhost.test.orders","ts_ms":1684733351963,"snapshot":"last","db":"test","sequence":"[null,\"26505352\"]","schema":"public","table":"orders","txId":729,"lsn":26505352,"xmin":null},"op":"r","ts_ms":1684733352110,"transaction":null}}"#;
680            let columns = get_temporal_test_columns();
681            let parser = build_parser(columns.clone()).await;
682            let [(op, row)]: [_; 1] = parse_one(parser, columns, data.to_vec())
683                .await
684                .try_into()
685                .unwrap();
686            assert_eq!(op, Op::Insert);
687            assert!(row[0].eq(&Some(ScalarImpl::Int32(0))));
688            assert!(row[1].eq(&Some(ScalarImpl::Time(Time::new(
689                NaiveTime::from_hms_micro_opt(11, 11, 11, 0).unwrap()
690            )))));
691            assert!(row[2].eq(&Some(ScalarImpl::Time(Time::new(
692                NaiveTime::from_hms_micro_opt(11, 11, 11, 10).unwrap()
693            )))));
694            assert!(row[3].eq(&Some(ScalarImpl::Time(Time::new(
695                NaiveTime::from_hms_micro_opt(11, 11, 11, 0).unwrap()
696            )))));
697            assert!(row[4].eq(&Some(ScalarImpl::Time(Time::new(
698                NaiveTime::from_hms_micro_opt(11, 11, 11, 10).unwrap()
699            )))));
700            assert!(row[5].eq(&Some(ScalarImpl::Timestamp(Timestamp::new(
701                "2011-11-11T11:11:11".parse().unwrap()
702            )))));
703            assert!(row[6].eq(&Some(ScalarImpl::Timestamp(Timestamp::new(
704                "2011-11-11T11:11:11.123456".parse().unwrap()
705            )))));
706            assert!(
707                row[9].eq(&Some(ScalarImpl::Interval(Interval::from_month_day_usec(
708                    14,
709                    3,
710                    14706780000
711                ))))
712            );
713            assert!(row[10].eq(&Some(ScalarImpl::Date(Date::new(
714                NaiveDate::from_ymd_opt(1999, 9, 9).unwrap()
715            )))));
716        }
717
718        #[tokio::test]
719        async fn test_numeric_types() {
720            // this test includes all supported numeric types, with the schema
721            // CREATE TABLE orders (
722            //     o_key integer,
723            //     o_smallint smallint,
724            //     o_integer integer,
725            //     o_bigint bigint,
726            //     o_real real,
727            //     o_double double precision,
728            //     o_numeric numeric,
729            //     o_numeric_6_3 numeric(6,3),
730            //     o_money money,
731            //     PRIMARY KEY (o_key)
732            // );
733            // this test covers an insert event on the table above
734            let data = br#"{"payload":{"before":null,"after":{"o_key":0,"o_smallint":32767,"o_integer":2147483647,"o_bigint":9223372036854775807,"o_real":9.999,"o_double":9.999999,"o_numeric":123456.789,"o_numeric_6_3":123.456,"o_money":123.12},"source":{"version":"1.9.7.Final","connector":"postgresql","name":"RW_CDC_localhost.test.orders","ts_ms":1684404343201,"snapshot":"last","db":"test","sequence":"[null,\"26519216\"]","schema":"public","table":"orders","txId":729,"lsn":26519216,"xmin":null},"op":"r","ts_ms":1684404343349,"transaction":null}}"#;
735            let columns = get_numeric_test_columns();
736            let parser = build_parser(columns.clone()).await;
737            let [(op, row)]: [_; 1] = parse_one(parser, columns, data.to_vec())
738                .await
739                .try_into()
740                .unwrap();
741            assert_eq!(op, Op::Insert);
742            assert!(row[0].eq(&Some(ScalarImpl::Int32(0))));
743            assert!(row[1].eq(&Some(ScalarImpl::Int16(32767))));
744            assert!(row[2].eq(&Some(ScalarImpl::Int32(2147483647))));
745            assert!(row[3].eq(&Some(ScalarImpl::Int64(9223372036854775807))));
746            assert!(row[4].eq(&Some(ScalarImpl::Float32((9.999).into()))));
747            assert!(row[5].eq(&Some(ScalarImpl::Float64((9.999999).into()))));
748            assert!(row[6].eq(&Some(ScalarImpl::Decimal("123456.7890".parse().unwrap()))));
749            assert!(row[7].eq(&Some(ScalarImpl::Decimal("123.456".parse().unwrap()))));
750            assert!(row[8].eq(&Some(ScalarImpl::Decimal("123.12".parse().unwrap()))));
751        }
752
753        #[tokio::test]
754        async fn test_other_types() {
755            // this test includes the remaining types, with the schema
756            // CREATE TABLE orders (
757            //     o_key integer,
758            //     o_boolean boolean,
759            //     o_bit bit,
760            //     o_bytea bytea,
761            //     o_json jsonb,
762            //     o_xml xml,
763            //     o_uuid uuid,
764            //     o_point point,
765            //     o_enum bear,
766            //     o_char char,
767            //     o_varchar varchar,
768            //     o_character character,
769            //     o_character_varying character varying,
770            //     PRIMARY KEY (o_key)
771            //  );
772            // this test covers an insert event on the table above
773            let data = br#"{"payload":{"before":null,"after":{"o_key":1,"o_boolean":false,"o_bit":true,"o_bytea":"ASNFZ4mrze8=","o_json":"{\"k1\": \"v1\", \"k2\": 11}","o_xml":"<!--hahaha-->","o_uuid":"60f14fe2-f857-404a-b586-3b5375b3259f","o_point":{"x":1.0,"y":2.0,"wkb":"AQEAAAAAAAAAAADwPwAAAAAAAABA","srid":null},"o_enum":"polar","o_char":"h","o_varchar":"ha","o_character":"h","o_character_varying":"hahaha"},"source":{"version":"1.9.7.Final","connector":"postgresql","name":"RW_CDC_localhost.test.orders","ts_ms":1684743927178,"snapshot":"last","db":"test","sequence":"[null,\"26524528\"]","schema":"public","table":"orders","txId":730,"lsn":26524528,"xmin":null},"op":"r","ts_ms":1684743927343,"transaction":null}}"#;
774            let columns = get_other_types_test_columns();
775            let parser = build_parser(columns.clone()).await;
776            let [(op, row)]: [_; 1] = parse_one(parser, columns, data.to_vec())
777                .await
778                .try_into()
779                .unwrap();
780            assert_eq!(op, Op::Insert);
781            assert!(row[0].eq(&Some(ScalarImpl::Int32(1))));
782            assert!(row[1].eq(&Some(ScalarImpl::Bool(false))));
783            assert!(row[2].eq(&Some(ScalarImpl::Bool(true))));
784            assert!(row[3].eq(&Some(ScalarImpl::Bytea(Box::new([
785                u8::from_str_radix("01", 16).unwrap(),
786                u8::from_str_radix("23", 16).unwrap(),
787                u8::from_str_radix("45", 16).unwrap(),
788                u8::from_str_radix("67", 16).unwrap(),
789                u8::from_str_radix("89", 16).unwrap(),
790                u8::from_str_radix("AB", 16).unwrap(),
791                u8::from_str_radix("CD", 16).unwrap(),
792                u8::from_str_radix("EF", 16).unwrap()
793            ])))));
794            assert_json_eq(&row[4], "{\"k1\": \"v1\", \"k2\": 11}");
795            assert!(row[5].eq(&Some(ScalarImpl::Utf8("<!--hahaha-->".into()))));
796            assert!(row[6].eq(&Some(ScalarImpl::Utf8(
797                "60f14fe2-f857-404a-b586-3b5375b3259f".into()
798            ))));
799            assert!(row[7].eq(&Some(ScalarImpl::Struct(StructValue::new(vec![
800                Some(ScalarImpl::Float32(1.into())),
801                Some(ScalarImpl::Float32(2.into()))
802            ])))));
803            assert!(row[8].eq(&Some(ScalarImpl::Utf8("polar".into()))));
804            assert!(row[9].eq(&Some(ScalarImpl::Utf8("h".into()))));
805            assert!(row[10].eq(&Some(ScalarImpl::Utf8("ha".into()))));
806            assert!(row[11].eq(&Some(ScalarImpl::Utf8("h".into()))));
807            assert!(row[12].eq(&Some(ScalarImpl::Utf8("hahaha".into()))));
808        }
809    }
810}
risingwave_connector/parser/debezium/simd_json_parser.rs

risingwave_connector/parser/debezium/
simd_json_parser.rs
