dry

Signed-off-by: Bugen Zhao <i@bugenzhao.com>
risingwavelabs · Mar 13, 2023 · c907a56 · c907a56
1 parent fdf7cb3
commit c907a56
Show file tree

Hide file tree

Showing 3 changed files with 3 additions and 12 deletions.
diff --git a/src/stream/src/executor/backfill.rs b/src/stream/src/executor/backfill.rs
@@ -398,10 +398,7 @@ where
     }
 
     fn mapping_watermark(watermark: Watermark, upstream_indices: &[usize]) -> Option<Watermark> {
-        upstream_indices
-            .iter()
-            .position(|&idx| idx == watermark.col_idx)
-            .map(|idx| watermark.with_idx(idx))
+        watermark.transform_with_indices(upstream_indices)
     }
 
     fn mapping_message(msg: Message, upstream_indices: &[usize]) -> Option<Message> {

diff --git a/src/stream/src/executor/chain.rs b/src/stream/src/executor/chain.rs
@@ -53,10 +53,7 @@ fn mapping_chunk(chunk: StreamChunk, upstream_indices: &[usize]) -> StreamChunk
 }
 
 fn mapping_watermark(watermark: Watermark, upstream_indices: &[usize]) -> Option<Watermark> {
-    upstream_indices
-        .iter()
-        .position(|&idx| idx == watermark.col_idx)
-        .map(|idx| watermark.with_idx(idx))
+    watermark.transform_with_indices(upstream_indices)
 }
 
 impl ChainExecutor {

diff --git a/src/stream/src/executor/rearranged_chain.rs b/src/stream/src/executor/rearranged_chain.rs
@@ -71,10 +71,7 @@ fn mapping(upstream_indices: &[usize], msg: Message) -> Option<Message> {
 }
 
 fn mapping_watermark(watermark: Watermark, upstream_indices: &[usize]) -> Option<Watermark> {
-    upstream_indices
-        .iter()
-        .position(|&idx| idx == watermark.col_idx)
-        .map(|idx| watermark.with_idx(idx))
+    watermark.transform_with_indices(upstream_indices)
 }
 
 #[derive(Debug)]