chore: codefmt

KKould · KKould · commit b012bb2270f8 · 2026-01-17T16:12:44.000+08:00
diff --git a/src/execution/dql/aggregate/stream_distinct.rs b/src/execution/dql/aggregate/stream_distinct.rs
@@ -64,7 +64,7 @@ impl<'a, T: Transaction + 'a> ReadExecutor<'a, T> for StreamDistinctExecutor {
 
                 if last_keys.as_ref() != Some(&group_keys) {
                     last_keys = Some(group_keys.clone());
-                    co.yield_(Ok(Tuple::new(None, group_keys))).await;
+                    co.yield_(Ok(Tuple::new(tuple.pk, group_keys))).await;
                 }
             }
         })
@@ -78,11 +78,14 @@ mod tests {
     use crate::execution::dql::aggregate::stream_distinct::StreamDistinctExecutor;
     use crate::execution::{try_collect, ReadExecutor};
     use crate::expression::ScalarExpression;
+    use crate::optimizer::heuristic::batch::HepBatchStrategy;
+    use crate::optimizer::heuristic::optimizer::HepOptimizerPipeline;
+    use crate::optimizer::rule::normalization::NormalizationRuleImpl;
     use crate::planner::operator::aggregate::AggregateOperator;
     use crate::planner::operator::values::ValuesOperator;
     use crate::planner::operator::Operator;
     use crate::planner::{Childrens, LogicalPlan};
-    use crate::storage::rocksdb::RocksStorage;
+    use crate::storage::rocksdb::{RocksStorage, RocksTransaction};
     use crate::storage::{StatisticsMetaCache, Storage, TableCache, ViewCache};
     use crate::types::value::DataValue;
     use crate::types::LogicalType;
@@ -113,6 +116,21 @@ mod tests {
         Ok((table_cache, view_cache, meta_cache, temp_dir, storage))
     }
 
+    fn optimize_exprs(plan: LogicalPlan) -> Result<LogicalPlan, DatabaseError> {
+        HepOptimizerPipeline::builder()
+            .before_batch(
+                "Expression Remapper".to_string(),
+                HepBatchStrategy::once_topdown(),
+                vec![
+                    NormalizationRuleImpl::BindExpressionPosition,
+                    NormalizationRuleImpl::EvaluatorBind,
+                ],
+            )
+            .build()
+            .instantiate(plan)
+            .find_best::<RocksTransaction>(None)
+    }
+
     #[test]
     fn stream_distinct_single_column_sorted() -> Result<(), DatabaseError> {
         let desc = ColumnDesc::new(LogicalType::Integer, None, false, None)?;
@@ -140,11 +158,16 @@ mod tests {
             agg_calls: vec![],
             is_distinct: true,
         };
+        let plan = LogicalPlan::new(Operator::Aggregate(agg), Childrens::Only(Box::new(input)));
+        let plan = optimize_exprs(plan)?;
+        let Operator::Aggregate(agg) = plan.operator else {
+            unreachable!()
+        };
 
         let (table_cache, view_cache, meta_cache, _temp_dir, storage) = build_test_storage()?;
         let mut transaction = storage.transaction()?;
         let tuples = try_collect(
-            StreamDistinctExecutor::from((agg, input))
+            StreamDistinctExecutor::from((agg, plan.childrens.pop_only()))
                 .execute((&table_cache, &view_cache, &meta_cache), &mut transaction),
         )?;
 
@@ -187,11 +210,16 @@ mod tests {
             agg_calls: vec![],
             is_distinct: true,
         };
+        let plan = LogicalPlan::new(Operator::Aggregate(agg), Childrens::Only(Box::new(input)));
+        let plan = optimize_exprs(plan)?;
+        let Operator::Aggregate(agg) = plan.operator else {
+            unreachable!()
+        };
 
         let (table_cache, view_cache, meta_cache, _temp_dir, storage) = build_test_storage()?;
         let mut transaction = storage.transaction()?;
         let tuples = try_collect(
-            StreamDistinctExecutor::from((agg, input))
+            StreamDistinctExecutor::from((agg, plan.childrens.pop_only()))
                 .execute((&table_cache, &view_cache, &meta_cache), &mut transaction),
         )?;
 
diff --git a/src/expression/simplify.rs b/src/expression/simplify.rs
@@ -100,8 +100,8 @@ impl VisitorMut<'_> for ConstantCalculator {
             ScalarExpression::TypeCast { expr: arg_expr, ty } => {
                 self.visit(arg_expr)?;
 
-                if let ScalarExpression::Constant(value) = arg_expr.as_ref() {
-                    let casted = value.clone().cast(ty)?;
+                if let ScalarExpression::Constant(value) = arg_expr.as_mut() {
+                    let casted = mem::replace(value, DataValue::Null).cast(ty)?;
                     let _ = mem::replace(expr, ScalarExpression::Constant(casted));
                 }
             }
diff --git a/tests/slt/where_by_index.slt b/tests/slt/where_by_index.slt
@@ -19,7 +19,7 @@ create index p_index on t1 (c1, c2);
 statement ok
 analyze table t1;
 
-query IIT
+query IIT   
 select * from t1 limit 10;
 ----
 0 1 2
@@ -256,3 +256,26 @@ select c2, c1 from t1 where c1 < 10 and c1 > 0 and c2 > 0 and c2 < 10;
 
 statement ok
 drop table t1;
+
+statement ok
+create table t_cover(id int primary key, c1 int, c2 int, c3 int);
+
+statement ok
+insert into t_cover values
+    (1, 1, 10, 11),
+    (2, 2, 20, 21),
+    (3, 2, 22, 23),
+    (4, 3, 30, 31);
+
+statement ok
+create index idx_cover on t_cover (c1, c2, c3);
+
+# composite index trailing columns cover (index columns > output columns)
+query II rowsort
+select c2, c3 from t_cover where c1 = 2;
+----
+20 21
+22 23
+
+statement ok
+drop table t_cover;
diff --git a/tpcc/src/backend/dual.rs b/tpcc/src/backend/dual.rs
@@ -290,226 +290,3 @@ fn sqlite_statement_spec(spec: &StatementSpec) -> StatementSpec {
         spec.clone()
     }
 }
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use crate::backend::{
-        BackendControl, BackendTransaction, ColumnType, DbParam, PreparedStatement, StatementSpec,
-    };
-    use chrono::NaiveDateTime;
-    use kite_sql::types::value::DataValue;
-    use rand::Rng;
-    use std::fs;
-    use std::path::PathBuf;
-
-    const DISTINCT_ORDER_BY_SQL: &str = "SELECT DISTINCT ol_i_id FROM order_line WHERE ol_w_id = ?1 AND ol_d_id = ?2 AND ol_o_id < ?3 AND ol_o_id >= (?4 - 20) ORDER BY ol_i_id";
-    const DISTINCT_DELIVERY_SQL: &str =
-        "SELECT DISTINCT ol_delivery_d FROM order_line WHERE ol_w_id = ?1 AND ol_d_id = ?2 ORDER BY ol_delivery_d";
-    const INT32_RESULT: [ColumnType; 1] = [ColumnType::Int32];
-    const DATETIME_RESULT: [ColumnType; 1] = [ColumnType::NullableDateTime];
-
-    struct TempKiteDir {
-        path: PathBuf,
-    }
-
-    impl TempKiteDir {
-        fn new() -> Result<Self, TpccError> {
-            let mut path = std::env::temp_dir();
-            let suffix: u64 = rand::thread_rng().gen();
-            path.push(format!("tpcc_dual_test_{suffix}"));
-            if path.exists() {
-                fs::remove_dir_all(&path)?;
-            }
-            Ok(Self { path })
-        }
-    }
-
-    impl Drop for TempKiteDir {
-        fn drop(&mut self) {
-            let _ = fs::remove_dir_all(&self.path);
-        }
-    }
-
-    fn prepare_single<B: BackendControl>(
-        backend: &B,
-        spec: StatementSpec,
-    ) -> Result<PreparedStatement, TpccError> {
-        let mut groups = backend.prepare_statements(&[vec![spec]])?;
-        Ok(groups
-            .pop()
-            .and_then(|mut group| group.pop())
-            .expect("missing prepared statement"))
-    }
-
-    fn collect_i32<B: BackendControl>(
-        backend: &B,
-        sql: &'static str,
-        params: &[DbParam],
-    ) -> Result<Vec<i32>, TpccError> {
-        let statement = prepare_single(
-            backend,
-            StatementSpec {
-                sql,
-                result_types: &INT32_RESULT,
-            },
-        )?;
-        let mut tx = backend.new_transaction()?;
-        let rows = {
-            let mut iter = tx.execute(&statement, params)?;
-            let mut rows = Vec::new();
-            while let Some(row) = iter.next() {
-                let tuple = row?;
-                let value = tuple.values[0].i32().expect("expected Int32 column");
-                rows.push(value);
-            }
-            rows
-        };
-        tx.commit()?;
-        Ok(rows)
-    }
-
-    fn collect_datetime<B: BackendControl>(
-        backend: &B,
-        sql: &'static str,
-        params: &[DbParam],
-    ) -> Result<Vec<Option<NaiveDateTime>>, TpccError> {
-        let statement = prepare_single(
-            backend,
-            StatementSpec {
-                sql,
-                result_types: &DATETIME_RESULT,
-            },
-        )?;
-        let mut tx = backend.new_transaction()?;
-        let rows = {
-            let mut iter = tx.execute(&statement, params)?;
-            let mut rows = Vec::new();
-            while let Some(row) = iter.next() {
-                let tuple = row?;
-                rows.push(tuple.values[0].datetime());
-            }
-            rows
-        };
-        tx.commit()?;
-        Ok(rows)
-    }
-
-    fn parse_dt(value: &str) -> NaiveDateTime {
-        NaiveDateTime::parse_from_str(value, "%Y-%m-%d %H:%M:%S").expect("invalid datetime literal")
-    }
-
-    #[test]
-    fn dual_stock_level_distinct_ordering_mismatch() -> Result<(), TpccError> {
-        let temp_dir = TempKiteDir::new()?;
-        let backend = DualBackend::new(
-            temp_dir
-                .path
-                .to_str()
-                .expect("temporary path should be valid utf-8"),
-        )?;
-
-        backend.execute_batch(
-            "create table order_line (
-                ol_o_id int not null,
-                ol_d_id tinyint not null,
-                ol_w_id smallint not null,
-                ol_number tinyint not null,
-                ol_i_id int,
-                ol_supply_w_id smallint,
-                ol_delivery_d datetime,
-                ol_quantity tinyint,
-                ol_amount decimal(6,2),
-                ol_dist_info char(24),
-                PRIMARY KEY(ol_w_id, ol_d_id, ol_o_id, ol_number)
-            );",
-        )?;
-        backend.execute_batch(
-            "CREATE INDEX fkey_order_line_1 ON order_line (ol_o_id, ol_d_id, ol_w_id);",
-        )?;
-
-        // Large distinct set makes accidental order matches vanishingly unlikely.
-        let i_ids = [
-            87181, 5901, 26250, 58438, 92124, 84344, 72959, 10014, 46998, 2731, 11001, 12002,
-            13003, 14004, 15005, 16006, 17007, 18008, 19009, 20010,
-        ];
-        let delivery_samples = [
-            None,
-            Some("2024-01-02 00:00:00"),
-            Some("2024-01-01 00:00:00"),
-            Some("2023-12-31 00:00:00"),
-        ];
-
-        for (idx, i_id) in i_ids.iter().enumerate() {
-            let o_id = idx + 1;
-            let delivery = delivery_samples
-                .get(idx)
-                .and_then(|value| *value)
-                .map(|value| format!("'{value}'"))
-                .unwrap_or_else(|| "NULL".to_string());
-            backend.execute_batch(&format!(
-                "insert into order_line values ({o_id}, 1, 1, 1, {i_id}, 1, {delivery}, 1, 1.00, 'dist')"
-            ))?;
-        }
-        backend.execute_batch(
-            "insert into order_line values (21, 1, 1, 1, 87181, 1, NULL, 1, 1.00, 'dist')",
-        )?;
-        backend.execute_batch(
-            "insert into order_line values (22, 2, 1, 1, 99999, 1, NULL, 1, 1.00, 'dist')",
-        )?;
-
-        let params: [DbParam; 4] = [
-            ("?1", DataValue::Int16(1)),
-            ("?2", DataValue::Int8(1)),
-            ("?3", DataValue::Int32(21)),
-            ("?4", DataValue::Int32(21)),
-        ];
-
-        let kite_unordered = collect_i32(&backend.kite, STOCK_LEVEL_DISTINCT_SQL, &params)?;
-        let sqlite_unordered = collect_i32(&backend.sqlite, STOCK_LEVEL_DISTINCT_SQLITE, &params)?;
-        let dual_unordered = collect_i32(&backend, STOCK_LEVEL_DISTINCT_SQL, &params)?;
-
-        let mut kite_sorted = kite_unordered.clone();
-        let mut sqlite_sorted = sqlite_unordered.clone();
-        kite_sorted.sort_unstable();
-        sqlite_sorted.sort_unstable();
-        assert_eq!(
-            kite_sorted, sqlite_sorted,
-            "distinct sets diverge; ordering is the only allowed difference"
-        );
-        assert_ne!(
-            kite_unordered, sqlite_unordered,
-            "unordered DISTINCT should not be compared row-by-row"
-        );
-        let mut dual_sorted = dual_unordered.clone();
-        dual_sorted.sort_unstable();
-        assert_eq!(
-            dual_sorted, kite_sorted,
-            "dual comparison should match KiteSQL distinct set"
-        );
-
-        let kite_ordered = collect_i32(&backend.kite, DISTINCT_ORDER_BY_SQL, &params)?;
-        let sqlite_ordered = collect_i32(&backend.sqlite, DISTINCT_ORDER_BY_SQL, &params)?;
-        assert_eq!(
-            kite_ordered, sqlite_ordered,
-            "explicit ORDER BY should align results across engines"
-        );
-
-        let params_delivery: [DbParam; 2] =
-            [("?1", DataValue::Int16(1)), ("?2", DataValue::Int8(1))];
-        let expected_delivery = vec![
-            None,
-            Some(parse_dt("2023-12-31 00:00:00")),
-            Some(parse_dt("2024-01-01 00:00:00")),
-            Some(parse_dt("2024-01-02 00:00:00")),
-        ];
-        let kite_delivery =
-            collect_datetime(&backend.kite, DISTINCT_DELIVERY_SQL, &params_delivery)?;
-        let sqlite_delivery =
-            collect_datetime(&backend.sqlite, DISTINCT_DELIVERY_SQL, &params_delivery)?;
-        assert_eq!(kite_delivery, expected_delivery);
-        assert_eq!(sqlite_delivery, expected_delivery);
-
-        Ok(())
-    }
-}
diff --git a/tpcc/src/load.rs b/tpcc/src/load.rs
@@ -785,9 +785,9 @@ impl Load {
 
                 let ol_dist_info = generate_string(rng, 24, 24);
                 let (ol_delivery_d, ol_amount) = if o_id > 2100 {
-                    ("null".to_string(), "0.00".to_string())
+                    ("null", "0.00".to_string())
                 } else {
-                    (date.clone(), format!("{:.2}", rng.gen_range(0.1..100.0)))
+                    (date.as_str(), format!("{:.2}", rng.gen_range(0.1..100.0)))
                 };
                 exec.execute_batch(&format!(
                     "insert into order_line values({}, {}, {}, {}, {}, {}, {}, {}, {}, '{}')",
diff --git a/tpcc/src/main.rs b/tpcc/src/main.rs
@@ -696,7 +696,7 @@ fn explain_tpcc() -> Result<(), DatabaseError> {
     use kite_sql::db::DataBaseBuilder;
     use kite_sql::types::tuple::create_table;
 
-    let database = DataBaseBuilder::path("./../kite_sql_tpcc").build()?;
+    let database = DataBaseBuilder::path("./kite_sql_tpcc").build()?;
     let mut tx = database.new_transaction()?;
 
     let customer_tuple = tx

Original file line number	Diff line number	Diff line change
`@@ -100,8 +100,8 @@ impl VisitorMut<'_> for ConstantCalculator {`
`100`	`100`	`ScalarExpression::TypeCast { expr: arg_expr, ty } => {`
`101`	`101`	`self.visit(arg_expr)?;`
`102`	`102`
`103`		`- if let ScalarExpression::Constant(value) = arg_expr.as_ref() {`
`104`		`- let casted = value.clone().cast(ty)?;`
	`103`	`+ if let ScalarExpression::Constant(value) = arg_expr.as_mut() {`
	`104`	`+ let casted = mem::replace(value, DataValue::Null).cast(ty)?;`
`105`	`105`	`let _ = mem::replace(expr, ScalarExpression::Constant(casted));`
`106`	`106`	`}`
`107`	`107`	`}`