using target_kind for group identification

Author: Davda-James

src/builder/analyzer.rs

@@ -1059,6 +1059,7 @@ pub async fn analyze_flow(
         let target_factory = get_target_factory(&target_kind)?;
         let analyzed_target_op_group = AnalyzedExportTargetOpGroup {
             target_factory,
+            target_kind: target_kind.clone(),
             op_idx: op_ids.export_op_ids,
         };
         export_ops_futs.extend(

src/builder/plan.rs

@@ -117,6 +117,7 @@ pub struct AnalyzedExportOp {
 
 pub struct AnalyzedExportTargetOpGroup {
     pub target_factory: Arc<dyn TargetFactory + Send + Sync>,
+    pub target_kind: String,
     pub op_idx: Vec<usize>,
 }
 

src/execution/evaluator.rs

@@ -322,7 +322,7 @@ async fn evaluate_child_op_scope(
     child_scope_entry: ScopeEntry<'_>,
     concurrency_controller: &concur_control::ConcurrencyController,
     memory: &EvaluationMemory,
-    operation_in_process_stats: Option<&crate::execution::stats::OperationInProcessStats>,
+    operation_in_process_stats: Option<&execution::stats::OperationInProcessStats>,
 ) -> Result<()> {
     let _permit = concurrency_controller
         .acquire(Some(|| {
@@ -356,15 +356,16 @@ async fn evaluate_op_scope(
     op_scope: &AnalyzedOpScope,
     scoped_entries: RefList<'_, &ScopeEntry<'_>>,
     memory: &EvaluationMemory,
-    operation_in_process_stats: Option<&crate::execution::stats::OperationInProcessStats>,
+    operation_in_process_stats: Option<&execution::stats::OperationInProcessStats>,
 ) -> Result<()> {
     let head_scope = *scoped_entries.head().unwrap();
     for reactive_op in op_scope.reactive_ops.iter() {
         match reactive_op {
             AnalyzedReactiveOp::Transform(op) => {
                 // Track transform operation start
                 if let Some(ref op_stats) = operation_in_process_stats {
-                    op_stats.start_processing(&op.name, 1);
+                    let transform_key = format!("transform/{}", op.name);
+                    op_stats.start_processing(&transform_key, 1);
                 }
 
                 let mut input_values = Vec::with_capacity(op.inputs.len());
@@ -399,7 +400,8 @@ async fn evaluate_op_scope(
 
                 // Track transform operation completion
                 if let Some(ref op_stats) = operation_in_process_stats {
-                    op_stats.finish_processing(&op.name, 1);
+                    let transform_key = format!("transform/{}", op.name);
+                    op_stats.finish_processing(&transform_key, 1);
                 }
 
                 result?
@@ -532,7 +534,7 @@ pub async fn evaluate_source_entry(
     src_eval_ctx: &SourceRowEvaluationContext<'_>,
     source_value: value::FieldValues,
     memory: &EvaluationMemory,
-    operation_in_process_stats: Option<&crate::execution::stats::OperationInProcessStats>,
+    operation_in_process_stats: Option<&execution::stats::OperationInProcessStats>,
 ) -> Result<EvaluateSourceEntryOutput> {
     let _permit = src_eval_ctx
         .import_op

src/execution/live_updater.rs

@@ -28,7 +28,7 @@ pub struct FlowLiveUpdater {
     join_set: Mutex<Option<JoinSet<Result<()>>>>,
     stats_per_task: Vec<Arc<stats::UpdateStats>>,
     /// Global tracking of in-process rows per operation
-    operation_in_process_stats: Arc<stats::OperationInProcessStats>,
+    pub operation_in_process_stats: Arc<stats::OperationInProcessStats>,
     recv_state: tokio::sync::Mutex<UpdateReceiveState>,
     num_remaining_tasks_rx: watch::Receiver<usize>,
 
@@ -415,23 +415,6 @@ impl FlowLiveUpdater {
         }
     }
 
-    /// Get the total number of rows currently being processed across all operations.
-    pub fn get_total_in_process_count(&self) -> i64 {
-        self.operation_in_process_stats.get_total_in_process_count()
-    }
-
-    /// Get the number of rows currently being processed for a specific operation.
-    pub fn get_operation_in_process_count(&self, operation_name: &str) -> i64 {
-        self.operation_in_process_stats
-            .get_operation_in_process_count(operation_name)
-    }
-
-    /// Get a snapshot of all operation in-process counts.
-    pub fn get_all_operations_in_process(&self) -> std::collections::HashMap<String, i64> {
-        self.operation_in_process_stats
-            .get_all_operations_in_process()
-    }
-
     pub async fn next_status_updates(&self) -> Result<FlowLiveUpdaterUpdates> {
         let mut recv_state = self.recv_state.lock().await;
         let recv_state = &mut *recv_state;

src/execution/row_indexer.rs

@@ -377,17 +377,11 @@ impl<'a> RowIndexer<'a> {
                     (!mutations_w_ctx.is_empty()).then(|| {
                         // Track export operation start
                         if let Some(ref op_stats) = self.operation_in_process_stats {
-                            for export_op_idx in &export_op_group.op_idx {
-                                let export_op = &self.src_eval_ctx.plan.export_ops[*export_op_idx];
-                                op_stats.start_processing(&export_op.name, 1);
-                            }
+                            let export_key = format!("export/{}", export_op_group.target_kind);
+                            op_stats.start_processing(&export_key, 1);
                         }
 
-                        let export_op_names: Vec<String> = export_op_group
-                            .op_idx
-                            .iter()
-                            .map(|idx| self.src_eval_ctx.plan.export_ops[*idx].name.clone())
-                            .collect();
+                        let export_key = format!("export/{}", export_op_group.target_kind);
                         let operation_in_process_stats = self.operation_in_process_stats;
 
                         async move {
@@ -398,9 +392,7 @@ impl<'a> RowIndexer<'a> {
 
                             // Track export operation completion
                             if let Some(ref op_stats) = operation_in_process_stats {
-                                for export_op_name in &export_op_names {
-                                    op_stats.finish_processing(export_op_name, 1);
-                                }
+                                op_stats.finish_processing(&export_key, 1);
                             }
 
                             result

src/execution/source_indexer.rs

@@ -326,8 +326,8 @@ impl SourceIndexingContext {
         let operation_name = {
             let plan_result = self.flow.get_execution_plan().await;
             match plan_result {
-                Ok(plan) => plan.import_ops[self.source_idx].name.clone(),
-                Err(_) => "unknown".to_string(),
+                Ok(plan) => format!("import/{}", plan.import_ops[self.source_idx].name),
+                Err(_) => "import/unknown".to_string(),
             }
         };
 
@@ -339,7 +339,8 @@ impl SourceIndexingContext {
             // Track that we're starting to process this row
             update_stats.processing.start(1);
             if let Some(ref op_stats) = operation_in_process_stats {
-                op_stats.start_processing(&import_op.name, 1);
+                let import_key = format!("import/{}", import_op.name);
+                op_stats.start_processing(&import_key, 1);
             }
 
             let eval_ctx = SourceRowEvaluationContext {