From 877bff7d10f839f80aa5708d93ec36afca87b812 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E5=BC=A0=E6=9E=97=E4=BC=9F?= <lewiszlw520@gmail.com>
Date: Tue, 30 Jan 2024 14:26:11 +0800
Subject: [PATCH] Refactor tuple

---
 bustubx/src/catalog/column.rs                |   6 +
 bustubx/src/common/scalar.rs                 |  22 ++
 bustubx/src/planner/physical_plan/insert.rs  |   2 +-
 bustubx/src/planner/physical_plan/project.rs |   5 +-
 bustubx/src/planner/physical_plan/values.rs  |   2 +-
 bustubx/src/storage/index.rs                 | 161 +++++++----
 bustubx/src/storage/index_page.rs            | 266 ++++++++++++-------
 bustubx/src/storage/table_heap.rs            |  94 +++++--
 bustubx/src/storage/table_page.rs            |  91 +++++--
 bustubx/src/storage/tuple.rs                 |  64 ++---
 10 files changed, 488 insertions(+), 225 deletions(-)

diff --git a/bustubx/src/catalog/column.rs b/bustubx/src/catalog/column.rs
index 896c168..23b25fa 100644
--- a/bustubx/src/catalog/column.rs
+++ b/bustubx/src/catalog/column.rs
@@ -14,6 +14,12 @@ pub struct Column {
     pub column_offset: usize,
 }
 
