feat: Add 'show tables' functionality in SQL executor (#61)

* feat: Add 'show tables' functionality in SQL executor

This commit introduces 'show tables' feature, allowing users to display table names in the database. Necessary changes were made in the execution module, database, binder, and storage to facilitate this feature. This includes addition of `ShowTables` struct and implementing the `Executor` trait for it, modifications in `Binder` to create a logical plan for 'show tables' commands, and adding `show_tables` function in storage to return a list of existing table names. Additionally, a 'ShowTablesOperator' was added in operator module to support operation, and new test cases were added to validate the modifications.

* Remove hardcoded table insertions in table_codec.rs

* feat(executor): add column count to `show_tables` function

Functions `show_tables` in the Storage trait and its implementations in Kip and Memory now return a tuple containing table name and column count. The functions `encode_root_table` and `decode_root_table` in table_codec have been updated for the new format. These changes improve utility of `show_tables` by providing more detailed information.

src/binder/mod.rs

@@ -8,6 +8,7 @@ mod delete;
 mod drop_table;
 mod truncate;
 mod distinct;
+mod show;
 
 use std::collections::BTreeMap;
 use sqlparser::ast::{Ident, ObjectName, ObjectType, SetExpr, Statement};
@@ -121,7 +122,10 @@ impl<S: Storage> Binder<S> {
             Statement::Truncate { table_name, .. } => {
                 self.bind_truncate(table_name).await?
             }
-            _ => unimplemented!(),
+            Statement::ShowTables { .. } => {
+                self.bind_show_tables()?
+            }
+            _ => return Err(BindError::UnsupportedStmt(stmt.to_string())),
         };
         Ok(plan)
     }

src/binder/show.rs

@@ -0,0 +1,19 @@
+use crate::binder::{Binder, BindError};
+use crate::planner::LogicalPlan;
+use crate::planner::operator::Operator;
+use crate::planner::operator::show::ShowTablesOperator;
+use crate::storage::Storage;
+
+impl<S: Storage> Binder<S> {
+    pub(crate) fn bind_show_tables(
+        &mut self,
+    ) -> Result<LogicalPlan, BindError> {
+        let plan = LogicalPlan {
+            operator: Operator::Show(
+                ShowTablesOperator {}
+            ),
+            childrens: vec![],
+        };
+        Ok(plan)
+    }
+}

src/catalog/column.rs

@@ -27,6 +27,16 @@ impl ColumnCatalog {
         }
     }
 
+    pub(crate) fn new_dummy(column_name: String)-> ColumnCatalog {
+        ColumnCatalog {
+            id: 0,
+            name: column_name,
+            table_name: None,
+            nullable: false,
+            desc: ColumnDesc::new(LogicalType::Varchar(None), false),
+        }
+    }
+
     pub(crate) fn datatype(&self) -> &LogicalType {
         &self.desc.column_datatype
     }

src/db.rs

@@ -301,6 +301,10 @@ mod test {
         println!("drop t1:");
         let _ = kipsql.run("drop table t1").await?;
 
+        println!("show tables:");
+        let tuples_show_tables = kipsql.run("show tables").await?;
+        println!("{}", create_table(&tuples_show_tables));
+
         Ok(())
     }
 }

src/execution/executor/mod.rs

@@ -1,6 +1,7 @@
 pub(crate) mod dql;
 pub(crate)mod ddl;
 pub(crate)mod dml;
+pub(crate) mod show;
 
 use futures::stream::BoxStream;
 use futures::TryStreamExt;
@@ -20,6 +21,7 @@ use crate::execution::executor::dql::projection::Projection;
 use crate::execution::executor::dql::seq_scan::SeqScan;
 use crate::execution::executor::dql::sort::Sort;
 use crate::execution::executor::dql::values::Values;
+use crate::execution::executor::show::show_table::ShowTables;
 use crate::execution::ExecutorError;
 use crate::planner::LogicalPlan;
 use crate::planner::operator::Operator;
@@ -103,6 +105,9 @@ pub fn build<S: Storage>(plan: LogicalPlan, storage: &S) -> BoxedExecutor {
         Operator::Truncate(op) => {
             Truncate::from(op).execute(storage)
         }
+        Operator::Show(op) => {
+            ShowTables::from(op).execute(storage)
+        }
     }
 }
 

src/execution/executor/show/mod.rs

@@ -0,0 +1,4 @@
+pub(crate) mod show_table;
+
+
+

src/execution/executor/show/show_table.rs

