feat: support timestamp precision on creating table (#1332)

* feat: support timestamp precision on creating table

* fix sqlness

* fix: substrait representation of different timestamp precision

src/sql/src/parser.rs

@@ -379,12 +379,15 @@ impl<'a> ParserContext<'a> {
 mod tests {
     use std::assert_matches::assert_matches;
 
+    use datatypes::prelude::ConcreteDataType;
     use sqlparser::ast::{
         Ident, ObjectName, Query as SpQuery, Statement as SpStatement, WildcardAdditionalOptions,
     };
     use sqlparser::dialect::GenericDialect;
 
     use super::*;
+    use crate::statements::create::CreateTable;
+    use crate::statements::sql_data_type_to_concrete_data_type;
 
     #[test]
     pub fn test_show_database_all() {
@@ -606,4 +609,54 @@ mod tests {
             ])))
         )
     }
+
+    fn test_timestamp_precision(sql: &str, expected_type: ConcreteDataType) {
+        match ParserContext::create_with_dialect(sql, &GenericDialect {})
+            .unwrap()
+            .pop()
+            .unwrap()
+        {
+            Statement::CreateTable(CreateTable { columns, .. }) => {
+                let ts_col = columns.get(0).unwrap();
+                assert_eq!(
+                    expected_type,
+                    sql_data_type_to_concrete_data_type(&ts_col.data_type).unwrap()
+                );
+            }
+            _ => unreachable!(),
+        }
+    }
+
+    #[test]
+    pub fn test_create_table_with_precision() {
+        test_timestamp_precision(
+            "create table demo (ts timestamp time index, cnt int);",
+            ConcreteDataType::timestamp_millisecond_datatype(),
+        );
+        test_timestamp_precision(
+            "create table demo (ts timestamp(0) time index, cnt int);",
+            ConcreteDataType::timestamp_second_datatype(),
+        );
+        test_timestamp_precision(
+            "create table demo (ts timestamp(3) time index, cnt int);",
+            ConcreteDataType::timestamp_millisecond_datatype(),
+        );
+        test_timestamp_precision(
+            "create table demo (ts timestamp(6) time index, cnt int);",
+            ConcreteDataType::timestamp_microsecond_datatype(),
+        );
+        test_timestamp_precision(
+            "create table demo (ts timestamp(9) time index, cnt int);",
+            ConcreteDataType::timestamp_nanosecond_datatype(),
+        );
+    }
+
+    #[test]
+    #[should_panic]
+    pub fn test_create_table_with_invalid_precision() {
+        test_timestamp_precision(
+            "create table demo (ts timestamp(1) time index, cnt int);",
+            ConcreteDataType::timestamp_millisecond_datatype(),
+        );
+    }
 }

src/sql/src/statements.rs

@@ -32,6 +32,7 @@ use common_base::bytes::Bytes;
 use common_time::Timestamp;
 use datatypes::prelude::ConcreteDataType;
 use datatypes::schema::{ColumnDefaultConstraint, ColumnSchema};
+use datatypes::types::TimestampType;
 use datatypes::value::Value;
 use snafu::{ensure, OptionExt, ResultExt};
 
@@ -304,7 +305,18 @@ pub fn sql_data_type_to_concrete_data_type(data_type: &SqlDataType) -> Result<Co
         SqlDataType::Date => Ok(ConcreteDataType::date_datatype()),
         SqlDataType::Varbinary(_) => Ok(ConcreteDataType::binary_datatype()),
         SqlDataType::Datetime(_) => Ok(ConcreteDataType::datetime_datatype()),
-        SqlDataType::Timestamp(_, _) => Ok(ConcreteDataType::timestamp_millisecond_datatype()),
+        SqlDataType::Timestamp(precision, _) => Ok(precision
+            .as_ref()
+            .map(|v| TimestampType::try_from(*v))
+            .transpose()
+            .map_err(|_| {
+                error::SqlTypeNotSupportedSnafu {
+                    t: data_type.clone(),
+                }
+                .build()
+            })?
+            .map(|t| ConcreteDataType::timestamp_datatype(t.unit()))
+            .unwrap_or(ConcreteDataType::timestamp_millisecond_datatype())),
         _ => error::SqlTypeNotSupportedSnafu {
             t: data_type.clone(),
         }