no async trait

src/daft-local-execution/Cargo.toml

@@ -1,5 +1,4 @@
 [dependencies]
-async-trait = {workspace = true}
 common-daft-config = {path = "../common/daft-config", default-features = false}
 common-display = {path = "../common/display", default-features = false}
 common-error = {path = "../common/error", default-features = false}

src/daft-local-execution/src/intermediate_ops/intermediate_op.rs

@@ -5,8 +5,6 @@ use common_error::DaftResult;
 use daft_micropartition::MicroPartition;
 use tracing::{info_span, instrument};
 
-use async_trait::async_trait;
-
 use crate::{
     channel::{create_channel, PipelineChannel, Receiver, Sender},
     pipeline::{PipelineNode, PipelineResultType},
@@ -100,7 +98,7 @@ impl IntermediateNode {
         Ok(())
     }
 
-    pub async fn spawn_workers(
+    pub fn spawn_workers(
         &self,
         num_workers: usize,
         destination_channel: &mut PipelineChannel,
@@ -185,7 +183,6 @@ impl TreeDisplay for IntermediateNode {
     }
 }
 
-#[async_trait]
 impl PipelineNode for IntermediateNode {
     fn children(&self) -> Vec<&dyn PipelineNode> {
         self.children.iter().map(|v| v.as_ref()).collect()
@@ -195,22 +192,21 @@ impl PipelineNode for IntermediateNode {
         self.intermediate_op.name()
     }
 
-    async fn start(
+    fn start(
         &mut self,
         maintain_order: bool,
         runtime_handle: &mut ExecutionRuntimeHandle,
     ) -> crate::Result<PipelineChannel> {
         let mut child_result_receivers = Vec::with_capacity(self.children.len());
         for child in self.children.iter_mut() {
-            let child_result_channel = child.start(maintain_order, runtime_handle).await?;
+            let child_result_channel = child.start(maintain_order, runtime_handle)?;
             child_result_receivers
                 .push(child_result_channel.get_receiver_with_stats(&self.runtime_stats));
         }
         let mut destination_channel = PipelineChannel::new(*NUM_CPUS, maintain_order);
 
-        let worker_senders = self
-            .spawn_workers(*NUM_CPUS, &mut destination_channel, runtime_handle)
-            .await;
+        let worker_senders =
+            self.spawn_workers(*NUM_CPUS, &mut destination_channel, runtime_handle);
         runtime_handle.spawn(
             Self::send_to_workers(
                 child_result_receivers,

src/daft-local-execution/src/pipeline.rs

@@ -16,7 +16,6 @@ use crate::{
     ExecutionRuntimeHandle, PipelineCreationSnafu,
 };
 
-use async_trait::async_trait;
 use common_display::{mermaid::MermaidDisplayVisitor, tree::TreeDisplay};
 use common_error::DaftResult;
 use daft_core::{
@@ -72,11 +71,10 @@ impl PipelineResultType {
     }
 }
 
-#[async_trait]
 pub trait PipelineNode: Sync + Send + TreeDisplay {
     fn children(&self) -> Vec<&dyn PipelineNode>;
     fn name(&self) -> &'static str;
-    async fn start(
+    fn start(
         &mut self,
         maintain_order: bool,
         runtime_handle: &mut ExecutionRuntimeHandle,

src/daft-local-execution/src/run.rs

@@ -136,10 +136,7 @@ pub fn run_local(
             .expect("Failed to create tokio runtime");
         runtime.block_on(async {
             let mut runtime_handle = ExecutionRuntimeHandle::new(cfg.default_morsel_size);
-            let mut receiver = pipeline
-                .start(true, &mut runtime_handle)
-                .await?
-                .get_receiver();
+            let mut receiver = pipeline.start(true, &mut runtime_handle)?.get_receiver();
             while let Some(val) = receiver.recv().await {
                 let _ = tx.send(val.as_data().clone()).await;
             }

src/daft-local-execution/src/sinks/blocking_sink.rs

@@ -11,7 +11,6 @@ use crate::{
     runtime_stats::RuntimeStatsContext,
     ExecutionRuntimeHandle,
 };
-use async_trait::async_trait;
 pub enum BlockingSinkStatus {
     NeedMoreInput,
     #[allow(dead_code)]
@@ -68,7 +67,6 @@ impl TreeDisplay for BlockingSinkNode {
     }
 }
 
-#[async_trait]
 impl PipelineNode for BlockingSinkNode {
     fn children(&self) -> Vec<&dyn PipelineNode> {
         vec![self.child.as_ref()]
@@ -78,15 +76,14 @@ impl PipelineNode for BlockingSinkNode {
         self.name
     }
 
-    async fn start(
+    fn start(
         &mut self,
         maintain_order: bool,
         runtime_handle: &mut ExecutionRuntimeHandle,
     ) -> crate::Result<PipelineChannel> {
         let child = self.child.as_mut();
         let mut child_results_receiver = child
-            .start(false, runtime_handle)
-            .await?
+            .start(false, runtime_handle)?
             .get_receiver_with_stats(&self.runtime_stats);
 
         let mut destination_channel = PipelineChannel::new(1, maintain_order);

src/daft-local-execution/src/sinks/streaming_sink.rs

@@ -9,7 +9,7 @@ use crate::{
     channel::PipelineChannel, pipeline::PipelineNode, runtime_stats::RuntimeStatsContext,
     ExecutionRuntimeHandle, NUM_CPUS,
 };
-use async_trait::async_trait;
+
 pub enum StreamSinkOutput {
     NeedMoreInput(Option<Arc<MicroPartition>>),
     #[allow(dead_code)]
@@ -73,7 +73,6 @@ impl TreeDisplay for StreamingSinkNode {
     }
 }
 
-#[async_trait]
 impl PipelineNode for StreamingSinkNode {
     fn children(&self) -> Vec<&dyn PipelineNode> {
         self.children.iter().map(|v| v.as_ref()).collect()
@@ -83,7 +82,7 @@ impl PipelineNode for StreamingSinkNode {
         self.name
     }
 
-    async fn start(
+    fn start(
         &mut self,
         maintain_order: bool,
         runtime_handle: &mut ExecutionRuntimeHandle,
@@ -92,7 +91,7 @@ impl PipelineNode for StreamingSinkNode {
             .children
             .get_mut(0)
             .expect("we should only have 1 child");
-        let child_results_channel = child.start(true, runtime_handle).await?;
+        let child_results_channel = child.start(true, runtime_handle)?;
         let mut child_results_receiver =
             child_results_channel.get_receiver_with_stats(&self.runtime_stats);
 

src/daft-local-execution/src/sources/source.rs

@@ -5,8 +5,6 @@ use daft_io::{IOStatsContext, IOStatsRef};
 use daft_micropartition::MicroPartition;
 use futures::{stream::BoxStream, StreamExt};
 
-use async_trait::async_trait;
-
 use crate::{
     channel::PipelineChannel, pipeline::PipelineNode, runtime_stats::RuntimeStatsContext,
     ExecutionRuntimeHandle,
@@ -62,15 +60,14 @@ impl TreeDisplay for SourceNode {
     }
 }
 
-#[async_trait]
 impl PipelineNode for SourceNode {
     fn name(&self) -> &'static str {
         self.source.name()
     }
     fn children(&self) -> Vec<&dyn PipelineNode> {
         vec![]
     }
-    async fn start(
+    fn start(
         &mut self,
         maintain_order: bool,
         runtime_handle: &mut ExecutionRuntimeHandle,