fix: show continuous stats during batch updates

src/execution/live_updater.rs

@@ -329,7 +329,40 @@ impl SourceUpdateTask {
         update_options: super::source_indexer::UpdateOptions,
     ) -> Result<()> {
         let update_stats = Arc::new(stats::UpdateStats::default());
-        source_indexing_context
+
+        // Spawn periodic stats reporting task if print_stats is enabled
+        let reporting_handle = if self.options.print_stats {
+            let update_stats_clone = update_stats.clone();
+            let update_title_owned = update_title.to_string();
+            let flow_name = self.flow.flow_instance.name.clone();
+            let import_op_name = self.import_op().name.clone();
+
+            let report_task = async move {
+                let mut interval = tokio::time::interval(REPORT_INTERVAL);
+                let mut last_stats = update_stats_clone.as_ref().clone();
+                interval.set_missed_tick_behavior(MissedTickBehavior::Delay);
+                interval.tick().await; // Skip first tick
+                loop {
+                    interval.tick().await;
+                    let current_stats = update_stats_clone.as_ref().clone();
+                    let delta = current_stats.delta(&last_stats);
+                    if delta.has_any_change() {
+                        // Print periodic progress (do NOT merge here, final report_stats will merge)
+                        println!(
+                            "{}.{} ({update_title_owned}): {}",
+                            flow_name, import_op_name, delta
+                        );
+                        last_stats = current_stats;
+                    }
+                }
+            };
+            Some(tokio::spawn(report_task))
+        } else {
+            None
+        };
+
+        // Run the actual update
+        let update_result = source_indexing_context
             .update(&self.pool, &update_stats, update_options)
             .await
             .with_context(|| {
@@ -338,12 +371,23 @@ impl SourceUpdateTask {
                     self.flow.flow_instance.name,
                     self.import_op().name
                 )
-            })?;
+            });
+
+        // Cancel the reporting task if it was spawned
+        if let Some(handle) = reporting_handle {
+            handle.abort();
+        }
+
+        // Check update result
+        update_result?;
+
         if update_stats.has_any_change() {
             self.status_tx.send_modify(|update| {
                 update.source_updates_num[self.source_idx] += 1;
             });
         }
+
+        // Report final stats
         self.report_stats(&update_stats, update_title);
         Ok(())
     }