+impl PartialEq for Column {
+    fn eq(&self, other: &Self) -> bool {
+        self.name == other.name && self.data_type == other.data_type
+    }
+}
+
 impl Column {
     pub fn new(name: String, data_type: DataType) -> Self {
         Self {
diff --git a/bustubx/src/common/scalar.rs b/bustubx/src/common/scalar.rs
index b2961d1..1d7a099 100644
--- a/bustubx/src/common/scalar.rs
+++ b/bustubx/src/common/scalar.rs
@@ -133,3 +133,25 @@ impl std::fmt::Display for ScalarValue {
         }
     }
 }
+
+macro_rules! impl_from_for_scalar {
+    ($ty:ty, $scalar:tt) => {
+        impl From<$ty> for ScalarValue {
+            fn from(value: $ty) -> Self {
+                ScalarValue::$scalar(Some(value))
+            }
+        }
+
+        impl From<Option<$ty>> for ScalarValue {
+            fn from(value: Option<$ty>) -> Self {
+                ScalarValue::$scalar(value)
+            }
+        }
+    };
+}
+
+impl_from_for_scalar!(i8, Int8);
+impl_from_for_scalar!(i16, Int16);
+impl_from_for_scalar!(i32, Int32);
+impl_from_for_scalar!(i64, Int64);
+impl_from_for_scalar!(bool, Boolean);
diff --git a/bustubx/src/planner/physical_plan/insert.rs b/bustubx/src/planner/physical_plan/insert.rs
index d870b60..5708c1e 100644
--- a/bustubx/src/planner/physical_plan/insert.rs
+++ b/bustubx/src/planner/physical_plan/insert.rs
@@ -52,7 +52,7 @@ impl VolcanoExecutor for PhysicalInsert {
                     let insert_rows = self.insert_rows.load(std::sync::atomic::Ordering::SeqCst);
                     self.insert_rows
                         .store(0, std::sync::atomic::Ordering::SeqCst);
-                    return Some(Tuple::from_values(
+                    return Some(Tuple::new(
                         Arc::new(self.output_schema()),
                         vec![ScalarValue::Int32(Some(insert_rows as i32))],
                     ));
diff --git a/bustubx/src/planner/physical_plan/project.rs b/bustubx/src/planner/physical_plan/project.rs
index be33baf..792fcad 100644
--- a/bustubx/src/planner/physical_plan/project.rs
+++ b/bustubx/src/planner/physical_plan/project.rs
@@ -34,9 +34,6 @@ impl VolcanoExecutor for PhysicalProject {
         for expr in &self.expressions {
             new_values.push(expr.evaluate(next_tuple.as_ref(), Some(&self.input.output_schema())));
         }
-        return Some(Tuple::from_values(
-            Arc::new(self.output_schema()),
-            new_values,
-        ));
+        return Some(Tuple::new(Arc::new(self.output_schema()), new_values));
     }
 }
diff --git a/bustubx/src/planner/physical_plan/values.rs b/bustubx/src/planner/physical_plan/values.rs
index 4bbe48f..f11c807 100644
--- a/bustubx/src/planner/physical_plan/values.rs
+++ b/bustubx/src/planner/physical_plan/values.rs
@@ -41,7 +41,7 @@ impl VolcanoExecutor for PhysicalValues {
             .fetch_add(1, std::sync::atomic::Ordering::SeqCst) as usize;
         if cursor < self.tuples.len() {
             let values = self.tuples[cursor].clone();
-            return Some(Tuple::from_values(Arc::new(self.output_schema()), values));
+            return Some(Tuple::new(Arc::new(self.output_schema()), values));
         } else {
             return None;
         }
diff --git a/bustubx/src/storage/index.rs b/bustubx/src/storage/index.rs
index 9929171..1a477b8 100644
--- a/bustubx/src/storage/index.rs
+++ b/bustubx/src/storage/index.rs
@@ -140,10 +140,7 @@ impl BPlusTreeIndex {
 
                 // internal page第一个kv对的key为空
                 new_internal_page.insert(
-                    Tuple::empty(
-                        self.index_metadata.key_schema.clone(),
-                        self.index_metadata.key_schema.fixed_len(),
-                    ),
+                    Tuple::empty(self.index_metadata.key_schema.clone()),
                     self.root_page_id,
                     &self.index_metadata.key_schema,
                 );
@@ -835,50 +832,65 @@ mod tests {
         let buffer_pool_manager = BufferPoolManager::new(1000, Arc::new(disk_manager));
         let mut index = BPlusTreeIndex::new(index_metadata, buffer_pool_manager, 2, 3);
 
-        index.insert(&Tuple::new(schema.clone(), vec![1, 1, 1]), Rid::new(1, 1));
+        index.insert(
+            &Tuple::new(schema.clone(), vec![1i8.into(), 2i16.into()]),
+            Rid::new(1, 1),
+        );
         assert_eq!(
             index
-                .get(&Tuple::new(schema.clone(), vec![1, 1, 1]))
+                .get(&Tuple::new(schema.clone(), vec![1i8.into(), 2i16.into()]))
                 .unwrap(),
             Rid::new(1, 1)
         );
         assert_eq!(index.root_page_id, 0);
         assert_eq!(index.buffer_pool_manager.replacer.size(), 1);
 
-        index.insert(&Tuple::new(schema.clone(), vec![2, 2, 2]), Rid::new(2, 2));
+        index.insert(
+            &Tuple::new(schema.clone(), vec![2i8.into(), 4i16.into()]),
+            Rid::new(2, 2),
+        );
         assert_eq!(
             index
-                .get(&Tuple::new(schema.clone(), vec![2, 2, 2]))
+                .get(&Tuple::new(schema.clone(), vec![2i8.into(), 4i16.into()]))
                 .unwrap(),
             Rid::new(2, 2)
         );
         assert_eq!(index.root_page_id, 0);
         assert_eq!(index.buffer_pool_manager.replacer.size(), 1);
 
-        index.insert(&Tuple::new(schema.clone(), vec![3, 3, 3]), Rid::new(3, 3));
+        index.insert(
+            &Tuple::new(schema.clone(), vec![3i8.into(), 6i16.into()]),
+            Rid::new(3, 3),
+        );
         assert_eq!(
             index
-                .get(&Tuple::new(schema.clone(), vec![3, 3, 3]))
+                .get(&Tuple::new(schema.clone(), vec![3i8.into(), 6i16.into()]))
                 .unwrap(),
             Rid::new(3, 3)
         );
         assert_eq!(index.root_page_id, 2);
         assert_eq!(index.buffer_pool_manager.replacer.size(), 3);
 
-        index.insert(&Tuple::new(schema.clone(), vec![4, 4, 4]), Rid::new(4, 4));
+        index.insert(
+            &Tuple::new(schema.clone(), vec![4i8.into(), 8i16.into()]),
+            Rid::new(4, 4),
+        );
         assert_eq!(
             index
-                .get(&Tuple::new(schema.clone(), vec![4, 4, 4]))
+                .get(&Tuple::new(schema.clone(), vec![4i8.into(), 8i16.into()]))
                 .unwrap(),
             Rid::new(4, 4)
         );
         assert_eq!(index.root_page_id, 2);
         assert_eq!(index.buffer_pool_manager.replacer.size(), 4);
 
-        index.insert(&Tuple::new(schema.clone(), vec![5, 5, 5]), Rid::new(5, 5));
+        index.insert(
+            &Tuple::new(schema.clone(), vec![5i8.into(), 10i16.into()]),
+            Rid::new(5, 5),
+        );
         assert_eq!(
             index
-                .get(&Tuple::new(schema.clone(), vec![5, 5, 5]))
+                .get(&Tuple::new(schema.clone(), vec![5i8.into(), 10i16.into()]))
                 .unwrap(),
             Rid::new(5, 5)
         );
@@ -905,79 +917,136 @@ mod tests {
         let buffer_pool_manager = BufferPoolManager::new(1000, Arc::new(disk_manager));
         let mut index = BPlusTreeIndex::new(index_metadata, buffer_pool_manager, 4, 5);
 
-        index.insert(&Tuple::new(schema.clone(), vec![1, 1, 1]), Rid::new(1, 1));
-        index.insert(&Tuple::new(schema.clone(), vec![2, 2, 2]), Rid::new(2, 2));
-        index.insert(&Tuple::new(schema.clone(), vec![3, 3, 3]), Rid::new(3, 3));
-        index.insert(&Tuple::new(schema.clone(), vec![4, 4, 4]), Rid::new(4, 4));
-        index.insert(&Tuple::new(schema.clone(), vec![5, 5, 5]), Rid::new(5, 5));
-        index.insert(&Tuple::new(schema.clone(), vec![6, 6, 6]), Rid::new(6, 6));
-        index.insert(&Tuple::new(schema.clone(), vec![7, 7, 7]), Rid::new(7, 7));
-        index.insert(&Tuple::new(schema.clone(), vec![8, 8, 8]), Rid::new(8, 8));
-        index.insert(&Tuple::new(schema.clone(), vec![9, 9, 9]), Rid::new(9, 9));
         index.insert(
-            &Tuple::new(schema.clone(), vec![10, 10, 10]),
+            &Tuple::new(schema.clone(), vec![1i8.into(), 1i16.into()]),
+            Rid::new(1, 1),
+        );
+        index.insert(
+            &Tuple::new(schema.clone(), vec![2i8.into(), 2i16.into()]),
+            Rid::new(2, 2),
+        );
+        index.insert(
+            &Tuple::new(schema.clone(), vec![3i8.into(), 3i16.into()]),
+            Rid::new(3, 3),
+        );
+        index.insert(
+            &Tuple::new(schema.clone(), vec![4i8.into(), 4i16.into()]),
+            Rid::new(4, 4),
+        );
+        index.insert(
+            &Tuple::new(schema.clone(), vec![5i8.into(), 5i16.into()]),
+            Rid::new(5, 5),
+        );
+        index.insert(
+            &Tuple::new(schema.clone(), vec![6i8.into(), 6i16.into()]),
+            Rid::new(6, 6),
+        );
+        index.insert(
+            &Tuple::new(schema.clone(), vec![7i8.into(), 7i16.into()]),
+            Rid::new(7, 7),
+        );
+        index.insert(
+            &Tuple::new(schema.clone(), vec![8i8.into(), 8i16.into()]),
+            Rid::new(8, 8),
+        );
+        index.insert(
+            &Tuple::new(schema.clone(), vec![9i8.into(), 9i16.into()]),
+            Rid::new(9, 9),
+        );
+        index.insert(
+            &Tuple::new(schema.clone(), vec![10i8.into(), 10i16.into()]),
             Rid::new(10, 10),
         );
         assert_eq!(index.buffer_pool_manager.replacer.size(), 5);
         assert_eq!(index.root_page_id, 2);
         index.print_tree();
 
-        index.delete(&Tuple::new(schema.clone(), vec![1, 1, 1]));
+        index.delete(&Tuple::new(schema.clone(), vec![1i8.into(), 1i16.into()]));
         assert_eq!(index.root_page_id, 2);
-        assert_eq!(index.get(&Tuple::new(schema.clone(), vec![1, 1, 1])), None);
+        assert_eq!(
+            index.get(&Tuple::new(schema.clone(), vec![1i8.into(), 1i16.into()])),
+            None
+        );
         assert_eq!(index.buffer_pool_manager.replacer.size(), 4);
 
-        index.delete(&Tuple::new(schema.clone(), vec![3, 3, 3]));
+        index.delete(&Tuple::new(schema.clone(), vec![3i8.into(), 3i16.into()]));
         assert_eq!(index.root_page_id, 2);
-        assert_eq!(index.get(&Tuple::new(schema.clone(), vec![3, 3, 3])), None);
+        assert_eq!(
+            index.get(&Tuple::new(schema.clone(), vec![3i8.into(), 3i16.into()])),
+            None
+        );
         assert_eq!(index.buffer_pool_manager.replacer.size(), 4);
 
-        index.delete(&Tuple::new(schema.clone(), vec![5, 5, 5]));
+        index.delete(&Tuple::new(schema.clone(), vec![5i8.into(), 5i16.into()]));
         assert_eq!(index.root_page_id, 2);
-        assert_eq!(index.get(&Tuple::new(schema.clone(), vec![5, 5, 5])), None);
+        assert_eq!(
+            index.get(&Tuple::new(schema.clone(), vec![5i8.into(), 5i16.into()])),
+            None
+        );
         assert_eq!(index.buffer_pool_manager.replacer.size(), 4);
 
-        index.delete(&Tuple::new(schema.clone(), vec![7, 7, 7]));
+        index.delete(&Tuple::new(schema.clone(), vec![7i8.into(), 7i16.into()]));
         assert_eq!(index.root_page_id, 2);
-        assert_eq!(index.get(&Tuple::new(schema.clone(), vec![7, 7, 7])), None);
+        assert_eq!(
+            index.get(&Tuple::new(schema.clone(), vec![7i8.into(), 7i16.into()])),
+            None
+        );
         assert_eq!(index.buffer_pool_manager.replacer.size(), 4);
 
-        index.delete(&Tuple::new(schema.clone(), vec![9, 9, 9]));
+        index.delete(&Tuple::new(schema.clone(), vec![9i8.into(), 9i16.into()]));
         assert_eq!(index.root_page_id, 2);
-        assert_eq!(index.get(&Tuple::new(schema.clone(), vec![9, 9, 9])), None);
+        assert_eq!(
+            index.get(&Tuple::new(schema.clone(), vec![9i8.into(), 9i16.into()])),
+            None
+        );
         assert_eq!(index.buffer_pool_manager.replacer.size(), 3);
 
-        index.delete(&Tuple::new(schema.clone(), vec![10, 10, 10]));
+        index.delete(&Tuple::new(schema.clone(), vec![10i8.into(), 10i16.into()]));
         assert_eq!(index.root_page_id, 2);
         assert_eq!(
-            index.get(&Tuple::new(schema.clone(), vec![10, 10, 10])),
+            index.get(&Tuple::new(schema.clone(), vec![10i8.into(), 10i16.into()])),
             None
         );
         assert_eq!(index.buffer_pool_manager.replacer.size(), 3);
 
-        index.delete(&Tuple::new(schema.clone(), vec![8, 8, 8]));
+        index.delete(&Tuple::new(schema.clone(), vec![8i8.into(), 8i16.into()]));
         assert_eq!(index.root_page_id, 0);
-        assert_eq!(index.get(&Tuple::new(schema.clone(), vec![8, 8, 8])), None);
+        assert_eq!(
+            index.get(&Tuple::new(schema.clone(), vec![8i8.into(), 8i16.into()])),
+            None
+        );
         assert_eq!(index.buffer_pool_manager.replacer.size(), 1);
 
-        index.delete(&Tuple::new(schema.clone(), vec![6, 6, 6]));
+        index.delete(&Tuple::new(schema.clone(), vec![6i8.into(), 6i16.into()]));
         assert_eq!(index.root_page_id, 0);
-        assert_eq!(index.get(&Tuple::new(schema.clone(), vec![6, 6, 6])), None);
+        assert_eq!(
+            index.get(&Tuple::new(schema.clone(), vec![6i8.into(), 6i16.into()])),
+            None
+        );
         assert_eq!(index.buffer_pool_manager.replacer.size(), 1);
 
-        index.delete(&Tuple::new(schema.clone(), vec![4, 4, 4]));
+        index.delete(&Tuple::new(schema.clone(), vec![4i8.into(), 4i16.into()]));
         assert_eq!(index.root_page_id, 0);
-        assert_eq!(index.get(&Tuple::new(schema.clone(), vec![4, 4, 4])), None);
+        assert_eq!(
+            index.get(&Tuple::new(schema.clone(), vec![4i8.into(), 4i16.into()])),
+            None
+        );
         assert_eq!(index.buffer_pool_manager.replacer.size(), 1);
 
-        index.delete(&Tuple::new(schema.clone(), vec![2, 2, 2]));
+        index.delete(&Tuple::new(schema.clone(), vec![2i8.into(), 2i16.into()]));
         assert_eq!(index.root_page_id, 0);
-        assert_eq!(index.get(&Tuple::new(schema.clone(), vec![2, 2, 2])), None);
+        assert_eq!(
+            index.get(&Tuple::new(schema.clone(), vec![2i8.into(), 2i16.into()])),
+            None
+        );
         assert_eq!(index.buffer_pool_manager.replacer.size(), 1);
 
-        index.delete(&Tuple::new(schema.clone(), vec![2, 2, 2]));
+        index.delete(&Tuple::new(schema.clone(), vec![2i8.into(), 2i16.into()]));
         assert_eq!(index.root_page_id, 0);
-        assert_eq!(index.get(&Tuple::new(schema.clone(), vec![2, 2, 2])), None);
+        assert_eq!(
+            index.get(&Tuple::new(schema.clone(), vec![2i8.into(), 2i16.into()])),
+            None
+        );
         assert_eq!(index.buffer_pool_manager.replacer.size(), 1);
     }
 }
diff --git a/bustubx/src/storage/index_page.rs b/bustubx/src/storage/index_page.rs
index 9dcfddf..ff7ad1c 100644
--- a/bustubx/src/storage/index_page.rs
+++ b/bustubx/src/storage/index_page.rs
@@ -235,7 +235,7 @@ impl BPlusTreeInternalPage {
                     self.array.remove(0);
                     self.current_size -= 1;
                     // 把第一个key置空
-                    self.array[0].0 = Tuple::empty(self.schema.clone(), self.array[0].0.data.len());
+                    self.array[0].0 = Tuple::empty(self.schema.clone());
                 } else {
                     self.array.remove(i as usize);
                     self.current_size -= 1;
@@ -589,6 +589,7 @@ impl BPlusTreeLeafPage {
 
 #[cfg(test)]
 mod tests {
+    use crate::common::ScalarValue;
     use crate::{
         catalog::{Column, DataType, Schema},
         common::rid::Rid,
@@ -606,14 +607,14 @@ mod tests {
             Column::new("a".to_string(), DataType::Int16),
         ]));
         let mut ori_page = BPlusTreeInternalPage::new(key_schema.clone(), 5);
-        ori_page.insert(Tuple::empty(key_schema.clone(), 3), 0, &key_schema);
+        ori_page.insert(Tuple::empty(key_schema.clone()), 0, &key_schema);
         ori_page.insert(
-            Tuple::new(key_schema.clone(), vec![1, 1, 1]),
+            Tuple::new(key_schema.clone(), vec![1i8.into(), 1i16.into()]),
             1,
             &key_schema,
         );
         ori_page.insert(
-            Tuple::new(key_schema.clone(), vec![2, 2, 2]),
+            Tuple::new(key_schema.clone(), vec![2i8.into(), 2i16.into()]),
             2,
             &key_schema,
         );
@@ -625,11 +626,14 @@ mod tests {
         assert_eq!(new_page.page_type, BPlusTreePageType::InternalPage);
         assert_eq!(new_page.current_size, 3);
         assert_eq!(new_page.max_size, 5);
-        assert_eq!(new_page.array[0].0.data, vec![0, 0, 0]);
+        assert_eq!(
+            new_page.array[0].0.data,
+            vec![ScalarValue::Int8(None), ScalarValue::Int16(None)]
+        );
         assert_eq!(new_page.array[0].1, 0);
-        assert_eq!(new_page.array[1].0.data, vec![1, 1, 1]);
+        assert_eq!(new_page.array[1].0.data, vec![1i8.into(), 1i16.into()]);
         assert_eq!(new_page.array[1].1, 1);
-        assert_eq!(new_page.array[2].0.data, vec![2, 2, 2]);
+        assert_eq!(new_page.array[2].0.data, vec![2i8.into(), 2i16.into()]);
         assert_eq!(new_page.array[2].1, 2);
     }
 
@@ -641,12 +645,12 @@ mod tests {
         ]));
         let mut ori_page = BPlusTreeLeafPage::new(key_schema.clone(), 5);
         ori_page.insert(
-            Tuple::new(key_schema.clone(), vec![1, 1, 1]),
+            Tuple::new(key_schema.clone(), vec![1i8.into(), 1i16.into()]),
             Rid::new(1, 1),
             &key_schema,
         );
         ori_page.insert(
-            Tuple::new(key_schema.clone(), vec![2, 2, 2]),
+            Tuple::new(key_schema.clone(), vec![2i8.into(), 2i16.into()]),
             Rid::new(2, 2),
             &key_schema,
         );
@@ -658,9 +662,9 @@ mod tests {
         assert_eq!(new_page.page_type, BPlusTreePageType::LeafPage);
         assert_eq!(new_page.current_size, 2);
         assert_eq!(new_page.max_size, 5);
-        assert_eq!(new_page.array[0].0.data, vec![1, 1, 1]);
+        assert_eq!(new_page.array[0].0.data, vec![1i8.into(), 1i16.into()]);
         assert_eq!(new_page.array[0].1, Rid::new(1, 1));
-        assert_eq!(new_page.array[1].0.data, vec![2, 2, 2]);
+        assert_eq!(new_page.array[1].0.data, vec![2i8.into(), 2i16.into()]);
         assert_eq!(new_page.array[1].1, Rid::new(2, 2));
     }
 
@@ -671,27 +675,26 @@ mod tests {
             Column::new("b".to_string(), DataType::Int16),
         ]));
         let mut internal_page = BPlusTreeInternalPage::new(key_schema.clone(), 3);
+        internal_page.insert(Tuple::empty(key_schema.clone()), 0, &key_schema);
         internal_page.insert(
-            Tuple::empty(key_schema.clone(), key_schema.fixed_len()),
-            0,
-            &key_schema,
-        );
-        internal_page.insert(
-            Tuple::new(key_schema.clone(), vec![2, 2, 2]),
+            Tuple::new(key_schema.clone(), vec![2i8.into(), 2i16.into()]),
             2,
             &key_schema,
         );
         internal_page.insert(
-            Tuple::new(key_schema.clone(), vec![1, 1, 1]),
+            Tuple::new(key_schema.clone(), vec![1i8.into(), 1i16.into()]),
             1,
             &key_schema,
         );
         assert_eq!(internal_page.current_size, 3);
-        assert_eq!(internal_page.array[0].0.data, vec![0, 0, 0]);
+        assert_eq!(
+            internal_page.array[0].0.data,
+            Tuple::empty(key_schema.clone()).data
+        );
         assert_eq!(internal_page.array[0].1, 0);
-        assert_eq!(internal_page.array[1].0.data, vec![1, 1, 1]);
+        assert_eq!(internal_page.array[1].0.data, vec![1i8.into(), 1i16.into()]);
         assert_eq!(internal_page.array[1].1, 1);
-        assert_eq!(internal_page.array[2].0.data, vec![2, 2, 2]);
+        assert_eq!(internal_page.array[2].0.data, vec![2i8.into(), 2i16.into()]);
         assert_eq!(internal_page.array[2].1, 2);
     }
 
@@ -703,26 +706,26 @@ mod tests {
         ]));
         let mut leaf_page = BPlusTreeLeafPage::new(key_schema.clone(), 3);
         leaf_page.insert(
-            Tuple::new(key_schema.clone(), vec![2, 2, 2]),
+            Tuple::new(key_schema.clone(), vec![2i8.into(), 2i16.into()]),
             Rid::new(2, 2),
             &key_schema,
         );
         leaf_page.insert(
-            Tuple::new(key_schema.clone(), vec![1, 1, 1]),
+            Tuple::new(key_schema.clone(), vec![1i8.into(), 1i16.into()]),
             Rid::new(1, 1),
             &key_schema,
         );
         leaf_page.insert(
-            Tuple::new(key_schema.clone(), vec![3, 3, 3]),
+            Tuple::new(key_schema.clone(), vec![3i8.into(), 3i16.into()]),
             Rid::new(3, 3),
             &key_schema,
         );
         assert_eq!(leaf_page.current_size, 3);
