Perf/tpcc order status

README.md

@@ -88,15 +88,17 @@ run `cargo run -p tpcc --release` to run tpcc
 - 32.0 GB
 - KIOXIA-EXCERIA PLUS G3 SSD
 - Tips: TPC-C currently only supports single thread
+
+All cases have been fully optimized.
 ```shell
 <90th Percentile RT (MaxRT)>
-   New-Order : 0.002  (0.005)
-     Payment : 0.001  (0.003)
-Order-Status : 0.057  (0.088)
+   New-Order : 0.002  (0.012)
+     Payment : 0.001  (0.002)
+Order-Status : 0.002  (0.019)
     Delivery : 0.001  (0.001)
- Stock-Level : 0.002  (0.006)
+ Stock-Level : 0.002  (0.018)
 <TpmC>
-11125 Tpmc
+37166 Tpmc
 ```
 #### 👉[check more](tpcc/README.md)
 

src/binder/aggregate.rs

@@ -98,7 +98,7 @@ impl<T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'_, '_, T, A>
                 return_orderby.push(SortField::new(
                     expr,
                     asc.is_none_or(|asc| asc),
-                    nulls_first.is_some_and(|first| first),
+                    nulls_first.unwrap_or(true),
                 ));
             }
             Some(return_orderby)

src/binder/alter_table.rs

