Merge branch 'sarvesh/infra' of https://github.com/cmu-db/optd into s…

…arvesh/infra

infra/src/types/operator/logical.rs

@@ -24,20 +24,20 @@ pub enum LogicalOperator<Link> {
 
 /// TODO Add docs.
 pub struct LogicalScanOperator<Link> {
-    table_name: String,
-    predicate: Link,
+    pub table_name: String,
+    pub predicate: Link,
 }
 
 /// TODO Add docs.
 pub struct LogicalFilterOperator<Link> {
-    child: Link,
-    predicate: Link,
+    pub child: Link,
+    pub predicate: Link,
 }
 
 /// TODO Add docs.
 pub struct LogicalJoinOperator<Link> {
-    join_type: (),
-    left: Link,
-    right: Link,
-    condition: Link,
+    pub join_type: (),
+    pub left: Link,
+    pub right: Link,
+    pub condition: Link,
 }

infra/src/types/operator/physical.rs

@@ -18,27 +18,27 @@
 /// [`PartialLogicalPlan`]: crate::plan::partial_logical_plan::PartialLogicalPlan
 /// [`PartialPhysicalPlan`]: crate::plan::partial_physical_plan::PartialPhysicalPlan
 pub enum PhysicalOperator<Link> {
-    Scan(TableScanOperator<Link>),
+    TableScan(TableScanOperator<Link>),
     Filter(PhysicalFilterOperator<Link>),
-    Join(HashJoinOperator<Link>),
+    HashJoin(HashJoinOperator<Link>),
 }
 
 /// TODO Add docs.
 pub struct TableScanOperator<Link> {
-    table_name: String,
-    predicate: Link,
+    pub table_name: String,
+    pub predicate: Link,
 }
 
 /// TODO Add docs.
 pub struct PhysicalFilterOperator<Link> {
-    child: Link,
-    predicate: Link,
+    pub child: Link,
+    pub predicate: Link,
 }
 
 /// TODO Add docs.
 pub struct HashJoinOperator<Link> {
-    join_type: (),
-    left: Link,
-    right: Link,
-    condition: Link,
+    pub join_type: (),
+    pub left: Link,
+    pub right: Link,
+    pub condition: Link,
 }

infra/src/types/plan/logical_plan.rs

@@ -13,7 +13,7 @@ use std::sync::Arc;
 /// operator).
 #[derive(Clone)]
 pub struct LogicalPlan {
-    root: Arc<LogicalOperator<LogicalLink>>,
+    pub root: Arc<LogicalOperator<LogicalLink>>,
 }
 
 /// A link in a [`LogicalPlan`] to a node.

infra/src/types/plan/physical_plan.rs

@@ -1,12 +1,29 @@
 use crate::types::operator::{
-    logical::LogicalOperator, physical::PhysicalOperator, ScalarOperator,
+    logical::LogicalOperator,
+    physical::{HashJoinOperator, PhysicalFilterOperator, PhysicalOperator, TableScanOperator},
+    ScalarOperator,
+};
+use datafusion::{
+    common::{arrow::datatypes::Schema, JoinType}, datasource::physical_plan::{CsvExecBuilder, FileScanConfig}, execution::object_store::ObjectStoreUrl, physical_plan::{
+        expressions::NoOp,
+        filter::FilterExec,
+        joins::{HashJoinExec, PartitionMode},
+        ExecutionPlan,
+    }
 };
 use std::sync::Arc;
 
 /// TODO Add docs.
 #[derive(Clone)]
 pub struct PhysicalPlan {
-    root: Arc<PhysicalOperator<PhysicalLink>>,
+    pub root: Arc<PhysicalOperator<PhysicalLink>>,
+}
+
+/// TODO This is hacky prototype code, DO NOT USE!
+impl PhysicalPlan {
+    fn as_datafusion_execution_plan(&self) -> Arc<dyn ExecutionPlan> {
+        self.root.as_datafusion_execution_plan()
+    }
 }
 
 /// TODO Add docs.
@@ -21,3 +38,67 @@ pub enum PhysicalLink {
 pub enum ScalarLink {
     ScalarNode(Arc<ScalarOperator<ScalarLink>>),
 }
+
+/// TODO This is hacky prototype code, DO NOT USE!
+impl PhysicalOperator<PhysicalLink> {
+    fn as_datafusion_execution_plan(&self) -> Arc<dyn ExecutionPlan> {
+        match self {
+            PhysicalOperator::TableScan(TableScanOperator {
+                table_name,
+                predicate,
+            }) => {
+                let object_store_url = ObjectStoreUrl::local_filesystem();
+                let schema = Arc::new(Schema::empty()); // TODO FIX THIS!
+                let file_scan_config = FileScanConfig::new(object_store_url, schema);
+
+                Arc::new(CsvExecBuilder::new(file_scan_config).build())
+            }
+            PhysicalOperator::Filter(PhysicalFilterOperator { child, predicate }) => {
+                let PhysicalLink::PhysicalNode(child_operator) = child else {
+                    unimplemented!("encountered a scalar operator");
+                };
+
+                let predicate_expr = Arc::new(NoOp {}); // TODO FIX THIS!
+                let child_plan = child_operator.as_datafusion_execution_plan();
+
+                Arc::new(FilterExec::try_new(predicate_expr, child_plan).unwrap())
+            }
+            PhysicalOperator::HashJoin(HashJoinOperator {
+                join_type,
+                left,
+                right,
+                condition,
+            }) => {
+                let PhysicalLink::PhysicalNode(left_operator) = left else {
+                    unimplemented!("encountered a scalar operator");
+                };
+                let PhysicalLink::PhysicalNode(right_operator) = right else {
+                    unimplemented!("encountered a scalar operator");
+                };
+
+                let left_plan = left_operator.as_datafusion_execution_plan();
+                let right_plan = right_operator.as_datafusion_execution_plan();
+                let condition_on = vec![]; // TODO FIX THIS!
+                let join_filter = None;
+                let join_type = JoinType::Inner;
+                let projection = None;
+                let partition_mode = PartitionMode::CollectLeft;
+                let null_equals_null = true;
+
+                Arc::new(
+                    HashJoinExec::try_new(
+                        left_plan,
+                        right_plan,
+                        condition_on,
+                        join_filter,
+                        &join_type,
+                        projection,
+                        partition_mode,
+                        null_equals_null,
+                    )
+                    .unwrap(),
+                )
+            }
+        }
+    }
+}