risingwave_stream/executor/iceberg_with_pk_index/
position_delete_merger.rs

1// Copyright 2026 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 anyhow::Context;
16use risingwave_common::id::SinkId;
17use risingwave_pb::connector_service::SinkMetadata;
18use risingwave_pb::stream_service::PbIcebergPkIndexSinkRole;
19
20use crate::executor::prelude::*;
21use crate::task::LocalBarrierManager;
22
23/// Trait abstracting position-delete file operations for testability.
24///
25/// Implementations are responsible for reading existing position deletes
26/// (V3 Puffin deletion vectors or V2 Parquet position-delete files),
27/// merging new delete positions, writing the resulting delete file, and
28/// returning the commit metadata for the current barrier.
29#[async_trait::async_trait]
30pub trait PositionDeleteHandler: Send + 'static {
31    fn write(&mut self, path: &str, pos: i64) -> StreamExecutorResult<()>;
32
33    async fn flush(&mut self) -> StreamExecutorResult<Option<SinkMetadata>>;
34}
35
36/// Position-delete merger executor for iceberg pk-index sink without Equality Delete.
37///
38/// This stateless executor receives [`file_path`, `position`] messages from the Writer Executor,
39/// merges them with existing position deletes, and reports the merged delete-file metadata to
40/// meta on each barrier. Depending on the table format version, the written delete file is a
41/// V3 Puffin deletion vector or a V2 file-scoped Parquet position-delete file.
42///
43/// The upstream plan shards messages by `file_path`, so each actor only merges delete
44/// positions for the files assigned to its shard.
45///
46/// Input schema: [`file_path`: Varchar, `position`: int64]
47/// Output: Barriers and watermarks only; no data chunks (terminal executor in the stream graph).
48pub struct PositionDeleteMergerExecutor<H>
49where
50    H: PositionDeleteHandler,
51{
52    actor_id: ActorId,
53    sink_id: SinkId,
54    local_barrier_manager: LocalBarrierManager,
55    input: Option<Executor>,
56    handler: H,
57}
58
59impl<H> PositionDeleteMergerExecutor<H>
60where
61    H: PositionDeleteHandler,
62{
63    pub fn new(
64        actor_id: ActorId,
65        sink_id: SinkId,
66        local_barrier_manager: LocalBarrierManager,
67        input: Executor,
68        handler: H,
69    ) -> Self {
70        Self {
71            actor_id,
72            sink_id,
73            local_barrier_manager,
74            input: Some(input),
75            handler,
76        }
77    }
78
79    #[try_stream(ok = Message, error = StreamExecutorError)]
80    async fn execute_inner(mut self) {
81        let mut input = self.input.take().unwrap().execute();
82
83        // Consume the first barrier.
84        let barrier = expect_first_barrier(&mut input).await?;
85        yield Message::Barrier(barrier);
86
87        #[for_await]
88        for msg in input {
89            match msg? {
90                Message::Chunk(chunk) => {
91                    for (op, row) in chunk.rows() {
92                        debug_assert_eq!(op, risingwave_common::array::Op::Insert);
93                        let file_path = row
94                            .datum_at(0)
95                            .map(|d| d.into_utf8())
96                            .context("file_path should not be null")?;
97                        let position = row
98                            .datum_at(1)
99                            .context("position should not be null")?
100                            .into_int64();
101                        self.handler.write(file_path, position)?;
102                    }
103                }
104                Message::Barrier(barrier) => {
105                    let mut metadata = None;
106                    if barrier.is_checkpoint() {
107                        metadata = self.handler.flush().await?;
108                    }
109
110                    if let Some(metadata) = metadata
111                        && metadata.metadata.is_some()
112                    {
113                        self.local_barrier_manager
114                            .report_iceberg_pk_index_sink_metadata(
115                                barrier.epoch,
116                                self.sink_id,
117                                self.actor_id,
118                                PbIcebergPkIndexSinkRole::PositionDeleteMerger,
119                                Some(metadata),
120                            );
121                    }
122
123                    yield Message::Barrier(barrier);
124                }
125                Message::Watermark(w) => {
126                    yield Message::Watermark(w);
127                }
128            }
129        }
130    }
131}
132
133impl<H> Execute for PositionDeleteMergerExecutor<H>
134where
135    H: PositionDeleteHandler,
136{
137    fn execute(self: Box<Self>) -> BoxedMessageStream {
138        self.execute_inner().boxed()
139    }
140}
141
142#[cfg(test)]
143mod tests {
144    use std::collections::BTreeSet;
145    use std::sync::{Arc, Mutex};
146
147    use hashbrown::HashMap;
148    use risingwave_common::array::{Array, ArrayBuilder, I64ArrayBuilder, Op, Utf8ArrayBuilder};
149    use risingwave_common::catalog::{Field, Schema};
150    use risingwave_common::id::SinkId;
151    use risingwave_common::types::DataType;
152    use risingwave_common::util::epoch::test_epoch;
153
154    use super::*;
155    use crate::executor::test_utils::MockSource;
156    use crate::task::LocalBarrierManager;
157
158    fn build_delete_position_chunk(positions: &[(&str, i64)]) -> StreamChunk {
159        let len = positions.len();
160        let mut file_path_builder = Utf8ArrayBuilder::new(len);
161        let mut position_builder = I64ArrayBuilder::new(len);
162
163        for (path, offset) in positions {
164            file_path_builder.append(Some(*path));
165            position_builder.append(Some(*offset));
166        }
167
168        StreamChunk::from_parts(
169            vec![Op::Insert; len],
170            risingwave_common::array::DataChunk::new(
171                vec![
172                    file_path_builder.finish().into_ref(),
173                    position_builder.finish().into_ref(),
174                ],
175                len,
176            ),
177        )
178    }
179
180    #[derive(Clone)]
181    struct PositionDeleteHandlerMock {
182        existing_dvs: Arc<Mutex<HashMap<String, BTreeSet<i64>>>>,
183        pending_dvs: Arc<Mutex<HashMap<String, BTreeSet<i64>>>>,
184        written_dvs: Arc<Mutex<HashMap<String, BTreeSet<i64>>>>,
185    }
186
187    impl PositionDeleteHandlerMock {
188        fn new() -> Self {
189            Self {
190                existing_dvs: Arc::new(Mutex::new(HashMap::new())),
191                pending_dvs: Arc::new(Mutex::new(HashMap::new())),
192                written_dvs: Arc::new(Mutex::new(HashMap::new())),
193            }
194        }
195
196        fn with_existing_dv(self, file_path: &str, positions: BTreeSet<i64>) -> Self {
197            self.existing_dvs
198                .lock()
199                .unwrap()
200                .insert(file_path.to_owned(), positions);
201            self
202        }
203    }
204
205    #[async_trait::async_trait]
206    impl PositionDeleteHandler for PositionDeleteHandlerMock {
207        fn write(&mut self, path: &str, pos: i64) -> StreamExecutorResult<()> {
208            self.pending_dvs
209                .lock()
210                .unwrap()
211                .entry_ref(path)
212                .or_default()
213                .insert(pos);
214            Ok(())
215        }
216
217        async fn flush(&mut self) -> StreamExecutorResult<Option<SinkMetadata>> {
218            let pending = {
219                let mut pending_dvs = self.pending_dvs.lock().unwrap();
220                std::mem::take(&mut *pending_dvs)
221            };
222            if pending.is_empty() {
223                return Ok(None);
224            }
225
226            let mut existing_dvs = self.existing_dvs.lock().unwrap();
227            let mut written_dvs = self.written_dvs.lock().unwrap();
228            for (file_path, positions) in pending {
229                let mut merged = existing_dvs.get(&file_path).cloned().unwrap_or_default();
230                merged.extend(positions);
231                existing_dvs.insert(file_path.clone(), merged.clone());
232                written_dvs.insert(file_path, merged);
233            }
234            // The mock asserts on side effects (`written_dvs`) rather than emitted
235            // metadata, so we still return Ok(None). The report-on-barrier path
236            // is exercised by the SLT integration tests.
237            // TODO: add unit test for report-on-barrier path once test infra is
238            // available to capture `LocalBarrierEvent`s on the receiver side.
239            Ok(None)
240        }
241    }
242
243    fn input_schema() -> Schema {
244        Schema::new(vec![
245            Field::unnamed(DataType::Varchar),
246            Field::unnamed(DataType::Int64),
247        ])
248    }
249
250    #[tokio::test]
251    async fn test_position_delete_merger_basic() {
252        let handler = PositionDeleteHandlerMock::new();
253        let written_dvs = handler.written_dvs.clone();
254
255        let (mut tx, source) = MockSource::channel();
256        let source = source.into_executor(input_schema(), vec![]);
257
258        let lbm = LocalBarrierManager::for_test();
259        let mut executor =
260            PositionDeleteMergerExecutor::new(123.into(), SinkId::new(0), lbm, source, handler)
261                .boxed()
262                .execute();
263
264        tx.push_barrier(test_epoch(1), false);
265        assert!(executor.next().await.unwrap().unwrap().is_barrier());
266
267        tx.push_chunk(build_delete_position_chunk(&[
268            ("file1.parquet", 0),
269            ("file1.parquet", 3),
270            ("file2.parquet", 1),
271        ]));
272        tx.push_barrier(test_epoch(2), false);
273
274        assert!(executor.next().await.unwrap().unwrap().is_barrier());
275
276        let dvs = written_dvs.lock().unwrap();
277        assert_eq!(dvs.get("file1.parquet").unwrap(), &BTreeSet::from([0, 3]));
278        assert_eq!(dvs.get("file2.parquet").unwrap(), &BTreeSet::from([1]));
279    }
280
281    #[tokio::test]
282    async fn test_position_delete_merger_merge_with_existing() {
283        let handler = PositionDeleteHandlerMock::new()
284            .with_existing_dv("file1.parquet", BTreeSet::from([0, 5, 10]));
285        let written_dvs = handler.written_dvs.clone();
286
287        let (mut tx, source) = MockSource::channel();
288        let source = source.into_executor(input_schema(), vec![]);
289
290        let lbm = LocalBarrierManager::for_test();
291        let mut executor =
292            PositionDeleteMergerExecutor::new(123.into(), SinkId::new(0), lbm, source, handler)
293                .boxed()
294                .execute();
295
296        tx.push_barrier(test_epoch(1), false);
297        assert!(executor.next().await.unwrap().unwrap().is_barrier());
298
299        tx.push_chunk(build_delete_position_chunk(&[
300            ("file1.parquet", 3),
301            ("file1.parquet", 5),
302            ("file1.parquet", 7),
303        ]));
304        tx.push_barrier(test_epoch(2), false);
305
306        assert!(executor.next().await.unwrap().unwrap().is_barrier());
307
308        let dvs = written_dvs.lock().unwrap();
309        assert_eq!(
310            dvs.get("file1.parquet").unwrap(),
311            &BTreeSet::from([0, 3, 5, 7, 10])
312        );
313    }
314
315    #[tokio::test]
316    async fn test_position_delete_merger_no_deletes() {
317        let handler = PositionDeleteHandlerMock::new();
318        let written_dvs = handler.written_dvs.clone();
319
320        let (mut tx, source) = MockSource::channel();
321        let source = source.into_executor(input_schema(), vec![]);
322
323        let lbm = LocalBarrierManager::for_test();
324        let mut executor =
325            PositionDeleteMergerExecutor::new(123.into(), SinkId::new(0), lbm, source, handler)
326                .boxed()
327                .execute();
328
329        tx.push_barrier(test_epoch(1), false);
330        assert!(executor.next().await.unwrap().unwrap().is_barrier());
331
332        tx.push_barrier(test_epoch(2), false);
333        assert!(executor.next().await.unwrap().unwrap().is_barrier());
334
335        assert!(written_dvs.lock().unwrap().is_empty());
336    }
337
338    #[tokio::test]
339    async fn test_position_delete_merger_multiple_epochs() {
340        let handler = PositionDeleteHandlerMock::new();
341        let written_dvs = handler.written_dvs.clone();
342
343        let (mut tx, source) = MockSource::channel();
344        let source = source.into_executor(input_schema(), vec![]);
345
346        let lbm = LocalBarrierManager::for_test();
347        let mut executor =
348            PositionDeleteMergerExecutor::new(123.into(), SinkId::new(0), lbm, source, handler)
349                .boxed()
350                .execute();
351
352        tx.push_barrier(test_epoch(1), false);
353        assert!(executor.next().await.unwrap().unwrap().is_barrier());
354
355        tx.push_chunk(build_delete_position_chunk(&[("file1.parquet", 0)]));
356        tx.push_barrier(test_epoch(2), false);
357        assert!(executor.next().await.unwrap().unwrap().is_barrier());
358        assert_eq!(
359            written_dvs.lock().unwrap().get("file1.parquet").unwrap(),
360            &BTreeSet::from([0])
361        );
362
363        tx.push_chunk(build_delete_position_chunk(&[("file1.parquet", 2)]));
364        tx.push_barrier(test_epoch(3), false);
365        assert!(executor.next().await.unwrap().unwrap().is_barrier());
366
367        assert_eq!(
368            written_dvs.lock().unwrap().get("file1.parquet").unwrap(),
369            &BTreeSet::from([0, 2])
370        );
371    }
372}
risingwave_stream/executor/iceberg_with_pk_index/position_delete_merger.rs

risingwave_stream/executor/iceberg_with_pk_index/
position_delete_merger.rs