-        assert_eq!(leaf_page.array[0].0.data, vec![1, 1, 1]);
+        assert_eq!(leaf_page.array[0].0.data, vec![1i8.into(), 1i16.into()]);
         assert_eq!(leaf_page.array[0].1, Rid::new(1, 1));
-        assert_eq!(leaf_page.array[1].0.data, vec![2, 2, 2]);
+        assert_eq!(leaf_page.array[1].0.data, vec![2i8.into(), 2i16.into()]);
         assert_eq!(leaf_page.array[1].1, Rid::new(2, 2));
-        assert_eq!(leaf_page.array[2].0.data, vec![3, 3, 3]);
+        assert_eq!(leaf_page.array[2].0.data, vec![3i8.into(), 3i16.into()]);
         assert_eq!(leaf_page.array[2].1, Rid::new(3, 3));
     }
 
@@ -733,67 +736,77 @@ mod tests {
             Column::new("b".to_string(), DataType::Int16),
         ]));
         let mut internal_page = BPlusTreeInternalPage::new(key_schema.clone(), 5);
+        internal_page.insert(Tuple::empty(key_schema.clone()), 0, &key_schema);
         internal_page.insert(
-            Tuple::empty(key_schema.clone(), key_schema.fixed_len()),
-            0,
-            &key_schema,
-        );
-        internal_page.insert(
-            Tuple::new(key_schema.clone(), vec![2, 2, 2]),
+            Tuple::new(key_schema.clone(), vec![2i8.into(), 2i16.into()]),
             2,
             &key_schema,
         );
         internal_page.insert(
-            Tuple::new(key_schema.clone(), vec![1, 1, 1]),
+            Tuple::new(key_schema.clone(), vec![1i8.into(), 1i16.into()]),
             1,
             &key_schema,
         );
         internal_page.insert(
-            Tuple::new(key_schema.clone(), vec![3, 3, 3]),
+            Tuple::new(key_schema.clone(), vec![3i8.into(), 3i16.into()]),
             3,
             &key_schema,
         );
         internal_page.insert(
-            Tuple::new(key_schema.clone(), vec![4, 4, 4]),
+            Tuple::new(key_schema.clone(), vec![4i8.into(), 4i16.into()]),
             4,
             &key_schema,
         );
 
         assert_eq!(
-            internal_page.look_up(&Tuple::new(key_schema.clone(), vec![0, 0, 0]), &key_schema),
+            internal_page.look_up(
+                &Tuple::new(key_schema.clone(), vec![0i8.into(), 0i16.into()]),
+                &key_schema
+            ),
             0
         );
         assert_eq!(
-            internal_page.look_up(&Tuple::new(key_schema.clone(), vec![3, 3, 3]), &key_schema),
+            internal_page.look_up(
+                &Tuple::new(key_schema.clone(), vec![3i8.into(), 3i16.into()]),
+                &key_schema
+            ),
             3
         );
         assert_eq!(
-            internal_page.look_up(&Tuple::new(key_schema.clone(), vec![5, 5, 5]), &key_schema),
+            internal_page.look_up(
+                &Tuple::new(key_schema.clone(), vec![5i8.into(), 5i16.into()]),
+                &key_schema
+            ),
             4
         );
 
         let mut internal_page = BPlusTreeInternalPage::new(key_schema.clone(), 2);