@@ -0,0 +1,52 @@
+use futures_async_stream::try_stream;
+use crate::execution::executor::{BoxedExecutor, Executor};
+use crate::execution::ExecutorError;
+use crate::planner::operator::show::ShowTablesOperator;
+use crate::storage::Storage;
+use crate::types::tuple::Tuple;
+use crate::catalog::ColumnCatalog;
+use crate::catalog::ColumnRef;
+use std::sync::Arc;
+use crate::types::value::{DataValue, ValueRef};
+
+pub struct ShowTables {
+    op: ShowTablesOperator,
+}
+
+impl From<ShowTablesOperator> for ShowTables {
+    fn from(op: ShowTablesOperator) -> Self {
+        ShowTables {
+            op
+        }
+    }
+}
+
+impl<S: Storage> Executor<S> for ShowTables {
+    fn execute(self, storage: &S) -> BoxedExecutor {
+        self._execute(storage.clone())
+    }
+}
+
+impl ShowTables {
+    #[try_stream(boxed, ok = Tuple, error = ExecutorError)]
+    pub async fn _execute<S: Storage>(self, storage: S) {
+        if let Some(tables) = storage.show_tables().await {
+            for (table,column_count) in tables {
+                let columns: Vec<ColumnRef> = vec![
+                    Arc::new(ColumnCatalog::new_dummy("TABLES".to_string())),
+                    Arc::new(ColumnCatalog::new_dummy("COLUMN_COUNT".to_string())),
+                ];
+                let values: Vec<ValueRef> = vec![
+                    Arc::new(DataValue::Utf8(Some(table))),
+                    Arc::new(DataValue::UInt32(Some(column_count as u32))),
+                ];
+
+                yield Tuple {
+                    id: None,
+                    columns,
+                    values,
+                };
+            }
+        }
+    }
+}

src/planner/operator/mod.rs

@@ -12,6 +12,7 @@ pub mod update;
 pub mod delete;
 pub mod drop_table;
 pub mod truncate;
+pub mod show;
 
 use itertools::Itertools;
 use crate::catalog::ColumnRef;
@@ -21,6 +22,7 @@ use crate::planner::operator::delete::DeleteOperator;
 use crate::planner::operator::drop_table::DropTableOperator;
 use crate::planner::operator::insert::InsertOperator;
 use crate::planner::operator::join::JoinCondition;
+use crate::planner::operator::show::ShowTablesOperator;
 use crate::planner::operator::truncate::TruncateOperator;
 use crate::planner::operator::update::UpdateOperator;
 use crate::planner::operator::values::ValuesOperator;
@@ -50,6 +52,8 @@ pub enum Operator {
     CreateTable(CreateTableOperator),
     DropTable(DropTableOperator),
     Truncate(TruncateOperator),
+    // Show
+    Show(ShowTablesOperator),
 }
 
 impl Operator {

src/planner/operator/show.rs

@@ -0,0 +1,2 @@
+#[derive(Debug, PartialEq, Clone)]
+pub struct ShowTablesOperator {}

src/storage/kip.rs

@@ -49,6 +49,11 @@ impl Storage for KipStorage {
         {
             self.inner.set(key, value).await?;
         }
+
+        let (k, v)= TableCodec::encode_root_table(table_name.as_str(), table.columns.len())
+            .ok_or(StorageError::Serialization)?;
+        self.inner.set(k, v).await?;
+
         self.cache.put(table_name.to_string(), table);
 
         Ok(table_name)
@@ -72,6 +77,9 @@ impl Storage for KipStorage {
         for col_key in col_keys {
             tx.remove(&col_key)?
         }
+        let (k, _) = TableCodec::encode_root_table(name.as_str(),0)
+            .ok_or(StorageError::Serialization)?;
+        tx.remove(&k)?;
         tx.commit().await?;
 
         let _ = self.cache.remove(name);
@@ -139,6 +147,24 @@ impl Storage for KipStorage {
 
         option
     }
+
+    async fn show_tables(&self) -> Option<Vec<(String,usize)>> {
+        let mut tables = vec![];
+        let (min, max) = TableCodec::root_table_bound();
+
+        let tx = self.inner.new_transaction().await;
+        let mut iter = tx.iter(Bound::Included(&min), Bound::Included(&max)).ok()?;
+
+        while let Some((key, value_option))  = iter.try_next().ok().flatten() {
+            if let Some(value) = value_option {
+                if let Some((table_name, column_count)) = TableCodec::decode_root_table(&key, &value) {
+                    tables.push((table_name,column_count));
+                }
+            }
+        }
+
+        Some(tables)
+    }
 }
 
 pub struct KipTable {

src/storage/memory.rs

@@ -114,6 +114,10 @@ impl Storage for MemStorage {
                 .get_table(name)
         }
     }
+
+    async fn show_tables(&self) -> Option<Vec<(String, usize)>> {
+        todo!()
+    }
 }
 
 unsafe impl Send for MemTable {

src/storage/mod.rs

@@ -25,6 +25,8 @@ pub trait Storage: Sync + Send + Clone + 'static {
 
     async fn table(&self, name: &String) -> Option<Self::TableType>;
     async fn table_catalog(&self, name: &String) -> Option<&TableCatalog>;
+
+    async fn show_tables(&self) -> Option<Vec<(String,usize)>>;
 }
 
 /// Optional bounds of the reader, of the form (offset, limit).
@@ -71,6 +73,10 @@ pub enum StorageError {
 
     #[error("The same primary key data already exists")]
     DuplicatePrimaryKey,
+
+    #[error("Serialization error")]
+    Serialization,
+
 }
 
 impl From<KernelError> for StorageError {

src/storage/table_codec.rs

@@ -98,6 +98,46 @@ impl TableCodec {
                     })
             })
     }
