perf(over window): pass through Updates not affecting window function outputs

e2e_test/over_window/generated/batch/main.slt.part

@@ -9,3 +9,4 @@ include ./expr_in_win_func/mod.slt.part
 include ./agg_in_win_func/mod.slt.part
 include ./opt_agg_then_join/mod.slt.part
 include ./with_filter/mod.slt.part
+include ./no_effect_updates/mod.slt.part

e2e_test/over_window/generated/batch/no_effect_updates/mod.slt.part

@@ -0,0 +1,94 @@
+# This file is generated by `gen.py`. Do not edit it manually!
+
+# Test handling of updates having no effect on window function outputs.
+
+statement ok
+create table t (
+      id int
+    , foo int
+    , bar int
+);
+
+statement ok
+create view v1 as
+select
+    *
+    , rank() over (partition by 1::int order by foo) as r1
+from t;
+
+statement ok
+create view v2 as
+select
+    *
+    , rank() over (partition by 1::int order by bar) as r2
+from t;
+
+statement ok
+insert into t values
+  (100001, 701, 805)
+, (100002, 700, 806)
+, (100003, 723, 807)
+, (100004, 702, 808);
+
+query iii
+select * from v1 order by r1, id;
+----
+100002 700 806 1
+100001 701 805 2
+100004 702 808 3
+100003 723 807 4
+
+query iii
+select * from v2 order by r2, id;
+----
+100001 701 805 1
+100002 700 806 2
+100003 723 807 3
+100004 702 808 4
+
+statement ok
+update t set foo = 733 where id = 100001;
+
+query iii
+select * from v1 order by r1, id;
+----
+100002 700 806 1
+100004 702 808 2
+100003 723 807 3
+100001 733 805 4
+
+query iii
+select * from v2 order by r2, id;
+----
+100001 733 805 1
+100002 700 806 2
+100003 723 807 3
+100004 702 808 4
+
+statement ok
+update t set bar = 804 where id = 100001;
+
+query iii
+select * from v1 order by r1, id;
+----
+100002 700 806 1
+100004 702 808 2
+100003 723 807 3
+100001 733 804 4
+
+query iii
+select * from v2 order by r2, id;
+----
+100001 733 804 1
+100002 700 806 2
+100003 723 807 3
+100004 702 808 4
+
+statement ok
+drop view v1;
+
+statement ok
+drop view v2;
+
+statement ok
+drop table t;

e2e_test/over_window/generated/streaming/main.slt.part

@@ -9,3 +9,4 @@ include ./expr_in_win_func/mod.slt.part
 include ./agg_in_win_func/mod.slt.part
 include ./opt_agg_then_join/mod.slt.part
 include ./with_filter/mod.slt.part
+include ./no_effect_updates/mod.slt.part

e2e_test/over_window/generated/streaming/no_effect_updates/mod.slt.part

@@ -0,0 +1,94 @@
+# This file is generated by `gen.py`. Do not edit it manually!
+
+# Test handling of updates having no effect on window function outputs.
+
+statement ok
+create table t (
+      id int
+    , foo int
+    , bar int
+);
+
+statement ok
+create materialized view v1 as
+select
+    *
+    , rank() over (partition by 1::int order by foo) as r1
+from t;
+
+statement ok
+create materialized view v2 as
+select
+    *
+    , rank() over (partition by 1::int order by bar) as r2
+from t;
+
+statement ok
+insert into t values
+  (100001, 701, 805)
+, (100002, 700, 806)
+, (100003, 723, 807)
+, (100004, 702, 808);
+
+query iii
+select * from v1 order by r1, id;
+----
+100002 700 806 1
+100001 701 805 2
+100004 702 808 3
+100003 723 807 4
+
+query iii
+select * from v2 order by r2, id;
+----
+100001 701 805 1
+100002 700 806 2
+100003 723 807 3
+100004 702 808 4
+
+statement ok
+update t set foo = 733 where id = 100001;
+
+query iii
+select * from v1 order by r1, id;
+----
+100002 700 806 1
+100004 702 808 2
+100003 723 807 3
+100001 733 805 4
+
+query iii
+select * from v2 order by r2, id;
+----
+100001 733 805 1
+100002 700 806 2
+100003 723 807 3
+100004 702 808 4
+
+statement ok
+update t set bar = 804 where id = 100001;
+
+query iii
+select * from v1 order by r1, id;
+----
+100002 700 806 1
+100004 702 808 2
+100003 723 807 3
+100001 733 804 4
+
+query iii
+select * from v2 order by r2, id;
+----
+100001 733 804 1
+100002 700 806 2
+100003 723 807 3
+100004 702 808 4
+
+statement ok
+drop materialized view v1;
+
+statement ok
+drop materialized view v2;
+
+statement ok
+drop table t;