+        internal_page.insert(Tuple::empty(key_schema.clone()), 0, &key_schema);
         internal_page.insert(
-            Tuple::empty(key_schema.clone(), key_schema.fixed_len()),
-            0,
-            &key_schema,
-        );
-        internal_page.insert(
-            Tuple::new(key_schema.clone(), vec![1, 1, 1]),
+            Tuple::new(key_schema.clone(), vec![1i8.into(), 1i16.into()]),
             1,
             &key_schema,
         );
 
         assert_eq!(
-            internal_page.look_up(&Tuple::new(key_schema.clone(), vec![0, 0, 0]), &key_schema),
+            internal_page.look_up(
+                &Tuple::new(key_schema.clone(), vec![0i8.into(), 0i16.into()]),
+                &key_schema
+            ),
             0
         );
         assert_eq!(
-            internal_page.look_up(&Tuple::new(key_schema.clone(), vec![1, 1, 1]), &key_schema),
+            internal_page.look_up(
+                &Tuple::new(key_schema.clone(), vec![1i8.into(), 1i16.into()]),
+                &key_schema
+            ),
             1
         );
         assert_eq!(
-            internal_page.look_up(&Tuple::new(key_schema.clone(), vec![2, 2, 2]), &key_schema),
+            internal_page.look_up(
+                &Tuple::new(key_schema.clone(), vec![2i8.into(), 2i16.into()]),
+                &key_schema
+            ),
             1
         );
     }
@@ -806,72 +819,99 @@ mod tests {
         ]));
         let mut leaf_page = BPlusTreeLeafPage::new(key_schema.clone(), 5);
         leaf_page.insert(
-            Tuple::new(key_schema.clone(), vec![2, 2, 2]),
+            Tuple::new(key_schema.clone(), vec![2i8.into(), 2i16.into()]),
             Rid::new(2, 2),
             &key_schema,
         );
         leaf_page.insert(
-            Tuple::new(key_schema.clone(), vec![1, 1, 1]),
+            Tuple::new(key_schema.clone(), vec![1i8.into(), 1i16.into()]),
             Rid::new(1, 1),
             &key_schema,
         );
         leaf_page.insert(
-            Tuple::new(key_schema.clone(), vec![3, 3, 3]),
+            Tuple::new(key_schema.clone(), vec![3i8.into(), 3i16.into()]),
             Rid::new(3, 3),
             &key_schema,
         );
         leaf_page.insert(
-            Tuple::new(key_schema.clone(), vec![5, 5, 5]),
+            Tuple::new(key_schema.clone(), vec![5i8.into(), 5i16.into()]),
             Rid::new(5, 5),
             &key_schema,
         );
         leaf_page.insert(
-            Tuple::new(key_schema.clone(), vec![4, 4, 4]),
+            Tuple::new(key_schema.clone(), vec![4i8.into(), 4i16.into()]),
             Rid::new(4, 4),
             &key_schema,
         );
         assert_eq!(
-            leaf_page.look_up(&Tuple::new(key_schema.clone(), vec![0, 0, 0]), &key_schema),
+            leaf_page.look_up(
+                &Tuple::new(key_schema.clone(), vec![0i8.into(), 0i16.into()]),
+                &key_schema
+            ),
             None
         );
         assert_eq!(
-            leaf_page.look_up(&Tuple::new(key_schema.clone(), vec![2, 2, 2]), &key_schema),
+            leaf_page.look_up(
+                &Tuple::new(key_schema.clone(), vec![2i8.into(), 2i16.into()]),
+                &key_schema
+            ),
             Some(Rid::new(2, 2))
         );
         assert_eq!(
-            leaf_page.look_up(&Tuple::new(key_schema.clone(), vec![3, 3, 3]), &key_schema),
+            leaf_page.look_up(
+                &Tuple::new(key_schema.clone(), vec![3i8.into(), 3i16.into()]),
+                &key_schema
+            ),
             Some(Rid::new(3, 3))
         );
         assert_eq!(
-            leaf_page.look_up(&Tuple::new(key_schema.clone(), vec![6, 6, 6]), &key_schema),
+            leaf_page.look_up(
+                &Tuple::new(key_schema.clone(), vec![6i8.into(), 6i16.into()]),
+                &key_schema
+            ),
             None
         );
 
         let mut leaf_page = BPlusTreeLeafPage::new(key_schema.clone(), 2);
         leaf_page.insert(
-            Tuple::new(key_schema.clone(), vec![2, 2, 2]),
+            Tuple::new(key_schema.clone(), vec![2i8.into(), 2i16.into()]),
             Rid::new(2, 2),
             &key_schema,
         );
         leaf_page.insert(
-            Tuple::new(key_schema.clone(), vec![1, 1, 1]),
+            Tuple::new(key_schema.clone(), vec![1i8.into(), 1i16.into()]),
             Rid::new(1, 1),
             &key_schema,
         );
         assert_eq!(
-            leaf_page.look_up(&Tuple::new(key_schema.clone(), vec![0, 0, 0]), &key_schema),
+            leaf_page.look_up(
+                &Tuple::new(
+                    key_schema.clone(),
+                    vec![ScalarValue::Int8(None), ScalarValue::Int16(None)]
+                ),
+                &key_schema
+            ),
             None
         );
         assert_eq!(
-            leaf_page.look_up(&Tuple::new(key_schema.clone(), vec![1, 1, 1]), &key_schema),
+            leaf_page.look_up(
+                &Tuple::new(key_schema.clone(), vec![1i8.into(), 1i16.into()]),
+                &key_schema
+            ),
             Some(Rid::new(1, 1))
         );
         assert_eq!(
-            leaf_page.look_up(&Tuple::new(key_schema.clone(), vec![2, 2, 2]), &key_schema),
+            leaf_page.look_up(
+                &Tuple::new(key_schema.clone(), vec![2i8.into(), 2i16.into()]),
+                &key_schema
+            ),
             Some(Rid::new(2, 2))
         );
         assert_eq!(
-            leaf_page.look_up(&Tuple::new(key_schema.clone(), vec![3, 3, 3]), &key_schema),
+            leaf_page.look_up(
+                &Tuple::new(key_schema.clone(), vec![3i8.into(), 3i16.into()]),
+                &key_schema
+            ),
             None
         );
     }
