Change back to non async for try_into_logical_plan

Author: kyotoYaho

datafusion/core/src/datasource/datasource.rs

@@ -86,5 +86,5 @@ pub trait TableProvider: Sync + Send {
 #[async_trait]
 pub trait TableProviderFactory: Sync + Send {
     /// Create a TableProvider with the given url
-    async fn create(&self, url: &str) -> Result<Arc<dyn TableProvider>>;
+    fn create(&self, url: &str) -> Result<Arc<dyn TableProvider>>;
 }

datafusion/core/src/execution/context.rs

@@ -429,7 +429,7 @@ impl SessionContext {
                     cmd.file_type
                 ))
             })?;
-        let table = (*factory).create(cmd.location.as_str()).await?;
+        let table = (*factory).create(cmd.location.as_str())?;
         self.register_table(cmd.name.as_str(), table)?;
         let plan = LogicalPlanBuilder::empty(false).build()?;
         Ok(Arc::new(DataFrame::new(self.state.clone(), &plan)))

datafusion/core/src/test_util.rs

@@ -328,10 +328,7 @@ pub struct TestTableFactory {}
 
 #[async_trait]
 impl TableProviderFactory for TestTableFactory {
-    async fn create(
-        &self,
-        url: &str,
-    ) -> datafusion_common::Result<Arc<dyn TableProvider>> {
+    fn create(&self, url: &str) -> datafusion_common::Result<Arc<dyn TableProvider>> {
         Ok(Arc::new(TestTableProvider {
             url: url.to_string(),
         }))

datafusion/proto/src/bytes/mod.rs

@@ -164,7 +164,7 @@ pub async fn logical_plan_from_bytes_with_extension_codec(
     let protobuf = protobuf::LogicalPlanNode::decode(bytes).map_err(|e| {
         DataFusionError::Plan(format!("Error decoding expr as protobuf: {}", e))
     })?;
-    protobuf.try_into_logical_plan(ctx, extension_codec).await
+    protobuf.try_into_logical_plan(ctx, extension_codec)
 }
 
 #[derive(Debug)]
@@ -189,7 +189,7 @@ impl LogicalExtensionCodec for DefaultExtensionCodec {
         ))
     }
 
-    async fn try_decode_table_provider(
+    fn try_decode_table_provider(
         &self,
         _buf: &[u8],
         _schema: SchemaRef,

datafusion/proto/src/lib.rs

@@ -172,7 +172,7 @@ mod roundtrip_tests {
             ))
         }
 
-        async fn try_decode_table_provider(
+        fn try_decode_table_provider(
             &self,
             buf: &[u8],
             _schema: SchemaRef,
@@ -189,7 +189,7 @@ mod roundtrip_tests {
                 .get("testtable")
                 .expect("Unable to find testtable factory")
                 .clone();
-            let provider = (*factory).create(msg.url.as_str()).await?;
+            let provider = (*factory).create(msg.url.as_str())?;
             Ok(provider)
         }
 
@@ -431,7 +431,7 @@ mod roundtrip_tests {
             }
         }
 
-        async fn try_decode_table_provider(
+        fn try_decode_table_provider(
             &self,
             _buf: &[u8],
             _schema: SchemaRef,

datafusion/proto/src/logical_plan.rs

@@ -86,7 +86,7 @@ pub trait AsLogicalPlan: Debug + Send + Sync + Clone {
         B: BufMut,
         Self: Sized;
 
-    async fn try_into_logical_plan(
+    fn try_into_logical_plan(
         &self,
         ctx: &SessionContext,
         extension_codec: &dyn LogicalExtensionCodec,
@@ -130,7 +130,7 @@ pub trait LogicalExtensionCodec: Debug + Send + Sync {
         buf: &mut Vec<u8>,
     ) -> Result<(), DataFusionError>;
 
-    async fn try_decode_table_provider(
+    fn try_decode_table_provider(
         &self,
         buf: &[u8],
         schema: SchemaRef,
@@ -170,7 +170,7 @@ impl LogicalExtensionCodec for DefaultLogicalExtensionCodec {
         ))
     }
 
-    async fn try_decode_table_provider(
+    fn try_decode_table_provider(
         &self,
         _buf: &[u8],
         _schema: SchemaRef,
@@ -196,7 +196,7 @@ impl LogicalExtensionCodec for DefaultLogicalExtensionCodec {
 macro_rules! into_logical_plan {
     ($PB:expr, $CTX:expr, $CODEC:expr) => {{
         if let Some(field) = $PB.as_ref() {
-            field.as_ref().try_into_logical_plan($CTX, $CODEC).await
+            field.as_ref().try_into_logical_plan($CTX, $CODEC)
         } else {
             Err(proto_error("Missing required field in protobuf"))
         }
@@ -301,7 +301,7 @@ impl AsLogicalPlan for LogicalPlanNode {
         })
     }
 
-    async fn try_into_logical_plan(
+    fn try_into_logical_plan(
         &self,
         ctx: &SessionContext,
         extension_codec: &dyn LogicalExtensionCodec,
@@ -487,9 +487,11 @@ impl AsLogicalPlan for LogicalPlanNode {
                     .iter()
                     .map(|expr| parse_expr(expr, ctx))
                     .collect::<Result<Vec<_>, _>>()?;
-                let provider = extension_codec
-                    .try_decode_table_provider(&scan.custom_table_data, schema, ctx)
-                    .await?;
+                let provider = extension_codec.try_decode_table_provider(
+                    &scan.custom_table_data,
+                    schema,
+                    ctx,
+                )?;
 
                 LogicalPlanBuilder::scan_with_filters(
                     &scan.table_name,
@@ -591,7 +593,7 @@ impl AsLogicalPlan for LogicalPlanNode {
                     .input.clone().ok_or_else(|| DataFusionError::Internal(String::from(
                     "Protobuf deserialization error, CreateViewNode has invalid LogicalPlan input.",
                 )))?
-                    .try_into_logical_plan(ctx, extension_codec).await?;
+                    .try_into_logical_plan(ctx, extension_codec)?;
                 let definition = if !create_view.definition.is_empty() {
                     Some(create_view.definition.clone())
                 } else {
@@ -716,7 +718,7 @@ impl AsLogicalPlan for LogicalPlanNode {
             LogicalPlanType::Union(union) => {
                 let mut input_plans: Vec<LogicalPlan> = vec![];
                 for i in union.inputs.iter() {
-                    let res = i.try_into_logical_plan(ctx, extension_codec).await?;
+                    let res = i.try_into_logical_plan(ctx, extension_codec)?;
                     input_plans.push(res);
                 }
 
@@ -744,7 +746,7 @@ impl AsLogicalPlan for LogicalPlanNode {
             LogicalPlanType::Extension(LogicalExtensionNode { node, inputs }) => {
                 let mut input_plans: Vec<LogicalPlan> = vec![];
                 for i in inputs.iter() {
-                    let res = i.try_into_logical_plan(ctx, extension_codec).await?;
+                    let res = i.try_into_logical_plan(ctx, extension_codec)?;
                     input_plans.push(res);
                 }