@@ -44,7 +44,7 @@ impl<T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'_, '_, T, A>
                 if_not_exists,
                 column_def,
             } => {
-                let plan = TableScanOperator::build(table_name.clone(), table, true);
+                let plan = TableScanOperator::build(table_name.clone(), table, true)?;
                 let column = self.bind_column(column_def, None)?;
 
                 if !is_valid_identifier(column.name()) {
@@ -66,7 +66,7 @@ impl<T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'_, '_, T, A>
                 if_exists,
                 ..
             } => {
-                let plan = TableScanOperator::build(table_name.clone(), table, true);
+                let plan = TableScanOperator::build(table_name.clone(), table, true)?;
                 let column_name = column_name.value.clone();
 
                 LogicalPlan::new(

src/binder/analyze.rs

@@ -40,7 +40,7 @@ impl<T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'_, '_, T, A>
             .ok_or(DatabaseError::TableNotFound)?;
         let index_metas = table.indexes.clone();
 
-        let scan_op = TableScanOperator::build(table_name.clone(), table, false);
+        let scan_op = TableScanOperator::build(table_name.clone(), table, false)?;
         Ok(LogicalPlan::new(
             Operator::Analyze(AnalyzeOperator {
                 table_name,

src/binder/copy.rs

@@ -118,7 +118,9 @@ impl<T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'_, '_, T, A>
                         target: ext_source,
                         schema_ref,
                     }),
-                    Childrens::Only(Box::new(TableScanOperator::build(table_name, table, false))),
+                    Childrens::Only(Box::new(TableScanOperator::build(
+                        table_name, table, false,
+                    )?)),
                 ))
             } else {
                 // COPY <dest_table> FROM <source_file>

src/binder/create_index.rs

@@ -49,7 +49,7 @@ impl<T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'_, '_, T, A>
             .source_and_bind(table_name.clone(), None, None, false)?
             .ok_or(DatabaseError::SourceNotFound)?;
         let plan = match source {
-            Source::Table(table) => TableScanOperator::build(table_name.clone(), table, true),
+            Source::Table(table) => TableScanOperator::build(table_name.clone(), table, true)?,
             Source::View(view) => LogicalPlan::clone(&view.plan),
         };
         let mut columns = Vec::with_capacity(exprs.len());

src/binder/delete.rs

@@ -51,7 +51,7 @@ impl<T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'_, '_, T, A>
                 .iter()
                 .map(|(_, column)| column.clone())
                 .collect_vec();
-            let mut plan = TableScanOperator::build(table_name.clone(), table, true);
+            let mut plan = TableScanOperator::build(table_name.clone(), table, true)?;
 
             if let Some(alias_idents) = alias_idents {
                 plan =

src/binder/select.rs

@@ -541,7 +541,7 @@ impl<'a: 'b, 'b, T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'
             .source_and_bind(table_name.clone(), table_alias.as_ref(), join_type, false)?
             .ok_or(DatabaseError::SourceNotFound)?;
         let mut plan = match source {
-            Source::Table(table) => TableScanOperator::build(table_name.clone(), table, with_pk),
+            Source::Table(table) => TableScanOperator::build(table_name.clone(), table, with_pk)?,
             Source::View(view) => LogicalPlan::clone(&view.plan),
         };
 

src/catalog/table.rs

@@ -38,7 +38,7 @@ pub struct TableCatalog {
     schema_ref: SchemaRef,
     primary_keys: Vec<(usize, ColumnRef)>,
     primary_key_indices: PrimaryKeyIndices,
-    primary_key_type: Option<LogicalType>,
+    primary_key_type: LogicalType,
 }
 
 //TODO: can add some like Table description and other information as attributes
@@ -99,6 +99,10 @@ impl TableCatalog {
         &self.primary_keys
     }
 
+    pub(crate) fn primary_keys_type(&self) -> &LogicalType {
+        &self.primary_key_type
+    }
+
     pub(crate) fn primary_keys_indices(&self) -> &PrimaryKeyIndices {
         &self.primary_key_indices
     }
@@ -144,23 +148,7 @@ impl TableCatalog {
         }
 
         let index_id = self.indexes.last().map(|index| index.id + 1).unwrap_or(0);
-        let pk_ty = self
-            .primary_key_type
-            .get_or_insert_with(|| {
-                let primary_keys = &self.primary_keys;
-
-                if primary_keys.len() == 1 {
-                    primary_keys[0].1.datatype().clone()
-                } else {
-                    LogicalType::Tuple(
-                        primary_keys
-                            .iter()
-                            .map(|(_, column)| column.datatype().clone())
-                            .collect_vec(),
-                    )
-                }
-            })
-            .clone();
+        let pk_ty = self.primary_key_type.clone();
 
         let mut val_tys = Vec::with_capacity(column_ids.len());
         for column_id in column_ids.iter() {
@@ -205,7 +193,7 @@ impl TableCatalog {
             schema_ref: Arc::new(vec![]),
             primary_keys: vec![],
             primary_key_indices: Default::default(),
-            primary_key_type: None,
+            primary_key_type: LogicalType::SqlNull,
         };
         let mut generator = Generator::new();
         for col_catalog in columns.into_iter() {
@@ -216,12 +204,26 @@ impl TableCatalog {
         let (primary_keys, primary_key_indices) =
             Self::build_primary_keys(&table_catalog.schema_ref);
 
+        table_catalog.primary_key_type = Self::build_primary_key_type(&primary_keys);
         table_catalog.primary_keys = primary_keys;
         table_catalog.primary_key_indices = primary_key_indices;
 
         Ok(table_catalog)
     }
 
+    fn build_primary_key_type(primary_keys: &[(usize, ColumnRef)]) -> LogicalType {
+        if primary_keys.len() == 1 {
+            primary_keys[0].1.datatype().clone()
+        } else {
+            LogicalType::Tuple(
+                primary_keys
+                    .iter()
+                    .map(|(_, column)| column.datatype().clone())
+                    .collect_vec(),
+            )
+        }
+    }
+
     pub(crate) fn reload(
         name: TableName,
         column_refs: Vec<ColumnRef>,
@@ -240,6 +242,7 @@ impl TableCatalog {
         }
         let schema_ref = Arc::new(column_refs.clone());
         let (primary_keys, primary_key_indices) = Self::build_primary_keys(&schema_ref);
+        let primary_key_type = Self::build_primary_key_type(&primary_keys);
 
         Ok(TableCatalog {
             name,
@@ -249,7 +252,7 @@ impl TableCatalog {
             schema_ref,
             primary_keys,
             primary_key_indices,
-            primary_key_type: None,
+            primary_key_type,
         })
     }