@@ -883,49 +923,63 @@ mod tests {
             Column::new("b".to_string(), DataType::Int16),
         ]));
         let mut internal_page = BPlusTreeInternalPage::new(key_schema.clone(), 5);
+        internal_page.insert(Tuple::empty(key_schema.clone()), 0, &key_schema);
         internal_page.insert(
-            Tuple::empty(key_schema.clone(), key_schema.fixed_len()),
-            0,
-            &key_schema,
-        );
-        internal_page.insert(
-            Tuple::new(key_schema.clone(), vec![2, 2, 2]),
+            Tuple::new(key_schema.clone(), vec![2i8.into(), 2i16.into()]),
             2,
             &key_schema,
         );
         internal_page.insert(
-            Tuple::new(key_schema.clone(), vec![1, 1, 1]),
+            Tuple::new(key_schema.clone(), vec![1i8.into(), 1i16.into()]),
             1,
             &key_schema,
         );
         internal_page.insert(
-            Tuple::new(key_schema.clone(), vec![3, 3, 3]),
+            Tuple::new(key_schema.clone(), vec![3i8.into(), 3i16.into()]),
             3,
             &key_schema,
         );
         internal_page.insert(
-            Tuple::new(key_schema.clone(), vec![4, 4, 4]),
+            Tuple::new(key_schema.clone(), vec![4i8.into(), 4i16.into()]),
             4,
             &key_schema,
         );
 
-        internal_page.delete(&Tuple::new(key_schema.clone(), vec![2, 2, 2]), &key_schema);
+        internal_page.delete(
+            &Tuple::new(key_schema.clone(), vec![2i8.into(), 2i16.into()]),
+            &key_schema,
+        );
         assert_eq!(internal_page.current_size, 4);
-        assert_eq!(internal_page.array[0].0.data, vec![0, 0, 0]);
+        assert_eq!(
+            internal_page.array[0].0.data,
+            vec![ScalarValue::Int8(None), ScalarValue::Int16(None)]
+        );
         assert_eq!(internal_page.array[0].1, 0);
-        assert_eq!(internal_page.array[1].0.data, vec![1, 1, 1]);
+        assert_eq!(internal_page.array[1].0.data, vec![1i8.into(), 1i16.into()]);
         assert_eq!(internal_page.array[1].1, 1);
-        assert_eq!(internal_page.array[2].0.data, vec![3, 3, 3]);
+        assert_eq!(internal_page.array[2].0.data, vec![3i8.into(), 3i16.into()]);
         assert_eq!(internal_page.array[2].1, 3);
-        assert_eq!(internal_page.array[3].0.data, vec![4, 4, 4]);
+        assert_eq!(internal_page.array[3].0.data, vec![4i8.into(), 4i16.into()]);
         assert_eq!(internal_page.array[3].1, 4);
-        internal_page.delete(&Tuple::new(key_schema.clone(), vec![4, 4, 4]), &key_schema);
+        internal_page.delete(
+            &Tuple::new(key_schema.clone(), vec![4i8.into(), 4i16.into()]),
+            &key_schema,
+        );
         assert_eq!(internal_page.current_size, 3);
-        internal_page.delete(&Tuple::new(key_schema.clone(), vec![3, 3, 3]), &key_schema);
+        internal_page.delete(
+            &Tuple::new(key_schema.clone(), vec![3i8.into(), 3i16.into()]),
+            &key_schema,
+        );
         assert_eq!(internal_page.current_size, 2);
-        internal_page.delete(&Tuple::new(key_schema.clone(), vec![1, 1, 1]), &key_schema);
+        internal_page.delete(
+            &Tuple::new(key_schema.clone(), vec![1i8.into(), 1i16.into()]),
+            &key_schema,
+        );
         assert_eq!(internal_page.current_size, 0);
-        internal_page.delete(&Tuple::new(key_schema.clone(), vec![1, 1, 1]), &key_schema);
+        internal_page.delete(
+            &Tuple::new(key_schema.clone(), vec![1i8.into(), 1i16.into()]),
+            &key_schema,
+        );
         assert_eq!(internal_page.current_size, 0);
     }
 
@@ -937,52 +991,70 @@ mod tests {
         ]));
         let mut leaf_page = BPlusTreeLeafPage::new(key_schema.clone(), 5);
         leaf_page.insert(
-            Tuple::new(key_schema.clone(), vec![2, 2, 2]),
+            Tuple::new(key_schema.clone(), vec![2i8.into(), 2i16.into()]),
             Rid::new(2, 2),
             &key_schema,
         );
         leaf_page.insert(
-            Tuple::new(key_schema.clone(), vec![1, 1, 1]),
+            Tuple::new(key_schema.clone(), vec![1i8.into(), 1i16.into()]),
             Rid::new(1, 1),
             &key_schema,
         );
         leaf_page.insert(
-            Tuple::new(key_schema.clone(), vec![3, 3, 3]),
+            Tuple::new(key_schema.clone(), vec![3i8.into(), 3i16.into()]),
             Rid::new(3, 3),
             &key_schema,
         );
         leaf_page.insert(
-            Tuple::new(key_schema.clone(), vec![5, 5, 5]),
+            Tuple::new(key_schema.clone(), vec![5i8.into(), 5i16.into()]),
             Rid::new(5, 5),
             &key_schema,
         );
         leaf_page.insert(
-            Tuple::new(key_schema.clone(), vec![4, 4, 4]),
+            Tuple::new(key_schema.clone(), vec![4i8.into(), 4i16.into()]),
             Rid::new(4, 4),
             &key_schema,
         );
 
-        leaf_page.delete(&Tuple::new(key_schema.clone(), vec![2, 2, 2]), &key_schema);
+        leaf_page.delete(
+            &Tuple::new(key_schema.clone(), vec![2i8.into(), 2i16.into()]),
+            &key_schema,
+        );
         assert_eq!(leaf_page.current_size, 4);
-        assert_eq!(leaf_page.array[0].0.data, vec![1, 1, 1]);
+        assert_eq!(leaf_page.array[0].0.data, vec![1i8.into(), 1i16.into()]);
         assert_eq!(leaf_page.array[0].1, Rid::new(1, 1));
-        assert_eq!(leaf_page.array[1].0.data, vec![3, 3, 3]);
+        assert_eq!(leaf_page.array[1].0.data, vec![3i8.into(), 3i16.into()]);
         assert_eq!(leaf_page.array[1].1, Rid::new(3, 3));
-        assert_eq!(leaf_page.array[2].0.data, vec![4, 4, 4]);
+        assert_eq!(leaf_page.array[2].0.data, vec![4i8.into(), 4i16.into()]);
         assert_eq!(leaf_page.array[2].1, Rid::new(4, 4));
-        assert_eq!(leaf_page.array[3].0.data, vec![5, 5, 5]);
+        assert_eq!(leaf_page.array[3].0.data, vec![5i8.into(), 5i16.into()]);
         assert_eq!(leaf_page.array[3].1, Rid::new(5, 5));
-        leaf_page.delete(&Tuple::new(key_schema.clone(), vec![3, 3, 3]), &key_schema);
+        leaf_page.delete(
+            &Tuple::new(key_schema.clone(), vec![3i8.into(), 3i16.into()]),
+            &key_schema,
+        );
         assert_eq!(leaf_page.current_size, 3);
