Fix: fix empty relation schema maintainence across passes (#118)

# Why we need this pr?
Previously. EmptyRelation's schema is derived from its logically
equivalent peer exprs. For example, the schema for `select * from
empty_relation` is deducted from `select * from t1 inner join t2 on
false`. And the empty relation expression will replace the original
inner join.

It works fine in the first pass as empty relation is always transformed
from expressions having proper schema(except for queries like `select
64;`, it is directly converted from datafusion and has empty schema).

But it will break adaptive design as the next time query is executed,
there's only empty relation left in the optimization tree and there's no
way to deduct schema properties from pure empty relation.

# Design
we serialize schema as emptyRelation's data value and it is stored in
its relnode, once we need the schema we fetch it from empty relation's
data.

# Major Changes
- schema prop maintains the type for ColumnRef typ and List typ
- adds convertion from Value to ConstantType to convert data to schema
field
- adds serialized data to EmptyRelation data so schema is keeped
- change EmptyRelation initialization method
- change columnExpr derivation rules for emptyRelation

---------

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

Author: AveryQi115

Diff for: Cargo.lock

@@ -151,6 +151,13 @@ impl Value {
             _ => panic!("Value is not a string"),
         }
     }
+
+    pub fn as_slice(&self) -> Arc<[u8]> {
+        match self {
+            Value::Serialized(i) => i.clone(),
+            _ => panic!("Value is not a serialized"),
+        }
+    }
 }
 
 /// A RelNode is consisted of a plan node type and some children.

Diff for: optd-core/src/rel_node.rs

