risingwave_connector/parser/
mod.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::collections::HashMap;
16use std::fmt::Debug;
17use std::sync::LazyLock;
18
19use anyhow::{Context, bail};
20use auto_enums::auto_enum;
21pub use avro::AvroParserConfig;
22pub use canal::*;
23pub use chunk_builder::{SourceStreamChunkBuilder, SourceStreamChunkRowWriter};
24use csv_parser::CsvParser;
25pub use debezium::*;
26use futures::{Future, Stream, StreamExt, TryFutureExt, TryStreamExt};
27use futures_async_stream::try_stream;
28pub use json_parser::*;
29pub use parquet_parser::ParquetParser;
30pub use protobuf::*;
31use risingwave_common::catalog::{
32    CDC_TABLE_NAME_COLUMN_NAME, Field, KAFKA_TIMESTAMP_COLUMN_NAME, Schema,
33};
34use risingwave_common::log::LogSuppressor;
35use risingwave_common::metrics::GLOBAL_ERROR_METRICS;
36use risingwave_common::row::OwnedRow;
37use risingwave_common::types::{Datum, DatumCow, DatumRef};
38use risingwave_common::util::tracing::InstrumentStream;
39use risingwave_connector_codec::decoder::avro::MapHandling;
40use thiserror_ext::AsReport;
41
42pub use self::mysql::{
43    mysql_datum_to_rw_datum, mysql_row_to_owned_row, mysql_row_to_owned_row_with_strict_pk,
44};
45use self::plain_parser::PlainParser;
46pub use self::postgres::{
47    postgres_cell_to_scalar_impl, postgres_cell_to_scalar_impl_strict, postgres_row_to_owned_row,
48    postgres_row_to_owned_row_with_strict_pk,
49};
50pub use self::sql_server::{
51    ScalarImplTiberiusWrapper, sql_server_row_to_owned_row,
52    sql_server_row_to_owned_row_with_strict_pk,
53};
54pub use self::unified::Access;
55pub use self::unified::json::{
56    BigintUnsignedHandlingMode, JsonAccess, TimeHandling, TimestampHandling, TimestamptzHandling,
57};
58use self::upsert_parser::UpsertParser;
59use crate::error::ConnectorResult;
60use crate::parser::maxwell::MaxwellParser;
61use crate::schema::schema_registry::SchemaRegistryConfig;
62use crate::source::monitor::GLOBAL_SOURCE_METRICS;
63use crate::source::{
64    BoxSourceMessageEventStream, SourceChunkStream, SourceColumnDesc, SourceColumnType,
65    SourceContext, SourceContextRef, SourceCtrlOpts, SourceMessageEvent, SourceMeta,
66    SourceReaderEvent,
67};
68
69fn decode_row_with_strict_pk(
70    connector_name: &str,
71    schema: &Schema,
72    pk_indices: &[usize],
73    mut decode: impl FnMut(usize, &Field) -> anyhow::Result<Datum>,
74    mut log_non_pk_error: impl FnMut(&str, anyhow::Error),
75) -> anyhow::Result<OwnedRow> {
76    if let Some(index) = pk_indices
77        .iter()
78        .copied()
79        .find(|index| *index >= schema.fields.len())
80    {
81        bail!(
82            "{connector_name} snapshot primary-key index {index} is out of bounds for {} columns",
83            schema.fields.len()
84        );
85    }
86
87    let mut datums = Vec::with_capacity(schema.fields.len());
88    for (index, field) in schema.fields.iter().enumerate() {
89        let is_pk = pk_indices.contains(&index);
90        let decode_result = decode(index, field);
91        let datum = if is_pk {
92            decode_result.with_context(|| {
93                format!(
94                    "failed to decode {connector_name} snapshot primary key `{}`",
95                    field.name
96                )
97            })?
98        } else {
99            match decode_result {
100                Ok(datum) => datum,
101                Err(err) => {
102                    log_non_pk_error(&field.name, err);
103                    None
104                }
105            }
106        };
107        if is_pk && datum.is_none() {
108            bail!(
109                "{connector_name} snapshot primary key `{}` cannot be NULL",
110                field.name
111            );
112        }
113        datums.push(datum);
114    }
115    Ok(OwnedRow::new(datums))
116}
117
118#[cfg(test)]
119mod strict_pk_tests {
120    use anyhow::anyhow;
121    use risingwave_common::row::Row;
122    use risingwave_common::types::{DataType, ScalarImpl};
123
124    use super::*;
125
126    fn test_schema() -> Schema {
127        Schema::new(vec![
128            Field::with_name(DataType::Int32, "id"),
129            Field::with_name(DataType::Int32, "payload"),
130        ])
131    }
132
133    #[test]
134    fn strict_pk_propagates_decode_error() {
135        let err = decode_row_with_strict_pk(
136            "test",
137            &test_schema(),
138            &[0],
139            |index, _| {
140                if index == 0 {
141                    Err(anyhow!("malformed integer"))
142                } else {
143                    Ok(Some(ScalarImpl::Int32(1)))
144                }
145            },
146            |_, _| unreachable!(),
147        )
148        .unwrap_err();
149
150        assert!(err.to_string().contains("snapshot primary key `id`"));
151        assert!(format!("{err:#}").contains("malformed integer"));
152    }
153
154    #[test]
155    fn strict_pk_rejects_decoded_null() {
156        let err = decode_row_with_strict_pk(
157            "test",
158            &test_schema(),
159            &[0],
160            |_, _| Ok(None),
161            |_, _| unreachable!(),
162        )
163        .unwrap_err();
164
165        assert!(
166            err.to_string()
167                .contains("snapshot primary key `id` cannot be NULL")
168        );
169    }
170
171    #[test]
172    fn strict_pk_keeps_non_pk_decode_lenient() {
173        let mut logged_columns = Vec::new();
174        let row = decode_row_with_strict_pk(
175            "test",
176            &test_schema(),
177            &[0],
178            |index, _| {
179                if index == 0 {
180                    Ok(Some(ScalarImpl::Int32(1)))
181                } else {
182                    Err(anyhow!("malformed payload"))
183                }
184            },
185            |name, _| logged_columns.push(name.to_owned()),
186        )
187        .unwrap();
188
189        assert!(row.datum_at(0).is_some());
190        assert!(row.datum_at(1).is_none());
191        assert_eq!(logged_columns, ["payload"]);
192    }
193}
194
195mod access_builder;
196pub mod additional_columns;
197mod avro;
198mod bytes_parser;
199mod canal;
200mod chunk_builder;
201mod config;
202mod csv_parser;
203mod debezium;
204mod json_parser;
205mod maxwell;
206mod mysql;
207pub mod parquet_parser;
208pub mod plain_parser;
209mod postgres;
210mod protobuf;
211pub mod scalar_adapter;
212mod sql_server;
213mod unified;
214mod upsert_parser;
215mod utils;
216
217use access_builder::{AccessBuilder, AccessBuilderImpl};
218pub use config::*;
219
220pub(crate) fn into_data_chunk_stream(
221    stream: impl Stream<Item = ConnectorResult<SourceReaderEvent>> + Send + 'static,
222) -> impl SourceChunkStream {
223    stream
224        .try_filter_map(|event| async move {
225            Ok(match event {
226                SourceReaderEvent::DataChunk(chunk) => Some(chunk),
227                SourceReaderEvent::SplitProgress(_) => None,
228            })
229        })
230        .boxed()
231}
232pub use debezium::DEBEZIUM_IGNORE_KEY;
233use debezium::schema_change::SchemaChangeEnvelope;
234pub use unified::{AccessError, AccessResult};
235
236/// The meta data of the original message for a row writer.
237///
238/// Extracted from the `SourceMessage`.
239#[derive(Clone, Copy, Debug)]
240pub struct MessageMeta<'a> {
241    source_meta: &'a SourceMeta,
242    split_id: &'a str,
243    offset: &'a str,
244}
245
246impl<'a> MessageMeta<'a> {
247    /// Extract the value for the given column.
248    ///
249    /// Returns `None` if the column is not a meta column.
250    fn value_for_column(self, desc: &SourceColumnDesc) -> DatumRef<'a> {
251        let datum: DatumRef<'_> = match desc.column_type {
252            // Row id columns are filled with `NULL` here and will be filled with the real
253            // row id generated by `RowIdGenExecutor` later.
254            SourceColumnType::RowId => None,
255            // Extract the offset from the meta data.
256            SourceColumnType::Offset => Some(self.offset.into()),
257            // Extract custom meta data per connector.
258            SourceColumnType::Meta if let SourceMeta::Kafka(kafka_meta) = self.source_meta => {
259                assert_eq!(
260                    desc.name.as_str(),
261                    KAFKA_TIMESTAMP_COLUMN_NAME,
262                    "unexpected kafka meta column name"
263                );
264                kafka_meta.extract_timestamp()
265            }
266            SourceColumnType::Meta if let SourceMeta::DebeziumCdc(cdc_meta) = self.source_meta => {
267                assert_eq!(
268                    desc.name.as_str(),
269                    CDC_TABLE_NAME_COLUMN_NAME,
270                    "unexpected cdc meta column name"
271                );
272                // For CDC sources, extract_table_name() strips the database prefix:
273                cdc_meta.extract_table_name()
274            }
275
276            // For other cases, return `None`.
277            SourceColumnType::Meta | SourceColumnType::Normal => return None,
278        };
279
280        datum
281    }
282}
283
284/// Transaction control message. Currently only used by Debezium messages.
285#[derive(Debug)]
286pub enum TransactionControl {
287    Begin { id: Box<str> },
288    Commit { id: Box<str> },
289}
290
291/// The result of parsing a message.
292#[derive(Debug)]
293pub enum ParseResult {
294    /// Some rows are parsed and written to the [`SourceStreamChunkRowWriter`].
295    Rows,
296    /// A transaction control message is parsed.
297    TransactionControl(TransactionControl),
298
299    /// A schema change message is parsed.
300    SchemaChange(SchemaChangeEnvelope),
301}
302
303#[derive(Clone, Copy, Debug, PartialEq)]
304pub enum ParserFormat {
305    CanalJson,
306    Csv,
307    Json,
308    Maxwell,
309    Debezium,
310    DebeziumMongo,
311    Upsert,
312    Plain,
313}
314
315/// `ByteStreamSourceParser` is the entrypoint abstraction for parsing messages.
316/// It consumes bytes of one individual message and produces parsed records.
317///
318/// It's used by [`ByteStreamSourceParserImpl::parse_stream_with_events`]. `pub` is for benchmark only.
319pub trait ByteStreamSourceParser: Send + Debug + Sized + 'static {
320    /// The column descriptors of the output chunk.
321    fn columns(&self) -> &[SourceColumnDesc];
322
323    /// The source context, used to report parsing error.
324    fn source_ctx(&self) -> &SourceContext;
325
326    /// The format of the specific parser.
327    fn parser_format(&self) -> ParserFormat;
328
329    /// Parse one record from the given `payload` and write rows to the `writer`.
330    ///
331    /// Returns error if **any** of the rows in the message failed to parse.
332    fn parse_one<'a>(
333        &'a mut self,
334        key: Option<Vec<u8>>,
335        payload: Option<Vec<u8>>,
336        writer: SourceStreamChunkRowWriter<'a>,
337    ) -> impl Future<Output = ConnectorResult<()>> + Send + 'a;
338
339    /// Parse one record from the given `payload`, either write rows to the `writer` or interpret it
340    /// as a transaction control message.
341    ///
342    /// The default implementation forwards to [`ByteStreamSourceParser::parse_one`] for
343    /// non-transactional sources.
344    ///
345    /// Returns error if **any** of the rows in the message failed to parse.
346    fn parse_one_with_txn<'a>(
347        &'a mut self,
348        key: Option<Vec<u8>>,
349        payload: Option<Vec<u8>>,
350        writer: SourceStreamChunkRowWriter<'a>,
351    ) -> impl Future<Output = ConnectorResult<ParseResult>> + Send + 'a {
352        self.parse_one(key, payload, writer)
353            .map_ok(|_| ParseResult::Rows)
354    }
355}
356
357#[easy_ext::ext(SourceParserIntoStreamExt)]
358impl<P: ByteStreamSourceParser> P {
359    pub fn parse_stream_with_events(
360        self,
361        msg_stream: BoxSourceMessageEventStream,
362    ) -> impl Stream<Item = ConnectorResult<SourceReaderEvent>> + Send {
363        let actor_id = self.source_ctx().actor_id;
364        let source_id = self.source_ctx().source_id.as_raw_id();
365
366        // The stream will be long-lived. We use `instrument_with` here to create
367        // a new span for the polling of each chunk.
368        let source_ctrl_opts = self.source_ctx().source_ctrl_opts;
369        parse_message_stream(self, msg_stream, source_ctrl_opts).instrument_with(
370            move || tracing::info_span!("source_parse_chunk", %actor_id, source_id),
371        )
372    }
373}
374
375// TODO: when upsert is disabled, how to filter those empty payload
376// Currently, an err is returned for non upsert with empty payload
377#[try_stream(ok = SourceReaderEvent, error = crate::error::ConnectorError)]
378async fn parse_message_stream<P: ByteStreamSourceParser>(
379    mut parser: P,
380    msg_stream: BoxSourceMessageEventStream,
381    source_ctrl_opts: SourceCtrlOpts,
382) {
383    let mut chunk_builder =
384        SourceStreamChunkBuilder::new(parser.columns().to_vec(), source_ctrl_opts);
385
386    let mut direct_cdc_event_lag_latency_metrics = HashMap::new();
387
388    #[for_await]
389    for event in msg_stream {
390        // It's possible that the split is not active, which means the next batch may arrive
391        // very lately, so we should prefer emitting all records in current batch before the end
392        // of each iteration, instead of merging them with the next batch. An exception is when
393        // a transaction is not committed yet, in which yield when the transaction is committed.
394
395        let batch = match event? {
396            SourceMessageEvent::Data(batch) => batch,
397            SourceMessageEvent::SplitProgress(progress) => {
398                yield SourceReaderEvent::SplitProgress(progress);
399                continue;
400            }
401        };
402        let batch_len = batch.len();
403        if batch_len == 0 {
404            continue;
405        }
406
407        let mut txn_started_in_last_batch = chunk_builder.is_in_transaction();
408        let process_time_ms = chrono::Utc::now().timestamp_millis();
409        let mut is_heartbeat_emitted = false;
410        for msg in batch {
411            if msg.is_cdc_heartbeat() {
412                if !is_heartbeat_emitted {
413                    tracing::debug!(offset = msg.offset, "handling a heartbeat message");
414                    chunk_builder.heartbeat(MessageMeta {
415                        source_meta: &msg.meta,
416                        split_id: &msg.split_id,
417                        offset: &msg.offset,
418                    });
419                    for chunk in chunk_builder.consume_ready_chunks() {
420                        yield SourceReaderEvent::DataChunk(chunk);
421                    }
422                    is_heartbeat_emitted = true;
423                }
424                continue;
425            }
426
427            // calculate process_time - event_time lag
428            if let SourceMeta::DebeziumCdc(msg_meta) = &msg.meta {
429                let lag_ms = process_time_ms - msg_meta.source_ts_ms;
430                // report to promethus
431                let full_table_name = msg_meta.full_table_name.clone();
432                let direct_cdc_event_lag_latency = direct_cdc_event_lag_latency_metrics
433                    .entry(full_table_name)
434                    .or_insert_with(|| {
435                        GLOBAL_SOURCE_METRICS
436                            .direct_cdc_event_lag_latency
437                            .with_guarded_label_values(&[&msg_meta.full_table_name])
438                    });
439                direct_cdc_event_lag_latency.observe(lag_ms as f64);
440            }
441
442            // Parse the message and write to the chunk builder, it's possible that the message
443            // contains multiple rows. When the chunk size reached the limit during parsing, the
444            // chunk builder may yield the chunk to `ready_chunks` and start a new chunk.
445            match parser
446                .parse_one_with_txn(
447                    msg.key,
448                    msg.payload,
449                    chunk_builder.row_writer().with_meta(MessageMeta {
450                        source_meta: &msg.meta,
451                        split_id: &msg.split_id,
452                        offset: &msg.offset,
453                    }),
454                )
455                .await
456            {
457                // It's possible that parsing multiple rows in a single message PARTIALLY failed.
458                // We still have to maintain the row number in this case.
459                res @ (Ok(ParseResult::Rows) | Err(_)) => {
460                    if let Err(error) = res {
461                        // TODO: not using tracing span to provide `split_id` and `offset` due to performance concern,
462                        //       see #13105
463                        static LOG_SUPPRESSOR: LazyLock<LogSuppressor> =
464                            LazyLock::new(LogSuppressor::default);
465                        if let Ok(suppressed_count) = LOG_SUPPRESSOR.check() {
466                            tracing::error!(
467                                error = %error.as_report(),
468                                split_id = &*msg.split_id,
469                                offset = msg.offset,
470                                suppressed_count,
471                                "failed to parse message, skipping"
472                            );
473                        }
474
475                        // report to error metrics
476                        let context = parser.source_ctx();
477                        GLOBAL_ERROR_METRICS.user_source_error.report([
478                            error.variant_name().to_owned(),
479                            context.source_id.to_string(),
480                            context.source_name.clone(),
481                            context.fragment_id.to_string(),
482                        ]);
483                    }
484
485                    for chunk in chunk_builder.consume_ready_chunks() {
486                        yield SourceReaderEvent::DataChunk(chunk);
487                    }
488                }
489
490                Ok(ParseResult::TransactionControl(txn_ctl)) => match txn_ctl {
491                    TransactionControl::Begin { id } => {
492                        chunk_builder.begin_transaction(id);
493                    }
494                    TransactionControl::Commit { id } => {
495                        chunk_builder.commit_transaction(id);
496                        assert!(!chunk_builder.is_in_transaction());
497
498                        if txn_started_in_last_batch {
499                            // If a transaction is across multiple batches, we yield the chunk
500                            // immediately after the transaction is committed.
501                            chunk_builder.finish_current_chunk();
502                            txn_started_in_last_batch = false;
503                        }
504
505                        for chunk in chunk_builder.consume_ready_chunks() {
506                            yield SourceReaderEvent::DataChunk(chunk);
507                        }
508                    }
509                },
510
511                Ok(ParseResult::SchemaChange(schema_change)) => {
512                    if schema_change.is_empty() {
513                        continue;
514                    }
515
516                    let (oneshot_tx, oneshot_rx) = tokio::sync::oneshot::channel();
517                    // we bubble up the schema change event to the source executor via channel,
518                    // and wait for the source executor to finish the schema change process before
519                    // parsing the following messages.
520                    if let Some(ref tx) = parser.source_ctx().schema_change_tx {
521                        tx.send((schema_change, oneshot_tx))
522                            .await
523                            .expect("send schema change to executor");
524                        match oneshot_rx.await {
525                            Ok(()) => {}
526                            Err(e) => {
527                                tracing::error!(error = %e.as_report(), "failed to wait for schema change");
528                            }
529                        }
530                    }
531                }
532            }
533        }
534
535        // Finish the remaining records in the batch.
536        if !chunk_builder.is_in_transaction() {
537            chunk_builder.finish_current_chunk();
538        }
539        for chunk in chunk_builder.consume_ready_chunks() {
540            yield SourceReaderEvent::DataChunk(chunk);
541        }
542    }
543}
544
545#[derive(Debug)]
546pub enum EncodingType {
547    Key,
548    Value,
549}
550
551/// The entrypoint of parsing. It parses `SourceMessage` stream (byte stream) into [`risingwave_common::array::StreamChunk`] stream.
552/// Used by [`crate::source::into_chunk_stream`].
553#[derive(Debug)]
554pub enum ByteStreamSourceParserImpl {
555    Csv(CsvParser),
556    Debezium(DebeziumParser),
557    Plain(PlainParser),
558    Upsert(UpsertParser),
559    DebeziumMongoJson(DebeziumMongoJsonParser),
560    Maxwell(MaxwellParser),
561    CanalJson(CanalJsonParser),
562}
563
564impl ByteStreamSourceParserImpl {
565    pub fn parse_stream_with_events(
566        self,
567        msg_stream: BoxSourceMessageEventStream,
568    ) -> impl Stream<Item = ConnectorResult<SourceReaderEvent>> + Send {
569        #[auto_enum(futures03::Stream)]
570        let stream = match self {
571            Self::Csv(parser) => parser.parse_stream_with_events(msg_stream),
572            Self::Debezium(parser) => parser.parse_stream_with_events(msg_stream),
573            Self::DebeziumMongoJson(parser) => parser.parse_stream_with_events(msg_stream),
574            Self::Maxwell(parser) => parser.parse_stream_with_events(msg_stream),
575            Self::CanalJson(parser) => parser.parse_stream_with_events(msg_stream),
576            Self::Plain(parser) => parser.parse_stream_with_events(msg_stream),
577            Self::Upsert(parser) => parser.parse_stream_with_events(msg_stream),
578        };
579        Box::pin(stream)
580    }
581}
582
583impl ByteStreamSourceParserImpl {
584    pub async fn create(
585        parser_config: ParserConfig,
586        source_ctx: SourceContextRef,
587    ) -> ConnectorResult<Self> {
588        let CommonParserConfig { rw_columns } = parser_config.common;
589        let protocol = &parser_config.specific.protocol_config;
590        let encode = &parser_config.specific.encoding_config;
591        match (protocol, encode) {
592            (ProtocolProperties::Plain, EncodingProperties::Csv(config)) => {
593                CsvParser::new(rw_columns, *config, source_ctx).map(Self::Csv)
594            }
595            (ProtocolProperties::DebeziumMongo, EncodingProperties::MongoJson(props)) => {
596                DebeziumMongoJsonParser::new(rw_columns, source_ctx, props.clone())
597                    .map(Self::DebeziumMongoJson)
598            }
599            (ProtocolProperties::Canal, EncodingProperties::Json(config)) => {
600                CanalJsonParser::new(rw_columns, source_ctx, config).map(Self::CanalJson)
601            }
602            (ProtocolProperties::Native, _) => unreachable!("Native parser should not be created"),
603            (ProtocolProperties::Upsert, _) => {
604                let parser =
605                    UpsertParser::new(parser_config.specific, rw_columns, source_ctx).await?;
606                Ok(Self::Upsert(parser))
607            }
608            (ProtocolProperties::Plain, _) => {
609                let parser =
610                    PlainParser::new(parser_config.specific, rw_columns, source_ctx).await?;
611                Ok(Self::Plain(parser))
612            }
613            (ProtocolProperties::Debezium(_), _) => {
614                let parser =
615                    DebeziumParser::new(parser_config.specific, rw_columns, source_ctx).await?;
616                Ok(Self::Debezium(parser))
617            }
618            (ProtocolProperties::Maxwell, _) => {
619                let parser =
620                    MaxwellParser::new(parser_config.specific, rw_columns, source_ctx).await?;
621                Ok(Self::Maxwell(parser))
622            }
623            _ => unreachable!(),
624        }
625    }
626
627    /// Create a parser for testing purposes.
628    pub fn create_for_test(parser_config: ParserConfig) -> ConnectorResult<Self> {
629        futures::executor::block_on(Self::create(parser_config, SourceContext::dummy().into()))
630    }
631}
632
633/// Test utilities for [`ByteStreamSourceParserImpl`].
634#[cfg(test)]
635pub mod test_utils {
636    use futures::StreamExt;
637    use itertools::Itertools;
638    use risingwave_common::array::StreamChunk;
639
640    use super::*;
641    use crate::source::SourceMessage;
642
643    #[easy_ext::ext(ByteStreamSourceParserImplTestExt)]
644    pub(crate) impl ByteStreamSourceParserImpl {
645        /// Parse the given payloads into a [`StreamChunk`].
646        async fn parse(self, payloads: Vec<Vec<u8>>) -> StreamChunk {
647            let source_messages = payloads
648                .into_iter()
649                .map(|p| SourceMessage {
650                    payload: (!p.is_empty()).then_some(p),
651                    ..SourceMessage::dummy()
652                })
653                .collect_vec();
654
655            into_data_chunk_stream(
656                self.parse_stream_with_events(
657                    futures::stream::once(async { Ok(SourceMessageEvent::Data(source_messages)) })
658                        .boxed(),
659                ),
660            )
661            .next()
662            .await
663            .unwrap()
664            .unwrap()
665        }
666
667        /// Parse the given key-value pairs into a [`StreamChunk`].
668        async fn parse_upsert(self, kvs: Vec<(Vec<u8>, Vec<u8>)>) -> StreamChunk {
669            let source_messages = kvs
670                .into_iter()
671                .map(|(k, v)| SourceMessage {
672                    key: (!k.is_empty()).then_some(k),
673                    payload: (!v.is_empty()).then_some(v),
674                    ..SourceMessage::dummy()
675                })
676                .collect_vec();
677
678            into_data_chunk_stream(
679                self.parse_stream_with_events(
680                    futures::stream::once(async { Ok(SourceMessageEvent::Data(source_messages)) })
681                        .boxed(),
682                ),
683            )
684            .next()
685            .await
686            .unwrap()
687            .unwrap()
688        }
689    }
690}
risingwave_connector/parser/mod.rs

risingwave_connector/parser/
mod.rs