-        leaf_page.delete(&Tuple::new(key_schema.clone(), vec![5, 5, 5]), &key_schema);
+        leaf_page.delete(
+            &Tuple::new(key_schema.clone(), vec![5i8.into(), 5i16.into()]),
+            &key_schema,
+        );
         assert_eq!(leaf_page.current_size, 2);
-        leaf_page.delete(&Tuple::new(key_schema.clone(), vec![1, 1, 1]), &key_schema);
+        leaf_page.delete(
+            &Tuple::new(key_schema.clone(), vec![1i8.into(), 1i16.into()]),
+            &key_schema,
+        );
         assert_eq!(leaf_page.current_size, 1);
-        assert_eq!(leaf_page.array[0].0.data, vec![4, 4, 4]);
+        assert_eq!(leaf_page.array[0].0.data, vec![4i8.into(), 4i16.into()]);
         assert_eq!(leaf_page.array[0].1, Rid::new(4, 4));
-        leaf_page.delete(&Tuple::new(key_schema.clone(), vec![4, 4, 4]), &key_schema);
+        leaf_page.delete(
+            &Tuple::new(key_schema.clone(), vec![4i8.into(), 4i16.into()]),
+            &key_schema,
+        );
         assert_eq!(leaf_page.current_size, 0);
-        leaf_page.delete(&Tuple::new(key_schema.clone(), vec![4, 4, 4]), &key_schema);
+        leaf_page.delete(
+            &Tuple::new(key_schema.clone(), vec![4i8.into(), 4i16.into()]),
+            &key_schema,
+        );
         assert_eq!(leaf_page.current_size, 0);
     }
 }
