refactor: Use TupleValueSerializable to simplify DataValue serialization (#289)

Author: KKould

src/binder/select.rs

@@ -159,10 +159,7 @@ impl<'a: 'b, 'b, T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'
         Ok(plan)
     }
 
-    fn bind_temp_values(
-        &mut self,
-        expr_rows: &Vec<Vec<Expr>>,
-    ) -> Result<LogicalPlan, DatabaseError> {
+    fn bind_temp_values(&mut self, expr_rows: &[Vec<Expr>]) -> Result<LogicalPlan, DatabaseError> {
         let values_len = expr_rows[0].len();
 
         let mut inferred_types: Vec<Option<LogicalType>> = vec![None; values_len];

src/catalog/table.rs

@@ -89,12 +89,6 @@ impl TableCatalog {
         &self.primary_key_indices
     }
 
-    pub(crate) fn types(&self) -> Vec<LogicalType> {
-        self.columns()
-            .map(|column| column.datatype().clone())
-            .collect_vec()
-    }
-
     /// Add a column to the table catalog.
     pub(crate) fn add_column(
         &mut self,

src/execution/ddl/add_column.rs

@@ -9,6 +9,7 @@ use crate::types::value::DataValue;
 use crate::{
     planner::operator::alter_table::add_column::AddColumnOperator, storage::Transaction, throw,
 };
+use itertools::Itertools;
 use std::ops::Coroutine;
 use std::ops::CoroutineState;
 use std::pin::Pin;
@@ -69,9 +70,14 @@ impl<'a, T: Transaction + 'a> WriteExecutor<'a, T> for AddColumn {
                 }
                 drop(coroutine);
 
+                let serializers = types.iter().map(|ty| ty.serializable()).collect_vec();
                 for tuple in tuples {
-                    throw!(unsafe { &mut (*transaction) }
-                        .append_tuple(table_name, tuple, &types, true));
+                    throw!(unsafe { &mut (*transaction) }.append_tuple(
+                        table_name,
+                        tuple,
+                        &serializers,
+                        true
+                    ));
                 }
                 let col_id = throw!(unsafe { &mut (*transaction) }.add_column(
                     cache.0,

src/execution/ddl/drop_column.rs

@@ -6,6 +6,7 @@ use crate::storage::{StatisticsMetaCache, TableCache, Transaction, ViewCache};
 use crate::throw;
 use crate::types::tuple::Tuple;
 use crate::types::tuple_builder::TupleBuilder;
+use itertools::Itertools;
 use std::ops::Coroutine;
 use std::ops::CoroutineState;
 use std::pin::Pin;
@@ -66,11 +67,12 @@ impl<'a, T: Transaction + 'a> WriteExecutor<'a, T> for DropColumn {
                         tuples.push(tuple);
                     }
                     drop(coroutine);
+                    let serializers = types.iter().map(|ty| ty.serializable()).collect_vec();
                     for tuple in tuples {
                         throw!(unsafe { &mut (*transaction) }.append_tuple(
                             &table_name,
                             tuple,
-                            &types,
+                            &serializers,
                             true
                         ));
                     }

src/execution/dml/copy_from_file.rs

@@ -5,8 +5,9 @@ use crate::execution::{Executor, WriteExecutor};
 use crate::planner::operator::copy_from_file::CopyFromFileOperator;
 use crate::storage::{StatisticsMetaCache, TableCache, Transaction, ViewCache};
 use crate::throw;
-use crate::types::tuple::{types, Tuple};
+use crate::types::tuple::Tuple;
 use crate::types::tuple_builder::TupleBuilder;
+use itertools::Itertools;
 use std::fs::File;
 use std::io::BufReader;
 use std::sync::mpsc;
@@ -33,7 +34,12 @@ impl<'a, T: Transaction + 'a> WriteExecutor<'a, T> for CopyFromFile {
         Box::new(
             #[coroutine]
             move || {
-                let types = types(&self.op.schema_ref);
+                let serializers = self
+                    .op
+                    .schema_ref
+                    .iter()
+                    .map(|column| column.datatype().serializable())
+                    .collect_vec();
                 let (tx, rx) = mpsc::channel();
                 let (tx1, rx1) = mpsc::channel();
                 // # Cancellation
@@ -50,7 +56,7 @@ impl<'a, T: Transaction + 'a> WriteExecutor<'a, T> for CopyFromFile {
                     throw!(unsafe { &mut (*transaction) }.append_tuple(
                         table.name(),
                         chunk,
-                        &types,
+                        &serializers,
                         false
                     ));
                     size += 1;

src/execution/dml/insert.rs

@@ -99,7 +99,10 @@ impl<'a, T: Transaction + 'a> WriteExecutor<'a, T> for Insert {
                         index_metas.push((index_meta, exprs));
                     }
 
-                    let types = table_catalog.types();
+                    let serializers = table_catalog
+                        .columns()
+                        .map(|column| column.datatype().serializable())
+                        .collect_vec();
                     let pk_indices = table_catalog.primary_keys_indices();
                     let mut coroutine = build_read(input, cache, transaction);
 
@@ -147,7 +150,7 @@ impl<'a, T: Transaction + 'a> WriteExecutor<'a, T> for Insert {
                         throw!(unsafe { &mut (*transaction) }.append_tuple(
                             &table_name,
                             tuple,
-                            &types,
+                            &serializers,
                             is_overwrite
                         ));
                         inserted_count += 1;

src/execution/dml/update.rs

@@ -8,10 +8,10 @@ use crate::planner::LogicalPlan;
 use crate::storage::{StatisticsMetaCache, TableCache, Transaction, ViewCache};
 use crate::throw;
 use crate::types::index::Index;
-use crate::types::tuple::types;
 use crate::types::tuple::Tuple;
 use crate::types::tuple_builder::TupleBuilder;
 use crate::types::value::DataValue;
+use itertools::Itertools;
 use std::collections::HashMap;
 use std::ops::Coroutine;
 use std::ops::CoroutineState;
@@ -62,7 +62,10 @@ impl<'a, T: Transaction + 'a> WriteExecutor<'a, T> for Update {
                 }
 
                 let input_schema = input.output_schema().clone();
-                let types = types(&input_schema);
+                let serializers = input_schema
+                    .iter()
+                    .map(|column| column.datatype().serializable())
+                    .collect_vec();
                 let mut updated_count = 0;
 
                 if let Some(table_catalog) =
@@ -133,7 +136,7 @@ impl<'a, T: Transaction + 'a> WriteExecutor<'a, T> for Update {
                         throw!(unsafe { &mut (*transaction) }.append_tuple(
                             &table_name,
                             tuple,
-                            &types,
+                            &serializers,
                             is_overwrite
                         ));
                         updated_count += 1;

src/execution/dql/index_scan.rs

@@ -50,7 +50,7 @@ impl<'a, T: Transaction + 'a> ReadExecutor<'a, T> for IndexScan {
                     columns,
                     self.index_by,
                     self.ranges,
-                    with_pk,
+                    with_pk
                 ));
 
                 while let Some(tuple) = throw!(iter.next_tuple()) {

src/execution/marco.rs renamed to src/execution/execute_macro.rs

@@ -1,7 +1,7 @@
 pub(crate) mod ddl;
 pub(crate) mod dml;
 pub(crate) mod dql;
-pub(crate) mod marco;
+pub(crate) mod execute_macro;
 
 use self::ddl::add_column::AddColumn;
 use self::dql::join::nested_loop_join::NestedLoopJoin;