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

risingwave_connector/parser/
mod.rs