diff --git a/bustubx/src/storage/table_heap.rs b/bustubx/src/storage/table_heap.rs
index 21059de..c50c1ba 100644
--- a/bustubx/src/storage/table_heap.rs
+++ b/bustubx/src/storage/table_heap.rs
@@ -214,7 +214,7 @@ mod tests {
     use std::sync::Arc;
     use tempfile::TempDir;
 
-    use crate::catalog::Schema;
+    use crate::catalog::{Column, DataType, Schema};
     use crate::{
         buffer::BufferPoolManager,
         storage::{table_heap::TableHeap, DiskManager, Tuple},
@@ -238,7 +238,10 @@ mod tests {
         let temp_dir = TempDir::new().unwrap();
         let temp_path = temp_dir.path().join("test.db");
 
-        let schema = Arc::new(Schema::empty());
+        let schema = Arc::new(Schema::new(vec![
+            Column::new("a".to_string(), DataType::Int8),
+            Column::new("b".to_string(), DataType::Int16),
+        ]));
         let disk_manager = DiskManager::try_new(&temp_path).unwrap();
         let buffer_pool_manager = BufferPoolManager::new(1000, Arc::new(disk_manager));
         let mut table_heap = TableHeap::new(schema.clone(), buffer_pool_manager);
@@ -248,17 +251,26 @@ mod tests {
             is_deleted: false,
         };
 
-        table_heap.insert_tuple(&meta, &Tuple::new(schema.clone(), vec![1; 2000]));
+        table_heap.insert_tuple(
+            &meta,
+            &Tuple::new(schema.clone(), vec![1i8.into(), 1i16.into()]),
+        );
         assert_eq!(table_heap.first_page_id, 0);
         assert_eq!(table_heap.last_page_id, 0);
         assert_eq!(table_heap.buffer_pool_manager.replacer.size(), 1);
 
-        table_heap.insert_tuple(&meta, &Tuple::new(schema.clone(), vec![1; 2000]));
+        table_heap.insert_tuple(
+            &meta,
+            &Tuple::new(schema.clone(), vec![2i8.into(), 2i16.into()]),
+        );
         assert_eq!(table_heap.first_page_id, 0);
         assert_eq!(table_heap.last_page_id, 0);
         assert_eq!(table_heap.buffer_pool_manager.replacer.size(), 1);
 
-        table_heap.insert_tuple(&meta, &Tuple::new(schema.clone(), vec![1; 2000]));
+        table_heap.insert_tuple(
+            &meta,
+            &Tuple::new(schema.clone(), vec![3i8.into(), 3i16.into()]),
+        );
         assert_eq!(table_heap.first_page_id, 0);
         assert_eq!(table_heap.last_page_id, 1);
         assert_eq!(table_heap.buffer_pool_manager.replacer.size(), 2);
@@ -269,7 +281,10 @@ mod tests {
         let temp_dir = TempDir::new().unwrap();
         let temp_path = temp_dir.path().join("test.db");
 
-        let schema = Arc::new(Schema::empty());
+        let schema = Arc::new(Schema::new(vec![
+            Column::new("a".to_string(), DataType::Int8),
+            Column::new("b".to_string(), DataType::Int16),
+        ]));
         let disk_manager = DiskManager::try_new(&temp_path).unwrap();
         let buffer_pool_manager = BufferPoolManager::new(1000, Arc::new(disk_manager));
         let mut table_heap = TableHeap::new(schema.clone(), buffer_pool_manager);
@@ -280,13 +295,22 @@ mod tests {
         };
 
         let rid1 = table_heap
-            .insert_tuple(&meta, &Tuple::new(schema.clone(), vec![1; 2000]))
+            .insert_tuple(
+                &meta,
+                &Tuple::new(schema.clone(), vec![1i8.into(), 1i16.into()]),
+            )
             .unwrap();
         let rid2 = table_heap
-            .insert_tuple(&meta, &Tuple::new(schema.clone(), vec![2; 2000]))
+            .insert_tuple(
+                &meta,
+                &Tuple::new(schema.clone(), vec![2i8.into(), 2i16.into()]),
+            )
             .unwrap();
         let rid3 = table_heap
-            .insert_tuple(&meta, &Tuple::new(schema.clone(), vec![3; 2000]))
+            .insert_tuple(
+                &meta,
+                &Tuple::new(schema.clone(), vec![3i8.into(), 3i16.into()]),
+            )
             .unwrap();
 
         let mut meta = table_heap.get_tuple_meta(rid2);
@@ -307,7 +331,10 @@ mod tests {
         let temp_dir = TempDir::new().unwrap();
         let temp_path = temp_dir.path().join("test.db");
 
-        let schema = Arc::new(Schema::empty());
+        let schema = Arc::new(Schema::new(vec![
+            Column::new("a".to_string(), DataType::Int8),
+            Column::new("b".to_string(), DataType::Int16),
+        ]));
         let disk_manager = DiskManager::try_new(&temp_path).unwrap();
         let buffer_pool_manager = BufferPoolManager::new(1000, Arc::new(disk_manager));
         let mut table_heap = TableHeap::new(schema.clone(), buffer_pool_manager);
@@ -318,7 +345,10 @@ mod tests {
             is_deleted: false,
         };
         let rid1 = table_heap
-            .insert_tuple(&meta1, &Tuple::new(schema.clone(), vec![1; 2000]))
+            .insert_tuple(
+                &meta1,
+                &Tuple::new(schema.clone(), vec![1i8.into(), 1i16.into()]),
+            )
             .unwrap();
         let meta2 = super::TupleMeta {
             insert_txn_id: 2,
@@ -326,7 +356,10 @@ mod tests {
             is_deleted: false,
         };
         let rid2 = table_heap
-            .insert_tuple(&meta2, &Tuple::new(schema.clone(), vec![2; 2000]))
+            .insert_tuple(
+                &meta2,
+                &Tuple::new(schema.clone(), vec![2i8.into(), 2i16.into()]),
+            )
             .unwrap();
         let meta3 = super::TupleMeta {
             insert_txn_id: 3,
@@ -334,20 +367,23 @@ mod tests {
             is_deleted: false,
         };
         let rid3 = table_heap
-            .insert_tuple(&meta3, &Tuple::new(schema.clone(), vec![3; 2000]))
+            .insert_tuple(
+                &meta3,
+                &Tuple::new(schema.clone(), vec![3i8.into(), 3i16.into()]),
+            )
             .unwrap();
 
         let (meta, tuple) = table_heap.get_tuple(rid1);
         assert_eq!(meta, meta1);
-        assert_eq!(tuple.data, vec![1; 2000]);
+        assert_eq!(tuple.data, vec![1i8.into(), 1i16.into()]);
 
         let (meta, tuple) = table_heap.get_tuple(rid2);
         assert_eq!(meta, meta2);
-        assert_eq!(tuple.data, vec![2; 2000]);
+        assert_eq!(tuple.data, vec![2i8.into(), 2i16.into()]);
 
         let (meta, tuple) = table_heap.get_tuple(rid3);
         assert_eq!(meta, meta3);
-        assert_eq!(tuple.data, vec![3; 2000]);
+        assert_eq!(tuple.data, vec![3i8.into(), 3i16.into()]);
 
         assert_eq!(table_heap.buffer_pool_manager.replacer.size(), 2);
     }
@@ -357,7 +393,10 @@ mod tests {
         let temp_dir = TempDir::new().unwrap();
         let temp_path = temp_dir.path().join("test.db");
 
-        let schema = Arc::new(Schema::empty());
+        let schema = Arc::new(Schema::new(vec![
+            Column::new("a".to_string(), DataType::Int8),
+            Column::new("b".to_string(), DataType::Int16),
+        ]));
 
         let disk_manager = DiskManager::try_new(&temp_path).unwrap();
         let buffer_pool_manager = BufferPoolManager::new(1000, Arc::new(disk_manager));
@@ -369,7 +408,10 @@ mod tests {
             is_deleted: false,
         };
         let rid1 = table_heap
-            .insert_tuple(&meta1, &Tuple::new(schema.clone(), vec![1; 2000]))
+            .insert_tuple(
+                &meta1,
+                &Tuple::new(schema.clone(), vec![1i8.into(), 1i16.into()]),
+            )
             .unwrap();
         let meta2 = super::TupleMeta {
             insert_txn_id: 2,
@@ -377,7 +419,10 @@ mod tests {
             is_deleted: false,
         };
         let rid2 = table_heap
-            .insert_tuple(&meta2, &Tuple::new(schema.clone(), vec![2; 2000]))
+            .insert_tuple(
+                &meta2,
+                &Tuple::new(schema.clone(), vec![2i8.into(), 2i16.into()]),
+            )
             .unwrap();
         let meta3 = super::TupleMeta {
             insert_txn_id: 3,
@@ -385,22 +430,25 @@ mod tests {
             is_deleted: false,
         };
         let rid3 = table_heap
-            .insert_tuple(&meta3, &Tuple::new(schema.clone(), vec![3; 2000]))
+            .insert_tuple(
+                &meta3,
+                &Tuple::new(schema.clone(), vec![3i8.into(), 3i16.into()]),
+            )
             .unwrap();
 
         let mut iterator = table_heap.iter(None, None);
 
         let (meta, tuple) = iterator.next(&mut table_heap).unwrap();
         assert_eq!(meta, meta1);
-        assert_eq!(tuple.data, vec![1; 2000]);
+        assert_eq!(tuple.data, vec![1i8.into(), 1i16.into()]);
 
         let (meta, tuple) = iterator.next(&mut table_heap).unwrap();
         assert_eq!(meta, meta2);
-        assert_eq!(tuple.data, vec![2; 2000]);
+        assert_eq!(tuple.data, vec![2i8.into(), 2i16.into()]);
 
         let (meta, tuple) = iterator.next(&mut table_heap).unwrap();
         assert_eq!(meta, meta3);
-        assert_eq!(tuple.data, vec![3; 2000]);
+        assert_eq!(tuple.data, vec![3i8.into(), 3i16.into()]);
 
         assert!(iterator.next(&mut table_heap).is_none());
 
diff --git a/bustubx/src/storage/table_page.rs b/bustubx/src/storage/table_page.rs
index 4f133ae..00fd514 100644
--- a/bustubx/src/storage/table_page.rs
+++ b/bustubx/src/storage/table_page.rs
@@ -98,7 +98,7 @@ impl TablePage {
 
         // Copy the tuple's data into the appropriate position within the page's data buffer.
         self.data[tuple_offset as usize..(tuple_offset + tuple.data.len() as u16) as usize]
-            .copy_from_slice(&tuple.data);
+            .copy_from_slice(&tuple.to_bytes());
         return Some(tuple_id);
     }
 
@@ -121,9 +121,9 @@ impl TablePage {
         }
 
         let (offset, size, meta) = self.tuple_info[tuple_id as usize];
-        let tuple = Tuple::new(
+        let tuple = Tuple::from_bytes(
             self.schema.clone(),
-            self.data[offset as usize..(offset + size) as usize].to_vec(),
+            &self.data[offset as usize..(offset + size) as usize].to_vec(),
         );
 
         return (meta, tuple);
@@ -219,7 +219,7 @@ impl TablePage {
 
 #[cfg(test)]
 mod tests {
-    use crate::catalog::Schema;
+    use crate::catalog::{Column, DataType, Schema};
     use crate::{
         common::{config::BUSTUBX_PAGE_SIZE, rid::Rid},
         storage::Tuple,
@@ -228,14 +228,20 @@ mod tests {
 
     #[test]
     pub fn test_table_page_insert() {
-        let schema = Arc::new(Schema::empty());
+        let schema = Arc::new(Schema::new(vec![
+            Column::new("a".to_string(), DataType::Int8),
+            Column::new("b".to_string(), DataType::Int16),
+        ]));
         let mut table_page = super::TablePage::new(schema.clone(), 0);
         let meta = super::TupleMeta {
             insert_txn_id: 0,
             delete_txn_id: 0,
             is_deleted: false,
         };
-        let tuple_id = table_page.insert_tuple(&meta, &Tuple::new(schema.clone(), vec![1, 1, 1]));
+        let tuple_id = table_page.insert_tuple(
+            &meta,
+            &Tuple::new(schema.clone(), vec![1i8.into(), 1i16.into()]),
+        );
         assert_eq!(tuple_id, Some(0));
         assert_eq!(table_page.num_tuples, 1);
         assert_eq!(table_page.num_deleted_tuples, 0);
@@ -247,7 +253,10 @@ mod tests {
         assert_eq!(table_page.tuple_info[tuple_id.unwrap() as usize].1, 3);
         assert_eq!(table_page.tuple_info[tuple_id.unwrap() as usize].2, meta);
 
-        let tuple_id = table_page.insert_tuple(&meta, &Tuple::new(schema.clone(), vec![1, 1, 1]));
+        let tuple_id = table_page.insert_tuple(
+            &meta,
+            &Tuple::new(schema.clone(), vec![1i8.into(), 1i16.into()]),
+        );
         assert_eq!(tuple_id, Some(1));
         assert_eq!(table_page.num_tuples, 2);
         assert_eq!(table_page.num_deleted_tuples, 0);
@@ -262,41 +271,65 @@ mod tests {
 
     #[test]
     pub fn test_table_page_get_tuple() {
-        let schema = Arc::new(Schema::empty());
+        let schema = Arc::new(Schema::new(vec![
+            Column::new("a".to_string(), DataType::Int8),
+            Column::new("b".to_string(), DataType::Int16),
+        ]));
         let mut table_page = super::TablePage::new(schema.clone(), 0);
         let meta = super::TupleMeta {
             insert_txn_id: 0,
             delete_txn_id: 0,
             is_deleted: false,
         };
-        let tuple_id = table_page.insert_tuple(&meta, &Tuple::new(schema.clone(), vec![1, 1, 1]));
+        let tuple_id = table_page.insert_tuple(
+            &meta,
+            &Tuple::new(schema.clone(), vec![1i8.into(), 1i16.into()]),
+        );
         assert_eq!(tuple_id, Some(0));
-        let tuple_id = table_page.insert_tuple(&meta, &Tuple::new(schema.clone(), vec![2, 2, 2]));
+        let tuple_id = table_page.insert_tuple(
+            &meta,
+            &Tuple::new(schema.clone(), vec![2i8.into(), 2i16.into()]),
+        );
         assert_eq!(tuple_id, Some(1));
-        let tuple_id = table_page.insert_tuple(&meta, &Tuple::new(schema.clone(), vec![3, 3, 3]));
+        let tuple_id = table_page.insert_tuple(
+            &meta,
+            &Tuple::new(schema.clone(), vec![3i8.into(), 3i16.into()]),
+        );
         assert_eq!(tuple_id, Some(2));
 
         let (tuple_meta, tuple) = table_page.get_tuple(&super::Rid::new(0, 0));
         assert_eq!(tuple_meta, meta);
-        assert_eq!(tuple.data, vec![1, 1, 1]);
+        assert_eq!(tuple.data, vec![1i8.into(), 1i16.into()]);
         let (tuple_meta, tuple) = table_page.get_tuple(&super::Rid::new(0, 1));
-        assert_eq!(tuple.data, vec![2, 2, 2]);
+        assert_eq!(tuple.data, vec![2i8.into(), 2i16.into()]);
         let (tuple_meta, tuple) = table_page.get_tuple(&super::Rid::new(0, 2));
-        assert_eq!(tuple.data, vec![3, 3, 3]);
+        assert_eq!(tuple.data, vec![3i8.into(), 3i16.into()]);
     }
 
     #[test]
     pub fn test_table_page_update_tuple_meta() {
-        let schema = Arc::new(Schema::empty());
+        let schema = Arc::new(Schema::new(vec![
+            Column::new("a".to_string(), DataType::Int8),
+            Column::new("b".to_string(), DataType::Int16),
+        ]));
         let mut table_page = super::TablePage::new(schema.clone(), 0);
         let meta = super::TupleMeta {
             insert_txn_id: 0,
             delete_txn_id: 0,
             is_deleted: false,
         };
-        let tuple_id = table_page.insert_tuple(&meta, &Tuple::new(schema.clone(), vec![1, 1, 1]));
-        let tuple_id = table_page.insert_tuple(&meta, &Tuple::new(schema.clone(), vec![2, 2, 2]));
-        let tuple_id = table_page.insert_tuple(&meta, &Tuple::new(schema.clone(), vec![3, 3, 3]));
+        let tuple_id = table_page.insert_tuple(
+            &meta,
+            &Tuple::new(schema.clone(), vec![1i8.into(), 1i16.into()]),
+        );
+        let tuple_id = table_page.insert_tuple(
+            &meta,
+            &Tuple::new(schema.clone(), vec![2i8.into(), 2i16.into()]),
+        );
+        let tuple_id = table_page.insert_tuple(
+            &meta,
+            &Tuple::new(schema.clone(), vec![3i8.into(), 3i16.into()]),
+        );
 
         let mut tuple_meta = table_page.get_tuple_meta(&super::Rid::new(0, 0));
         tuple_meta.is_deleted = true;
@@ -312,16 +345,28 @@ mod tests {
 
     #[test]
     pub fn test_table_page_from_to_bytes() {
-        let schema = Arc::new(Schema::empty());
+        let schema = Arc::new(Schema::new(vec![
+            Column::new("a".to_string(), DataType::Int8),
+            Column::new("b".to_string(), DataType::Int16),
+        ]));
         let mut table_page = super::TablePage::new(schema.clone(), 1);
         let meta = super::TupleMeta {
             insert_txn_id: 0,
             delete_txn_id: 0,
             is_deleted: false,
         };
-        let tuple_id1 = table_page.insert_tuple(&meta, &Tuple::new(schema.clone(), vec![1, 1, 1]));
-        let tuple_id2 = table_page.insert_tuple(&meta, &Tuple::new(schema.clone(), vec![2, 2, 2]));
-        let tuple_id3 = table_page.insert_tuple(&meta, &Tuple::new(schema.clone(), vec![3, 3, 3]));
+        let tuple_id1 = table_page.insert_tuple(
+            &meta,
+            &Tuple::new(schema.clone(), vec![1i8.into(), 1i16.into()]),
+        );
+        let tuple_id2 = table_page.insert_tuple(
+            &meta,
+            &Tuple::new(schema.clone(), vec![2i8.into(), 2i16.into()]),
+        );
+        let tuple_id3 = table_page.insert_tuple(
+            &meta,
+            &Tuple::new(schema.clone(), vec![3i8.into(), 3i16.into()]),
+        );
 
         let bytes = table_page.to_bytes();
         let table_page2 = super::TablePage::from_bytes(schema.clone(), &bytes);
@@ -347,6 +392,6 @@ mod tests {
 
         let (tuple_meta, tuple) = table_page2.get_tuple(&Rid::new(0, tuple_id2.unwrap() as u32));
         assert_eq!(tuple_meta, meta);
-        assert_eq!(tuple.data, vec![2, 2, 2]);
+        assert_eq!(tuple.data, vec![2i8.into(), 2i16.into()]);
     }
 }
diff --git a/bustubx/src/storage/tuple.rs b/bustubx/src/storage/tuple.rs
index 256cfe2..0cde17c 100644
--- a/bustubx/src/storage/tuple.rs
+++ b/bustubx/src/storage/tuple.rs
@@ -12,31 +12,33 @@ pub struct TupleMeta {
 #[derive(Debug, Clone)]
 pub struct Tuple {
     pub schema: SchemaRef,
-    pub data: Vec<u8>,
+    pub data: Vec<ScalarValue>,
 }
 
 impl Tuple {
-    pub fn new(schema: SchemaRef, data: Vec<u8>) -> Self {
+    pub fn new(schema: SchemaRef, data: Vec<ScalarValue>) -> Self {
         Self { schema, data }
     }
 
-    pub fn empty(schema: SchemaRef, size: usize) -> Self {
-        Self {
-            schema,
-            data: vec![0; size],
-        }
-    }
-
-    pub fn from_values(schema: SchemaRef, values: Vec<ScalarValue>) -> Self {
+    pub fn empty(schema: SchemaRef) -> Self {
         let mut data = vec![];
-        for value in values {
-            data.extend(value.to_bytes());
+        for col in schema.columns.iter() {
+            data.push(ScalarValue::new_empty(col.data_type));
         }
         Self { schema, data }
     }
 
     pub fn from_bytes(schema: SchemaRef, raw: &[u8]) -> Self {
-        let data = raw.to_vec();
+        let mut data = vec![];
+        let mut raw_data = raw.to_vec();
+        for col in schema.columns.iter() {
+            data.push(ScalarValue::from_bytes(raw_data.as_ref(), col.data_type));
+            raw_data = raw_data
+                .into_iter()
+                .skip(col.data_type.type_size())
+                .into_iter()
+                .collect::<Vec<u8>>();
+        }
         Self { schema, data }
     }
 
@@ -54,14 +56,15 @@ impl Tuple {
     }
 
     pub fn is_zero(&self) -> bool {
-        // Iterate over each element in the 'data' vector using the 'iter' method.
-        // The closure '|&x| x == 0' checks if each element is equal to 0.
-        // The 'all' method returns 'true' if the closure returns 'true' for all elements.
-        self.data.iter().all(|&x| x == 0)
+        self.data.iter().all(|x| x.is_null())
     }
 
     pub fn to_bytes(&self) -> Vec<u8> {
-        self.data.clone()
+        let mut bytes = vec![];
+        for v in self.data.iter() {
+            bytes.extend(v.to_bytes());
+        }
+        bytes
     }
 
     pub fn all_values(&self, schema: &Schema) -> Vec<ScalarValue> {
@@ -88,13 +91,14 @@ impl Tuple {
     }
 
     pub fn get_value_by_col(&self, column: ColumnRef) -> ScalarValue {
-        let offset = column.column_offset;
-        let len = column.data_type.type_size();
-        // Intercept the byte sequence starting from offset,
-        // and get length len from data as the current col row bytes.
-        let raw = &self.data[offset..offset + len];
-
-        ScalarValue::from_bytes(raw, column.data_type)
+        let (idx, col) = self
+            .schema
+            .columns
+            .iter()
+            .enumerate()
+            .find(|c| c.1 == &column)
+            .unwrap();
+        self.data.get(idx).unwrap().clone()
     }
 
     // TODO 比较索引key大小
@@ -129,11 +133,11 @@ mod tests {
             Column::new("a".to_string(), DataType::Int8),
             Column::new("b".to_string(), DataType::Int16),
         ]));
-        let tuple1 = super::Tuple::new(schema.clone(), vec![1u8, 1, 1]);
-        let tuple2 = super::Tuple::new(schema.clone(), vec![1u8, 1, 1]);
-        let tuple3 = super::Tuple::new(schema.clone(), vec![1u8, 2, 1]);
-        let tuple4 = super::Tuple::new(schema.clone(), vec![2u8, 1, 1]);
-        let tuple5 = super::Tuple::new(schema.clone(), vec![1u8, 0, 1]);
+        let tuple1 = super::Tuple::new(schema.clone(), vec![1i8.into(), 2i16.into()]);
+        let tuple2 = super::Tuple::new(schema.clone(), vec![1i8.into(), 2i16.into()]);
+        let tuple3 = super::Tuple::new(schema.clone(), vec![1i8.into(), 3i16.into()]);
+        let tuple4 = super::Tuple::new(schema.clone(), vec![2i8.into(), 2i16.into()]);
+        let tuple5 = super::Tuple::new(schema.clone(), vec![1i8.into(), 1i16.into()]);
 
         assert_eq!(tuple1.compare(&tuple2, &schema), std::cmp::Ordering::Equal);
         assert_eq!(tuple1.compare(&tuple3, &schema), std::cmp::Ordering::Less);