Make DistributedTaskContext Copy

It's two usize fields with no heap data, so Copy is a free refinement
over Clone. Drop the now-redundant `.clone()` calls at internal sites
that pass it by value (recursion helpers, worker metrics callbacks).

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: homa31

src/common/recursion.rs

@@ -85,21 +85,19 @@ impl TreeNodeExt for Arc<dyn ExecutionPlan> {
             ctx: DistributedTaskContext,
             f: &mut F,
         ) -> Result<TreeNodeRecursion> {
-            f(plan, ctx.clone())?.visit_children(|| {
+            f(plan, ctx)?.visit_children(|| {
                 if let Some(ciu) = plan.as_any().downcast_ref::<ChildrenIsolatorUnionExec>() {
                     // Just recurse to children that will actually get executed by this
                     // ChildrenIsolatorUnionExec.
                     ciu.task_idx_map[ctx.task_index].iter().apply_until_stop(
-                        |(child_i, child_ctx)| {
-                            recurse(&ciu.children[*child_i], child_ctx.clone(), f)
-                        },
+                        |(child_i, child_ctx)| recurse(&ciu.children[*child_i], *child_ctx, f),
                     )
                 } else if plan.is_network_boundary() {
                     Ok(TreeNodeRecursion::Continue)
                 } else {
                     plan.children()
                         .into_iter()
-                        .apply_until_stop(|child| recurse(child, ctx.clone(), f))
+                        .apply_until_stop(|child| recurse(child, ctx, f))
                 }
             })
         }
@@ -126,7 +124,7 @@ impl TreeNodeExt for Arc<dyn ExecutionPlan> {
                     tnr: TreeNodeRecursion::Jump,
                 });
             };
-            let transformed = f(node, dt_ctx.clone())?;
+            let transformed = f(node, dt_ctx)?;
             if transformed.tnr == TreeNodeRecursion::Stop {
                 return Ok(transformed);
             }
@@ -142,11 +140,11 @@ impl TreeNodeExt for Arc<dyn ExecutionPlan> {
             if let Some(ciu) = node.as_any().downcast_ref::<ChildrenIsolatorUnionExec>() {
                 let mut child_ctxs = vec![None; ciu.children.len()];
                 for (child_idx, child_ctx) in &ciu.task_idx_map[dt_ctx.task_index] {
-                    child_ctxs[*child_idx] = Some(child_ctx.clone());
+                    child_ctxs[*child_idx] = Some(*child_ctx);
                 }
                 stack.extend(child_ctxs.into_iter().rev());
             } else {
-                stack.extend(node.children().iter().map(|_| Some(dt_ctx.clone())).rev());
+                stack.extend(node.children().iter().map(|_| Some(dt_ctx)).rev());
             }
             Ok(transformed)
         })

src/stage.rs

@@ -139,7 +139,7 @@ impl Stage {
     }
 }
 
-#[derive(Debug, Clone, PartialEq)]
+#[derive(Debug, Clone, Copy, PartialEq)]
 pub struct DistributedTaskContext {
     pub task_index: usize,
     pub task_count: usize,

src/worker/impl_execute_task.rs

@@ -64,7 +64,7 @@ pub(crate) async fn execute_local_task(
     let plan = task_data.plan;
     let task_ctx = task_data.task_ctx;
     let d_cfg = DistributedConfig::from_config_options(task_ctx.session_config().options())?;
-    let d_ctx = DistributedTaskContext::from_ctx(&task_ctx).as_ref().clone();
+    let d_ctx = *DistributedTaskContext::from_ctx(&task_ctx).as_ref();
 
     let send_metrics = d_cfg.collect_metrics;
     let partition_count = plan.properties().partitioning.partition_count();
@@ -91,7 +91,6 @@ pub(crate) async fn execute_local_task(
         let metrics_tx = Arc::clone(&task_data.metrics_tx);
         let task_data_metrics = Arc::clone(&task_data.task_data_metrics);
         let key = key.clone();
-        let d_ctx = d_ctx.clone();
         let stream = on_drop_stream(stream, move || {
             // Stream was dropped before fully consumed -- see https://github.com/datafusion-contrib/datafusion-distributed/issues/412
             // Send metrics via the coordinator channel so they are not lost.