cherry pick perf(over window): series of performance improvements to release-2.1

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;

grafana/risingwave-dev-dashboard.dashboard.py

@@ -1321,16 +1321,20 @@ def section_streaming_actors(outer_panels: Panels):
                     ],
                 ),
                 panels.timeseries_actor_ops(
-                    "Over Window Executor Compute Count",
+                    "Over Window Executor State Computation",
                     "",
                     [
+                        panels.target(
+                            f"sum(rate({table_metric('stream_over_window_accessed_entry_count')}[$__rate_interval])) by (table_id, fragment_id)",
+                            "accessed entry count - table {{table_id}} fragment {{fragment_id}}",
+                        ),
                         panels.target(
                             f"sum(rate({table_metric('stream_over_window_compute_count')}[$__rate_interval])) by (table_id, fragment_id)",
                             "compute count - table {{table_id}} fragment {{fragment_id}}",
                         ),
                         panels.target(
-                            f"sum(rate({table_metric('stream_over_window_same_result_count')}[$__rate_interval])) by (table_id, fragment_id)",
-                            "same result count - table {{table_id}} fragment {{fragment_id}}",
+                            f"sum(rate({table_metric('stream_over_window_same_output_count')}[$__rate_interval])) by (table_id, fragment_id)",
+                            "same output count - table {{table_id}} fragment {{fragment_id}}",
                         ),
                     ],
                 ),

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/expr/core/src/window_function/kind.rs

@@ -13,14 +13,15 @@
 // limitations under the License.
 
 use anyhow::Context;
+use enum_as_inner::EnumAsInner;
 use parse_display::{Display, FromStr};
 use risingwave_common::bail;
 
 use crate::aggregate::AggType;
 use crate::Result;
 
 /// Kind of window functions.
-#[derive(Debug, Display, FromStr /* for builtin */, Clone, PartialEq, Eq, Hash)]
+#[derive(Debug, Display, FromStr /* for builtin */, Clone, PartialEq, Eq, Hash, EnumAsInner)]
 #[display(style = "snake_case")]
 pub enum WindowFuncKind {
     // General-purpose window functions.
@@ -64,7 +65,7 @@ impl WindowFuncKind {
 }
 
 impl WindowFuncKind {
-    pub fn is_rank(&self) -> bool {
+    pub fn is_numbering(&self) -> bool {
         matches!(self, Self::RowNumber | Self::Rank | Self::DenseRank)
     }
 }