rename child_task_metrics to input_task_metrics

Author: jayshrivastava

src/flight_service/do_get.rs

@@ -231,12 +231,12 @@ pub fn collect_and_create_metrics_flight_data(
         })
         .collect::<Result<Vec<_>, _>>()?;
     result
-        .child_task_metrics
+        .input_task_metrics
         .insert(stage_key, proto_task_metrics);
 
     // Serialize the metrics for all tasks.
     let mut task_metrics_set = vec![];
-    for (stage_key, metrics) in result.child_task_metrics.into_iter() {
+    for (stage_key, metrics) in result.input_task_metrics.into_iter() {
         task_metrics_set.push(TaskMetrics {
             stage_key: Some(stage_key),
             metrics,

src/metrics/task_metrics_collector.rs

@@ -20,17 +20,17 @@ use std::sync::Arc;
 pub struct TaskMetricsCollector {
     /// metrics contains the metrics for the current task.
     task_metrics: Vec<MetricsSet>,
-    /// child_task_metrics contains metrics for tasks from child [StageExec]s if they were
+    /// input_task_metrics contains metrics for tasks from child [StageExec]s if they were
     /// collected.
-    child_task_metrics: HashMap<StageKey, Vec<MetricsSetProto>>,
+    input_task_metrics: HashMap<StageKey, Vec<MetricsSetProto>>,
 }
 
 /// MetricsCollectorResult is the result of collecting metrics from a task.
 pub struct MetricsCollectorResult {
     // metrics is a collection of metrics for a task ordered using a pre-order traversal of the task's plan.
     pub task_metrics: Vec<MetricsSet>,
-    // child_task_metrics contains metrics for child tasks if they were collected.
-    pub child_task_metrics: HashMap<StageKey, Vec<MetricsSetProto>>,
+    // input_task_metrics contains metrics for child tasks if they were collected.
+    pub input_task_metrics: HashMap<StageKey, Vec<MetricsSetProto>>,
 }
 
 impl TreeNodeRewriter for TaskMetricsCollector {
@@ -62,7 +62,7 @@ impl TreeNodeRewriter for TaskMetricsCollector {
             for mut entry in metrics_collection.iter_mut() {
                 let stage_key = entry.key().clone();
                 let task_metrics = std::mem::take(entry.value_mut()); // Avoid copy.
-                match self.child_task_metrics.get(&stage_key) {
+                match self.input_task_metrics.get(&stage_key) {
                     // There should never be two NetworkShuffleExec with metrics for the same stage_key.
                     // By convention, the NetworkShuffleExec which runs the last partition in a task should be
                     // sent metrics (the NetworkShuffleExec tracks it for us).
@@ -73,7 +73,7 @@ impl TreeNodeRewriter for TaskMetricsCollector {
                         );
                     }
                     None => {
-                        self.child_task_metrics
+                        self.input_task_metrics
                             .insert(stage_key.clone(), task_metrics);
                     }
                 }
@@ -98,7 +98,7 @@ impl TaskMetricsCollector {
     pub fn new() -> Self {
         Self {
             task_metrics: Vec::new(),
-            child_task_metrics: HashMap::new(),
+            input_task_metrics: HashMap::new(),
         }
     }
 
@@ -113,7 +113,7 @@ impl TaskMetricsCollector {
         plan.rewrite(&mut self)?;
         Ok(MetricsCollectorResult {
             task_metrics: self.task_metrics,
-            child_task_metrics: self.child_task_metrics,
+            input_task_metrics: self.input_task_metrics,
         })
     }
 }
@@ -265,7 +265,7 @@ mod tests {
         // Collect metrics for all tasks from the root StageExec.
         let collector = TaskMetricsCollector::new();
         let result = collector.collect(stage_exec.plan.clone()).unwrap();
-        let mut actual_collected_metrics = result.child_task_metrics;
+        let mut actual_collected_metrics = result.input_task_metrics;
         actual_collected_metrics.insert(
             StageKey {
                 query_id: stage_exec.query_id.to_string(),