fix(test): remove auto setting implicit flush from recovery test (#16211

)

Signed-off-by: Runji Wang <[email protected]>

ci/workflows/pull-request.yml

@@ -525,7 +525,7 @@ steps:
     retry: *auto-retry
 
   - label: "recovery test (deterministic simulation)"
-    command: "TEST_NUM=8 KILL_RATE=0.4 BACKGROUND_DDL_RATE=0.0 ci/scripts/deterministic-recovery-test.sh"
+    command: "TEST_NUM=8 KILL_RATE=1.0 BACKGROUND_DDL_RATE=0.0 ci/scripts/deterministic-recovery-test.sh"
     if: |
       !(build.pull_request.labels includes "ci/skip-ci") && build.env("CI_STEPS") == null
       || build.pull_request.labels includes "ci/run-recovery-test-deterministic-simulation"

e2e_test/ddl/index.slt

@@ -123,6 +123,9 @@ create table t (j jsonb);
 statement ok
 insert into t values ('{"k": "abc" }'::jsonb);
 
+statement ok
+flush;
+
 statement ok
 create index idx on t(j->'k');
 

e2e_test/streaming/basic.slt

@@ -31,6 +31,9 @@ insert into t5 values ('+inf');
 statement ok
 insert into t5 values ('-inf');
 
+statement ok
+flush;
+
 statement ok
 create materialized view mv1 as select * from t1;
 

e2e_test/streaming/bug_fixes/mv_on_singleton_mv_4153.slt

@@ -4,6 +4,9 @@ create table t (v int);
 statement ok
 insert into t values (666), (233), (233);
 
+statement ok
+flush;
+
 # Create singleton mview
 statement ok
 create materialized view mv as select v from t order by v limit 10;

e2e_test/streaming/demo/ecommerce.slt

@@ -19,6 +19,9 @@ values
     ('62', '803', 2443.632235, '2022-07-19 12:31:22.07'),
     ('63', '74', 390.383406, '2022-07-19 12:31:22.07');
 
+statement ok
+flush;
+
 statement ok
 CREATE MATERIALIZED VIEW order_details AS
 SELECT
@@ -30,9 +33,6 @@ FROM
 GROUP BY
     order_id;
 
-statement ok
-flush;
-
 query IRR rowsort
 SELECT * FROM order_details;
 ----
@@ -76,6 +76,9 @@ INSERT INTO parcel_events
 ('56', '2022-07-20 12:31:22.07', 'parcel_shipped'),
 ('76', '2022-08-10 12:31:22.07', 'parcel_shipped');
 
+statement ok
+flush;
+
 statement ok
 CREATE MATERIALIZED VIEW abnormal_delivery AS
 SELECT
@@ -103,9 +106,6 @@ FROM
 where
     t2.event_timestamp - t1.event_timestamp > interval '7 days';
 
-statement ok
-flush;
-
 query IT rowsort
 SELECT * FROM abnormal_delivery;
 ----

e2e_test/streaming/demo/metric_analysis.slt

@@ -38,6 +38,9 @@ INSERT INTO tcp_metrics VALUES
 ('812b4ba287f5ee0bc9d43bbf5bbe87fb', '3', '2022-06-08 09:39:54', 'srtt', 'all', '599.9202978109153'),
 ('812b4ba287f5ee0bc9d43bbf5bbe87fb', '3', '2022-06-08 09:39:54', 'download_speed', 'all', '149.7339243859982');
 
+statement ok
+FLUSH;
+
 statement ok
 CREATE MATERIALIZED VIEW high_util_tcp_metrics AS
 SELECT

e2e_test/streaming/natural_and_cross_join.slt

@@ -23,6 +23,9 @@ INSERT INTO departments (department_name, department_id) VALUES
     ('Engineering', 1),
     ('HR', 2);
 
+statement ok
+flush
+
 # Create materialized view with NATURAL JOIN
 statement ok
 CREATE MATERIALIZED VIEW employee_department_natural_join AS
@@ -37,9 +40,6 @@ SELECT e.employee_name, d.department_name
 FROM employees e CROSS JOIN departments d
 ORDER BY e.employee_name, d.department_name;
 
-statement ok
-flush
-
 # Test NATURAL JOIN
 query TT rowsort
 SELECT employee_name, department_name FROM employee_department_natural_join;

e2e_test/streaming/non_strict_mode.slt

@@ -10,6 +10,9 @@ create table t(x int);
 statement ok
 insert into t values (0),(1),(2),(NULL);
 
+statement ok
+flush;
+
 statement ok
 create materialized view mv_proj as select x, 10/x as v from t;
 
