fix: background mview without backfilling should not tracked in barri…

…er runtime info (#19355)

src/meta/src/barrier/context/recovery.rs

@@ -64,7 +64,6 @@ impl GlobalBarrierWorkerContextImpl {
         Ok(())
     }
 
-    // FIXME: didn't consider Values here
     async fn recover_background_mv_progress(
         &self,
     ) -> MetaResult<HashMap<TableId, (String, TableFragments)>> {
@@ -82,7 +81,14 @@ impl GlobalBarrierWorkerContextImpl {
                 .get_job_fragments_by_id(mview.table_id)
                 .await?;
             let table_fragments = TableFragments::from_protobuf(table_fragments);
-            mview_map.insert(table_id, (mview.definition.clone(), table_fragments));
+            if table_fragments.tracking_progress_actor_ids().is_empty() {
+                // If there's no tracking actor in the mview, we can finish the job directly.
+                mgr.catalog_controller
+                    .finish_streaming_job(mview.table_id, None)
+                    .await?;
+            } else {
+                mview_map.insert(table_id, (mview.definition.clone(), table_fragments));
+            }
         }
 
         // If failed, enter recovery mode.

src/meta/src/manager/metadata.rs

@@ -723,7 +723,7 @@ impl MetadataManager {
 }
 
 impl MetadataManager {
-    /// Wait for job finishing notification in `TrackingJob::pre_finish`.
+    /// Wait for job finishing notification in `TrackingJob::finish`.
     /// The progress is updated per barrier.
     pub(crate) async fn wait_streaming_job_finished(
         &self,

src/meta/src/model/stream.rs

@@ -401,14 +401,6 @@ impl TableFragments {
             .cloned()
     }
 
-    /// Returns actors that contains backfill executors.
-    pub fn backfill_actor_ids(&self) -> HashSet<ActorId> {
-        Self::filter_actor_ids(self, |fragment_type_mask| {
-            (fragment_type_mask & FragmentTypeFlag::StreamScan as u32) != 0
-        })
-        .collect()
-    }
-
     pub fn snapshot_backfill_actor_ids(&self) -> HashSet<ActorId> {
         Self::filter_actor_ids(self, |mask| {
             (mask & FragmentTypeFlag::SnapshotBackfillStreamScan as u32) != 0