feat(executor): DynamicFilter write right table state exactly once - only if actor owns vnode = 0 (#3783)

* initial

* stash

* do-appply-planner-test

* fmt

* clippy

* license

* minor

* stash

* fix required dist, output indices

* fmt

* stash

* revert

* do-apply-planner-test

* success!

* more e2e

* minor cleanup

* stash

* fix

* fix

* fix

* minor

* minor

* madsim

* clippy

* fmt

* minor

* fix

* fix, apply code review suggestions

* fix

* fix, improve

* simplify

* clippy

* refactor into sub functions

* clippy

* minor

* fix

* fix: rowsort

* fmt

* use literal in expression

* clippy

* fmt

* use madsim btreemap

* minor

* fmt

* respect previous behaviour - return empty message

* bump-ci

* use correct enum

* add state table

* fmt

* prepare for cache policy

* fmt

* minor

* minor

* minor

* Update src/stream/src/executor/managed_state/dynamic_filter.rs

add license

Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>

* fmt

* add state table for RHS

* minor

* fmt

* add right_table to executor

* minor

* fmt

* fix

* clippy

* fmt

* fix double write

* use correct

* try to write on every msg

* fix weird omissions

* bump

* minor

* revert q18, q20

* revert q4

* remove inserts

* only insert for right table

* try to insert at barrier only

* add asserts

* debug

* fix

* fmt

* only write on vnode = 0

* minor

* minor

* use bool as we need to read from RHS in the future... (for recovery)...

Co-authored-by: TennyZhuang <zty0826@gmail.com>
Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>

Author: 3 people

src/stream/src/executor/dynamic_filter.rs

@@ -47,6 +47,7 @@ pub struct DynamicFilterExecutor<S: StateStore> {
     comparator: ExprNodeType,
     range_cache: RangeCache<S>,
     right_table: StateTable<S>,
+    is_right_table_writer: bool,
     actor_id: u64,
     schema: Schema,
     metrics: Arc<StreamingMetrics>,
@@ -63,6 +64,7 @@ impl<S: StateStore> DynamicFilterExecutor<S> {
         comparator: ExprNodeType,
         state_table_l: StateTable<S>,
         state_table_r: StateTable<S>,
+        is_right_table_writer: bool,
         actor_id: u64,
         metrics: Arc<StreamingMetrics>,
     ) -> Self {
@@ -76,6 +78,7 @@ impl<S: StateStore> DynamicFilterExecutor<S> {
             comparator,
             range_cache: RangeCache::new(state_table_l, 0, usize::MAX),
             right_table: state_table_r,
+            is_right_table_writer,
             actor_id,
             metrics,
             schema,
@@ -240,6 +243,7 @@ impl<S: StateStore> DynamicFilterExecutor<S> {
 
         let mut prev_epoch_value: Option<Datum> = None;
         let mut current_epoch_value: Option<Datum> = None;
+        let mut current_epoch_row = None;
         let mut epoch: u64 = 0;
 
         let aligned_stream = barrier_align(
@@ -288,11 +292,10 @@ impl<S: StateStore> DynamicFilterExecutor<S> {
                             Op::UpdateInsert | Op::Insert => {
                                 last_is_insert = true;
                                 current_epoch_value = Some(row.value_at(0).to_owned_datum());
-                                // self.right_table.insert(row.to_owned_row())?;
+                                current_epoch_row = Some(row.to_owned_row());
                             }
-                            Op::UpdateDelete | Op::Delete => {
+                            _ => {
                                 last_is_insert = false;
-                                // self.right_table.delete(row.to_owned_row())?;
                             }
                         }
                     }
@@ -330,7 +333,14 @@ impl<S: StateStore> DynamicFilterExecutor<S> {
                             yield Message::Chunk(chunk);
                         }
                     }
-                    self.right_table.commit(epoch).await?;
+
+                    if self.is_right_table_writer {
+                        if let Some(row) = current_epoch_row.take() {
+                            assert_eq!(epoch, barrier.epoch.prev);
+                            self.right_table.insert(row)?;
+                            self.right_table.commit(epoch).await?;
+                        }
+                    }
 
                     self.range_cache.flush().await?;
 

src/stream/src/from_proto/dynamic_filter.rs

@@ -52,6 +52,9 @@ impl ExecutorBuilder for DynamicFilterExecutorBuilder {
             ));
         }
 
+        // Only write the RHS value if this actor is in charge of vnode 0
+        let is_right_table_writer = vnodes.is_set(0)?;
+
         let state_table_l =
             StateTable::from_table_catalog(node.get_left_table()?, store.clone(), Some(vnodes));
 
@@ -66,6 +69,7 @@ impl ExecutorBuilder for DynamicFilterExecutorBuilder {
             comparator,
             state_table_l,
             state_table_r,
+            is_right_table_writer,
             params.actor_id as u64,
             params.executor_stats,
         )))