remove into function in datafusion-repr, add schema conversion in bridge repo instead

Signed-off-by: AveryQi115 <[email protected]>

Author: AveryQi115

Cargo.lock

@@ -3,7 +3,7 @@ use std::{collections::HashMap, sync::Arc};
 use anyhow::{bail, Context, Result};
 use async_recursion::async_recursion;
 use datafusion::{
-    arrow::datatypes::{Schema, SchemaRef},
+    arrow::datatypes::{DataType, Field, Schema, SchemaRef},
     datasource::source_as_provider,
     logical_expr::Operator,
     physical_expr,
@@ -33,6 +33,27 @@ use optd_datafusion_repr::{
 
 use crate::{physical_collector::CollectorExec, OptdPlanContext};
 
+// TODO: current DataType and ConstantType are not 1 to 1 mapping
+// optd schema stores constantType from data type in catalog.get
+// for decimal128, the precision is lost
+fn from_optd_schema(optd_schema: &OptdSchema) -> Schema {
+    let match_type = |typ: &ConstantType| match typ {
+        ConstantType::Any => unimplemented!(),
+        ConstantType::Bool => DataType::Boolean,
+        ConstantType::Int => DataType::Int64,
+        ConstantType::Date => DataType::Date32,
+        ConstantType::Decimal => DataType::Float64,
+        ConstantType::Utf8String => DataType::Utf8,
+    };
+    let fields: Vec<_> = optd_schema
+        .0
+        .iter()
+        .enumerate()
+        .map(|(i, typ)| Field::new(&format!("c{}", i), match_type(typ), false))
+        .collect();
+    Schema::new(fields)
+}
+
 impl OptdPlanContext<'_> {
     #[async_recursion]
     async fn from_optd_table_scan(
@@ -449,7 +470,7 @@ impl OptdPlanContext<'_> {
             }
             OptRelNodeTyp::PhysicalEmptyRelation => {
                 let physical_node = PhysicalEmptyRelation::from_rel_node(rel_node).unwrap();
-                let datafusion_schema: Schema = schema.into();
+                let datafusion_schema: Schema = from_optd_schema(&schema);
                 Ok(Arc::new(datafusion::physical_plan::empty::EmptyExec::new(
                     physical_node.produce_one_row(),
                     Arc::new(datafusion_schema),
@@ -461,7 +482,6 @@ impl OptdPlanContext<'_> {
     }
 
     pub async fn from_optd(&mut self, root_rel: OptRelNodeRef) -> Result<Arc<dyn ExecutionPlan>> {
-        println!("{}", root_rel);
         self.from_optd_plan_node(PlanNode::from_rel_node(root_rel).unwrap())
             .await
     }

optd-datafusion-bridge/src/from_optd.rs

@@ -7,8 +7,6 @@ edition = "2021"
 
 [dependencies]
 anyhow = "1"
-arrow-schema = "*"
-datafusion = "32.0.0"
 num-traits = "0.2"
 num-derive = "0.2"
 tracing = "0.1"

optd-datafusion-repr/Cargo.toml

@@ -106,7 +106,7 @@ impl CostModel<OptRelNodeTyp> for OptCostModel {
                     .unwrap_or(1) as f64;
                 Self::cost(row_cnt, 0.0, row_cnt)
             }
-            OptRelNodeTyp::PhysicalEmptyRelation => Self::cost(1.0, 0.01, 0.0),
+            OptRelNodeTyp::PhysicalEmptyRelation => Self::cost(0.5, 0.01, 0.0),
             OptRelNodeTyp::PhysicalFilter => {
                 let (row_cnt, _, _) = Self::cost_tuple(&children[0]);
                 let (_, compute_cost, _) = Self::cost_tuple(&children[1]);

optd-datafusion-repr/src/cost/base_cost.rs

@@ -1,12 +1,6 @@
-use std::{
-    collections::HashMap,
-    sync::{Arc, Mutex},
-};
-
 use optd_core::property::PropertyBuilder;
 
 use crate::plan_nodes::{ConstantType, OptRelNodeTyp};
-use datafusion::arrow::datatypes::{DataType, Field, Schema as DatafusionSchema};
 
 #[derive(Clone, Debug)]
 pub struct Schema(pub Vec<ConstantType>);
@@ -18,29 +12,6 @@ impl Schema {
     }
 }
 
-impl Into<DatafusionSchema> for Schema {
-    // TODO: current DataType and ConstantType are not 1 to 1 mapping
-    // optd schema stores constantType from data type in catalog.get
-    // for decimal128, the precision is lost
-    fn into(self) -> DatafusionSchema {
-        let match_type = |typ: &ConstantType| match typ {
-            ConstantType::Any => unimplemented!(),
-            ConstantType::Bool => DataType::Boolean,
-            ConstantType::Int => DataType::Int64,
-            ConstantType::Date => DataType::Date32,
-            ConstantType::Decimal => DataType::Float64,
-            ConstantType::Utf8String => DataType::Utf8,
-        };
-        let fields: Vec<_> = self
-            .0
-            .iter()
-            .enumerate()
-            .map(|(i, typ)| Field::new(&format!("c{}", i), match_type(typ), false))
-            .collect();
-        DatafusionSchema::new(fields)
-    }
-}
-
 pub trait Catalog: Send + Sync + 'static {
     fn get(&self, name: &str) -> Schema;
 }