feat: implement DESCRIBE TABLE (#558)

Also need to support describe table in other catalog/schema

Cargo.lock

@@ -4376,6 +4376,7 @@ dependencies = [
  "metrics",
  "num",
  "num-traits",
+ "once_cell",
  "paste",
  "rand 0.8.5",
  "serde",

src/datanode/src/instance/sql.rs

@@ -115,6 +115,11 @@ impl Instance {
             Statement::ShowTables(stmt) => {
                 self.sql_handler.execute(SqlRequest::ShowTables(stmt)).await
             }
+            Statement::DescribeTable(stmt) => {
+                self.sql_handler
+                    .execute(SqlRequest::DescribeTable(stmt))
+                    .await
+            }
             Statement::ShowCreateTable(_stmt) => {
                 unimplemented!("SHOW CREATE TABLE is unimplemented yet");
             }

src/datanode/src/sql.rs

@@ -16,8 +16,9 @@
 
 use catalog::CatalogManagerRef;
 use common_query::Output;
-use query::sql::{show_databases, show_tables};
+use query::sql::{describe_table, show_databases, show_tables};
 use snafu::{OptionExt, ResultExt};
+use sql::statements::describe::DescribeTable;
 use sql::statements::show::{ShowDatabases, ShowTables};
 use table::engine::{EngineContext, TableEngineRef, TableReference};
 use table::requests::*;
@@ -37,6 +38,7 @@ pub enum SqlRequest {
     Alter(AlterTableRequest),
     ShowDatabases(ShowDatabases),
     ShowTables(ShowTables),
+    DescribeTable(DescribeTable),
 }
 
 // Handler to execute SQL except query
@@ -65,6 +67,9 @@ impl SqlHandler {
             SqlRequest::ShowTables(stmt) => {
                 show_tables(stmt, self.catalog_manager.clone()).context(error::ExecuteSqlSnafu)
             }
+            SqlRequest::DescribeTable(stmt) => {
+                describe_table(stmt, self.catalog_manager.clone()).context(error::ExecuteSqlSnafu)
+            }
         }
     }
 

src/datatypes/src/schema/constraint.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::fmt::{Display, Formatter};
 use std::sync::Arc;
 
 use common_time::util;
@@ -53,6 +54,15 @@ impl TryFrom<ColumnDefaultConstraint> for Vec<u8> {
     }
 }
 
+impl Display for ColumnDefaultConstraint {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        match self {
+            ColumnDefaultConstraint::Function(expr) => write!(f, "{}", expr),
+            ColumnDefaultConstraint::Value(v) => write!(f, "{}", v),
+        }
+    }
+}
+
 impl ColumnDefaultConstraint {
     /// Returns a default null constraint.
     pub fn null_value() -> ColumnDefaultConstraint {

src/datatypes/src/value.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use std::cmp::Ordering;
+use std::fmt::{Display, Formatter};
 
 use common_base::bytes::{Bytes, StringBytes};
 use common_time::date::Date;
@@ -62,6 +63,47 @@ pub enum Value {
     List(ListValue),
 }
 
+impl Display for Value {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Value::Null => write!(f, "{}", self.data_type().name()),
+            Value::Boolean(v) => write!(f, "{}", v),
+            Value::UInt8(v) => write!(f, "{}", v),
+            Value::UInt16(v) => write!(f, "{}", v),
+            Value::UInt32(v) => write!(f, "{}", v),
+            Value::UInt64(v) => write!(f, "{}", v),
+            Value::Int8(v) => write!(f, "{}", v),
+            Value::Int16(v) => write!(f, "{}", v),
+            Value::Int32(v) => write!(f, "{}", v),
+            Value::Int64(v) => write!(f, "{}", v),
+            Value::Float32(v) => write!(f, "{}", v),
+            Value::Float64(v) => write!(f, "{}", v),
+            Value::String(v) => write!(f, "{}", v.as_utf8()),
+            Value::Binary(v) => {
+                let hex = v
+                    .iter()
+                    .map(|b| format!("{:02x}", b))
+                    .collect::<Vec<String>>()
+                    .join("");
+                write!(f, "{}", hex)
+            }
+            Value::Date(v) => write!(f, "{}", v),
+            Value::DateTime(v) => write!(f, "{}", v),
+            Value::Timestamp(v) => write!(f, "{}", v.to_iso8601_string()),
+            Value::List(v) => {
+                let default = Box::new(vec![]);
+                let items = v.items().as_ref().unwrap_or(&default);
+                let items = items
+                    .iter()
+                    .map(|i| i.to_string())
+                    .collect::<Vec<String>>()
+                    .join(", ");
+                write!(f, "{}[{}]", v.datatype.name(), items)
+            }
+        }
+    }
+}
+
 impl Value {
     /// Returns data type of the value.
     ///
@@ -624,6 +666,7 @@ impl<'a> PartialOrd for ListValueRef<'a> {
 #[cfg(test)]
 mod tests {
     use arrow::datatypes::DataType as ArrowDataType;
+    use num_traits::Float;
 
     use super::*;
 
@@ -1033,7 +1076,7 @@ mod tests {
         );
         assert_eq!(
             serde_json::Value::Number(5000i32.into()),
-            to_json(Value::Date(common_time::date::Date::new(5000)))
+            to_json(Value::Date(Date::new(5000)))
         );
         assert_eq!(
             serde_json::Value::Number(5000i64.into()),
@@ -1217,4 +1260,51 @@ mod tests {
         );
         assert_eq!(ValueRef::String(hello), ValueRef::from(hello));
     }
+
+    #[test]
+    fn test_display() {
+        assert_eq!(Value::Null.to_string(), "Null");
+        assert_eq!(Value::UInt8(8).to_string(), "8");
+        assert_eq!(Value::UInt16(16).to_string(), "16");
+        assert_eq!(Value::UInt32(32).to_string(), "32");
+        assert_eq!(Value::UInt64(64).to_string(), "64");
+        assert_eq!(Value::Int8(-8).to_string(), "-8");
+        assert_eq!(Value::Int16(-16).to_string(), "-16");
+        assert_eq!(Value::Int32(-32).to_string(), "-32");
+        assert_eq!(Value::Int64(-64).to_string(), "-64");
+        assert_eq!(Value::Float32((-32.123).into()).to_string(), "-32.123");
+        assert_eq!(Value::Float64((-64.123).into()).to_string(), "-64.123");
+        assert_eq!(Value::Float64(OrderedF64::infinity()).to_string(), "inf");
+        assert_eq!(Value::Float64(OrderedF64::nan()).to_string(), "NaN");
+        assert_eq!(Value::String(StringBytes::from("123")).to_string(), "123");
+        assert_eq!(
+            Value::Binary(Bytes::from(vec![1, 2, 3])).to_string(),
+            "010203"
+        );
+        assert_eq!(Value::Date(Date::new(0)).to_string(), "1970-01-01");
+        assert_eq!(
+            Value::DateTime(DateTime::new(0)).to_string(),
+            "1970-01-01 00:00:00"
+        );
+        assert_eq!(
+            Value::Timestamp(Timestamp::new(1000, TimeUnit::Millisecond)).to_string(),
+            "1970-01-01 00:00:01+0000"
+        );
+        assert_eq!(
+            Value::List(ListValue::new(
+                Some(Box::new(vec![Value::Int8(1), Value::Int8(2)])),
+                ConcreteDataType::int8_datatype(),
+            ))
+            .to_string(),
+            "Int8[1, 2]"
+        );
+        assert_eq!(
+            Value::List(ListValue::new(
+                Some(Box::new(vec![])),
+                ConcreteDataType::timestamp_datatype(TimeUnit::Millisecond),
+            ))
+            .to_string(),
+            "Timestamp[]"
+        );
+    }
 }

src/frontend/src/instance.rs

@@ -608,7 +608,9 @@ impl SqlQueryHandler for Instance {
                     .context(server_error::ExecuteQuerySnafu { query })
             }
 
-            Statement::ShowDatabases(_) | Statement::ShowTables(_) => self
+            Statement::ShowDatabases(_)
+            | Statement::ShowTables(_)
+            | Statement::DescribeTable(_) => self
                 .handle_select(Select::Sql(query.to_string()), stmt)
                 .await
                 .map_err(BoxedError::new)

src/frontend/src/instance/distributed.rs

@@ -30,7 +30,7 @@ use meta_client::rpc::{
     CreateRequest as MetaCreateRequest, Partition as MetaPartition, PutRequest, RouteResponse,
     TableName, TableRoute,
 };
-use query::sql::{show_databases, show_tables};
+use query::sql::{describe_table, show_databases, show_tables};
 use query::{QueryEngineFactory, QueryEngineRef};
 use snafu::{ensure, OptionExt, ResultExt};
 use sql::statements::create::Partitions;
@@ -141,6 +141,8 @@ impl DistInstance {
                 .context(error::ExecuteSqlSnafu { sql }),
             Statement::ShowTables(stmt) => show_tables(stmt, self.catalog_manager.clone())
                 .context(error::ExecuteSqlSnafu { sql }),
+            Statement::DescribeTable(stmt) => describe_table(stmt, self.catalog_manager.clone())
+                .context(error::ExecuteSqlSnafu { sql }),
             _ => unreachable!(),
         }
     }

src/query/Cargo.toml

@@ -24,6 +24,7 @@ datatypes = { path = "../datatypes" }
 futures = "0.3"
 futures-util = "0.3"
 metrics = "0.20"
+once_cell = "1.10"
 serde = "1.0"
 serde_json = "1.0"
 snafu = { version = "0.7", features = ["backtraces"] }

src/query/src/datafusion/planner.rs

@@ -66,6 +66,7 @@ where
             Statement::ShowTables(_)
             | Statement::ShowDatabases(_)
             | Statement::ShowCreateTable(_)
+            | Statement::DescribeTable(_)
             | Statement::CreateTable(_)
             | Statement::CreateDatabase(_)
             | Statement::Alter(_)

src/query/src/error.rs

@@ -44,6 +44,9 @@ pub enum InnerError {
         backtrace: Backtrace,
     },
 
+    #[snafu(display("Table not found: {}", table))]
+    TableNotFound { table: String, backtrace: Backtrace },
+
     #[snafu(display("Failed to do vector computation, source: {}", source))]
     VectorComputation {
         #[snafu(backtrace)]
@@ -62,9 +65,10 @@ impl ErrorExt for InnerError {
         use InnerError::*;
 
         match self {
-            UnsupportedExpr { .. } | CatalogNotFound { .. } | SchemaNotFound { .. } => {
-                StatusCode::InvalidArguments
-            }
+            UnsupportedExpr { .. }
+            | CatalogNotFound { .. }
+            | SchemaNotFound { .. }
+            | TableNotFound { .. } => StatusCode::InvalidArguments,
             Catalog { source } => source.status_code(),
             VectorComputation { source } => source.status_code(),
             CreateRecordBatch { source } => source.status_code(),

src/query/src/sql.rs

@@ -21,13 +21,57 @@ use common_recordbatch::RecordBatches;
 use datatypes::prelude::*;
 use datatypes::schema::{ColumnSchema, Schema};
 use datatypes::vectors::{Helper, StringVector};
+use once_cell::sync::Lazy;
 use snafu::{ensure, OptionExt, ResultExt};
+use sql::statements::describe::DescribeTable;
 use sql::statements::show::{ShowDatabases, ShowKind, ShowTables};
 
 use crate::error::{self, Result};
 
 const SCHEMAS_COLUMN: &str = "Schemas";
 const TABLES_COLUMN: &str = "Tables";
+const COLUMN_NAME_COLUMN: &str = "Field";
+const COLUMN_TYPE_COLUMN: &str = "Type";
+const COLUMN_NULLABLE_COLUMN: &str = "Null";
+const COLUMN_DEFAULT_COLUMN: &str = "Default";
+const COLUMN_SEMANTIC_TYPE_COLUMN: &str = "Semantic Type";
+
+const SEMANTIC_TYPE_PRIMARY_KEY: &str = "PRIMARY KEY";
+const SEMANTIC_TYPE_VALUE: &str = "VALUE";
+const SEMANTIC_TYPE_TIME_INDEX: &str = "TIME INDEX";
+
+const NULLABLE_YES: &str = "YES";
+const NULLABLE_NO: &str = "NO";
+
+static DESCRIBE_TABLE_OUTPUT_SCHEMA: Lazy<Arc<Schema>> = Lazy::new(|| {
+    Arc::new(Schema::new(vec![
+        ColumnSchema::new(
+            COLUMN_NAME_COLUMN,
+            ConcreteDataType::string_datatype(),
+            false,
+        ),
+        ColumnSchema::new(
+            COLUMN_TYPE_COLUMN,
+            ConcreteDataType::string_datatype(),
+            false,
+        ),
+        ColumnSchema::new(
+            COLUMN_NULLABLE_COLUMN,
+            ConcreteDataType::string_datatype(),
+            false,
+        ),
+        ColumnSchema::new(
+            COLUMN_DEFAULT_COLUMN,
+            ConcreteDataType::string_datatype(),
+            false,
+        ),
+        ColumnSchema::new(
+            COLUMN_SEMANTIC_TYPE_COLUMN,
+            ConcreteDataType::string_datatype(),
+            false,
+        ),
+    ]))
+});
 
 pub fn show_databases(stmt: ShowDatabases, catalog_manager: CatalogManagerRef) -> Result<Output> {
     // TODO(LFC): supports WHERE
@@ -93,3 +137,302 @@ pub fn show_tables(stmt: ShowTables, catalog_manager: CatalogManagerRef) -> Resu
         .context(error::CreateRecordBatchSnafu)?;
     Ok(Output::RecordBatches(records))
 }
+
+pub fn describe_table(stmt: DescribeTable, catalog_manager: CatalogManagerRef) -> Result<Output> {
+    let catalog = stmt.catalog_name.as_str();
+    let schema = stmt.schema_name.as_str();
+    catalog_manager
+        .catalog(catalog)
+        .context(error::CatalogSnafu)?
+        .context(error::CatalogNotFoundSnafu { catalog })?;
+    let schema = catalog_manager
+        .schema(catalog, schema)
+        .context(error::CatalogSnafu)?
+        .context(error::SchemaNotFoundSnafu { schema })?;
+    let table = schema
+        .table(&stmt.table_name)
+        .context(error::CatalogSnafu)?
+        .context(error::TableNotFoundSnafu {
+            table: &stmt.table_name,
+        })?;
+
+    let table_info = table.table_info();
+    let columns_schemas = table_info.meta.schema.column_schemas();
+    let columns = vec![
+        describe_column_names(columns_schemas),
+        describe_column_types(columns_schemas),
+        describe_column_nullables(columns_schemas),
+        describe_column_defaults(columns_schemas),
+        describe_column_semantic_types(columns_schemas, &table_info.meta.primary_key_indices),
+    ];
+    let records = RecordBatches::try_from_columns(DESCRIBE_TABLE_OUTPUT_SCHEMA.clone(), columns)
+        .context(error::CreateRecordBatchSnafu)?;
+    Ok(Output::RecordBatches(records))
+}
+
+fn describe_column_names(columns_schemas: &[ColumnSchema]) -> VectorRef {
+    Arc::new(StringVector::from_iterator(
+        columns_schemas.iter().map(|cs| cs.name.as_str()),
+    ))
+}
+
+fn describe_column_types(columns_schemas: &[ColumnSchema]) -> VectorRef {
+    Arc::new(StringVector::from_iterator(
+        columns_schemas.iter().map(|cs| cs.data_type.name()),
+    ))
+}
+
+fn describe_column_nullables(columns_schemas: &[ColumnSchema]) -> VectorRef {
+    Arc::new(StringVector::from_iterator(columns_schemas.iter().map(
+        |cs| {
+            if cs.is_nullable() {
+                NULLABLE_YES
+            } else {
+                NULLABLE_NO
+            }
+        },
+    )))
+}
+
+fn describe_column_defaults(columns_schemas: &[ColumnSchema]) -> VectorRef {
+    Arc::new(StringVector::from(
+        columns_schemas
+            .iter()
+            .map(|cs| {
+                cs.default_constraint()
+                    .map_or(String::from(""), |dc| dc.to_string())
+            })
+            .collect::<Vec<String>>(),
+    ))
+}
+
+fn describe_column_semantic_types(
+    columns_schemas: &[ColumnSchema],
+    primary_key_indices: &[usize],
+) -> VectorRef {
+    Arc::new(StringVector::from(
+        columns_schemas
+            .iter()
+            .enumerate()
+            .map(|(i, cs)| {
+                if primary_key_indices.contains(&i) {
+                    String::from(SEMANTIC_TYPE_PRIMARY_KEY)
+                } else if cs.is_time_index() {
+                    String::from(SEMANTIC_TYPE_TIME_INDEX)
+                } else {
+                    String::from(SEMANTIC_TYPE_VALUE)
+                }
+            })
+            .collect::<Vec<String>>(),
+    ))
+}
+
+#[cfg(test)]
+mod test {
+    use std::sync::Arc;
+
+    use arrow::array::PrimitiveArray;
+    use catalog::local::{MemoryCatalogManager, MemoryCatalogProvider, MemorySchemaProvider};
+    use catalog::{CatalogList, CatalogManagerRef, CatalogProvider, SchemaProvider};
+    use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
+    use common_error::ext::ErrorExt;
+    use common_query::Output;
+    use common_recordbatch::{RecordBatch, RecordBatches};
+    use common_time::timestamp::TimeUnit;
+    use datatypes::prelude::ConcreteDataType;
+    use datatypes::schema::{ColumnDefaultConstraint, ColumnSchema, Schema, SchemaRef};
+    use datatypes::vectors::{StringVector, TimestampVector, UInt32Vector, VectorRef};
+    use snafu::ResultExt;
+    use sql::statements::describe::DescribeTable;
+    use table::test_util::MemTable;
+
+    use crate::error;
+    use crate::error::Result;
+    use crate::sql::{
+        describe_table, DESCRIBE_TABLE_OUTPUT_SCHEMA, NULLABLE_NO, NULLABLE_YES,
+        SEMANTIC_TYPE_TIME_INDEX, SEMANTIC_TYPE_VALUE,
+    };
+
+    #[test]
+    fn test_describe_table_catalog_not_found() -> Result<()> {
+        let catalog_name = DEFAULT_CATALOG_NAME.to_string();
+        let schema_name = DEFAULT_SCHEMA_NAME.to_string();
+        let table_name = "test_table";
+        let table_schema = SchemaRef::new(Schema::new(vec![ColumnSchema::new(
+            "test_col",
+            ConcreteDataType::uint32_datatype(),
+            false,
+        )]));
+        let data = vec![Arc::new(UInt32Vector::from_vec(vec![0])) as _];
+        let catalog_manager =
+            prepare_describe_table(&catalog_name, &schema_name, table_name, table_schema, data);
+
+        let stmt = DescribeTable::new("unknown".to_string(), schema_name, table_name.to_string());
+
+        let err = describe_table(stmt, catalog_manager).err().unwrap();
+        let err = err.as_any().downcast_ref::<error::InnerError>().unwrap();
+
+        if let error::InnerError::CatalogNotFound { catalog, .. } = err {
+            assert_eq!(catalog, "unknown");
+        } else {
+            panic!("describe table returned incorrect error");
+        }
+
+        Ok(())
+    }
+
+    #[test]
+    fn test_describe_table_schema_not_found() -> Result<()> {
+        let catalog_name = DEFAULT_CATALOG_NAME.to_string();
+        let schema_name = DEFAULT_SCHEMA_NAME.to_string();
+        let table_name = "test_table";
+        let table_schema = SchemaRef::new(Schema::new(vec![ColumnSchema::new(
+            "test_col",
+            ConcreteDataType::uint32_datatype(),
+            false,
+        )]));
+        let data = vec![Arc::new(UInt32Vector::from_vec(vec![0])) as _];
+        let catalog_manager =
+            prepare_describe_table(&catalog_name, &schema_name, table_name, table_schema, data);
+
+        let stmt = DescribeTable::new(catalog_name, "unknown".to_string(), table_name.to_string());
+
+        let err = describe_table(stmt, catalog_manager).err().unwrap();
+        let err = err.as_any().downcast_ref::<error::InnerError>().unwrap();
+
+        if let error::InnerError::SchemaNotFound { schema, .. } = err {
+            assert_eq!(schema, "unknown");
+        } else {
+            panic!("describe table returned incorrect error");
+        }
+
+        Ok(())
+    }
+
+    #[test]
+    fn test_describe_table_table_not_found() -> Result<()> {
+        let catalog_name = DEFAULT_CATALOG_NAME.to_string();
+        let schema_name = DEFAULT_SCHEMA_NAME.to_string();
+        let table_name = "test_table";
+        let table_schema = SchemaRef::new(Schema::new(vec![ColumnSchema::new(
+            "test_col",
+            ConcreteDataType::uint32_datatype(),
+            false,
+        )]));
+        let data = vec![Arc::new(UInt32Vector::from_vec(vec![0])) as _];
+        let catalog_manager =
+            prepare_describe_table(&catalog_name, &schema_name, table_name, table_schema, data);
+
+        let stmt = DescribeTable::new(catalog_name, schema_name, "unknown".to_string());
+
+        let err = describe_table(stmt, catalog_manager).err().unwrap();
+        let err = err.as_any().downcast_ref::<error::InnerError>().unwrap();
+
+        if let error::InnerError::TableNotFound { table, .. } = err {
+            assert_eq!(table, "unknown");
+        } else {
+            panic!("describe table returned incorrect error");
+        }
+
+        Ok(())
+    }
+
+    #[test]
+    fn test_describe_table_multiple_columns() -> Result<()> {
+        let catalog_name = DEFAULT_CATALOG_NAME;
+        let schema_name = DEFAULT_SCHEMA_NAME;
+        let table_name = "test_table";
+        let schema = vec![
+            ColumnSchema::new("t1", ConcreteDataType::uint32_datatype(), true),
+            ColumnSchema::new(
+                "t2",
+                ConcreteDataType::timestamp_datatype(TimeUnit::Millisecond),
+                false,
+            )
+            .with_default_constraint(Some(ColumnDefaultConstraint::Function(String::from(
+                "current_timestamp()",
+            ))))
+            .unwrap()
+            .with_time_index(true),
+        ];
+        let data = vec![
+            Arc::new(UInt32Vector::from_vec(vec![0])) as _,
+            Arc::new(TimestampVector::new(PrimitiveArray::from_vec(vec![0]))) as _,
+        ];
+        let expected_columns = vec![
+            Arc::new(StringVector::from(vec!["t1", "t2"])) as _,
+            Arc::new(StringVector::from(vec!["UInt32", "Timestamp"])) as _,
+            Arc::new(StringVector::from(vec![NULLABLE_YES, NULLABLE_NO])) as _,
+            Arc::new(StringVector::from(vec!["", "current_timestamp()"])) as _,
+            Arc::new(StringVector::from(vec![
+                SEMANTIC_TYPE_VALUE,
+                SEMANTIC_TYPE_TIME_INDEX,
+            ])) as _,
+        ];
+
+        describe_table_test_by_schema(
+            catalog_name,
+            schema_name,
+            table_name,
+            schema,
+            data,
+            expected_columns,
+        )
+    }
+
+    fn describe_table_test_by_schema(
+        catalog_name: &str,
+        schema_name: &str,
+        table_name: &str,
+        schema: Vec<ColumnSchema>,
+        data: Vec<VectorRef>,
+        expected_columns: Vec<VectorRef>,
+    ) -> Result<()> {
+        let table_schema = SchemaRef::new(Schema::new(schema));
+        let catalog_manager =
+            prepare_describe_table(catalog_name, schema_name, table_name, table_schema, data);
+
+        let expected =
+            RecordBatches::try_from_columns(DESCRIBE_TABLE_OUTPUT_SCHEMA.clone(), expected_columns)
+                .context(error::CreateRecordBatchSnafu)?;
+
+        let stmt = DescribeTable::new(
+            catalog_name.to_string(),
+            schema_name.to_string(),
+            table_name.to_string(),
+        );
+        if let Output::RecordBatches(res) = describe_table(stmt, catalog_manager)? {
+            assert_eq!(res.take(), expected.take());
+        } else {
+            panic!("describe table must return record batch");
+        }
+
+        Ok(())
+    }
+
+    fn prepare_describe_table(
+        catalog_name: &str,
+        schema_name: &str,
+        table_name: &str,
+        table_schema: SchemaRef,
+        data: Vec<VectorRef>,
+    ) -> CatalogManagerRef {
+        let record_batch = RecordBatch::new(table_schema, data).unwrap();
+        let table = Arc::new(MemTable::new(table_name, record_batch));
+
+        let schema_provider = Arc::new(MemorySchemaProvider::new());
+        let catalog_provider = Arc::new(MemoryCatalogProvider::new());
+        let catalog_manager = Arc::new(MemoryCatalogManager::default());
+        schema_provider
+            .register_table(table_name.to_string(), table)
+            .unwrap();
+        catalog_provider
+            .register_schema(schema_name.to_string(), schema_provider)
+            .unwrap();
+        catalog_manager
+            .register_catalog(catalog_name.to_string(), catalog_provider)
+            .unwrap();
+
+        catalog_manager
+    }
+}

src/sql/src/parser.rs

@@ -21,8 +21,10 @@ use sqlparser::tokenizer::{Token, Tokenizer};
 use crate::error::{
     self, InvalidDatabaseNameSnafu, InvalidTableNameSnafu, Result, SyntaxSnafu, TokenizerSnafu,
 };
+use crate::statements::describe::DescribeTable;
 use crate::statements::show::{ShowCreateTable, ShowDatabases, ShowKind, ShowTables};
 use crate::statements::statement::Statement;
+use crate::statements::table_idents_to_full_name;
 
 /// GrepTime SQL parser context, a simple wrapper for Datafusion SQL parser.
 pub struct ParserContext<'a> {
@@ -85,6 +87,11 @@ impl<'a> ParserContext<'a> {
                         self.parse_show()
                     }
 
+                    Keyword::DESCRIBE | Keyword::DESC => {
+                        self.parser.next_token();
+                        self.parse_describe()
+                    }
+
                     Keyword::INSERT => self.parse_insert(),
 
                     Keyword::SELECT | Keyword::WITH | Keyword::VALUES => self.parse_query(),
@@ -217,6 +224,39 @@ impl<'a> ParserContext<'a> {
         Ok(Statement::ShowTables(ShowTables { kind, database }))
     }
 
+    /// Parses DESCRIBE statements
+    fn parse_describe(&mut self) -> Result<Statement> {
+        if self.matches_keyword(Keyword::TABLE) {
+            self.parser.next_token();
+            self.parse_describe_table()
+        } else {
+            self.unsupported(self.peek_token_as_string())
+        }
+    }
+
+    fn parse_describe_table(&mut self) -> Result<Statement> {
+        let table_idents =
+            self.parser
+                .parse_object_name()
+                .with_context(|_| error::UnexpectedSnafu {
+                    sql: self.sql,
+                    expected: "a table name",
+                    actual: self.peek_token_as_string(),
+                })?;
+        ensure!(
+            !table_idents.0.is_empty(),
+            InvalidTableNameSnafu {
+                name: table_idents.to_string(),
+            }
+        );
+        let (catalog_name, schema_name, table_name) = table_idents_to_full_name(&table_idents)?;
+        Ok(Statement::DescribeTable(DescribeTable {
+            catalog_name,
+            schema_name,
+            table_name,
+        }))
+    }
+
     fn parse_explain(&mut self) -> Result<Statement> {
         todo!()
     }

src/sql/src/statements.rs

@@ -14,6 +14,7 @@
 
 pub mod alter;
 pub mod create;
+pub mod describe;
 pub mod insert;
 pub mod query;
 pub mod show;

src/sql/src/statements/describe.rs

@@ -0,0 +1,102 @@
+// Copyright 2022 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+/// SQL structure for `DESCRIBE TABLE`.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct DescribeTable {
+    pub catalog_name: String,
+    pub schema_name: String,
+    pub table_name: String,
+}
+
+impl DescribeTable {
+    /// Creates a statement for `DESCRIBE TABLE`
+    pub fn new(catalog_name: String, schema_name: String, table_name: String) -> Self {
+        DescribeTable {
+            catalog_name,
+            schema_name,
+            table_name,
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::assert_matches::assert_matches;
+
+    use sqlparser::dialect::GenericDialect;
+
+    use crate::parser::ParserContext;
+    use crate::statements::statement::Statement;
+
+    #[test]
+    pub fn test_describe_table() {
+        let sql = "DESCRIBE TABLE test";
+        let stmts: Vec<Statement> =
+            ParserContext::create_with_dialect(sql, &GenericDialect {}).unwrap();
+        assert_eq!(1, stmts.len());
+        assert_matches!(&stmts[0], Statement::DescribeTable { .. });
+        match &stmts[0] {
+            Statement::DescribeTable(show) => {
+                assert_eq!(show.table_name.as_str(), "test");
+            }
+            _ => {
+                unreachable!();
+            }
+        }
+    }
+
+    #[test]
+    pub fn test_describe_schema_table() {
+        let sql = "DESCRIBE TABLE test_schema.test";
+        let stmts: Vec<Statement> =
+            ParserContext::create_with_dialect(sql, &GenericDialect {}).unwrap();
+        assert_eq!(1, stmts.len());
+        assert_matches!(&stmts[0], Statement::DescribeTable { .. });
+        match &stmts[0] {
+            Statement::DescribeTable(show) => {
+                assert_eq!(show.schema_name.as_str(), "test_schema");
+                assert_eq!(show.table_name.as_str(), "test");
+            }
+            _ => {
+                unreachable!();
+            }
+        }
+    }
+
+    #[test]
+    pub fn test_describe_catalog_schema_table() {
+        let sql = "DESCRIBE TABLE test_catalog.test_schema.test";
+        let stmts: Vec<Statement> =
+            ParserContext::create_with_dialect(sql, &GenericDialect {}).unwrap();
+        assert_eq!(1, stmts.len());
+        assert_matches!(&stmts[0], Statement::DescribeTable { .. });
+        match &stmts[0] {
+            Statement::DescribeTable(show) => {
+                assert_eq!(show.catalog_name.as_str(), "test_catalog");
+                assert_eq!(show.schema_name.as_str(), "test_schema");
+                assert_eq!(show.table_name.as_str(), "test");
+            }
+            _ => {
+                unreachable!();
+            }
+        }
+    }
+
+    #[test]
+    pub fn test_describe_missing_table_name() {
+        let sql = "DESCRIBE TABLE";
+        ParserContext::create_with_dialect(sql, &GenericDialect {}).unwrap_err();
+    }
+}

src/sql/src/statements/statement.rs

@@ -17,6 +17,7 @@ use sqlparser::parser::ParserError;
 
 use crate::statements::alter::AlterTable;
 use crate::statements::create::{CreateDatabase, CreateTable};
+use crate::statements::describe::DescribeTable;
 use crate::statements::insert::Insert;
 use crate::statements::query::Query;
 use crate::statements::show::{ShowCreateTable, ShowDatabases, ShowTables};
@@ -40,6 +41,8 @@ pub enum Statement {
     ShowTables(ShowTables),
     // SHOW CREATE TABLE
     ShowCreateTable(ShowCreateTable),
+    // DESCRIBE TABLE
+    DescribeTable(DescribeTable),
 }
 
 /// Converts Statement to sqlparser statement
@@ -57,6 +60,9 @@ impl TryFrom<Statement> for SpStatement {
             Statement::ShowCreateTable(_) => Err(ParserError::ParserError(
                 "sqlparser does not support SHOW CREATE TABLE query.".to_string(),
             )),
+            Statement::DescribeTable(_) => Err(ParserError::ParserError(
+                "sqlparser does not support DESCRIBE TABLE query.".to_string(),
+            )),
             Statement::Query(s) => Ok(SpStatement::Query(Box::new(s.inner))),
             Statement::Insert(i) => Ok(i.inner),
             Statement::CreateDatabase(_) | Statement::CreateTable(_) | Statement::Alter(_) => {