fix(dyn-filter): filter left changes according to watermark before writing state table

proto/stream_plan.proto

@@ -480,7 +480,7 @@ message DynamicFilterNode {
   // When the condition changes, we will tell downstream to insert the LHS records which now match the condition.
   // If this is false, we need to store RHS records which match the condition in the internal table.
   // When the condition changes, we will tell downstream to delete the LHS records which now no longer match the condition.
-  bool condition_always_relax = 5;
+  bool condition_always_relax = 5 [deprecated = true];
 }
 
 // Delta join with two indexes. This is a pseudo plan node generated on frontend. On meta

src/common/src/util/stream_graph_visitor.rs

@@ -133,12 +133,7 @@ pub fn visit_stream_node_tables_inner<F>(
                 optional!(node.memo_table, "TemporalJoinMemo");
             }
             NodeBody::DynamicFilter(node) => {
-                if node.condition_always_relax {
-                    always!(node.left_table, "DynamicFilterLeftNotSatisfy");
-                } else {
-                    always!(node.left_table, "DynamicFilterLeft");
-                }
-
+                always!(node.left_table, "DynamicFilterLeft");
                 always!(node.right_table, "DynamicFilterRight");
             }
 

src/frontend/src/optimizer/plan_node/stream_dynamic_filter.rs

@@ -36,7 +36,6 @@ pub struct StreamDynamicFilter {
     pub base: PlanBase<Stream>,
     core: generic::DynamicFilter<PlanRef>,
     cleaned_by_watermark: bool,
-    condition_always_relax: bool,
 }
 
 impl StreamDynamicFilter {
@@ -67,7 +66,6 @@ impl StreamDynamicFilter {
             base,
             core,
             cleaned_by_watermark,
-            condition_always_relax,
         }
     }
 
@@ -100,7 +98,13 @@ impl StreamDynamicFilter {
                     // downstream. See `derive_watermark_columns`.
                     true
                 }
-                _ => false,
+                ExprType::LessThan | ExprType::LessThanOrEqual => {
+                    // For <= and <, watermark on rhs means all rows older than the watermark should already be emitted,
+                    // and future lhs inputs should be directly passed to downstream without any state table operation.
+                    // So, the state table can be cleaned up.
+                    true
+                }
+                _ => unreachable!(),
             }
         } else {
             false
@@ -124,12 +128,6 @@ impl Distill for StreamDynamicFilter {
                 Pretty::display(&self.cleaned_by_watermark),
             ));
         }
-        if self.condition_always_relax {
-            vec.push((
-                "condition_always_relax",
-                Pretty::display(&self.condition_always_relax),
-            ));
-        }
         childless_record(
             plan_node_name!(
                 "StreamDynamicFilter",
@@ -172,12 +170,13 @@ impl StreamNode for StreamDynamicFilter {
         let right = self.right();
         let right_table = infer_right_internal_table_catalog(right.plan_base())
             .with_id(state.gen_table_id_wrapped());
+        #[allow(deprecated)]
         NodeBody::DynamicFilter(DynamicFilterNode {
             left_key: left_index as u32,
             condition,
             left_table: Some(left_table.to_internal_table_prost()),
             right_table: Some(right_table.to_internal_table_prost()),
-            condition_always_relax: self.condition_always_relax,
+            condition_always_relax: false, // deprecated
         })
     }
 }

src/storage/src/store.rs

@@ -403,6 +403,7 @@ pub trait LocalStateStore: StaticSendSync {
         read_options: ReadOptions,
     ) -> impl Future<Output = StorageResult<Self::RevIter<'_>>> + Send + '_;
 
+    /// Get last persisted watermark for a given vnode.
     fn get_table_watermark(&self, vnode: VirtualNode) -> Option<Bytes>;
 
     /// Inserts a key-value entry associated with a given `epoch` into the state store.

src/stream/src/common/table/mod.rs

@@ -14,7 +14,6 @@
 
 pub mod state_table;
 mod state_table_cache;
-mod watermark;
 
 #[cfg(test)]
 pub mod test_state_table;