e2e_test/over_window/templates/main.slt.part

@@ -7,3 +7,4 @@ include ./expr_in_win_func/mod.slt.part
 include ./agg_in_win_func/mod.slt.part
 include ./opt_agg_then_join/mod.slt.part
 include ./with_filter/mod.slt.part
+include ./no_effect_updates/mod.slt.part

e2e_test/over_window/templates/no_effect_updates/mod.slt.part

@@ -0,0 +1,92 @@
+# Test handling of updates having no effect on window function outputs.
+
+statement ok
+create table t (
+      id int
+    , foo int
+    , bar int
+);
+
+statement ok
+create $view_type v1 as
+select
+    *
+    , rank() over (partition by 1::int order by foo) as r1
+from t;
+
+statement ok
+create $view_type v2 as
+select
+    *
+    , rank() over (partition by 1::int order by bar) as r2
+from t;
+
+statement ok
+insert into t values
+  (100001, 701, 805)
+, (100002, 700, 806)
+, (100003, 723, 807)
+, (100004, 702, 808);
+
+query iii
+select * from v1 order by r1, id;
+----
+100002 700 806 1
+100001 701 805 2
+100004 702 808 3
+100003 723 807 4
+
+query iii
+select * from v2 order by r2, id;
+----
+100001 701 805 1
+100002 700 806 2
+100003 723 807 3
+100004 702 808 4
+
+statement ok
+update t set foo = 733 where id = 100001;
+
+query iii
+select * from v1 order by r1, id;
+----
+100002 700 806 1
+100004 702 808 2
+100003 723 807 3
+100001 733 805 4
+
+query iii
+select * from v2 order by r2, id;
+----
+100001 733 805 1
+100002 700 806 2
+100003 723 807 3
+100004 702 808 4
+
+statement ok
+update t set bar = 804 where id = 100001;
+
+query iii
+select * from v1 order by r1, id;
+----
+100002 700 806 1
+100004 702 808 2
+100003 723 807 3
+100001 733 804 4
+
+query iii
+select * from v2 order by r2, id;
+----
+100001 733 804 1
+100002 700 806 2
+100003 723 807 3
+100004 702 808 4
+
+statement ok
+drop $view_type v1;
+
+statement ok
+drop $view_type v2;
+
+statement ok
+drop table t;

src/common/src/array/stream_record.rs

@@ -28,7 +28,7 @@ pub enum RecordType {
 }
 
 /// Generic type to represent a row change.
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Copy)]
 pub enum Record<R: Row> {
     Insert { new_row: R },
     Delete { old_row: R },

src/stream/src/common/table/state_table.rs

@@ -644,16 +644,16 @@ where
     pub async fn get_encoded_row(&self, pk: impl Row) -> StreamExecutorResult<Option<Bytes>> {
         assert!(pk.len() <= self.pk_indices.len());
 
-        if self.prefix_hint_len != 0 {
-            debug_assert_eq!(self.prefix_hint_len, pk.len());
-        }
-
         let serialized_pk =
             serialize_pk_with_vnode(&pk, &self.pk_serde, self.compute_vnode_by_pk(&pk));
 
         let prefix_hint = if self.prefix_hint_len != 0 && self.prefix_hint_len == pk.len() {
             Some(serialized_pk.slice(VirtualNode::SIZE..))
         } else {
+            #[cfg(debug_assertions)]
+            if self.prefix_hint_len != 0 {
+                warn!("prefix_hint_len is not equal to pk.len(), may not be able to utilize bloom filter");
+            }
             None
         };