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

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

@@ -1274,6 +1274,24 @@ def section_streaming_actors(outer_panels):
                         ),
                     ],
                 ),
+                panels.timeseries_actor_ops(
+                    "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_output_count')}[$__rate_interval])) by (table_id, fragment_id)",
+                            "same output count - table {{table_id}} fragment {{fragment_id}}",
+                        ),
+                    ],
+                ),
                 panels.timeseries_percentage(
                     "Executor Cache Miss Ratio",
                     "",

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

@@ -12,14 +12,15 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use enum_as_inner::EnumAsInner;
 use parse_display::{Display, FromStr};
 use risingwave_common::bail;
 
 use crate::aggregate::AggKind;
 use crate::Result;
 
 /// Kind of window functions.
-#[derive(Debug, Display, FromStr, Clone, PartialEq, Eq, Hash)]
+#[derive(Debug, Display, FromStr, Clone, PartialEq, Eq, Hash, EnumAsInner)]
 #[display(style = "snake_case")]
 pub enum WindowFuncKind {
     // General-purpose window functions.
@@ -62,7 +63,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)
     }
 }

src/expr/core/src/window_function/states.rs

@@ -86,6 +86,17 @@ impl WindowStates {
         }
         Ok(())
     }
+
+    /// Slide all windows forward, until the current key is `curr_key`, ignoring the output and evict hints.
+    /// After this method, `self.curr_key() == Some(curr_key)`.
+    /// `curr_key` must exist in the `WindowStates`.
+    pub fn just_slide_to(&mut self, curr_key: &StateKey) -> Result<()> {
+        // TODO(rc): with the knowledge of the old output, we can "jump" to the `curr_key` directly for some window function kind
+        while self.curr_key() != Some(curr_key) {
+            self.just_slide()?;
+        }
+        Ok(())
+    }
 }
 
 impl Deref for WindowStates {

src/frontend/src/optimizer/plan_node/logical_over_window.rs

@@ -278,7 +278,7 @@ impl LogicalOverWindow {
         let rewritten_selected_items = over_window_builder.rewrite_selected_items(select_exprs)?;
 
         for window_func in &window_functions {
-            if window_func.kind.is_rank() && window_func.order_by.sort_exprs.is_empty() {
+            if window_func.kind.is_numbering() && window_func.order_by.sort_exprs.is_empty() {
                 return Err(ErrorCode::InvalidInputSyntax(format!(
                     "window rank function without order by: {:?}",
                     window_func

src/frontend/src/optimizer/rule/over_window_split_rule.rs

@@ -36,7 +36,7 @@ impl Rule for OverWindowSplitRule {
             .iter()
             .enumerate()
             .map(|(idx, func)| {
-                let func_seq = if func.kind.is_rank() {
+                let func_seq = if func.kind.is_numbering() {
                     rank_func_seq += 1;
                     rank_func_seq
                 } else {