feat: reuse typed prepared plans and reduce execution overhead

kite_sql_serde_macros/src/projection.rs

@@ -108,7 +108,7 @@ pub(crate) fn handle(ast: DeriveInput) -> Result<TokenStream, Error> {
             ) -> ::std::result::Result<::std::vec::Vec<::kite_sql::planner::ExprRef>, ::kite_sql::errors::DatabaseError>
             where
                 T: ::kite_sql::storage::Transaction,
-                A: AsRef<[(&'static str, ::kite_sql::types::value::DataValue)]>,
+                A: AsRef<[(usize, ::kite_sql::types::LogicalType)]>,
             {
                 Ok(::std::vec![
                     #(#projection_exprs.into_scalar()),*

kite_sql_serde_macros/src/reference_serialization.rs

@@ -127,7 +127,7 @@ pub(crate) fn handle(ast: DeriveInput) -> Result<TokenStream, Error> {
 
             quote! {
                 impl crate::serdes::ReferenceSerialization for #struct_name {
-                    fn encode<W: std::io::Write, A: crate::planner::MetaArena>(
+                    fn encode<W: std::io::Write, A: crate::planner::MetaArena + ?Sized>(
                         &self,
                         writer: &mut W,
                         is_direct: bool,
@@ -141,7 +141,7 @@ pub(crate) fn handle(ast: DeriveInput) -> Result<TokenStream, Error> {
                         Ok(())
                     }
 
-                    fn decode<T: crate::storage::Transaction, R: std::io::Read, A: crate::planner::MetaArena>(
+                    fn decode<T: crate::storage::Transaction, R: std::io::Read, A: crate::planner::MetaArena + ?Sized>(
                         reader: &mut R,
                         drive: Option<&crate::serdes::ReferenceDecodeContext<'_, T>>,
                         reference_tables: &crate::serdes::ReferenceTables,
@@ -208,7 +208,7 @@ pub(crate) fn handle(ast: DeriveInput) -> Result<TokenStream, Error> {
 
             quote! {
                 impl crate::serdes::ReferenceSerialization for #struct_name {
-                    fn encode<W: std::io::Write, A: crate::planner::MetaArena>(
+                    fn encode<W: std::io::Write, A: crate::planner::MetaArena + ?Sized>(
                         &self,
                         writer: &mut W,
                         is_direct: bool,
@@ -222,7 +222,7 @@ pub(crate) fn handle(ast: DeriveInput) -> Result<TokenStream, Error> {
                         Ok(())
                     }
 
-                    fn decode<T: crate::storage::Transaction, R: std::io::Read, A: crate::planner::MetaArena>(
+                    fn decode<T: crate::storage::Transaction, R: std::io::Read, A: crate::planner::MetaArena + ?Sized>(
                         reader: &mut R,
                         drive: Option<&crate::serdes::ReferenceDecodeContext<'_, T>>,
                         reference_tables: &crate::serdes::ReferenceTables,

src/binder/aggregate.rs

@@ -16,9 +16,10 @@ use super::{Binder, QueryBindStep};
 use crate::errors::DatabaseError;
 use crate::expression::visitor::{walk_expr, ExprVisitor};
 use crate::expression::visitor_mut::{walk_mut_expr, ExprVisitorMut};
+use crate::planner::MetaArena;
 use crate::planner::{ExprRef, LogicalPlan, PlanArena};
 use crate::storage::Transaction;
-use crate::types::value::DataValue;
+use crate::types::LogicalType;
 use crate::{
     expression::ScalarExpression,
     planner::operator::{aggregate::AggregateOperator, sort::SortField},
@@ -28,8 +29,8 @@ struct AggregateCallCollector<'a> {
     agg_calls: &'a mut Vec<ExprRef>,
 }
 
-impl ExprVisitor<PlanArena<'_>> for AggregateCallCollector<'_> {
-    fn visit(&mut self, expr: ExprRef, arena: &PlanArena<'_>) -> Result<(), DatabaseError> {
+impl ExprVisitor<dyn MetaArena + '_> for AggregateCallCollector<'_> {
+    fn visit(&mut self, expr: ExprRef, arena: &(dyn MetaArena + '_)) -> Result<(), DatabaseError> {
         match arena.expression(expr) {
             ScalarExpression::AggCall { .. } => self.agg_calls.push(expr),
             ScalarExpression::Alias { expr, .. } => self.visit(*expr, arena)?,
@@ -40,7 +41,7 @@ impl ExprVisitor<PlanArena<'_>> for AggregateCallCollector<'_> {
     }
 }
 
-impl<T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'_, '_, T, A> {
+impl<T: Transaction, A: AsRef<[(usize, LogicalType)]>> Binder<'_, '_, T, A> {
     pub fn bind_aggregate(
         &mut self,
         children: LogicalPlan,
@@ -214,7 +215,7 @@ impl<T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'_, '_, T, A>
         &mut self,
         select_list: &mut [ExprRef],
         expr: ExprRef,
-        arena: &mut PlanArena<'_>,
+        arena: &mut (dyn MetaArena + '_),
     ) -> Result<(), DatabaseError> {
         if let ScalarExpression::Alias { alias, .. } = arena.expression(expr) {
             if let Some(i) = select_list.iter().position(|inner_expr| {
@@ -275,16 +276,16 @@ impl<'a> HavingOrderByValidator<'a> {
         }
     }
 
-    fn agg_miss(expr: ExprRef, arena: &PlanArena<'_>) -> DatabaseError {
+    fn agg_miss(expr: ExprRef, arena: &dyn MetaArena) -> DatabaseError {
         DatabaseError::AggMiss(format!(
             "expression '{}' must appear in the GROUP BY clause or be used in an aggregate function",
             expr.output_name(arena)
         ))
     }
 }
 
-impl ExprVisitor<PlanArena<'_>> for HavingOrderByValidator<'_> {
-    fn visit(&mut self, expr: ExprRef, arena: &PlanArena<'_>) -> Result<(), DatabaseError> {
+impl ExprVisitor<dyn MetaArena + '_> for HavingOrderByValidator<'_> {
+    fn visit(&mut self, expr: ExprRef, arena: &(dyn MetaArena + '_)) -> Result<(), DatabaseError> {
         let contains = |expressions: &[ExprRef]| {
             expressions
                 .iter()
@@ -334,7 +335,7 @@ impl<'a> AggregateOutputBinder<'a> {
     fn output_ref(
         &mut self,
         expr: ExprRef,
-        arena: &mut PlanArena<'_>,
+        arena: &mut dyn MetaArena,
     ) -> Result<Option<ScalarExpression>, DatabaseError> {
         let output_count = self.agg_calls.len() + self.group_by_exprs.len();
         self.agg_calls
@@ -370,7 +371,7 @@ impl ExprVisitorMut for AggregateOutputBinder<'_> {
     fn visit(
         &mut self,
         expr: &mut ExprRef,
-        arena: &mut PlanArena<'_>,
+        arena: &mut (dyn MetaArena + '_),
     ) -> Result<(), DatabaseError> {
         if let ScalarExpression::Alias {
             alias: crate::expression::AliasType::Name(_),
@@ -401,7 +402,6 @@ mod tests {
     use crate::expression::{AliasType, BinaryOperator, ScalarExpression};
     use crate::planner::{ExprRef, PlanArena};
     use crate::storage::Storage;
-    use crate::types::value::DataValue;
     use crate::types::LogicalType;
 
     fn test_column(arena: &mut PlanArena, name: &str, ty: LogicalType) -> ColumnRef {
@@ -504,7 +504,7 @@ mod tests {
         let scala_functions = Default::default();
         let table_functions = Default::default();
         let transaction = tables.storage.transaction()?;
-        let args: [(&'static str, DataValue); 0] = [];
+        let args: [(usize, LogicalType); 0] = [];
         let mut binder = Binder::new(
             BinderContext::new(
                 &tables.table_cache,

src/binder/alter_table.rs

@@ -23,10 +23,9 @@ use crate::planner::operator::alter_table::drop_column::DropColumnOperator;
 use crate::planner::operator::Operator;
 use crate::planner::{Childrens, LogicalPlan};
 use crate::storage::Transaction;
-use crate::types::value::DataValue;
 use crate::types::LogicalType;
 
-impl<T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'_, '_, T, A> {
+impl<T: Transaction, A: AsRef<[(usize, LogicalType)]>> Binder<'_, '_, T, A> {
     pub(crate) fn bind_add_column(
         &mut self,
         table_name: TableName,

src/binder/analyze.rs

@@ -20,9 +20,9 @@ use crate::planner::operator::table_scan::TableScanOperator;
 use crate::planner::operator::Operator;
 use crate::planner::{Childrens, LogicalPlan};
 use crate::storage::Transaction;
-use crate::types::value::DataValue;
+use crate::types::LogicalType;
 
-impl<T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'_, '_, T, A> {
+impl<T: Transaction, A: AsRef<[(usize, LogicalType)]>> Binder<'_, '_, T, A> {
     pub(crate) fn bind_analyze(
         &mut self,
         table_name: TableName,

src/binder/copy.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use crate::types::LogicalType;
 use std::path::PathBuf;
 use std::str::FromStr;
 
@@ -64,7 +65,7 @@ impl FromStr for ExtSource {
     }
 }
 
-impl<T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'_, '_, T, A> {
+impl<T: Transaction, A: AsRef<[(usize, LogicalType)]>> Binder<'_, '_, T, A> {
     pub(super) fn bind_copy_to_file(
         &mut self,
         target: ExtSource,

src/binder/create_index.rs

@@ -21,9 +21,9 @@ use crate::planner::operator::Operator;
 use crate::planner::{Childrens, LogicalPlan};
 use crate::storage::Transaction;
 use crate::types::index::IndexType;
-use crate::types::value::DataValue;
+use crate::types::LogicalType;
 
-impl<T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'_, '_, T, A> {
+impl<T: Transaction, A: AsRef<[(usize, LogicalType)]>> Binder<'_, '_, T, A> {
     pub(crate) fn bind_create_index_source(
         &mut self,
         table_name: TableName,

src/binder/create_table.rs

@@ -19,10 +19,10 @@ use crate::planner::operator::create_table::CreateTableOperator;
 use crate::planner::operator::Operator;
 use crate::planner::{Childrens, LogicalPlan};
 use crate::storage::Transaction;
-use crate::types::value::DataValue;
+use crate::types::LogicalType;
 use std::collections::HashSet;
 
-impl<T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'_, '_, T, A> {
+impl<T: Transaction, A: AsRef<[(usize, LogicalType)]>> Binder<'_, '_, T, A> {
     // TODO: TableConstraint
     pub(crate) fn bind_create_table(
         &mut self,

src/binder/create_view.rs

@@ -21,9 +21,9 @@ use crate::planner::operator::create_view::CreateViewOperator;
 use crate::planner::operator::Operator;
 use crate::planner::{Childrens, ExprRef, LogicalPlan};
 use crate::storage::Transaction;
-use crate::types::value::DataValue;
+use crate::types::LogicalType;
 
-impl<T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'_, '_, T, A> {
+impl<T: Transaction, A: AsRef<[(usize, LogicalType)]>> Binder<'_, '_, T, A> {
     pub(crate) fn bind_create_view(
         &mut self,
         view_name: TableName,

src/binder/delete.rs

@@ -19,9 +19,9 @@ use crate::planner::operator::delete::DeleteOperator;
 use crate::planner::operator::Operator;
 use crate::planner::{Childrens, LogicalPlan};
 use crate::storage::Transaction;
-use crate::types::value::DataValue;
+use crate::types::LogicalType;
 
-impl<T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'_, '_, T, A> {
+impl<T: Transaction, A: AsRef<[(usize, LogicalType)]>> Binder<'_, '_, T, A> {
     pub(crate) fn bind_delete(
         &mut self,
         table_name: TableName,

src/binder/describe.rs

@@ -19,9 +19,9 @@ use crate::planner::operator::describe::DescribeOperator;
 use crate::planner::operator::Operator;
 use crate::planner::{Childrens, LogicalPlan};
 use crate::storage::Transaction;
-use crate::types::value::DataValue;
+use crate::types::LogicalType;
 
-impl<T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'_, '_, T, A> {
+impl<T: Transaction, A: AsRef<[(usize, LogicalType)]>> Binder<'_, '_, T, A> {
     pub(crate) fn bind_describe(
         &mut self,
         table_name: TableName,

src/binder/distinct.rs

@@ -18,11 +18,12 @@ use crate::expression::visitor_mut::{walk_mut_expr, ExprVisitorMut};
 use crate::expression::ScalarExpression;
 use crate::planner::operator::aggregate::AggregateOperator;
 use crate::planner::operator::sort::SortField;
-use crate::planner::{ExprRef, LogicalPlan, PlanArena};
+use crate::planner::MetaArena;
+use crate::planner::{ExprRef, LogicalPlan};
 use crate::storage::Transaction;
-use crate::types::value::DataValue;
+use crate::types::LogicalType;
 
-impl<T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'_, '_, T, A> {
+impl<T: Transaction, A: AsRef<[(usize, LogicalType)]>> Binder<'_, '_, T, A> {
     pub fn bind_distinct(
         &mut self,
         children: LogicalPlan,
@@ -83,7 +84,7 @@ impl<'a> DistinctOutputBinder<'a> {
         Self { select_list }
     }
 
-    fn output_ref(&mut self, expr: ExprRef, arena: &mut PlanArena<'_>) -> Option<ScalarExpression> {
+    fn output_ref(&mut self, expr: ExprRef, arena: &mut dyn MetaArena) -> Option<ScalarExpression> {
         self.select_list
             .iter()
             .position(|candidate| {
@@ -103,7 +104,7 @@ impl ExprVisitorMut for DistinctOutputBinder<'_> {
     fn visit(
         &mut self,
         expr: &mut ExprRef,
-        arena: &mut PlanArena<'_>,
+        arena: &mut (dyn MetaArena + '_),
     ) -> Result<(), DatabaseError> {
         if let ScalarExpression::Alias {
             alias: crate::expression::AliasType::Name(_),

src/binder/drop_index.rs

@@ -19,9 +19,9 @@ use crate::planner::operator::drop_index::DropIndexOperator;
 use crate::planner::operator::Operator;
 use crate::planner::{Childrens, LogicalPlan};
 use crate::storage::Transaction;
-use crate::types::value::DataValue;
+use crate::types::LogicalType;
 
-impl<T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'_, '_, T, A> {
+impl<T: Transaction, A: AsRef<[(usize, LogicalType)]>> Binder<'_, '_, T, A> {
     pub(crate) fn bind_drop_index(
         &mut self,
         table_name: TableName,

src/binder/drop_table.rs

@@ -19,9 +19,9 @@ use crate::planner::operator::drop_table::DropTableOperator;
 use crate::planner::operator::Operator;
 use crate::planner::{Childrens, LogicalPlan};
 use crate::storage::Transaction;
-use crate::types::value::DataValue;
+use crate::types::LogicalType;
 
-impl<T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'_, '_, T, A> {
+impl<T: Transaction, A: AsRef<[(usize, LogicalType)]>> Binder<'_, '_, T, A> {
     pub(crate) fn bind_drop_table(
         &mut self,
         table_name: TableName,

src/binder/drop_view.rs

@@ -19,9 +19,9 @@ use crate::planner::operator::drop_view::DropViewOperator;
 use crate::planner::operator::Operator;
 use crate::planner::{Childrens, LogicalPlan};
 use crate::storage::Transaction;
-use crate::types::value::DataValue;
+use crate::types::LogicalType;
 
-impl<T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'_, '_, T, A> {
+impl<T: Transaction, A: AsRef<[(usize, LogicalType)]>> Binder<'_, '_, T, A> {
     pub(crate) fn bind_drop_view(
         &mut self,
         view_name: TableName,

src/binder/explain.rs

@@ -17,9 +17,9 @@ use crate::errors::DatabaseError;
 use crate::planner::operator::Operator;
 use crate::planner::{Childrens, LogicalPlan};
 use crate::storage::Transaction;
-use crate::types::value::DataValue;
+use crate::types::LogicalType;
 
-impl<T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'_, '_, T, A> {
+impl<T: Transaction, A: AsRef<[(usize, LogicalType)]>> Binder<'_, '_, T, A> {
     pub(crate) fn bind_explain(&mut self, plan: LogicalPlan) -> Result<LogicalPlan, DatabaseError> {
         Ok(LogicalPlan::new(
             Operator::Explain,

src/binder/expr.rs

@@ -38,7 +38,7 @@ macro_rules! try_default {
     };
 }
 
-impl<'a, T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'a, '_, T, A> {
+impl<'a, T: Transaction, A: AsRef<[(usize, LogicalType)]>> Binder<'a, '_, T, A> {
     fn find_column_in_schema<'schema>(
         schema_ref: impl IntoIterator<Item = &'schema ColumnRef>,
         arena: &PlanArena,

src/binder/insert.rs

@@ -18,21 +18,22 @@ use crate::errors::DatabaseError;
 use crate::planner::operator::insert::InsertOperator;
 use crate::planner::operator::values::ValuesOperator;
 use crate::planner::operator::Operator;
-use crate::planner::{Childrens, LogicalPlan};
+use crate::planner::{Childrens, ExprRef, LogicalPlan};
 use crate::storage::Transaction;
 use crate::types::tuple::Schema;
-use crate::types::value::DataValue;
+use crate::types::LogicalType;
 
-impl<T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'_, '_, T, A> {
+impl<T: Transaction, A: AsRef<[(usize, LogicalType)]>> Binder<'_, '_, T, A> {
     pub(crate) fn bind_insert_values(
         &mut self,
         table_name: TableName,
         schema_ref: Schema,
-        rows: Vec<Vec<DataValue>>,
+        rows: Vec<ExprRef>,
+        row_count: usize,
         is_overwrite: bool,
         is_mapping_by_name: bool,
     ) -> Result<LogicalPlan, DatabaseError> {
-        let values_plan = self.bind_values(rows, schema_ref);
+        let values_plan = self.bind_values(rows, row_count, schema_ref);
 
         Ok(LogicalPlan::new(
             Operator::Insert(InsertOperator {
@@ -62,11 +63,12 @@ impl<T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'_, '_, T, A>
 
     pub(crate) fn bind_values(
         &mut self,
-        rows: Vec<Vec<DataValue>>,
+        rows: Vec<ExprRef>,
+        row_count: usize,
         schema_ref: Schema,
     ) -> LogicalPlan {
         LogicalPlan::new(
-            Operator::Values(ValuesOperator { rows, schema_ref }),
+            Operator::Values(ValuesOperator::new(rows, row_count, schema_ref)),
             Childrens::None,
         )
     }