recovered stream job requires checkpoint barrier

src/meta/src/barrier/mod.rs

@@ -234,7 +234,7 @@ impl CheckpointControl {
     async fn finish_commands(&mut self, checkpoint: bool) -> MetaResult<bool> {
         for command in self
             .finished_commands
-            .extract_if(|c| checkpoint || c.is_barrier())
+            .extract_if(|c| checkpoint || !c.is_checkpoint())
         {
             // The command is ready to finish. We can now call `pre_finish`.
             command.pre_finish().await?;

src/meta/src/barrier/progress.rs

@@ -159,10 +159,12 @@ impl TrackingJob {
         }
     }
 
-    pub(crate) fn is_barrier(&self) -> bool {
+    pub(crate) fn is_checkpoint(&self) -> bool {
         match self {
             TrackingJob::Recovered(_) => true,
-            TrackingJob::New(command) => command.context.kind.is_barrier(),
+            TrackingJob::New(command) => {
+                command.context.kind.is_initial() || command.context.kind.is_checkpoint()
+            }
         }
     }
 

src/tests/simulation/tests/integration_tests/recovery/background_ddl.rs

@@ -113,6 +113,10 @@ async fn test_background_mv_barrier_recovery() -> Result<()> {
         .run("SELECT v1 FROM t WHERE v1 NOT IN (SELECT v1 FROM mv1)")
         .await?;
     tracing::debug!(missing_rows);
+    let missing_rows_with_row_id = session
+        .run("SELECT _row_id, v1 FROM t WHERE v1 NOT IN (SELECT v1 FROM mv1)")
+        .await?;
+    tracing::debug!(missing_rows_with_row_id);
 
     assert_eq!(t_count, mv1_count);