diff --git a/src/binder.rs b/src/binder.rs
index 65cba89..4d535b9 100644
--- a/src/binder.rs
+++ b/src/binder.rs
@@ -934,7 +934,7 @@ mod tests {
                     BoundBaseTableReferenceAST {
                         table_name: "t1".to_string(),
                         alias: Some("_t1".to_string()),
-                        first_page_id: PageID(3),
+                        first_page_id: PageID(5),
                         schema: Schema {
                             columns: vec![
                                 Column {
@@ -1017,7 +1017,7 @@ mod tests {
                         left: Box::new(BoundTableReferenceAST::Base(BoundBaseTableReferenceAST {
                             table_name: "t1".to_string(),
                             alias: None,
-                            first_page_id: PageID(3),
+                            first_page_id: PageID(5),
                             schema: Schema {
                                 columns: vec![
                                     Column {
@@ -1034,7 +1034,7 @@ mod tests {
                         right: Box::new(BoundTableReferenceAST::Base(BoundBaseTableReferenceAST {
                             table_name: "t2".to_string(),
                             alias: Some("_t2".to_string()),
-                            first_page_id: PageID(4),
+                            first_page_id: PageID(6),
                             schema: Schema {
                                 columns: vec![
                                     Column {
@@ -1174,7 +1174,7 @@ mod tests {
                                 BoundBaseTableReferenceAST {
                                     table_name: "t1".to_string(),
                                     alias: None,
-                                    first_page_id: PageID(3),
+                                    first_page_id: PageID(5),
                                     schema: Schema {
                                         columns: vec![
                                             Column {
@@ -1293,7 +1293,7 @@ mod tests {
                         data_type: Some(DataType::Varchar),
                     }),
                 ],
-                first_page_id: PageID(3),
+                first_page_id: PageID(5),
                 table_schema: Schema {
                     columns: vec![
                         Column {
@@ -1328,7 +1328,7 @@ mod tests {
                 table_reference: BoundBaseTableReferenceAST {
                     table_name: "t1".to_string(),
                     alias: None,
-                    first_page_id: PageID(3),
+                    first_page_id: PageID(5),
                     schema: Schema {
                         columns: vec![
                             Column {
@@ -1379,7 +1379,7 @@ mod tests {
                 table_reference: BoundBaseTableReferenceAST {
                     table_name: "t1".to_string(),
                     alias: None,
-                    first_page_id: PageID(3),
+                    first_page_id: PageID(5),
                     schema: Schema {
                         columns: vec![
                             Column {
diff --git a/src/catalog.rs b/src/catalog.rs
index c1aee58..5816926 100644
--- a/src/catalog.rs
+++ b/src/catalog.rs
@@ -7,8 +7,10 @@ use crate::{
     common::{PageID, TransactionID},
     concurrency::TransactionManager,
     lock::LockManager,
-    log::{LogManager, LogRecordBody, NewTablePage},
-    page::table_page::TABLE_PAGE_PAGE_TYPE,
+    log::{LogManager, LogRecordBody, NewBPlusTreeLeafPage, NewTablePage},
+    page::{
+        b_plus_tree_leaf_page::B_PLUS_TREE_LEAF_PAGE_PAGE_TYPE, table_page::TABLE_PAGE_PAGE_TYPE,
+    },
     table::TableHeap,
     value::{integer::IntegerValue, varchar::VarcharValue, Value},
 };
@@ -60,6 +62,7 @@ pub struct Catalog {
     lock_manager: Arc<RwLock<LockManager>>,
     log_manager: Arc<Mutex<LogManager>>,
     next_table_id: u32,
+    next_index_id: u32,
 }
 
 impl Catalog {
@@ -75,11 +78,13 @@ impl Catalog {
             lock_manager,
             log_manager,
             next_table_id: 0,
+            next_index_id: 0,
         }
     }
     pub fn bootstrap(&mut self, init: bool) -> Result<()> {
         if !init {
             self.set_next_table_id()?;
+            self.set_next_index_id()?;
             return Ok(());
         }
         let txn_id = self
@@ -102,6 +107,18 @@ impl Catalog {
             txn_id,
             SYSTEM_COLUMNS_FIRST_PAGE_ID,
         )?;
+        self.create_system_table(
+            "system_indexes",
+            &Self::system_indexes_schema(),
+            txn_id,
+            SYSTEM_INDEXES_FIRST_PAGE_ID,
+        )?;
+        self.create_system_table(
+            "system_index_columns",
+            &Self::system_index_columns_schema(),
+            txn_id,
+            SYSTEM_INDEX_COLUMNS_FIRST_PAGE_ID,
+        )?;
         self.transaction_manager
             .lock()
             .map_err(|_| anyhow::anyhow!("lock error"))?
@@ -172,6 +189,70 @@ impl Catalog {
         }
         Ok(())
     }
+    pub fn create_index(
+        &mut self,
+        name: &str,
+        table_name: &str,
+        column_names: &[String],
+        txn_id: TransactionID,
+    ) -> Result<()> {
+        let page = self
+            .buffer_pool_manager
+            .lock()
+            .map_err(|_| anyhow::anyhow!("lock error"))?
+            .new_page(B_PLUS_TREE_LEAF_PAGE_PAGE_TYPE)?;
+        let page_id = page
+            .read()
+            .map_err(|_| anyhow::anyhow!("lock error"))?
+            .page_id();
+        let lsn = self
+            .log_manager
+            .lock()
+            .map_err(|_| anyhow::anyhow!("lock error"))?
+            .append(
+                txn_id,
+                LogRecordBody::NewBPlusTreeLeafPage(NewBPlusTreeLeafPage { page_id }),
+            )?;
+        page.write()
+            .map_err(|_| anyhow::anyhow!("lock error"))?
+            .with_b_plus_tree_leaf_page_mut(|page| page.set_lsn(lsn));
+        self.buffer_pool_manager
+            .lock()
+            .map_err(|_| anyhow::anyhow!("lock error"))?
+            .unpin_page(
+                page.read()
+                    .map_err(|_| anyhow::anyhow!("lock error"))?
+                    .page_id(),
+                true,
+            )?;
+        let mut system_indexes_table =
+            self.system_table_heap(PageID(SYSTEM_INDEXES_FIRST_PAGE_ID.0), txn_id);
+        let index_id = self.next_index_id;
+        let values = vec![
+            Value::Integer(IntegerValue(index_id as i64)),
+            Value::Varchar(VarcharValue(name.to_string())),
+            Value::Varchar(VarcharValue(table_name.to_string())),
+            Value::Integer(IntegerValue(
+                page.read()
+                    .map_err(|_| anyhow::anyhow!("lock error"))?
+                    .page_id()
+                    .0 as i64,
+            )),
+        ];
+        system_indexes_table.insert(&values)?;
+        self.next_index_id += 1;
+        let mut system_index_columns_table =
+            self.system_table_heap(PageID(SYSTEM_INDEX_COLUMNS_FIRST_PAGE_ID.0), txn_id);
+        for (i, column_name) in column_names.iter().enumerate() {
+            let values = vec![
+                Value::Integer(IntegerValue(index_id as i64)),
+                Value::Varchar(VarcharValue(column_name.to_string())),
+                Value::Integer(IntegerValue(i as i64)),
+            ];
+            system_index_columns_table.insert(&values)?;
+        }
+        Ok(())
+    }
     pub fn get_first_page_id_by_table_name(
         &self,
         table_name: &str,
@@ -321,6 +402,30 @@ impl Catalog {
             .commit(txn_id)?;
         Ok(())
     }
+    fn set_next_index_id(&mut self) -> Result<()> {
+        let mut max_index_id = 0;
+        let txn_id = self
+            .transaction_manager
+            .lock()
+            .map_err(|_| anyhow::anyhow!("lock error"))?
+            .begin()?;
+        let system_indexes_table =
+            self.system_table_heap(PageID(SYSTEM_INDEXES_FIRST_PAGE_ID.0), txn_id);
+        for tuple in system_indexes_table.iter() {
+            let values = tuple.values(&Self::system_indexes_schema());
+            if let Value::Integer(IntegerValue(index_id)) = values[0] {
+                if index_id > max_index_id {
+                    max_index_id = index_id;
+                }
+            }
+        }
+        self.next_table_id = (max_index_id as u32) + 1;
+        self.transaction_manager
+            .lock()
+            .map_err(|_| anyhow::anyhow!("lock error"))?
+            .commit(txn_id)?;
+        Ok(())
+    }
     fn get_table_id_by_table_name(&self, table_name: &str, txn_id: TransactionID) -> Result<u32> {
         let system_tables_table =
             self.system_table_heap(PageID(SYSTEM_TABLES_FIRST_PAGE_ID.0), txn_id);
@@ -377,12 +482,49 @@ impl Catalog {
             ],
         }
     }
+    pub fn system_indexes_schema() -> Schema {
+        Schema {
+            columns: vec![
+                Column {
+                    name: "id".to_string(),
+                    data_type: DataType::Integer,
+                },
+                Column {
+                    name: "name".to_string(),
+                    data_type: DataType::Varchar,
+                },
+                Column {
+                    name: "table_name".to_string(),
+                    data_type: DataType::Varchar,
+                },
+                Column {
+                    name: "first_page_id".to_string(),
+                    data_type: DataType::Integer,
+                },
+            ],
+        }
+    }
+    pub fn system_index_columns_schema() -> Schema {
+        Schema {
+            columns: vec![
+                Column {
+                    name: "index_id".to_string(),
+                    data_type: DataType::Integer,
+                },
+                Column {
+                    name: "column_name".to_string(),
+                    data_type: DataType::Varchar,
+                },
+            ],
+        }
+    }
 }
 
-const SYSTEM_TABLE_COUNT: usize = 2;
+const SYSTEM_TABLE_COUNT: usize = 4;
 const SYSTEM_TABLES_FIRST_PAGE_ID: PageID = PageID(1);
 const SYSTEM_COLUMNS_FIRST_PAGE_ID: PageID = PageID(2);
-// const SYSTEM_INDEXES_FIRST_PAGE_ID: PageID = PageID(3);
+const SYSTEM_INDEXES_FIRST_PAGE_ID: PageID = PageID(3);
+const SYSTEM_INDEX_COLUMNS_FIRST_PAGE_ID: PageID = PageID(4);
 
 #[cfg(test)]
 mod tests {
diff --git a/src/executor.rs b/src/executor.rs
index 861b7c3..837d12d 100644
--- a/src/executor.rs
+++ b/src/executor.rs
@@ -64,11 +64,11 @@ impl ExecutorEngine {
     }
     fn create_executor(&self, plan: &Plan) -> Executor {
         match plan {
-            Plan::SeqScan(plan) => Executor::SeqScan(SeqScanExecutor {
-                plan: plan.clone(),
+            Plan::SeqScan(seq_scan_plan) => Executor::SeqScan(SeqScanExecutor {
+                plan: seq_scan_plan.clone(),
                 executor_context: &self.context,
                 table_iterator: TableHeap::new(
-                    plan.first_page_id,
+                    seq_scan_plan.first_page_id,
                     self.context.buffer_pool_manager.clone(),
                     self.context.transaction_manager.clone(),
                     self.context.lock_manager.clone(),
@@ -77,42 +77,46 @@ impl ExecutorEngine {
                 )
                 .iter(),
             }),
-            Plan::Filter(plan) => Executor::Filter(FilterExecutor {
-                plan: plan.clone(),
-                child: Box::new(self.create_executor(&plan.child)),
+            Plan::IndexScan(_) => unimplemented!(),
+            Plan::Filter(filter_plan) => Executor::Filter(FilterExecutor {
+                plan: filter_plan.clone(),
+                child: Box::new(self.create_executor(&plan.children()[0])),
                 executor_context: &self.context,
             }),
-            Plan::Project(plan) => Executor::Project(ProjectExecutor {
-                plan: plan.clone(),
-                child: Box::new(self.create_executor(&plan.child)),
+            Plan::Project(project_plan) => Executor::Project(ProjectExecutor {
+                plan: project_plan.clone(),
+                child: Box::new(self.create_executor(&plan.children()[0])),
                 executor_context: &self.context,
             }),
-            Plan::NestedLoopJoin(plan) => {
+            Plan::NestedLoopJoin(nested_loop_join_plan) => {
                 let children = plan
-                    .children
+                    .children()
                     .iter()
                     .map(|child| self.create_executor(child))
                     .collect::<Vec<_>>();
                 Executor::NestedLoopJoin(NestedLoopJoinExecutor {
-                    plan: plan.clone(),
+                    plan: nested_loop_join_plan.clone(),
                     children,
                     tuples: vec![],
                     executor_context: &self.context,
-                    matched_statuses: vec![false; plan.children.len() - 1],
-                    in_guard_statuses: vec![false; plan.children.len() - 1],
+                    matched_statuses: vec![false; plan.children().len() - 1],
+                    in_guard_statuses: vec![false; plan.children().len() - 1],
                 })
             }
-            Plan::Aggregate(plan) => {
+            Plan::Aggregate(aggregate_plan) => {
                 let mut aggregate_tables = vec![];
-                for _ in 0..plan.aggregate_functions.len() {
+                for _ in 0..aggregate_plan.aggregate_functions.len() {
                     aggregate_tables.push(AggregateTable::new());
                 }
                 Executor::Aggregate(AggregateExecutor {
-                    plan: plan.clone(),
-                    child: Box::new(self.create_executor(&plan.child)),
+                    plan: aggregate_plan.clone(),
+                    child: Box::new(self.create_executor(&plan.children()[0])),
                     executor_context: &self.context,
-                    aggregate_table_value: if plan.group_by.is_empty() {
-                        AggregateTableValue::Value(vec![vec![]; plan.aggregate_functions.len()])
+                    aggregate_table_value: if aggregate_plan.group_by.is_empty() {
+                        AggregateTableValue::Value(vec![
+                            vec![];
+                            aggregate_plan.aggregate_functions.len()
+                        ])
                     } else {
                         AggregateTableValue::Table(AggregateTable::new())
                     },
@@ -120,30 +124,30 @@ impl ExecutorEngine {
                     index: 0,
                 })
             }
-            Plan::Sort(plan) => Executor::Sort(SortExecutor {
-                plan: plan.clone(),
-                child: Box::new(self.create_executor(&plan.child)),
+            Plan::Sort(sort_plan) => Executor::Sort(SortExecutor {
+                plan: sort_plan.clone(),
+                child: Box::new(self.create_executor(&plan.children()[0])),
                 executor_context: &self.context,
                 result: vec![],
                 cursor: 0,
             }),
-            Plan::Limit(plan) => Executor::Limit(limit_executor::LimitExecutor {
-                plan: plan.clone(),
-                child: Box::new(self.create_executor(&plan.child)),
+            Plan::Limit(limit_plan) => Executor::Limit(limit_executor::LimitExecutor {
+                plan: limit_plan.clone(),
+                child: Box::new(self.create_executor(&plan.children()[0])),
                 executor_context: &self.context,
                 result: vec![],
                 cursor: 0,
             }),
-            Plan::EmptyRow(plan) => Executor::EmptyRow(EmptyRowExecutor {
-                plan: plan.clone(),
+            Plan::EmptyRow(empty_plan) => Executor::EmptyRow(EmptyRowExecutor {
+                plan: empty_plan.clone(),
                 executor_context: &self.context,
                 returned: false,
             }),
-            Plan::Insert(plan) => Executor::Insert(InsertExecutor {
-                plan: plan.clone(),
+            Plan::Insert(insert_plan) => Executor::Insert(InsertExecutor {
+                plan: insert_plan.clone(),
                 executor_context: &self.context,
                 table_heap: TableHeap::new(
-                    plan.first_page_id,
+                    insert_plan.first_page_id,
                     self.context.buffer_pool_manager.clone(),
                     self.context.transaction_manager.clone(),
                     self.context.lock_manager.clone(),
@@ -153,12 +157,12 @@ impl ExecutorEngine {
                 count: 0,
                 executed: false,
             }),
-            Plan::Delete(plan) => Executor::Delete(DeleteExecutor {
-                plan: plan.clone(),
-                child: Box::new(self.create_executor(&plan.child)),
+            Plan::Delete(delete_plan) => Executor::Delete(DeleteExecutor {
+                plan: delete_plan.clone(),
+                child: Box::new(self.create_executor(&plan.children()[0])),
                 executor_context: &self.context,
                 table_heap: TableHeap::new(
-                    plan.first_page_id,
+                    delete_plan.first_page_id,
                     self.context.buffer_pool_manager.clone(),
                     self.context.transaction_manager.clone(),
                     self.context.lock_manager.clone(),
@@ -168,12 +172,12 @@ impl ExecutorEngine {
                 count: 0,
                 executed: false,
             }),
-            Plan::Update(plan) => Executor::Update(UpdateExecutor {
-                plan: plan.clone(),
-                child: Box::new(self.create_executor(&plan.child)),
+            Plan::Update(update_plan) => Executor::Update(UpdateExecutor {
+                plan: update_plan.clone(),
+                child: Box::new(self.create_executor(&plan.children()[0])),
                 executor_context: &self.context,
                 table_heap: TableHeap::new(
-                    plan.first_page_id,
+                    update_plan.first_page_id,
                     self.context.buffer_pool_manager.clone(),
                     self.context.transaction_manager.clone(),
                     self.context.lock_manager.clone(),
diff --git a/src/instance.rs b/src/instance.rs
index d2a8b2d..dcf5f20 100644
--- a/src/instance.rs
+++ b/src/instance.rs
@@ -15,7 +15,7 @@ use crate::{
     executor::{ExecutorContext, ExecutorEngine},
     lock::LockManager,
     log::LogManager,
-    parser::{CreateTableStatementAST, StatementAST},
+    parser::{CreateIndexStatementAST, CreateTableStatementAST, StatementAST},
     plan::Planner,
     recovery::RecoveryManager,
     value::Value,
@@ -105,6 +105,21 @@ impl Instance {
             .map_err(|e| anyhow::anyhow!("{}", e))?
             .create_table(&statement.table_name, &schema, txn_id)
     }
+    pub fn create_index(
+        &self,
+        statement: &CreateIndexStatementAST,
+        txn_id: TransactionID,
+    ) -> Result<()> {
+        self.catalog
+            .lock()
+            .map_err(|e| anyhow::anyhow!("{}", e))?
+            .create_index(
+                &statement.index_name,
+                &statement.table_name,
+                &statement.column_names,
+                txn_id,
+            )
+    }
 
     // DDL
     pub fn execute(
diff --git a/src/log.rs b/src/log.rs
index 020ad99..507b834 100644
--- a/src/log.rs
+++ b/src/log.rs
@@ -110,6 +110,7 @@ pub enum LogRecordBody {
     DeleteFromTablePage(DeleteFromTablePage),
     SetNextPageID(SetNextPageID),
     NewTablePage(NewTablePage),
+    NewBPlusTreeLeafPage(NewBPlusTreeLeafPage),
 }
 
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -130,6 +131,10 @@ pub struct SetNextPageID {
 pub struct NewTablePage {
     pub page_id: PageID,
 }
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct NewBPlusTreeLeafPage {
+    pub page_id: PageID,
+}
 
 impl From<&[u8]> for LogRecordBody {
     fn from(bytes: &[u8]) -> Self {
@@ -144,6 +149,7 @@ impl From<&[u8]> for LogRecordBody {
             4 => LogRecordBody::DeleteFromTablePage(DeleteFromTablePage::from(&bytes[4..])),
             5 => LogRecordBody::SetNextPageID(SetNextPageID::from(&bytes[4..])),
             6 => LogRecordBody::NewTablePage(NewTablePage::from(&bytes[4..])),
+            7 => LogRecordBody::NewBPlusTreeLeafPage(NewBPlusTreeLeafPage::from(&bytes[4..])),
             _ => panic!("invalid log record type id"),
         }
     }
@@ -177,6 +183,10 @@ impl LogRecordBody {
                 buffer.extend_from_slice(&(6u32).to_be_bytes());
                 buffer.extend_from_slice(&body.serialize());
             }
+            LogRecordBody::NewBPlusTreeLeafPage(body) => {
+                buffer.extend_from_slice(&(7u32).to_be_bytes());
+                buffer.extend_from_slice(&body.serialize());
+            }
         }
         buffer
     }
@@ -189,6 +199,7 @@ impl LogRecordBody {
             LogRecordBody::DeleteFromTablePage(body) => 4 + body.size(),
             LogRecordBody::SetNextPageID(body) => 4 + body.size(),
             LogRecordBody::NewTablePage(body) => 4 + body.size(),
+            LogRecordBody::NewBPlusTreeLeafPage(body) => 4 + body.size(),
         }
     }
 }
@@ -284,6 +295,24 @@ impl NewTablePage {
         4
     }
 }
+impl From<&[u8]> for NewBPlusTreeLeafPage {
+    fn from(bytes: &[u8]) -> Self {
+        let mut buffer = [0u8; 4];
+        buffer.copy_from_slice(&bytes[0..4]);
+        let page_id = PageID(u32::from_be_bytes(buffer));
+        NewBPlusTreeLeafPage { page_id }
+    }
+}
+impl NewBPlusTreeLeafPage {
+    fn serialize(&self) -> Vec<u8> {
+        let mut buffer = Vec::new();
+        buffer.extend_from_slice(&self.page_id.0.to_be_bytes());
+        buffer
+    }
+    fn size(&self) -> usize {
+        4
+    }
+}
 
 #[cfg(test)]
 mod tests {
@@ -316,6 +345,10 @@ mod tests {
             TransactionID(2),
             LogRecordBody::NewTablePage(NewTablePage { page_id: PageID(2) }),
         )?;
+        log_manager.append(
+            TransactionID(2),
+            LogRecordBody::NewBPlusTreeLeafPage(NewBPlusTreeLeafPage { page_id: PageID(3) }),
+        )?;
         log_manager.append(
             TransactionID(2),
             LogRecordBody::SetNextPageID(SetNextPageID {
@@ -327,7 +360,7 @@ mod tests {
         log_manager.flush()?;
 
         let mut log_manager = LogManager::new(log_file_path.to_str().unwrap())?;
-        assert_eq!(log_manager.next_lsn, LSN(9));
+        assert_eq!(log_manager.next_lsn, LSN(10));
         let records = log_manager.read()?;
         assert_eq!(records[0].lsn, LSN(1));
         assert_eq!(records[0].txn_id, TransactionID(1));
@@ -362,14 +395,20 @@ mod tests {
         assert_eq!(records[6].txn_id, TransactionID(2));
         assert_eq!(
             records[6].body,
+            LogRecordBody::NewBPlusTreeLeafPage(NewBPlusTreeLeafPage { page_id: PageID(3) })
+        );
+        assert_eq!(records[7].lsn, LSN(8));
+        assert_eq!(records[7].txn_id, TransactionID(2));
+        assert_eq!(
+            records[7].body,
             LogRecordBody::SetNextPageID(SetNextPageID {
                 page_id: PageID(1),
                 next_page_id: PageID(2),
             })
         );
-        assert_eq!(records[7].lsn, LSN(8));
-        assert_eq!(records[7].txn_id, TransactionID(2));
-        assert_eq!(records[7].body, LogRecordBody::AbortTransaction);
+        assert_eq!(records[8].lsn, LSN(9));
+        assert_eq!(records[8].txn_id, TransactionID(2));
+        assert_eq!(records[8].body, LogRecordBody::AbortTransaction);
 
         Ok(())
     }
diff --git a/src/page.rs b/src/page.rs
index 1717516..b65f8c1 100644
--- a/src/page.rs
+++ b/src/page.rs
@@ -1,6 +1,9 @@
-use crate::common::PageID;
+use crate::common::{PageID, INVALID_PAGE_ID};
 
-use self::table_page::{TablePage, TABLE_PAGE_PAGE_TYPE};
+use self::{
+    b_plus_tree_leaf_page::{BPlusTreeLeafPage, B_PLUS_TREE_LEAF_PAGE_PAGE_TYPE},
+    table_page::{TablePage, TABLE_PAGE_PAGE_TYPE},
+};
 
 pub mod b_plus_tree_internal_page;
 pub mod b_plus_tree_leaf_page;
@@ -14,8 +17,10 @@ const PAGE_ID_SIZE: usize = 4;
 #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
 pub struct PageType(pub u32);
 
+#[derive(Debug)]
 pub enum Page {
     Table(TablePage),
+    BPlusTreeLeaf(BPlusTreeLeafPage),
 }
 impl Page {
     pub fn with_table_page<F, R>(&self, f: F) -> R
@@ -24,6 +29,7 @@ impl Page {
     {
         match self {
             Page::Table(table_page) => f(table_page),
+            _ => panic!("page type not supported"),
         }
     }
     pub fn with_table_page_mut<F, R>(&mut self, f: F) -> R
@@ -32,11 +38,34 @@ impl Page {
     {
         match self {
             Page::Table(table_page) => f(table_page),
+            _ => panic!("page type not supported"),
+        }
+    }
+    pub fn with_b_plus_tree_leaf_page<F, R>(&self, f: F) -> R
+    where
+        F: FnOnce(&BPlusTreeLeafPage) -> R,
+    {
+        match self {
+            Page::BPlusTreeLeaf(b_plus_tree_leaf_page) => f(b_plus_tree_leaf_page),
+            _ => panic!("page type not supported"),
+        }
+    }
+    pub fn with_b_plus_tree_leaf_page_mut<F, R>(&mut self, f: F) -> R
+    where
+        F: FnOnce(&mut BPlusTreeLeafPage) -> R,
+    {
+        match self {
+            Page::BPlusTreeLeaf(b_plus_tree_leaf_page) => f(b_plus_tree_leaf_page),
+            _ => panic!("page type not supported"),
         }
     }
     pub fn new(page_id: PageID, page_type: PageType) -> Self {
         match page_type {
             TABLE_PAGE_PAGE_TYPE => Page::Table(TablePage::new(page_id)),
+            // TODO:
+            B_PLUS_TREE_LEAF_PAGE_PAGE_TYPE => {
+                Page::BPlusTreeLeaf(BPlusTreeLeafPage::new(page_id, INVALID_PAGE_ID, None))
+            }
             _ => panic!("page type not supported"),
         }
     }
@@ -46,11 +75,13 @@ impl Page {
     pub fn data(&self) -> &[u8] {
         match self {
             Page::Table(table_page) => &table_page.data,
+            Page::BPlusTreeLeaf(b_plus_tree_leaf_page) => &b_plus_tree_leaf_page.data,
         }
     }
     pub fn page_id(&self) -> PageID {
         match self {
             Page::Table(table_page) => table_page.page_id(),
+            Page::BPlusTreeLeaf(b_plus_tree_leaf_page) => b_plus_tree_leaf_page.page_id(),
         }
     }
 }
diff --git a/src/page/b_plus_tree_leaf_page.rs b/src/page/b_plus_tree_leaf_page.rs
index 385e3b9..7a9a123 100644
--- a/src/page/b_plus_tree_leaf_page.rs
+++ b/src/page/b_plus_tree_leaf_page.rs
@@ -36,6 +36,7 @@ const LINE_POINTER_SIZE: usize = LINE_POINTER_OFFSET_SIZE + LINE_POINTER_SIZE_SI
 // RID
 const VALUE_SIZE: usize = 8;
 
+#[derive(Debug)]
 pub struct BPlusTreeLeafPage {
     pub data: Box<[u8]>,
 }
@@ -68,6 +69,11 @@ impl BPlusTreeLeafPage {
         BPlusTreeLeafPage { data: data.into() }
     }
 
+    pub fn page_id(&self) -> PageID {
+        let mut bytes = [0u8; 4];
+        bytes.copy_from_slice(&self.data[PAGE_ID_OFFSET..(PAGE_ID_OFFSET + PAGE_ID_SIZE)]);
+        PageID(u32::from_le_bytes(bytes))
+    }
     pub fn lsn(&self) -> LSN {
         let mut buf = [0u8; 8];
         buf.copy_from_slice(&self.data[LSN_OFFSET..(LSN_OFFSET + LSN_SIZE)]);
diff --git a/src/page/table_page.rs b/src/page/table_page.rs
index 9f1d714..04a8ed3 100644
--- a/src/page/table_page.rs
+++ b/src/page/table_page.rs
@@ -27,6 +27,7 @@ const LINE_POINTER_OFFSET_SIZE: usize = 4;
 const LINE_POINTER_SIZE_SIZE: usize = 4;
 const LINE_POINTER_SIZE: usize = LINE_POINTER_OFFSET_SIZE + LINE_POINTER_SIZE_SIZE;
 
+#[derive(Debug)]
 pub struct TablePage {
     pub data: Box<[u8]>,
 }
diff --git a/src/plan.rs b/src/plan.rs
index de932b3..0d4c425 100644
--- a/src/plan.rs
+++ b/src/plan.rs
@@ -14,6 +14,7 @@ use crate::{
 #[derive(Debug, PartialEq, Eq, Clone)]
 pub enum Plan {
     SeqScan(SeqScanPlan),
+    IndexScan(IndexScanPlan),
     Filter(FilterPlan),
     Project(ProjectPlan),
     NestedLoopJoin(NestedLoopJoinPlan),
@@ -29,6 +30,7 @@ impl Plan {
     pub fn schema(&self) -> &Schema {
         match self {
             Plan::SeqScan(plan) => &plan.schema,
+            Plan::IndexScan(plan) => &plan.schema,
             Plan::Filter(plan) => &plan.schema,
             Plan::Project(plan) => &plan.schema,
             Plan::NestedLoopJoin(plan) => &plan.schema,
@@ -41,6 +43,22 @@ impl Plan {
             Plan::Update(plan) => &plan.schema,
         }
     }
+    pub fn children(&self) -> Vec<Box<Plan>> {
+        match self {
+            Plan::SeqScan(_) => vec![],
+            Plan::IndexScan(_) => vec![],
+            Plan::Filter(plan) => vec![plan.child.clone()],
+            Plan::Project(plan) => vec![plan.child.clone()],
+            Plan::NestedLoopJoin(plan) => plan.children.clone(),
+            Plan::Aggregate(plan) => vec![plan.child.clone()],
+            Plan::Sort(plan) => vec![plan.child.clone()],
+            Plan::Limit(plan) => vec![plan.child.clone()],
+            Plan::EmptyRow(_) => vec![],
+            Plan::Insert(_) => vec![],
+            Plan::Delete(plan) => vec![plan.child.clone()],
+            Plan::Update(plan) => vec![plan.child.clone()],
+        }
+    }
 }
 #[derive(Debug, PartialEq, Eq, Clone)]
 pub struct SeqScanPlan {
@@ -48,6 +66,11 @@ pub struct SeqScanPlan {
     pub schema: Schema,
 }
 #[derive(Debug, PartialEq, Eq, Clone)]
+pub struct IndexScanPlan {
+    pub root_page_id: PageID,
+    pub schema: Schema,
+}
+#[derive(Debug, PartialEq, Eq, Clone)]
 pub struct FilterPlan {
     pub condition: BoundExpressionAST,
     pub schema: Schema,
@@ -429,7 +452,7 @@ mod tests {
         assert_eq!(
             plan,
             Plan::Delete(DeletePlan {
-                first_page_id: PageID(3),
+                first_page_id: PageID(5),
                 schema: Schema {
                     columns: vec![Column {
                         name: "__delete_count".to_owned(),
@@ -465,7 +488,7 @@ mod tests {
                         ],
                     },
                     child: Box::new(Plan::SeqScan(SeqScanPlan {
-                        first_page_id: PageID(3),
+                        first_page_id: PageID(5),
                         schema: Schema {
                             columns: vec![
                                 Column {
diff --git a/src/recovery.rs b/src/recovery.rs
index 23fd1b1..38e6017 100644
--- a/src/recovery.rs
+++ b/src/recovery.rs
@@ -4,9 +4,9 @@ use anyhow::Result;
 
 use crate::{
     buffer::BufferPoolManager,
-    common::TransactionID,
+    common::{TransactionID, INVALID_PAGE_ID},
     log::{LogRecord, LogRecordBody},
-    page::{table_page::TablePage, Page},
+    page::{b_plus_tree_leaf_page::BPlusTreeLeafPage, table_page::TablePage, Page},
 };
 
 pub struct RecoveryManager {
@@ -104,6 +104,17 @@ impl RecoveryManager {
                         .map_err(|_| anyhow::anyhow!("lock error"))?
                         .init_page_for_recovery(body.page_id, page)?;
                 }
+                LogRecordBody::NewBPlusTreeLeafPage(ref body) => {
+                    // TODO:
+                    let mut b_plus_tree_leaf_page =
+                        BPlusTreeLeafPage::new(body.page_id, INVALID_PAGE_ID, None);
+                    b_plus_tree_leaf_page.set_lsn(log_record.lsn);
+                    let page = Page::BPlusTreeLeaf(b_plus_tree_leaf_page);
+                    self.buffer_pool_manager
+                        .lock()
+                        .map_err(|_| anyhow::anyhow!("lock error"))?
+                        .init_page_for_recovery(body.page_id, page)?;
+                }
                 LogRecordBody::SetNextPageID(ref body) => {
                     let page = self
                         .buffer_pool_manager
diff --git a/src/server.rs b/src/server.rs
index 996acd7..69a0fad 100644
--- a/src/server.rs
+++ b/src/server.rs
@@ -180,10 +180,11 @@ impl Session {
                         format!("table {} created", ast.table_name)
                     }
                     StatementAST::CreateIndex(ast) => {
-                        format!(
-                            "create index is not implemented yet ({:?}, {:?}, {:?})",
-                            ast.index_name, ast.table_name, ast.column_names
-                        )
+                        self.instance
+                            .write()
+                            .map_err(|_| anyhow!("lock error"))?
+                            .create_index(&ast, txn_id)?;
+                        format!("index {} created", ast.index_name)
                     }
                     _ => {
                         let (rows, schema) = self