diff --git a/bustubx/src/catalog/catalog.rs b/bustubx/src/catalog/catalog.rs
index 6e79e1c..b359359 100644
--- a/bustubx/src/catalog/catalog.rs
+++ b/bustubx/src/catalog/catalog.rs
@@ -198,9 +198,9 @@ mod tests {
 
         let table_name = "test_table1".to_string();
         let schema = Arc::new(Schema::new(vec![
-            Column::new("a".to_string(), DataType::Int8),
-            Column::new("b".to_string(), DataType::Int16),
-            Column::new("c".to_string(), DataType::Int32),
+            Column::new("a".to_string(), DataType::Int8, true),
+            Column::new("b".to_string(), DataType::Int16, true),
+            Column::new("c".to_string(), DataType::Int32, true),
         ]));
         let table_info = catalog
             .create_table(table_name.clone(), schema.clone())
@@ -211,9 +211,9 @@ mod tests {
 
         let table_name = "test_table2".to_string();
         let schema = Arc::new(Schema::new(vec![
-            Column::new("d".to_string(), DataType::Int32),
-            Column::new("e".to_string(), DataType::Int16),
-            Column::new("f".to_string(), DataType::Int8),
+            Column::new("d".to_string(), DataType::Int32, true),
+            Column::new("e".to_string(), DataType::Int16, true),
+            Column::new("f".to_string(), DataType::Int8, true),
         ]));
         let table_info = catalog
             .create_table(table_name.clone(), schema.clone())
@@ -236,17 +236,17 @@ mod tests {
 
         let table_name1 = "test_table1".to_string();
         let schema = Arc::new(Schema::new(vec![
-            Column::new("a".to_string(), DataType::Int8),
-            Column::new("b".to_string(), DataType::Int16),
-            Column::new("c".to_string(), DataType::Int32),
+            Column::new("a".to_string(), DataType::Int8, true),
+            Column::new("b".to_string(), DataType::Int16, true),
+            Column::new("c".to_string(), DataType::Int32, true),
         ]));
         let _ = catalog.create_table(table_name1.clone(), schema);
 
         let table_name2 = "test_table2".to_string();
         let schema = Arc::new(Schema::new(vec![
-            Column::new("d".to_string(), DataType::Int32),
-            Column::new("e".to_string(), DataType::Int16),
-            Column::new("f".to_string(), DataType::Int8),
+            Column::new("d".to_string(), DataType::Int32, true),
+            Column::new("e".to_string(), DataType::Int16, true),
+            Column::new("f".to_string(), DataType::Int8, true),
         ]));
         let _ = catalog.create_table(table_name2.clone(), schema);
 
@@ -286,9 +286,9 @@ mod tests {
 
         let table_name = "test_table1".to_string();
         let schema = Arc::new(Schema::new(vec![
-            Column::new("a".to_string(), DataType::Int8),
-            Column::new("b".to_string(), DataType::Int16),
-            Column::new("c".to_string(), DataType::Int32),
+            Column::new("a".to_string(), DataType::Int8, true),
+            Column::new("b".to_string(), DataType::Int16, true),
+            Column::new("c".to_string(), DataType::Int32, true),
         ]));
         let _ = catalog.create_table(table_name.clone(), schema);
 
diff --git a/bustubx/src/catalog/column.rs b/bustubx/src/catalog/column.rs
index 49e4090..0bc3532 100644
--- a/bustubx/src/catalog/column.rs
+++ b/bustubx/src/catalog/column.rs
@@ -21,12 +21,11 @@ impl PartialEq for Column {
 impl Eq for Column {}
 
 impl Column {
-    // TODO set nullable
-    pub fn new(name: String, data_type: DataType) -> Self {
+    pub fn new(name: String, data_type: DataType, nullable: bool) -> Self {
         Self {
             name,
             data_type,
-            nullable: false,
+            nullable,
         }
     }
 }
diff --git a/bustubx/src/catalog/schema.rs b/bustubx/src/catalog/schema.rs
index 0f3bf6c..6770eef 100644
--- a/bustubx/src/catalog/schema.rs
+++ b/bustubx/src/catalog/schema.rs
@@ -9,7 +9,7 @@ pub type SchemaRef = Arc<Schema>;
 lazy_static::lazy_static! {
     pub static ref EMPTY_SCHEMA_REF: SchemaRef = Arc::new(Schema::empty());
     pub static ref INSERT_OUTPUT_SCHEMA_REF: SchemaRef = Arc::new(Schema::new(
-        vec![Column::new("insert_rows".to_string(), DataType::Int32)]
+        vec![Column::new("insert_rows".to_string(), DataType::Int32, false)]
     ));
 }
 
diff --git a/bustubx/src/expression/column.rs b/bustubx/src/expression/column.rs
index 732fe75..80ddcea 100644
--- a/bustubx/src/expression/column.rs
+++ b/bustubx/src/expression/column.rs
@@ -31,11 +31,11 @@ impl ExprTrait for ColumnExpr {
     }
 
     fn to_column(&self, input_schema: &Schema) -> BustubxResult<Column> {
-        Ok(Column {
-            name: self.name.clone(),
-            data_type: self.data_type(input_schema)?,
-            nullable: self.nullable(input_schema)?,
-        })
+        Ok(Column::new(
+            self.name.clone(),
+            self.data_type(input_schema)?,
+            self.nullable(input_schema)?,
+        ))
     }
 }
 
diff --git a/bustubx/src/expression/literal.rs b/bustubx/src/expression/literal.rs
index 8ec36c3..121cf83 100644
--- a/bustubx/src/expression/literal.rs
+++ b/bustubx/src/expression/literal.rs
@@ -27,6 +27,7 @@ impl ExprTrait for Literal {
         Ok(Column::new(
             format!("{}", self.value),
             self.data_type(input_schema)?,
+            self.nullable(input_schema)?,
         ))
     }
 }
diff --git a/bustubx/src/planner/logical_planner/plan_create_table.rs b/bustubx/src/planner/logical_planner/plan_create_table.rs
index 07398be..ad56588 100644
--- a/bustubx/src/planner/logical_planner/plan_create_table.rs
+++ b/bustubx/src/planner/logical_planner/plan_create_table.rs
@@ -17,6 +17,7 @@ impl<'a> LogicalPlanner<'a> {
             columns.push(Column::new(
                 col_def.name.value.clone(),
                 (&col_def.data_type).try_into()?,
+                false,
             ))
         }
         Ok(LogicalPlan::CreateTable(CreateTable { name, columns }))
diff --git a/bustubx/src/planner/logical_planner/plan_set_expr.rs b/bustubx/src/planner/logical_planner/plan_set_expr.rs
index 1fd85cd..72b2b3c 100644
--- a/bustubx/src/planner/logical_planner/plan_set_expr.rs
+++ b/bustubx/src/planner/logical_planner/plan_set_expr.rs
@@ -1,4 +1,4 @@
-use crate::catalog::{Column, Schema};
+use crate::catalog::{Column, Schema, EMPTY_SCHEMA_REF};
 use crate::expression::{Alias, ColumnExpr, Expr, ExprTrait};
 use crate::planner::logical_plan::JoinType;
 use crate::planner::logical_plan::{
@@ -258,7 +258,8 @@ impl LogicalPlanner<'_> {
         for (idx, item) in first_row.iter().enumerate() {
             columns.push(Column::new(
                 idx.to_string(),
-                item.data_type(&Schema::empty())?,
+                item.data_type(&EMPTY_SCHEMA_REF)?,
+                item.nullable(&EMPTY_SCHEMA_REF)?,
             ))
         }
 
diff --git a/bustubx/src/storage/index.rs b/bustubx/src/storage/index.rs
index bbcda5d..5d39721 100644
--- a/bustubx/src/storage/index.rs
+++ b/bustubx/src/storage/index.rs
@@ -795,10 +795,10 @@ mod tests {
             "test_index".to_string(),
             "test_table".to_string(),
             Arc::new(Schema::new(vec![
-                Column::new("a".to_string(), DataType::Int8),
-                Column::new("b".to_string(), DataType::Int16),
-                Column::new("c".to_string(), DataType::Int8),
-                Column::new("d".to_string(), DataType::Int16),
+                Column::new("a".to_string(), DataType::Int8, false),
+                Column::new("b".to_string(), DataType::Int16, false),
+                Column::new("c".to_string(), DataType::Int8, false),
+                Column::new("d".to_string(), DataType::Int16, false),
             ])),
             vec![1, 3],
         );
@@ -819,8 +819,8 @@ mod tests {
         let temp_path = temp_dir.path().join("test.db");
 
         let schema = Arc::new(Schema::new(vec![
-            Column::new("a".to_string(), DataType::Int8),
-            Column::new("b".to_string(), DataType::Int16),
+            Column::new("a".to_string(), DataType::Int8, false),
+            Column::new("b".to_string(), DataType::Int16, false),
         ]));
         let index_metadata = IndexMetadata::new(
             "test_index".to_string(),
@@ -904,8 +904,8 @@ mod tests {
         let temp_path = temp_dir.path().join("test.db");
 
         let schema = Arc::new(Schema::new(vec![
-            Column::new("a".to_string(), DataType::Int8),
-            Column::new("b".to_string(), DataType::Int16),
+            Column::new("a".to_string(), DataType::Int8, false),
+            Column::new("b".to_string(), DataType::Int16, false),
         ]));
         let index_metadata = IndexMetadata::new(
             "test_index".to_string(),
diff --git a/bustubx/src/storage/index_page.rs b/bustubx/src/storage/index_page.rs
index 3b50c6f..d6fb96a 100644
--- a/bustubx/src/storage/index_page.rs
+++ b/bustubx/src/storage/index_page.rs
@@ -597,8 +597,8 @@ mod tests {
     #[test]
     pub fn test_internal_page_from_to_bytes() {
         let key_schema = Arc::new(Schema::new(vec![
-            Column::new("a".to_string(), DataType::Int8),
-            Column::new("a".to_string(), DataType::Int16),
+            Column::new("a".to_string(), DataType::Int8, false),
+            Column::new("a".to_string(), DataType::Int16, false),
         ]));
         let mut ori_page = BPlusTreeInternalPage::new(key_schema.clone(), 5);
         ori_page.insert(Tuple::empty(key_schema.clone()), 0, &key_schema);
@@ -634,8 +634,8 @@ mod tests {
     #[test]
     pub fn test_leaf_page_from_to_bytes() {
         let key_schema = Arc::new(Schema::new(vec![
-            Column::new("a".to_string(), DataType::Int8),
-            Column::new("a".to_string(), DataType::Int16),
+            Column::new("a".to_string(), DataType::Int8, false),
+            Column::new("a".to_string(), DataType::Int16, false),
         ]));
         let mut ori_page = BPlusTreeLeafPage::new(key_schema.clone(), 5);
         ori_page.insert(
@@ -665,8 +665,8 @@ mod tests {
     #[test]
     pub fn test_internal_page_insert() {
         let key_schema = Arc::new(Schema::new(vec![
-            Column::new("a".to_string(), DataType::Int8),
-            Column::new("b".to_string(), DataType::Int16),
+            Column::new("a".to_string(), DataType::Int8, false),
+            Column::new("b".to_string(), DataType::Int16, false),
         ]));
         let mut internal_page = BPlusTreeInternalPage::new(key_schema.clone(), 3);
         internal_page.insert(Tuple::empty(key_schema.clone()), 0, &key_schema);
@@ -695,8 +695,8 @@ mod tests {
     #[test]
     pub fn test_leaf_page_insert() {
         let key_schema = Arc::new(Schema::new(vec![
-            Column::new("a".to_string(), DataType::Int8),
-            Column::new("b".to_string(), DataType::Int16),
+            Column::new("a".to_string(), DataType::Int8, false),
+            Column::new("b".to_string(), DataType::Int16, false),
         ]));
         let mut leaf_page = BPlusTreeLeafPage::new(key_schema.clone(), 3);
         leaf_page.insert(
@@ -726,8 +726,8 @@ mod tests {
     #[test]
     pub fn test_internal_page_look_up() {
         let key_schema = Arc::new(Schema::new(vec![
-            Column::new("a".to_string(), DataType::Int8),
-            Column::new("b".to_string(), DataType::Int16),
+            Column::new("a".to_string(), DataType::Int8, false),
+            Column::new("b".to_string(), DataType::Int16, false),
         ]));
         let mut internal_page = BPlusTreeInternalPage::new(key_schema.clone(), 5);
         internal_page.insert(Tuple::empty(key_schema.clone()), 0, &key_schema);
@@ -808,8 +808,8 @@ mod tests {
     #[test]
     pub fn test_leaf_page_look_up() {
         let key_schema = Arc::new(Schema::new(vec![
-            Column::new("a".to_string(), DataType::Int8),
-            Column::new("b".to_string(), DataType::Int16),
+            Column::new("a".to_string(), DataType::Int8, false),
+            Column::new("b".to_string(), DataType::Int16, false),
         ]));
         let mut leaf_page = BPlusTreeLeafPage::new(key_schema.clone(), 5);
         leaf_page.insert(
@@ -913,8 +913,8 @@ mod tests {
     #[test]
     pub fn test_internal_page_delete() {
         let key_schema = Arc::new(Schema::new(vec![
-            Column::new("a".to_string(), DataType::Int8),
-            Column::new("b".to_string(), DataType::Int16),
+            Column::new("a".to_string(), DataType::Int8, false),
+            Column::new("b".to_string(), DataType::Int16, false),
         ]));
         let mut internal_page = BPlusTreeInternalPage::new(key_schema.clone(), 5);
         internal_page.insert(Tuple::empty(key_schema.clone()), 0, &key_schema);
@@ -980,8 +980,8 @@ mod tests {
     #[test]
     pub fn test_leaf_page_delete() {
         let key_schema = Arc::new(Schema::new(vec![
-            Column::new("a".to_string(), DataType::Int8),
-            Column::new("b".to_string(), DataType::Int16),
+            Column::new("a".to_string(), DataType::Int8, false),
+            Column::new("b".to_string(), DataType::Int16, false),
         ]));
         let mut leaf_page = BPlusTreeLeafPage::new(key_schema.clone(), 5);
         leaf_page.insert(
diff --git a/bustubx/src/storage/table_heap.rs b/bustubx/src/storage/table_heap.rs
index bb790d0..fc03792 100644
--- a/bustubx/src/storage/table_heap.rs
+++ b/bustubx/src/storage/table_heap.rs
@@ -235,8 +235,8 @@ mod tests {
         let temp_path = temp_dir.path().join("test.db");
 
         let schema = Arc::new(Schema::new(vec![
-            Column::new("a".to_string(), DataType::Int8),
-            Column::new("b".to_string(), DataType::Int16),
+            Column::new("a".to_string(), DataType::Int8, false),
+            Column::new("b".to_string(), DataType::Int16, false),
         ]));
         let disk_manager = DiskManager::try_new(&temp_path).unwrap();
         let buffer_pool_manager = BufferPoolManager::new(1000, Arc::new(disk_manager));
@@ -275,8 +275,8 @@ mod tests {
         let temp_path = temp_dir.path().join("test.db");
 
         let schema = Arc::new(Schema::new(vec![
-            Column::new("a".to_string(), DataType::Int8),
-            Column::new("b".to_string(), DataType::Int16),
+            Column::new("a".to_string(), DataType::Int8, false),
+            Column::new("b".to_string(), DataType::Int16, false),
         ]));
         let disk_manager = DiskManager::try_new(&temp_path).unwrap();
         let buffer_pool_manager = BufferPoolManager::new(1000, Arc::new(disk_manager));
@@ -324,8 +324,8 @@ mod tests {
         let temp_path = temp_dir.path().join("test.db");
 
         let schema = Arc::new(Schema::new(vec![
-            Column::new("a".to_string(), DataType::Int8),
-            Column::new("b".to_string(), DataType::Int16),
+            Column::new("a".to_string(), DataType::Int8, false),
+            Column::new("b".to_string(), DataType::Int16, false),
         ]));
         let disk_manager = DiskManager::try_new(&temp_path).unwrap();
         let buffer_pool_manager = BufferPoolManager::new(1000, Arc::new(disk_manager));
@@ -384,8 +384,8 @@ mod tests {
         let temp_path = temp_dir.path().join("test.db");
 
         let schema = Arc::new(Schema::new(vec![
-            Column::new("a".to_string(), DataType::Int8),
-            Column::new("b".to_string(), DataType::Int16),
+            Column::new("a".to_string(), DataType::Int8, false),
+            Column::new("b".to_string(), DataType::Int16, false),
         ]));
 
         let disk_manager = DiskManager::try_new(&temp_path).unwrap();
diff --git a/bustubx/src/storage/table_page.rs b/bustubx/src/storage/table_page.rs
index 30304a1..c5e1d17 100644
--- a/bustubx/src/storage/table_page.rs
+++ b/bustubx/src/storage/table_page.rs
@@ -227,8 +227,8 @@ mod tests {
     #[test]
     pub fn test_table_page_insert() {
         let schema = Arc::new(Schema::new(vec![
-            Column::new("a".to_string(), DataType::Int8),
-            Column::new("b".to_string(), DataType::Int16),
+            Column::new("a".to_string(), DataType::Int8, false),
+            Column::new("b".to_string(), DataType::Int16, false),
         ]));
         let mut table_page = super::TablePage::new(schema.clone(), 0);
         let meta = super::TupleMeta {
@@ -270,8 +270,8 @@ mod tests {
     #[test]
     pub fn test_table_page_get_tuple() {
         let schema = Arc::new(Schema::new(vec![
-            Column::new("a".to_string(), DataType::Int8),
-            Column::new("b".to_string(), DataType::Int16),
+            Column::new("a".to_string(), DataType::Int8, false),
+            Column::new("b".to_string(), DataType::Int16, false),
         ]));
         let mut table_page = super::TablePage::new(schema.clone(), 0);
         let meta = super::TupleMeta {
@@ -307,8 +307,8 @@ mod tests {
     #[test]
     pub fn test_table_page_update_tuple_meta() {
         let schema = Arc::new(Schema::new(vec![
-            Column::new("a".to_string(), DataType::Int8),
-            Column::new("b".to_string(), DataType::Int16),
+            Column::new("a".to_string(), DataType::Int8, false),
+            Column::new("b".to_string(), DataType::Int16, false),
         ]));
         let mut table_page = super::TablePage::new(schema.clone(), 0);
         let meta = super::TupleMeta {
@@ -344,8 +344,8 @@ mod tests {
     #[test]
     pub fn test_table_page_from_to_bytes() {
         let schema = Arc::new(Schema::new(vec![
-            Column::new("a".to_string(), DataType::Int8),
-            Column::new("b".to_string(), DataType::Int16),
+            Column::new("a".to_string(), DataType::Int8, false),
+            Column::new("b".to_string(), DataType::Int16, false),
         ]));
         let mut table_page = super::TablePage::new(schema.clone(), 1);
         let meta = super::TupleMeta {
diff --git a/bustubx/src/storage/tuple.rs b/bustubx/src/storage/tuple.rs
index ce1db9d..71ab554 100644
--- a/bustubx/src/storage/tuple.rs
+++ b/bustubx/src/storage/tuple.rs
@@ -123,8 +123,8 @@ mod tests {
     #[test]
     pub fn test_compare() {
         let schema = Arc::new(Schema::new(vec![
-            Column::new("a".to_string(), DataType::Int8),
-            Column::new("b".to_string(), DataType::Int16),
+            Column::new("a".to_string(), DataType::Int8, false),
+            Column::new("b".to_string(), DataType::Int16, false),
         ]));
         let tuple1 = super::Tuple::new(schema.clone(), vec![1i8.into(), 2i16.into()]);
         let tuple2 = super::Tuple::new(schema.clone(), vec![1i8.into(), 2i16.into()]);