risingwave_connector/sink/formatter/
append_only.rs

1// Copyright 2025 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use risingwave_common::array::Op;
16
17use super::{Result, SinkFormatter, StreamChunk};
18use crate::sink::encoder::RowEncoder;
19use crate::tri;
20
21pub struct AppendOnlyFormatter<KE, VE> {
22    key_encoder: Option<KE>,
23    val_encoder: VE,
24}
25
26impl<KE, VE> AppendOnlyFormatter<KE, VE> {
27    pub fn new(key_encoder: Option<KE>, val_encoder: VE) -> Self {
28        Self {
29            key_encoder,
30            val_encoder,
31        }
32    }
33}
34
35impl<KE: RowEncoder, VE: RowEncoder> SinkFormatter for AppendOnlyFormatter<KE, VE> {
36    type K = KE::Output;
37    type V = VE::Output;
38
39    fn format_chunk(
40        &self,
41        chunk: &StreamChunk,
42    ) -> impl Iterator<Item = Result<(Option<Self::K>, Option<Self::V>)>> {
43        std::iter::from_coroutine(
44            #[coroutine]
45            || {
46                for (op, row) in chunk.rows() {
47                    if op != Op::Insert {
48                        continue;
49                    }
50                    let event_key_object = match &self.key_encoder {
51                        Some(key_encoder) => Some(tri!(key_encoder.encode(row))),
52                        None => None,
53                    };
54                    let event_object = Some(tri!(self.val_encoder.encode(row)));
55
56                    yield Ok((event_key_object, event_object))
57                }
58            },
59        )
60    }
61}
risingwave_connector/sink/formatter/append_only.rs

risingwave_connector/sink/formatter/
append_only.rs
