test: snapshot testing for stream executors

.config/hakari.toml

@@ -29,4 +29,10 @@ workspace-members = [
     "risingwave_object_store",
     "risingwave_bench",
 ]
-third-party = [{ name = "opendal" }, { name = "criterion" }]
+third-party = [
+    { name = "opendal" },
+    # These are solely dev-dependencies. Unifying them may slow down build.
+    { name = "criterion" },
+    { name = "console" },
+    { name = "similar" },
+]

.config/nextest.toml

@@ -1,5 +1,5 @@
 [profile.default]
-retries = 5
+retries = 0
 slow-timeout = { period = "5s" }
 status-level = "all"
 final-status-level = "slow"

src/stream/Cargo.toml

@@ -80,7 +80,10 @@ workspace-hack = { path = "../workspace-hack" }
 [dev-dependencies]
 assert_matches = "1"
 criterion = { version = "0.4", features = ["async_tokio", "async"] }
+insta = "1.29" 
 risingwave_hummock_test = { path = "../storage/hummock_test", features = ["test"] }
+serde = { version = "1.0", features = ["derive"] }
+serde_yaml = "0.9"
 tracing-test = "0.2"
 
 [[bench]]

src/stream/src/executor/over_window/eowc.rs

@@ -494,9 +494,7 @@ mod tests {
     use std::sync::atomic::AtomicU64;
     use std::sync::Arc;
 
-    use risingwave_common::array::StreamChunk;
     use risingwave_common::catalog::{ColumnDesc, ColumnId, Field, Schema, TableId};
-    use risingwave_common::test_prelude::StreamChunkTestExt;
     use risingwave_common::types::DataType;
     use risingwave_common::util::sort_util::OrderType;
     use risingwave_expr::agg::{AggArgs, AggKind};
@@ -506,7 +504,8 @@ mod tests {
 
     use super::{EowcOverWindowExecutor, EowcOverWindowExecutorArgs};
     use crate::common::table::state_table::StateTable;
-    use crate::executor::test_utils::{MessageSender, MockSource, StreamExecutorTestExt};
+    use crate::executor::test_utils::snapshot::executor_snapshot;
+    use crate::executor::test_utils::{MessageSender, MockSource};
     use crate::executor::{ActorContext, BoxedMessageStream, Executor};
 
     async fn create_executor<S: StateStore>(
@@ -580,75 +579,37 @@ mod tests {
             },
         ];
 
-        {
-            // test basic
-            let (mut tx, mut over_window) = create_executor(calls.clone(), store.clone()).await;
-
-            tx.push_barrier(1, false);
-            over_window.expect_barrier().await;
-
-            tx.push_chunk(StreamChunk::from_pretty(
-                " I T  I   i
-                + 1 p1 100 10
-                + 1 p1 101 16
-                + 4 p2 200 20",
-            ));
-            assert_eq!(1, over_window.expect_watermark().await.val.into_int64());
-            assert_eq!(
-                over_window.expect_chunk().await,
-                StreamChunk::from_pretty(
-                    " I T  I   i  i  i
-                    + 1 p1 100 10 .  16"
-                )
-            );
-
-            tx.push_chunk(StreamChunk::from_pretty(
-                " I T  I   i
-                + 5 p1 102 18
-                + 7 p2 201 22
-                + 8 p3 300 33",
-            ));
-            // NOTE: no watermark message here, since watermark(1) was already received
-            assert_eq!(
-                over_window.expect_chunk().await,
-                StreamChunk::from_pretty(
-                    " I T  I   i  i  i
-                    + 1 p1 101 16 10 18
-                    + 4 p2 200 20 .  22"
-                )
-            );
-
-            tx.push_barrier(2, false);
-            over_window.expect_barrier().await;
-        }
-
-        {
-            // test recovery
-            let (mut tx, mut over_window) = create_executor(calls.clone(), store.clone()).await;
-
-            tx.push_barrier(3, false);
-            over_window.expect_barrier().await;
-
-            tx.push_chunk(StreamChunk::from_pretty(
-                " I  T  I   i
-                + 10 p1 103 13
-                + 12 p2 202 28
-                + 13 p3 301 39",
-            ));
-            assert_eq!(5, over_window.expect_watermark().await.val.into_int64());
-            assert_eq!(
-                over_window.expect_chunk().await,
-                StreamChunk::from_pretty(
-                    " I T  I   i  i  i
-                    + 5 p1 102 18 16 13
-                    + 7 p2 201 22 20 28
-                    + 8 p3 300 33 .  39"
-                )
-            );
-
-            tx.push_barrier(4, false);
-            over_window.expect_barrier().await;
-        }
+        insta::assert_snapshot!(
+            executor_snapshot(
+                Box::new(async || create_executor(calls.clone(), store.clone()).await),
+                r###"
+- barrier
+# FIXME: the header cannot be aligned. What's the correct YAML format?
+- !chunk |-
+    I T  I   i
+    + 1 p1 100 10
+    + 1 p1 101 16
+    + 4 p2 200 20
+- !chunk |-
+    I T  I   i
+    + 5 p1 102 18
+    + 7 p2 201 22
+    + 8 p3 300 33
+# TODO: preserve comment in snapshot?
+# NOTE: no watermark message here, since watermark(1) was already received
+- barrier
+- recovery
+- barrier
+- !chunk |-
+    I  T  I   i
+    + 10 p1 103 13
+    + 12 p2 202 28
+    + 13 p3 301 39
+- barrier
+"###,
+            )
+            .await
+        );
     }
 
     #[tokio::test]
@@ -661,26 +622,19 @@ mod tests {
             frame: Frame::rows(FrameBound::Preceding(1), FrameBound::Following(1)),
         }];
 
