perf: use byteorder on DataValue::to_raw & DataValue::from_raw

Author: KKould

Cargo.toml

@@ -35,11 +35,10 @@ harness = false
 [dependencies]
 ahash                 = { version = "0.8" }
 bincode               = { version = "1" }
-bytes                 = { version = "1" }
+byteorder             = { version = "1" }
 chrono                = { version = "0.4" }
 comfy-table           = { version = "7" }
 csv                   = { version = "1" }
-encode_unicode        = { version = "1" }
 dirs                  = { version = "5" }
 itertools             = { version = "0.12" }
 ordered-float         = { version = "4" }
@@ -68,10 +67,8 @@ tokio                 = { version = "1.36", features = ["full"], optional = true
 
 
 [dev-dependencies]
-cargo-tarpaulin       = { version = "0.27" }
 criterion             = { version = "0.5", features = ["html_reports"] }
 indicatif             = { version = "0.17" }
-rand_distr            = { version = "0.4" }
 tempfile              = { version = "3.10" }
 # Benchmark
 sqlite                = { version = "0.34" }

README.md

@@ -73,13 +73,13 @@ run `cargo run -p tpcc --release` to run tpcc
 - Tips: TPC-C currently only supports single thread
 ```shell
 <90th Percentile RT (MaxRT)>
-   New-Order : 0.003  (0.012)
-     Payment : 0.001  (0.003)
-Order-Status : 0.054  (0.188)
-    Delivery : 0.021  (0.049)
- Stock-Level : 0.004  (0.006)
+   New-Order : 0.003  (0.031)
+     Payment : 0.001  (0.028)
+Order-Status : 0.053  (0.144)
+    Delivery : 0.020  (0.032)
+ Stock-Level : 0.004  (0.007)
 <TpmC>
-7345 Tpmc
+7457 Tpmc
 ```
 #### 👉[check more](tpcc/README.md)
 

src/errors.rs

@@ -4,7 +4,7 @@ use crate::types::LogicalType;
 use chrono::ParseError;
 use sqlparser::parser::ParserError;
 use std::num::{ParseFloatError, ParseIntError, TryFromIntError};
-use std::str::ParseBoolError;
+use std::str::{ParseBoolError, Utf8Error};
 use std::string::FromUtf8Error;
 
 #[derive(thiserror::Error, Debug)]
@@ -161,6 +161,12 @@ pub enum DatabaseError {
     UnsupportedBinaryOperator(LogicalType, BinaryOperator),
     #[error("unsupported statement: {0}")]
     UnsupportedStmt(String),
+    #[error("utf8: {0}")]
+    Utf8(
+        #[source]
+        #[from]
+        Utf8Error,
+    ),
     #[error("values length not match, expect {0}, got {1}")]
     ValuesLenMismatch(usize, usize),
     #[error("the view already exists")]

src/serdes/char.rs

@@ -1,7 +1,6 @@
 use crate::errors::DatabaseError;
 use crate::serdes::{ReferenceSerialization, ReferenceTables};
 use crate::storage::{TableCache, Transaction};
-use encode_unicode::CharExt;
 use std::io::{Read, Write};
 
 impl ReferenceSerialization for char {
@@ -11,20 +10,21 @@ impl ReferenceSerialization for char {
         _: bool,
         _: &mut ReferenceTables,
     ) -> Result<(), DatabaseError> {
-        let (bytes, _) = self.to_utf8_array();
+        let mut buf = [0u8; 2];
+        self.encode_utf8(&mut buf);
 
-        Ok(writer.write_all(&bytes)?)
+        Ok(writer.write_all(&buf)?)
     }
 
     fn decode<T: Transaction, R: Read>(
         reader: &mut R,
         _: Option<(&T, &TableCache)>,
         _: &ReferenceTables,
     ) -> Result<Self, DatabaseError> {
-        let mut buf = [0u8; 4];
+        let mut buf = [0u8; 2];
         reader.read_exact(&mut buf)?;
 
         // SAFETY
-        Ok(char::from_utf8_array(buf).unwrap())
+        Ok(std::str::from_utf8(&buf)?.chars().next().unwrap())
     }
 }

src/serdes/data_value.rs

@@ -1,118 +1,7 @@
-use crate::errors::DatabaseError;
-use crate::serdes::{ReferenceSerialization, ReferenceTables};
-use crate::storage::{TableCache, Transaction};
+use crate::implement_serialization_by_bincode;
 use crate::types::value::DataValue;
-use crate::types::LogicalType;
-use std::io::{Read, Write};
 
-impl DataValue {
-    // FIXME: redundant code
-    pub(crate) fn inner_encode<W: Write>(
-        &self,
-        writer: &mut W,
-        ty: &LogicalType,
-    ) -> Result<(), DatabaseError> {
-        writer.write_all(&[if self.is_null() { 0u8 } else { 1u8 }])?;
-
-        if self.is_null() {
-            return Ok(());
-        }
-        if let DataValue::Tuple(values) = self {
-            match values {
-                None => writer.write_all(&[0u8])?,
-                Some((values, is_upper)) => {
-                    writer.write_all(&[1u8])?;
-                    writer.write_all(&(values.len() as u32).to_le_bytes())?;
-                    for value in values.iter() {
-                        value.inner_encode(writer, &value.logical_type())?
-                    }
-                    writer.write_all(&[if *is_upper { 1u8 } else { 0u8 }])?;
-                }
-            }
-
-            return Ok(());
-        }
-        if ty.raw_len().is_none() {
-            let mut bytes = Vec::new();
-            writer.write_all(&(self.to_raw(&mut bytes)? as u32).to_le_bytes())?;
-            writer.write_all(&bytes)?;
-        } else {
-            let _ = self.to_raw(writer)?;
-        }
-
-        Ok(())
-    }
-
-    pub(crate) fn inner_decode<R: Read>(
-        reader: &mut R,
-        ty: &LogicalType,
-    ) -> Result<Self, DatabaseError> {
-        let mut bytes = [0u8; 1];
-        reader.read_exact(&mut bytes)?;
-        if bytes[0] == 0 {
-            return Ok(DataValue::none(ty));
-        }
-        if let LogicalType::Tuple(types) = ty {
-            let mut bytes = [0u8; 1];
-            reader.read_exact(&mut bytes)?;
-            let values = match bytes[0] {
-                0 => None,
-                1 => {
-                    let mut bytes = [0u8; 4];
-                    reader.read_exact(&mut bytes)?;
-                    let len = u32::from_le_bytes(bytes) as usize;
-                    let mut vec = Vec::with_capacity(len);
-
-                    for ty in types.iter() {
-                        vec.push(Self::inner_decode(reader, ty)?);
-                    }
-                    let mut bytes = [0u8];
-                    reader.read_exact(&mut bytes)?;
-                    Some((vec, bytes[0] == 1))
-                }
-                _ => unreachable!(),
-            };
-
-            return Ok(DataValue::Tuple(values));
-        }
-        let value_len = match ty.raw_len() {
-            None => {
-                let mut bytes = [0u8; 4];
-                reader.read_exact(&mut bytes)?;
-                u32::from_le_bytes(bytes) as usize
-            }
-            Some(len) => len,
-        };
-        let mut buf = vec![0u8; value_len];
-        reader.read_exact(&mut buf)?;
-
-        Ok(DataValue::from_raw(&buf, ty))
-    }
-}
-
-impl ReferenceSerialization for DataValue {
-    fn encode<W: Write>(
-        &self,
-        writer: &mut W,
-        is_direct: bool,
-        reference_tables: &mut ReferenceTables,
-    ) -> Result<(), DatabaseError> {
-        let ty = self.logical_type();
-        ty.encode(writer, is_direct, reference_tables)?;
-
-        self.inner_encode(writer, &ty)
-    }
-
-    fn decode<T: Transaction, R: Read>(
-        reader: &mut R,
-        drive: Option<(&T, &TableCache)>,
-        reference_tables: &ReferenceTables,
-    ) -> Result<Self, DatabaseError> {
-        let logical_type = LogicalType::decode(reader, drive, reference_tables)?;
-
-        Self::inner_decode(reader, &logical_type)
-    }
-}
+implement_serialization_by_bincode!(DataValue);
 
 #[cfg(test)]
 pub(crate) mod test {

src/serdes/mod.rs

@@ -33,37 +33,20 @@ macro_rules! implement_serialization_by_bincode {
             fn encode<W: std::io::Write>(
                 &self,
                 writer: &mut W,
-                is_direct: bool,
-                reference_tables: &mut $crate::serdes::ReferenceTables,
+                _: bool,
+                _: &mut $crate::serdes::ReferenceTables,
             ) -> Result<(), $crate::errors::DatabaseError> {
-                let bytes = bincode::serialize(self)?;
-                $crate::serdes::ReferenceSerialization::encode(
-                    &bytes.len(),
-                    writer,
-                    is_direct,
-                    reference_tables,
-                )?;
-                std::io::Write::write_all(writer, &bytes)?;
+                bincode::serialize_into(writer, self)?;
 
                 Ok(())
             }
 
             fn decode<T: $crate::storage::Transaction, R: std::io::Read>(
                 reader: &mut R,
-                drive: Option<(&T, &$crate::storage::TableCache)>,
-                reference_tables: &$crate::serdes::ReferenceTables,
+                _: Option<(&T, &$crate::storage::TableCache)>,
+                _: &$crate::serdes::ReferenceTables,
             ) -> Result<Self, $crate::errors::DatabaseError> {
-                let mut buf = vec![
-                    0u8;
-                    <usize as $crate::serdes::ReferenceSerialization>::decode(
-                        reader,
-                        drive,
-                        reference_tables
-                    )?
-                ];
-                std::io::Read::read_exact(reader, &mut buf)?;
-
-                Ok(bincode::deserialize::<Self>(&buf)?)
+                Ok(bincode::deserialize_from(reader)?)
             }
         }
     };

src/storage/mod.rs

@@ -9,13 +9,12 @@ use crate::errors::DatabaseError;
 use crate::expression::range_detacher::Range;
 use crate::optimizer::core::statistics_meta::{StatisticMetaLoader, StatisticsMeta};
 use crate::serdes::ReferenceTables;
-use crate::storage::table_codec::TableCodec;
+use crate::storage::table_codec::{Bytes, TableCodec};
 use crate::types::index::{Index, IndexId, IndexMetaRef, IndexType};
 use crate::types::tuple::{Tuple, TupleId};
 use crate::types::value::DataValue;
 use crate::types::{ColumnId, LogicalType};
 use crate::utils::lru::SharedLruCache;
-use bytes::Bytes;
 use itertools::Itertools;
 use std::collections::Bound;
 use std::io::Cursor;
@@ -64,8 +63,9 @@ pub trait Transaction: Sized {
         let pk_indices = table.primary_keys_indices();
         let table_types = table.types();
         if columns.is_empty() {
-            let (i, column) = &table.primary_keys()[0];
-            columns.push((*i, column.clone()));
+            for (i, column) in table.primary_keys() {
+                columns.push((*i, column.clone()));
+            }
         }
         let mut tuple_columns = Vec::with_capacity(columns.len());
         let mut projections = Vec::with_capacity(columns.len());
@@ -535,7 +535,7 @@ pub trait Transaction: Sized {
         // Tips: only `Column`, `IndexMeta`, `TableMeta`
         while let Some((key, value)) = column_iter.try_next().ok().flatten() {
             if key.starts_with(&table_min) {
-                let mut cursor = Cursor::new(value.as_ref());
+                let mut cursor = Cursor::new(value);
                 columns.push(TableCodec::decode_column::<Self, _>(
                     &mut cursor,
                     &reference_tables,
@@ -693,15 +693,18 @@ impl<T: Transaction> IndexImplParams<'_, T> {
     ) -> Result<Option<Tuple>, DatabaseError> {
         let key = TableCodec::encode_tuple_key(self.table_name, tuple_id)?;
 
-        Ok(self.tx.get(&key)?.map(|bytes| {
-            TableCodec::decode_tuple(
-                &self.table_types,
-                pk_indices,
-                &self.projections,
-                &self.tuple_schema_ref,
-                &bytes,
-            )
-        }))
+        self.tx
+            .get(&key)?
+            .map(|bytes| {
+                TableCodec::decode_tuple(
+                    &self.table_types,
+                    pk_indices,
+                    &self.projections,
+                    &self.tuple_schema_ref,
+                    &bytes,
+                )
+            })
+            .transpose()
     }
 }
 
@@ -760,13 +763,13 @@ impl<T: Transaction> IndexImpl<T> for PrimaryKeyIndexImpl {
         pk_indices: &PrimaryKeyIndices,
         params: &IndexImplParams<T>,
     ) -> Result<Tuple, DatabaseError> {
-        Ok(TableCodec::decode_tuple(
+        TableCodec::decode_tuple(
             &params.table_types,
             pk_indices,
             &params.projections,
             &params.tuple_schema_ref,
             bytes,
-        ))
+        )
     }
 
     fn eq_to_res<'a>(
@@ -786,7 +789,8 @@ impl<T: Transaction> IndexImpl<T> for PrimaryKeyIndexImpl {
                     &params.tuple_schema_ref,
                     &bytes,
                 )
-            });
+            })
+            .transpose()?;
         Ok(IndexResult::Tuple(tuple))
     }
 
@@ -804,7 +808,7 @@ fn secondary_index_lookup<T: Transaction>(
     pk_indices: &PrimaryKeyIndices,
     params: &IndexImplParams<T>,
 ) -> Result<Tuple, DatabaseError> {
-    let tuple_id = TableCodec::decode_index(bytes, &params.index_meta.pk_ty)?;
+    let tuple_id = TableCodec::decode_index(bytes)?;
     params
         .get_tuple_by_id(pk_indices, &tuple_id)?
         .ok_or(DatabaseError::TupleIdNotFound(tuple_id))
@@ -829,7 +833,7 @@ impl<T: Transaction> IndexImpl<T> for UniqueIndexImpl {
         let Some(bytes) = params.tx.get(&self.bound_key(params, value)?)? else {
             return Ok(IndexResult::Tuple(None));
         };
-        let tuple_id = TableCodec::decode_index(&bytes, &params.index_meta.pk_ty)?;
+        let tuple_id = TableCodec::decode_index(&bytes)?;
         let tuple = params
             .get_tuple_by_id(pk_indices, &tuple_id)?
             .ok_or(DatabaseError::TupleIdNotFound(tuple_id))?;
@@ -950,7 +954,7 @@ impl<'a, T: Transaction + 'a> Iter for TupleIter<'a, T> {
                 &self.projections,
                 &self.tuple_columns,
                 &value,
-            );
+            )?;
 
             if let Some(num) = self.limit.as_mut() {
                 num.sub_assign(1);