diff --git a/ci/scripts/backfill-test.sh b/ci/scripts/backfill-test.sh
index 039e8bee94865..1a4ed37fb9d7f 100755
--- a/ci/scripts/backfill-test.sh
+++ b/ci/scripts/backfill-test.sh
@@ -30,12 +30,9 @@ git config --global --add safe.directory /risingwave
 
 download_and_prepare_rw "$profile" common
 
-echo "--- e2e, ci-backfill, build"
-cargo make ci-start ci-backfill
-
 ################ TESTS
 
-echo "--- e2e, ci-backfill, run backfill test"
+echo "--- Run backfill tests"
 ./ci/scripts/run-backfill-tests.sh
 
 echo "--- Kill cluster"
diff --git a/ci/scripts/run-backfill-tests.sh b/ci/scripts/run-backfill-tests.sh
index 77a6ef1a5510c..cf00dfe3dd0a9 100755
--- a/ci/scripts/run-backfill-tests.sh
+++ b/ci/scripts/run-backfill-tests.sh
@@ -16,6 +16,7 @@
 set -euo pipefail
 
 PARENT_PATH=$(dirname "${BASH_SOURCE[0]}")
+TEST_DIR=$PWD/e2e_test
 
 run_sql_file() {
   psql -h localhost -p 4566 -d dev -U root -f "$@"
@@ -29,15 +30,99 @@ flush() {
   run_sql "FLUSH;"
 }
 
-run_sql_file "$PARENT_PATH"/sql/backfill/create_base_table.sql
+# Test snapshot and upstream read.
+test_snapshot_and_upstream_read() {
+  echo "--- e2e, ci-backfill, test_snapshot_and_upstream_read"
+  cargo make ci-start ci-backfill
 
-# Provide snapshot
-run_sql_file "$PARENT_PATH"/sql/backfill/insert.sql
-run_sql_file "$PARENT_PATH"/sql/backfill/insert.sql &
-run_sql_file "$PARENT_PATH"/sql/backfill/create_mv.sql &
+  run_sql_file "$PARENT_PATH"/sql/backfill/create_base_table.sql
 
-wait
+  # Provide snapshot
+  run_sql_file "$PARENT_PATH"/sql/backfill/insert.sql
 
-run_sql_file "$PARENT_PATH"/sql/backfill/select.sql </dev/null
+  # Provide updates ...
+  run_sql_file "$PARENT_PATH"/sql/backfill/insert.sql &
 
-echo "Backfill tests complete"
+  # ... and concurrently create mv.
+  run_sql_file "$PARENT_PATH"/sql/backfill/create_mv.sql &
+
+  wait
+
+  run_sql_file "$PARENT_PATH"/sql/backfill/select.sql </dev/null
+
+  echo "Backfill tests complete"
+}
+
+# Test background ddl recovery
+test_background_ddl_recovery() {
+  echo "--- e2e, ci-3streaming-2serving-3fe, test background ddl"
+  cargo make ci-start ci-1cn-1fe-with-recovery
+
+  # Test before recovery
+  sqllogictest -d dev -h localhost -p 4566 "$TEST_DIR/background_ddl/recovery/create.slt"
+  sqllogictest -d dev -h localhost -p 4566 "$TEST_DIR/background_ddl/recovery/validate.slt"
+
+  OLD_PROGRESS=$(run_sql "SHOW JOBS;" | grep -E -o "[0-9]{1,2}\.[0-9]{1,2}")
+
+  # Restart
+  cargo make ci-kill
+  cargo make dev ci-1cn-1fe-with-recovery
+
+  # Test after recovery
+  sqllogictest -d dev -h localhost -p 4566 "$TEST_DIR/background_ddl/recovery/validate.slt"
+
+  # Recover the mview progress
+  sleep 3
+
+  NEW_PROGRESS=$(run_sql "SHOW JOBS;" | grep -E -o "[0-9]{1,2}\.[0-9]{1,2}")
+
+  if [[ $OLD_PROGRESS < $NEW_PROGRESS ]]; then
+    echo "OK: $OLD_PROGRESS smaller than $NEW_PROGRESS"
+  else
+    echo "FAILED: $OLD_PROGRESS larger or equal to $NEW_PROGRESS"
+  fi
+
+  sleep 60
+
+  # Test after backfill finished
+  sqllogictest -d dev -h localhost -p 4566 "$TEST_DIR/background_ddl/recovery/validate_finished.slt"
+
+  sqllogictest -d dev -h localhost -p 4566 "$TEST_DIR/background_ddl/recovery/drop.slt"
+
+  cargo make ci-kill
+}
+
+# Test foreground ddl should not recover
+test_foreground_ddl_no_recover() {
+  echo "--- e2e, ci-3streaming-2serving-3fe, test background ddl"
+  cargo make ci-start ci-1cn-1fe-with-recovery
+
+  # Test before recovery
+  sqllogictest -d dev -h localhost -p 4566 "$TEST_DIR/background_ddl/foreground/create_base_table.slt"
+  run_sql "CREATE MATERIALIZED VIEW m1 as select * FROM t;" &
+  sleep 3
+  sqllogictest -d dev -h localhost -p 4566 "$TEST_DIR/background_ddl/foreground/validate.slt"
+
+  # Restart
+  cargo make ci-kill
+  cargo make dev ci-1cn-1fe-with-recovery
+
+  # Leave sometime for recovery
+  sleep 5
+
+  # Test after recovery
+  sqllogictest -d dev -h localhost -p 4566 "$TEST_DIR/background_ddl/foreground/validate_restart.slt"
+
+  sqllogictest -d dev -h localhost -p 4566 "$TEST_DIR/background_ddl/foreground/drop.slt"
+
+  cargo make ci-kill
+}
+
+main() {
+  set -euo pipefail
+  # test_snapshot_and_upstream_read
+  # test_background_ddl_recovery
+  test_foreground_ddl_no_recover
+}
+
+main
diff --git a/ci/scripts/run-e2e-test.sh b/ci/scripts/run-e2e-test.sh
index e88704ca7a22f..dc6620ecbbe9f 100755
--- a/ci/scripts/run-e2e-test.sh
+++ b/ci/scripts/run-e2e-test.sh
@@ -66,7 +66,7 @@ RUST_LOG="info,risingwave_stream=info,risingwave_batch=info,risingwave_storage=i
 cluster_start
 sqllogictest -p 4566 -d dev './e2e_test/ddl/**/*.slt' --junit "batch-ddl-${profile}"
 if [[ $mode != "standalone" ]]; then
-  sqllogictest -p 4566 -d dev './e2e_test/background_ddl/**/*.slt' --junit "batch-ddl-${profile}"
+  sqllogictest -p 4566 -d dev './e2e_test/background_ddl/*.slt' --junit "batch-ddl-${profile}"
 else
   echo "Skipping background_ddl test for $mode"
 fi
diff --git a/e2e_test/background_ddl/foreground/create_base_table.slt b/e2e_test/background_ddl/foreground/create_base_table.slt
new file mode 100644
index 0000000000000..2c950e77d3c28
--- /dev/null
+++ b/e2e_test/background_ddl/foreground/create_base_table.slt
@@ -0,0 +1,5 @@
+statement ok
+CREATE TABLE t(v1 int);
+
+statement ok
+INSERT INTO t SELECT * FROM generate_series(1, 500000);
diff --git a/e2e_test/background_ddl/foreground/drop.slt b/e2e_test/background_ddl/foreground/drop.slt
new file mode 100644
index 0000000000000..08b5401cd970b
--- /dev/null
+++ b/e2e_test/background_ddl/foreground/drop.slt
@@ -0,0 +1,2 @@
+statement ok
+DROP TABLE t;
\ No newline at end of file
diff --git a/e2e_test/background_ddl/foreground/validate.slt b/e2e_test/background_ddl/foreground/validate.slt
new file mode 100644
index 0000000000000..92aa20940ac3a
--- /dev/null
+++ b/e2e_test/background_ddl/foreground/validate.slt
@@ -0,0 +1,4 @@
+query I
+select count(*) from rw_catalog.rw_ddl_progress;
+----
+1
diff --git a/e2e_test/background_ddl/foreground/validate_restart.slt b/e2e_test/background_ddl/foreground/validate_restart.slt
new file mode 100644
index 0000000000000..1bcf4fbd7af37
--- /dev/null
+++ b/e2e_test/background_ddl/foreground/validate_restart.slt
@@ -0,0 +1,4 @@
+query I
+select count(*) from rw_catalog.rw_ddl_progress;
+----
+0
diff --git a/e2e_test/background_ddl/recovery/create.slt b/e2e_test/background_ddl/recovery/create.slt
new file mode 100644
index 0000000000000..428a52e29f1e9
--- /dev/null
+++ b/e2e_test/background_ddl/recovery/create.slt
@@ -0,0 +1,11 @@
+statement ok
+CREATE TABLE t(v1 int);
+
+statement ok
+INSERT INTO t SELECT * FROM generate_series(1, 500000);
+
+statement ok
+SET BACKGROUND_DDL=true;
+
+statement ok
+CREATE MATERIALIZED VIEW m1 as SELECT * FROM t;
\ No newline at end of file
diff --git a/e2e_test/background_ddl/recovery/drop.slt b/e2e_test/background_ddl/recovery/drop.slt
new file mode 100644
index 0000000000000..d73c030b90ad9
--- /dev/null
+++ b/e2e_test/background_ddl/recovery/drop.slt
@@ -0,0 +1,5 @@
+statement ok
+DROP MATERIALIZED VIEW m1;
+
+statement ok
+DROP TABLE t;
diff --git a/e2e_test/background_ddl/recovery/validate.slt b/e2e_test/background_ddl/recovery/validate.slt
new file mode 100644
index 0000000000000..153f27293d5c7
--- /dev/null
+++ b/e2e_test/background_ddl/recovery/validate.slt
@@ -0,0 +1,4 @@
+query I
+SELECT count(*) FROM t;
+----
+500000
\ No newline at end of file
diff --git a/e2e_test/background_ddl/recovery/validate_finished.slt b/e2e_test/background_ddl/recovery/validate_finished.slt
new file mode 100644
index 0000000000000..3b0e427ac454f
--- /dev/null
+++ b/e2e_test/background_ddl/recovery/validate_finished.slt
@@ -0,0 +1,4 @@
+query I
+SELECT count(*) FROM m1;
+----
+500000
\ No newline at end of file