-        let (mut tx, mut over_window) = create_executor(calls.clone(), store.clone()).await;
-
-        tx.push_barrier(1, false);
-        over_window.expect_barrier().await;
-
-        tx.push_chunk(StreamChunk::from_pretty(
-            " I T  I   i
-            + 1 p1 100 10
-            + 1 p1 101 16
-            + 4 p1 102 20",
-        ));
-        assert_eq!(1, over_window.expect_watermark().await.val.into_int64());
-        let chunk = over_window.expect_chunk().await;
-        assert_eq!(
-            chunk,
-            StreamChunk::from_pretty(
-                " I T  I   i  I
-                + 1 p1 100 10 26
-                + 1 p1 101 16 46"
+        insta::assert_snapshot!(
+            executor_snapshot(
+                Box::new(async || create_executor(calls.clone(), store.clone()).await),
+                r###"
+- barrier
+- !chunk |-
+    I T  I   i
+    + 1 p1 100 10
+    + 1 p1 101 16
+    + 4 p1 102 20
+"###
             )
+            .await
         );
     }
 }

src/stream/src/executor/over_window/snapshots/risingwave_stream__executor__over_window__eowc__tests__over_window.snap

@@ -0,0 +1,61 @@
+---
+source: src/stream/src/executor/over_window/eowc.rs
+expression: "executor_snapshot(Box::new(async ||\n                create_executor(calls.clone(), store.clone()).await),\n        SnapshotEvent::parse(inputs)).await"
+---
+- input: barrier
+  output:
+  - barrier
+- input: !chunk |-
+    +---+---+----+-----+----+
+    | + | 1 | p1 | 100 | 10 |
+    | + | 1 | p1 | 101 | 16 |
+    | + | 4 | p2 | 200 | 20 |
+    +---+---+----+-----+----+
+  output:
+  - !watermark
+    col_idx: 0
+    val: 1
+  - !chunk |-
+    +---+---+----+-----+----+---+----+
+    | + | 1 | p1 | 100 | 10 |   | 16 |
+    +---+---+----+-----+----+---+----+
+- input: !chunk |-
+    +---+---+----+-----+----+
+    | + | 5 | p1 | 102 | 18 |
+    | + | 7 | p2 | 201 | 22 |
+    | + | 8 | p3 | 300 | 33 |
+    +---+---+----+-----+----+
+  output:
+  - !chunk |-
+    +---+---+----+-----+----+----+----+
+    | + | 1 | p1 | 101 | 16 | 10 | 18 |
+    | + | 4 | p2 | 200 | 20 |    | 22 |
+    +---+---+----+-----+----+----+----+
+- input: barrier
+  output:
+  - barrier
+- input: recovery
+  output: []
+- input: barrier
+  output:
+  - barrier
+- input: !chunk |-
+    +---+----+----+-----+----+
+    | + | 10 | p1 | 103 | 13 |
+    | + | 12 | p2 | 202 | 28 |
+    | + | 13 | p3 | 301 | 39 |
+    +---+----+----+-----+----+
+  output:
+  - !watermark
+    col_idx: 0
+    val: 5
+  - !chunk |-
+    +---+---+----+-----+----+----+----+
+    | + | 5 | p1 | 102 | 18 | 16 | 13 |
+    | + | 7 | p2 | 201 | 22 | 20 | 28 |
+    | + | 8 | p3 | 300 | 33 |    | 39 |
+    +---+---+----+-----+----+----+----+
+- input: barrier
+  output:
+  - barrier
+

src/stream/src/executor/over_window/snapshots/risingwave_stream__executor__over_window__eowc__tests__over_window_aggregate.snap

@@ -0,0 +1,23 @@
+---
+source: src/stream/src/executor/over_window/eowc.rs
+expression: "executor_snapshot(Box::new(async ||\n                create_executor(calls.clone(), store.clone()).await),\n        r###\"\n- barrier\n- !chunk |-\n    I T  I   i\n    + 1 p1 100 10\n    + 1 p1 101 16\n    + 4 p1 102 20\n\"###).await"
+---
+- input: barrier
+  output:
+  - barrier
+- input: !chunk |-
+    +---+---+----+-----+----+
+    | + | 1 | p1 | 100 | 10 |
+    | + | 1 | p1 | 101 | 16 |
+    | + | 4 | p1 | 102 | 20 |
+    +---+---+----+-----+----+
+  output:
+  - !watermark
+    col_idx: 0
+    val: 1
+  - !chunk |-
+    +---+---+----+-----+----+----+
+    | + | 1 | p1 | 100 | 10 | 26 |
+    | + | 1 | p1 | 101 | 16 | 46 |
+    +---+---+----+-----+----+----+
+