@@ -548,10 +548,6 @@ impl OptdPlanContext<'_> {
         node: PlanNode,
         meta: &RelNodeMetaMap,
     ) -> Result<Arc<dyn ExecutionPlan>> {
-        let mut schema = OptdSchema { fields: vec![] };
-        if node.typ() == OptRelNodeTyp::PhysicalEmptyRelation {
-            schema = node.schema(self.optimizer.unwrap().optd_optimizer());
-        }
         let rel_node = node.into_rel_node();
 
         let group_id = meta
@@ -599,6 +595,7 @@ impl OptdPlanContext<'_> {
             }
             OptRelNodeTyp::PhysicalEmptyRelation => {
                 let physical_node = PhysicalEmptyRelation::from_rel_node(rel_node).unwrap();
+                let schema = physical_node.empty_relation_schema();
                 let datafusion_schema: Schema = from_optd_schema(schema);
                 Arc::new(datafusion::physical_plan::empty::EmptyExec::new(
                     physical_node.produce_one_row(),

Diff for: optd-datafusion-bridge/src/from_optd.rs

@@ -12,6 +12,7 @@ use optd_datafusion_repr::plan_nodes::{
     LogicalEmptyRelation, LogicalFilter, LogicalJoin, LogicalLimit, LogicalProjection, LogicalScan,
     LogicalSort, OptRelNode, OptRelNodeRef, OptRelNodeTyp, PlanNode, SortOrderExpr, SortOrderType,
 };
+use optd_datafusion_repr::properties::schema::Schema as OPTDSchema;
 
 use crate::OptdPlanContext;
 
@@ -366,7 +367,12 @@ impl OptdPlanContext<'_> {
         &mut self,
         node: &logical_plan::EmptyRelation,
     ) -> Result<LogicalEmptyRelation> {
-        Ok(LogicalEmptyRelation::new(node.produce_one_row))
+        // empty_relation from datafusion always have an empty schema
+        let empty_schema = OPTDSchema { fields: vec![] };
+        Ok(LogicalEmptyRelation::new(
+            node.produce_one_row,
+            empty_schema,
+        ))
     }
 
     fn conv_into_optd_limit(&mut self, node: &logical_plan::Limit) -> Result<LogicalLimit> {

Diff for: optd-datafusion-bridge/src/into_optd.rs

@@ -22,3 +22,5 @@ datafusion-expr = "32.0.0"
 async-trait = "0.1"
 datafusion = "32.0.0"
 assert_approx_eq = "1.1.0"
+serde = { version = "1.0", features = ["derive"] }
+bincode = "1.3.3"

Diff for: optd-datafusion-repr/Cargo.toml

@@ -23,7 +23,7 @@ use optd_core::{
 
 pub use agg::{LogicalAgg, PhysicalAgg};
 pub use apply::{ApplyType, LogicalApply};
-pub use empty_relation::{LogicalEmptyRelation, PhysicalEmptyRelation};
+pub use empty_relation::{EmptyRelationData, LogicalEmptyRelation, PhysicalEmptyRelation};
 pub use expr::{
     BetweenExpr, BinOpExpr, BinOpType, CastExpr, ColumnRefExpr, ConstantExpr, ConstantType,
     DataTypeExpr, ExprList, FuncExpr, FuncType, InListExpr, LikeExpr, LogOpExpr, LogOpType,

Diff for: optd-datafusion-repr/src/plan_nodes.rs

@@ -1,11 +1,16 @@
 use pretty_xmlish::Pretty;
 
+use bincode;
 use optd_core::rel_node::{RelNode, RelNodeMetaMap, Value};
+use serde::{Deserialize, Serialize};
+use std::sync::Arc;
 
 use crate::explain::Insertable;
 
 use super::{replace_typ, OptRelNode, OptRelNodeRef, OptRelNodeTyp, PlanNode};
 
+use crate::properties::schema::Schema;
+
 #[derive(Clone, Debug)]
 pub struct LogicalEmptyRelation(pub PlanNode);
 
@@ -29,25 +34,49 @@ impl OptRelNode for LogicalEmptyRelation {
     }
 }
 
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct EmptyRelationData {
+    pub produce_one_row: bool,
+    pub schema: Schema,
+}
+
 impl LogicalEmptyRelation {
-    pub fn new(produce_one_row: bool) -> LogicalEmptyRelation {
+    pub fn new(produce_one_row: bool, schema: Schema) -> LogicalEmptyRelation {
+        let data = EmptyRelationData {
+            produce_one_row,
+            schema,
+        };
+        let serialized_data: Arc<[u8]> = bincode::serialize(&data).unwrap().into_iter().collect();
         LogicalEmptyRelation(PlanNode(
             RelNode {
                 typ: OptRelNodeTyp::EmptyRelation,
                 children: vec![],
-                data: Some(Value::Bool(produce_one_row)),
+                data: Some(Value::Serialized(serialized_data)),
             }
             .into(),
         ))
     }
 
-    pub fn produce_one_row(&self) -> bool {
-        self.clone()
+    fn get_data(&self) -> EmptyRelationData {
+        let serialized_data = self
+            .clone()
             .into_rel_node()
             .data
             .as_ref()
             .unwrap()
-            .as_bool()
+            .as_slice();
+
+        bincode::deserialize(serialized_data.as_ref()).unwrap()
+    }
+
+    pub fn empty_relation_schema(&self) -> Schema {
+        let data = self.get_data();
+        data.schema
+    }
+
+    pub fn produce_one_row(&self) -> bool {
+        let data = self.get_data();
+        data.produce_one_row
     }
 }
 
@@ -80,12 +109,25 @@ impl PhysicalEmptyRelation {
         Self(node)
     }
 
-    pub fn produce_one_row(&self) -> bool {
-        self.clone()
+    fn get_data(&self) -> EmptyRelationData {
+        let serialized_data = self
+            .clone()
             .into_rel_node()
             .data
             .as_ref()
             .unwrap()
-            .as_bool()
+            .as_slice();
+
+        bincode::deserialize(serialized_data.as_ref()).unwrap()
+    }
+
+    pub fn produce_one_row(&self) -> bool {
+        let data = self.get_data();
+        data.produce_one_row
+    }
+
+    pub fn empty_relation_schema(&self) -> Schema {
+        let data = self.get_data();
+        data.schema
     }
 }

Diff for: optd-datafusion-repr/src/plan_nodes/empty_relation.rs

@@ -3,6 +3,7 @@ use std::{fmt::Display, sync::Arc};
 use arrow_schema::DataType;
 use itertools::Itertools;
 use pretty_xmlish::Pretty;
+use serde::{Deserialize, Serialize};
 
 use optd_core::rel_node::{RelNode, RelNodeMetaMap, Value};
 
@@ -65,7 +66,7 @@ impl OptRelNode for ExprList {
     }
 }
 
-#[derive(Copy, Clone, PartialEq, Eq, Hash, Debug)]
+#[derive(Copy, Clone, PartialEq, Eq, Hash, Debug, Serialize, Deserialize)]
 pub enum ConstantType {
     Bool,
     Utf8String,
@@ -84,12 +85,9 @@ pub enum ConstantType {
     Any,
 }
 
-#[derive(Clone, Debug)]
-pub struct ConstantExpr(pub Expr);
-
-impl ConstantExpr {
-    pub fn new(value: Value) -> Self {
-        let typ = match &value {
+impl ConstantType {
+    pub fn get_data_type_from_value(value: &Value) -> Self {
+        match value {
             Value::Bool(_) => ConstantType::Bool,
             Value::String(_) => ConstantType::Utf8String,
             Value::UInt8(_) => ConstantType::UInt8,
@@ -102,7 +100,16 @@ impl ConstantExpr {
             Value::Int64(_) => ConstantType::Int64,
             Value::Float(_) => ConstantType::Float64,
             _ => unimplemented!(),
-        };
+        }
+    }
+}
+
+#[derive(Clone, Debug)]
+pub struct ConstantExpr(pub Expr);
+
+impl ConstantExpr {
+    pub fn new(value: Value) -> Self {
+        let typ = ConstantType::get_data_type_from_value(&value);
         Self::new_with_type(value, typ)
     }
 

Diff for: optd-datafusion-repr/src/plan_nodes/expr.rs

@@ -1,2 +1,4 @@
 pub mod column_ref;
 pub mod schema;
+
+const DEFAULT_NAME: &str = "unnamed";

Diff for: optd-datafusion-repr/src/properties.rs

@@ -2,9 +2,10 @@ use std::{ops::Deref, sync::Arc};
 
 use optd_core::property::PropertyBuilder;
 
-use crate::plan_nodes::OptRelNodeTyp;
+use crate::plan_nodes::{EmptyRelationData, OptRelNodeTyp};
 
 use super::schema::Catalog;
+use super::DEFAULT_NAME;
 
 #[derive(Clone, Debug)]
 pub enum ColumnRef {
@@ -55,6 +56,19 @@ impl PropertyBuilder<OptRelNodeTyp> for ColumnRefPropertyBuilder {
                     })
                     .collect()
             }
+            OptRelNodeTyp::EmptyRelation => {
+                let data = data.unwrap().as_slice();
+                let empty_relation_data: EmptyRelationData =
+                    bincode::deserialize(data.as_ref()).unwrap();
+                let schema = empty_relation_data.schema;
+                let column_cnt = schema.fields.len();
+                (0..column_cnt)
+                    .map(|i| ColumnRef::BaseTableColumnRef {
+                        table: DEFAULT_NAME.to_string(),
+                        col_idx: i,
+                    })
+                    .collect()
+            }
             OptRelNodeTyp::ColumnRef => {
                 let col_ref_idx = data.unwrap().as_u64();
                 // this is always safe since col_ref_idx was initially a usize in ColumnRefExpr::new()
@@ -114,7 +128,6 @@ impl PropertyBuilder<OptRelNodeTyp> for ColumnRefPropertyBuilder {
             | OptRelNodeTyp::BinOp(_)
             | OptRelNodeTyp::DataType(_)
             | OptRelNodeTyp::Between
-            | OptRelNodeTyp::EmptyRelation
             | OptRelNodeTyp::Like
             | OptRelNodeTyp::InList => {
                 vec![ColumnRef::Derived]

Diff for: optd-datafusion-repr/src/properties/column_ref.rs

@@ -1,16 +1,18 @@
+use serde::{Deserialize, Serialize};
 use std::sync::Arc;
 
 use optd_core::property::PropertyBuilder;
 
-use crate::plan_nodes::{ConstantType, OptRelNodeTyp};
+use super::DEFAULT_NAME;
+use crate::plan_nodes::{ConstantType, EmptyRelationData, OptRelNodeTyp};
 
-#[derive(Clone, Debug)]
+#[derive(Clone, Debug, Serialize, Deserialize)]
 pub struct Field {
     pub name: String,
     pub typ: ConstantType,
     pub nullable: bool,
 }
-#[derive(Clone, Debug)]
+#[derive(Clone, Debug, Serialize, Deserialize)]
 pub struct Schema {
     pub fields: Vec<Field>,
 }
@@ -61,23 +63,33 @@ impl PropertyBuilder<OptRelNodeTyp> for SchemaPropertyBuilder {
                 schema.fields.extend(schema2.fields);
                 schema
             }
-            OptRelNodeTyp::List => {
-                // TODO: calculate real is_nullable for aggregations
+            OptRelNodeTyp::EmptyRelation => {
+                let data = data.unwrap().as_slice();
+                let empty_relation_data: EmptyRelationData =
+                    bincode::deserialize(data.as_ref()).unwrap();
+                empty_relation_data.schema
+            }
+            OptRelNodeTyp::ColumnRef => {
+                let data_typ = ConstantType::get_data_type_from_value(&data.unwrap());
                 Schema {
-                    fields: vec![
-                        Field {
-                            name: "unnamed".to_string(),
-                            typ: ConstantType::Any,
-                            nullable: true
-                        };
-                        children.len()
-                    ],
+                    fields: vec![Field {
+                        name: DEFAULT_NAME.to_string(),
+                        typ: data_typ,
+                        nullable: true,
+                    }],
+                }
+            }
+            OptRelNodeTyp::List => {
+                let mut fields = vec![];
+                for child in children {
+                    fields.extend(child.fields.clone());
                 }
+                Schema { fields }
             }
             OptRelNodeTyp::LogOp(_) => Schema {
                 fields: vec![
                     Field {
-                        name: "unnamed".to_string(),
+                        name: DEFAULT_NAME.to_string(),
                         typ: ConstantType::Any,
                         nullable: true
                     };