@@ -60,6 +63,9 @@ create table t(x varchar);
 statement ok
 insert into t values ('two'), ('4');
 
+statement ok
+flush;
+
 statement ok
 create materialized view mv_coalesce as select coalesce(x::int, 0) as v from t;
 

e2e_test/streaming/rate_limit/basic.slt

@@ -29,6 +29,9 @@ DROP MATERIALIZED VIEW m;
 statement ok
 delete from t1;
 
+statement ok
+FLUSH;
+
 # Test with small rate_limit. In this case, the stream chunk needs to be split
 
 statement ok

e2e_test/streaming/time_window.slt

@@ -12,6 +12,9 @@ insert into t1 values
     (7, 1, 6, '2022-01-01 10:51:00'),
     (8, 3, 8, '2022-01-01 11:02:00');
 
+statement ok
+flush;
+
 statement ok
 create materialized view mv_tumble as
 select * from tumble(t1, created_at, interval '30' minute);
@@ -44,9 +47,6 @@ select uid, sum(v) as sum_v, window_start
 from hop(t1, created_at, interval '15' minute, interval '30' minute)
 group by window_start, uid;
 
-statement ok
-flush;
-
 query IITTT
 select row_id, uid, created_at, window_start, window_end
 from mv_tumble order by row_id, window_start;

e2e_test/streaming/time_window_join.slt

@@ -12,6 +12,9 @@ insert into LeftTable values (timestamp '2020-04-15 12:02', 1, 'L1'), (timestamp
 statement ok
 insert into RightTable values (timestamp '2020-04-15 12:01', 2, 'R2'), (timestamp '2020-04-15 12:04', 3, 'R3'), (timestamp '2020-04-15 12:05', 4, 'R4');
 
+statement ok
+flush;
+
 statement ok
 CREATE MATERIALIZED VIEW mv_window_join AS
 SELECT L.num AS L_Num, L.id AS L_Id, R.num AS R_Num, R.id AS R_Id,
@@ -22,9 +25,6 @@ FULL JOIN
 TUMBLE(RightTable, row_time, interval '5' minute) AS R
 ON L.num = R.num AND L.window_start = R.window_start AND L.window_end = R.window_end;
 
-statement ok
-flush;
-
 query ITITTT rowsort
 SELECT * FROM mv_window_join;
 ----

src/tests/simulation/src/client.rs

@@ -129,10 +129,6 @@ impl RisingWave {
                 tracing::error!("postgres connection error: {e}");
             }
         });
-        // for recovery
-        client
-            .simple_query("SET RW_IMPLICIT_FLUSH TO true;")
-            .await?;
         // replay all SET statements
         for stmt in SetStmtsIterator::new(set_stmts) {
             client.simple_query(&stmt).await?;

src/tests/simulation/src/slt.rs

@@ -67,25 +67,22 @@ enum SqlCmd {
 }
 
 impl SqlCmd {
-    // We won't kill during insert/update/delete/alter since the atomicity is not guaranteed.
-    // Notice that `create table as` is also not atomic in our system.
-    // TODO: For `SqlCmd::Alter`, since table fragment and catalog commit for table schema change
-    // are not transactional, we can't kill during `alter table add/drop columns` for now, will
-    // remove it until transactional commit of table fragment and catalog is supported.
-    fn ignore_kill(&self) -> bool {
+    fn allow_kill(&self) -> bool {
         matches!(
             self,
-            SqlCmd::Dml
-                | SqlCmd::Flush
-                | SqlCmd::Alter
-                | SqlCmd::Create {
-                    is_create_table_as: true,
-                    ..
-                }
-                | SqlCmd::CreateSink {
-                    is_sink_into_table: true,
-                }
+            SqlCmd::Create {
+                // `create table as` is also not atomic in our system.
+                is_create_table_as: false,
+                ..
+            } | SqlCmd::CreateSink {
+                is_sink_into_table: false,
+            } | SqlCmd::CreateMaterializedView { .. }
+                | SqlCmd::Drop
         )
+        // We won't kill during insert/update/delete/alter since the atomicity is not guaranteed.
+        // TODO: For `SqlCmd::Alter`, since table fragment and catalog commit for table schema change
+        // are not transactional, we can't kill during `alter table add/drop columns` for now, will
+        // remove it until transactional commit of table fragment and catalog is supported.
     }
 }
 
@@ -282,7 +279,7 @@ pub async fn run_slt_task(
                 background_ddl_enabled = background_ddl_setting;
             };
 
-            if cmd.ignore_kill() {
+            if !cmd.allow_kill() {
                 for i in 0usize.. {
                     let delay = Duration::from_secs(1 << i);
                     if let Err(err) = tester