+
+
+    /// Key: RootCatalog_0_TableName
+    /// Value: ColumnCount
+    pub fn encode_root_table(table_name: &str,column_count:usize) -> Option<(Bytes, Bytes)> {
+        let key = format!(
+            "RootCatalog_{}_{}",
+            BOUND_MIN_TAG,
+            table_name,
+        );
+
+        bincode::serialize(&column_count).ok()
+            .map(|bytes| {
+                (Bytes::from(key.into_bytes()), Bytes::from(bytes))
+            })
+    }
+
+    // TODO: value is reserved for saving meta-information
+    pub fn decode_root_table(key: &[u8], bytes: &[u8]) -> Option<(String,usize)> {
+        String::from_utf8(key.to_owned()).ok()?
+            .split("_")
+            .nth(2)
+            .and_then(|table_name| {
+                bincode::deserialize::<usize>(bytes).ok()
+                    .and_then(|name| {
+                        Some((table_name.to_string(), name))
+                    })
+            })
+    }
+
+    pub fn root_table_bound() -> (Vec<u8>, Vec<u8>) {
+        let op = |bound_id| {
+            format!(
+                "RootCatalog_{}",
+                bound_id,
+            )
+        };
+
+        (op(BOUND_MIN_TAG).into_bytes(), op(BOUND_MAX_TAG).into_bytes())
+    }
 }
 
 #[cfg(test)]
@@ -153,6 +193,25 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn test_root_catalog() {
+        let (table_catalog, _) = build_table_codec();
+        let (key, bytes) = TableCodec::encode_root_table(&table_catalog.name,2).unwrap();
+
+        assert_eq!(
+            String::from_utf8(key.to_vec()).ok().unwrap(),
+            format!(
+                "RootCatalog_0_{}",
+                table_catalog.name,
+            )
+        );
+
+        let (table_name, column_count) = TableCodec::decode_root_table(&key, &bytes).unwrap();
+
+        assert_eq!(table_name, table_catalog.name.as_str());
+        assert_eq!(column_count, 2);
+    }
+
     #[test]
     fn test_table_codec_column() {
         let (table_catalog, _) = build_table_codec();
@@ -240,4 +299,27 @@ mod tests {
         assert_eq!(String::from_utf8(vec[1].clone()).unwrap(), "T1_Data_0_0000000000000000001");
         assert_eq!(String::from_utf8(vec[2].clone()).unwrap(), "T1_Data_0_0000000000000000002");
     }
+
+    #[test]
+    fn test_root_codec_name_bound(){
+        let mut set = BTreeSet::new();
+        let op = |str: &str| {
+            str.to_string().into_bytes()
+        };
+
+        set.insert(op("RootCatalog_0_T0"));
+        set.insert(op("RootCatalog_0_T1"));
+        set.insert(op("RootCatalog_0_T2"));
+
+        let (min, max) = TableCodec::root_table_bound();
+
+        let vec = set
+            .range::<Vec<u8>, (Bound<&Vec<u8>>, Bound<&Vec<u8>>)>((Bound::Included(&min), Bound::Included(&max)))
+            .collect_vec();
+
+        assert_eq!(String::from_utf8(vec[0].clone()).unwrap(), "RootCatalog_0_T0");
+        assert_eq!(String::from_utf8(vec[1].clone()).unwrap(), "RootCatalog_0_T1");
+        assert_eq!(String::from_utf8(vec[2].clone()).unwrap(), "RootCatalog_0_T2");
+
+    }
 }