diff --git a/Cargo.lock b/Cargo.lock
index 7ca7d99b95..51052b6e50 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1923,6 +1923,8 @@ dependencies = [
  "datatypes",
  "serde",
  "snafu",
+ "sqlparser 0.34.0",
+ "sqlparser_derive 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "statrs",
  "tokio",
 ]
@@ -9043,10 +9045,12 @@ dependencies = [
  "datatypes",
  "hex",
  "itertools 0.10.5",
+ "lazy_static",
  "once_cell",
  "regex",
  "snafu",
  "sqlparser 0.34.0",
+ "sqlparser_derive 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "table",
 ]
 
diff --git a/src/common/query/Cargo.toml b/src/common/query/Cargo.toml
index e4c44b9586..60f6a8003d 100644
--- a/src/common/query/Cargo.toml
+++ b/src/common/query/Cargo.toml
@@ -16,6 +16,8 @@ datafusion.workspace = true
 datatypes = { workspace = true }
 serde.workspace = true
 snafu.workspace = true
+sqlparser.workspace = true
+sqlparser_derive = "0.1"
 statrs = "0.16"
 
 [dev-dependencies]
diff --git a/src/common/query/src/lib.rs b/src/common/query/src/lib.rs
index 5d7922ade2..95d1aed2fe 100644
--- a/src/common/query/src/lib.rs
+++ b/src/common/query/src/lib.rs
@@ -26,6 +26,7 @@ pub mod logical_plan;
 pub mod physical_plan;
 pub mod prelude;
 mod signature;
+use sqlparser_derive::{Visit, VisitMut};
 
 // sql output
 pub enum Output {
@@ -48,7 +49,7 @@ impl Debug for Output {
 
 pub use datafusion::physical_plan::ExecutionPlan as DfPhysicalPlan;
 
-#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Visit, VisitMut)]
 pub enum AddColumnLocation {
     First,
     After { column_name: String },
diff --git a/src/operator/src/expr_factory.rs b/src/operator/src/expr_factory.rs
index b9fda8224e..c45ab659c6 100644
--- a/src/operator/src/expr_factory.rs
+++ b/src/operator/src/expr_factory.rs
@@ -128,11 +128,11 @@ pub(crate) async fn create_external_expr(
 
     let mut table_options = create.options;
 
-    let (object_store, files) = prepare_file_table_files(&table_options)
+    let (object_store, files) = prepare_file_table_files(&table_options.map)
         .await
         .context(PrepareFileTableSnafu)?;
 
-    let file_column_schemas = infer_file_table_schema(&object_store, &files, &table_options)
+    let file_column_schemas = infer_file_table_schema(&object_store, &files, &table_options.map)
         .await
         .context(InferFileTableSchemaSnafu)?
         .column_schemas;
@@ -157,7 +157,7 @@ pub(crate) async fn create_external_expr(
         files,
         file_column_schemas,
     };
-    let _ = table_options.insert(
+    table_options.insert(
         FILE_TABLE_META_KEY.to_string(),
         serde_json::to_string(&meta).context(EncodeJsonSnafu)?,
     );
@@ -172,7 +172,7 @@ pub(crate) async fn create_external_expr(
         time_index,
         primary_keys,
         create_if_not_exists: create.if_not_exists,
-        table_options,
+        table_options: table_options.map,
         table_id: None,
         engine: create.engine.to_string(),
     };
diff --git a/src/operator/src/statement.rs b/src/operator/src/statement.rs
index 5cb0a2e4d3..e6b46d26a6 100644
--- a/src/operator/src/statement.rs
+++ b/src/operator/src/statement.rs
@@ -21,7 +21,6 @@ mod dml;
 mod show;
 mod tql;
 
-use std::collections::HashMap;
 use std::str::FromStr;
 use std::sync::Arc;
 
@@ -43,6 +42,7 @@ use session::context::QueryContextRef;
 use snafu::{OptionExt, ResultExt};
 use sql::statements::copy::{CopyDatabaseArgument, CopyTable, CopyTableArgument};
 use sql::statements::statement::Statement;
+use sql::statements::OptionMap;
 use sqlparser::ast::ObjectName;
 use table::engine::TableReference;
 use table::requests::{CopyDatabaseRequest, CopyDirection, CopyTableRequest};
@@ -259,8 +259,8 @@ fn to_copy_table_request(stmt: CopyTable, query_ctx: QueryContextRef) -> Result<
         schema_name,
         table_name,
         location,
-        with,
-        connection,
+        with: with.map,
+        connection: connection.map,
         pattern,
         direction,
         // we copy the whole table by default.
@@ -292,14 +292,14 @@ fn to_copy_database_request(
         catalog_name,
         schema_name: database_name,
         location: arg.location,
-        with: arg.with,
-        connection: arg.connection,
+        with: arg.with.map,
+        connection: arg.connection.map,
         time_range,
     })
 }
 
 /// Extracts timestamp from a [HashMap<String, String>] with given key.
-fn extract_timestamp(map: &HashMap<String, String>, key: &str) -> Result<Option<Timestamp>> {
+fn extract_timestamp(map: &OptionMap, key: &str) -> Result<Option<Timestamp>> {
     map.get(key)
         .map(|v| {
             Timestamp::from_str(v)
diff --git a/src/servers/src/error.rs b/src/servers/src/error.rs
index bec041d210..049ded45c6 100644
--- a/src/servers/src/error.rs
+++ b/src/servers/src/error.rs
@@ -65,12 +65,7 @@ pub enum Error {
         source: std::io::Error,
     },
 
-    #[snafu(display(
-        "Failed to execute query, source: {}, query: {}, location: {}",
-        source,
-        query,
-        location
-    ))]
+    #[snafu(display("Failed to execute query, source: {}, query: {}", source, query))]
     ExecuteQuery {
         query: String,
         location: Location,
@@ -279,7 +274,7 @@ pub enum Error {
         source: query::error::Error,
     },
 
-    #[snafu(display("Failed to get param types, source: {source}, location: {location}"))]
+    #[snafu(display("Failed to get param types, source: {source}"))]
     GetPreparedStmtParams {
         source: query::error::Error,
         location: Location,
diff --git a/src/sql/Cargo.toml b/src/sql/Cargo.toml
index 5d271470aa..3e17e1fbe3 100644
--- a/src/sql/Cargo.toml
+++ b/src/sql/Cargo.toml
@@ -15,10 +15,12 @@ datafusion-sql.workspace = true
 datatypes = { workspace = true }
 hex = "0.4"
 itertools.workspace = true
+lazy_static.workspace = true
 once_cell.workspace = true
 regex.workspace = true
 snafu = { version = "0.7", features = ["backtraces"] }
 sqlparser.workspace = true
+sqlparser_derive = "0.1"
 table = { workspace = true }
 
 [dev-dependencies]
diff --git a/src/sql/src/ast.rs b/src/sql/src/ast.rs
index a72d7965b7..e1944de98a 100644
--- a/src/sql/src/ast.rs
+++ b/src/sql/src/ast.rs
@@ -13,7 +13,7 @@
 // limitations under the License.
 
 pub use sqlparser::ast::{
-    visit_expressions_mut, BinaryOperator, ColumnDef, ColumnOption, ColumnOptionDef, DataType,
-    Expr, Function, FunctionArg, FunctionArgExpr, Ident, ObjectName, SqlOption, TableConstraint,
-    TimezoneInfo, Value, VisitMut, Visitor,
+    visit_expressions_mut, visit_statements_mut, BinaryOperator, ColumnDef, ColumnOption,
+    ColumnOptionDef, DataType, Expr, Function, FunctionArg, FunctionArgExpr, Ident, ObjectName,
+    SqlOption, TableConstraint, TimezoneInfo, Value, Visit, VisitMut, Visitor, VisitorMut,
 };
diff --git a/src/sql/src/parser.rs b/src/sql/src/parser.rs
index 9b9409b2f5..913d2ce760 100644
--- a/src/sql/src/parser.rs
+++ b/src/sql/src/parser.rs
@@ -22,6 +22,7 @@ use crate::ast::{Expr, ObjectName};
 use crate::error::{self, Result, SyntaxSnafu};
 use crate::parsers::tql_parser;
 use crate::statements::statement::Statement;
+use crate::statements::transform_statements;
 
 /// GrepTime SQL parser context, a simple wrapper for Datafusion SQL parser.
 pub struct ParserContext<'a> {
@@ -58,6 +59,8 @@ impl<'a> ParserContext<'a> {
             expecting_statement_delimiter = true;
         }
 
+        transform_statements(&mut stmts)?;
+
         Ok(stmts)
     }
 
diff --git a/src/sql/src/parsers/copy_parser.rs b/src/sql/src/parsers/copy_parser.rs
index fc3573fa19..34b3a80cfd 100644
--- a/src/sql/src/parsers/copy_parser.rs
+++ b/src/sql/src/parsers/copy_parser.rs
@@ -62,8 +62,8 @@ impl<'a> ParserContext<'a> {
         let (with, connection, location) = self.parse_copy_to()?;
         Ok(CopyDatabaseArgument {
             database_name,
-            with,
-            connection,
+            with: with.into(),
+            connection: connection.into(),
             location,
         })
     }
@@ -82,8 +82,8 @@ impl<'a> ParserContext<'a> {
             let (with, connection, location) = self.parse_copy_to()?;
             Ok(CopyTable::To(CopyTableArgument {
                 table_name,
-                with,
-                connection,
+                with: with.into(),
+                connection: connection.into(),
                 location,
             }))
         } else {
@@ -308,7 +308,10 @@ mod tests {
                     if let Some(expected_pattern) = test.expected_pattern {
                         assert_eq!(copy_table.pattern().unwrap(), expected_pattern);
                     }
-                    assert_eq!(copy_table.connection.clone(), test.expected_connection);
+                    assert_eq!(
+                        copy_table.connection.clone(),
+                        test.expected_connection.into()
+                    );
                 }
                 _ => unreachable!(),
             }
@@ -348,7 +351,10 @@ mod tests {
                 Statement::Copy(crate::statements::copy::Copy::CopyTable(CopyTable::To(
                     copy_table,
                 ))) => {
-                    assert_eq!(copy_table.connection.clone(), test.expected_connection);
+                    assert_eq!(
+                        copy_table.connection.clone(),
+                        test.expected_connection.into()
+                    );
                 }
                 _ => unreachable!(),
             }
@@ -374,7 +380,7 @@ mod tests {
             [("format".to_string(), "parquet".to_string())]
                 .into_iter()
                 .collect::<HashMap<_, _>>(),
-            stmt.with
+            stmt.with.map
         );
 
         assert_eq!(
@@ -384,7 +390,7 @@ mod tests {
             ]
             .into_iter()
             .collect::<HashMap<_, _>>(),
-            stmt.connection
+            stmt.connection.map
         );
     }
 }
diff --git a/src/sql/src/parsers/create_parser.rs b/src/sql/src/parsers/create_parser.rs
index 15ca790bd2..51b5ac48b5 100644
--- a/src/sql/src/parsers/create_parser.rs
+++ b/src/sql/src/parsers/create_parser.rs
@@ -37,7 +37,9 @@ use crate::statements::create::{
     CreateDatabase, CreateExternalTable, CreateTable, PartitionEntry, Partitions, TIME_INDEX,
 };
 use crate::statements::statement::Statement;
-use crate::statements::{sql_data_type_to_concrete_data_type, sql_value_to_value};
+use crate::statements::{
+    get_data_type_by_alias_name, sql_data_type_to_concrete_data_type, sql_value_to_value,
+};
 use crate::util::parse_option_string;
 
 pub const ENGINE: &str = "ENGINE";
@@ -106,7 +108,7 @@ impl<'a> ParserContext<'a> {
             name: table_name,
             columns,
             constraints,
-            options,
+            options: options.into(),
             if_not_exists,
             engine,
         }))
@@ -374,8 +376,11 @@ impl<'a> ParserContext<'a> {
                     msg: "time index column can't be null",
                 }
             );
+
+            // The timestamp type may be an alias type, we have to retrieve the actual type.
+            let data_type = get_real_timestamp_type(&column.data_type);
             ensure!(
-                matches!(column.data_type, DataType::Timestamp(_, _)),
+                matches!(data_type, DataType::Timestamp(_, _)),
                 InvalidColumnOptionSnafu {
                     name: column.name.to_string(),
                     msg: "time index column data type should be timestamp",
@@ -653,8 +658,9 @@ fn validate_time_index(create_table: &CreateTable) -> Result<()> {
             ),
         })?;
 
+    let time_index_data_type = get_real_timestamp_type(&time_index_column.data_type);
     ensure!(
-        matches!(time_index_column.data_type, DataType::Timestamp(_, _)),
+        matches!(time_index_data_type, DataType::Timestamp(_, _)),
         InvalidColumnOptionSnafu {
             name: time_index_column.name.to_string(),
             msg: "time index column data type should be timestamp",
@@ -664,6 +670,19 @@ fn validate_time_index(create_table: &CreateTable) -> Result<()> {
     Ok(())
 }
 
+fn get_real_timestamp_type(data_type: &DataType) -> DataType {
+    match data_type {
+        DataType::Custom(name, tokens) if name.0.len() == 1 && tokens.is_empty() => {
+            if let Some(real_type) = get_data_type_by_alias_name(name.0[0].value.as_str()) {
+                real_type
+            } else {
+                data_type.clone()
+            }
+        }
+        _ => data_type.clone(),
+    }
+}
+
 fn validate_partitions(columns: &[ColumnDef], partitions: &Partitions) -> Result<()> {
     let partition_columns = ensure_partition_columns_defined(columns, partitions)?;
 
@@ -881,7 +900,7 @@ mod tests {
             match &stmts[0] {
                 Statement::CreateExternalTable(c) => {
                     assert_eq!(c.name.to_string(), test.expected_table_name.to_string());
-                    assert_eq!(c.options, test.expected_options);
+                    assert_eq!(c.options, test.expected_options.into());
                     assert_eq!(c.if_not_exists, test.expected_if_not_exist);
                     assert_eq!(c.engine, test.expected_engine);
                 }
@@ -895,7 +914,7 @@ mod tests {
         let sql = "CREATE EXTERNAL TABLE city (
             host string,
             ts int64,
-            cpu float64 default 0,
+            cpu float32 default 0,
             memory float64,
             TIME INDEX (ts),
             PRIMARY KEY(ts, host)
@@ -911,13 +930,13 @@ mod tests {
         match &stmts[0] {
             Statement::CreateExternalTable(c) => {
                 assert_eq!(c.name.to_string(), "city");
-                assert_eq!(c.options, options);
+                assert_eq!(c.options, options.into());
 
                 let columns = &c.columns;
                 assert_column_def(&columns[0], "host", "STRING");
-                assert_column_def(&columns[1], "ts", "int64");
-                assert_column_def(&columns[2], "cpu", "float64");
-                assert_column_def(&columns[3], "memory", "float64");
+                assert_column_def(&columns[1], "ts", "BIGINT");
+                assert_column_def(&columns[2], "cpu", "FLOAT");
+                assert_column_def(&columns[3], "memory", "DOUBLE");
 
                 let constraints = &c.constraints;
                 assert_matches!(
@@ -1423,7 +1442,7 @@ ENGINE=mito";
         let sql = r"create table demo(
                              host string,
                              ts timestamp,
-                             cpu float64 default 0,
+                             cpu float32 default 0,
                              memory float64,
                              TIME INDEX (ts),
                              PRIMARY KEY(ts, host)) engine=mito
@@ -1440,8 +1459,9 @@ ENGINE=mito";
                 let columns = &c.columns;
                 assert_column_def(&columns[0], "host", "STRING");
                 assert_column_def(&columns[1], "ts", "TIMESTAMP");
-                assert_column_def(&columns[2], "cpu", "float64");
-                assert_column_def(&columns[3], "memory", "float64");
+                assert_column_def(&columns[2], "cpu", "FLOAT");
+                assert_column_def(&columns[3], "memory", "DOUBLE");
+
                 let constraints = &c.constraints;
                 assert_matches!(
                     &constraints[0],
diff --git a/src/sql/src/statements.rs b/src/sql/src/statements.rs
index 1510335c77..1e4f03ad9f 100644
--- a/src/sql/src/statements.rs
+++ b/src/sql/src/statements.rs
@@ -20,10 +20,12 @@ pub mod describe;
 pub mod drop;
 pub mod explain;
 pub mod insert;
+mod option_map;
 pub mod query;
 pub mod show;
 pub mod statement;
 pub mod tql;
+mod transform;
 pub mod truncate;
 
 use std::str::FromStr;
@@ -38,7 +40,9 @@ use datatypes::prelude::ConcreteDataType;
 use datatypes::schema::{ColumnDefaultConstraint, ColumnSchema, COMMENT_KEY};
 use datatypes::types::TimestampType;
 use datatypes::value::{OrderedF32, OrderedF64, Value};
+pub use option_map::OptionMap;
 use snafu::{ensure, OptionExt, ResultExt};
+pub use transform::{get_data_type_by_alias_name, transform_statements};
 
 use crate::ast::{
     ColumnDef, ColumnOption, ColumnOptionDef, DataType as SqlDataType, Expr, TimezoneInfo,
diff --git a/src/sql/src/statements/alter.rs b/src/sql/src/statements/alter.rs
index 132f9368c1..cf3dc1bf91 100644
--- a/src/sql/src/statements/alter.rs
+++ b/src/sql/src/statements/alter.rs
@@ -14,8 +14,9 @@
 
 use common_query::AddColumnLocation;
 use sqlparser::ast::{ColumnDef, Ident, ObjectName, TableConstraint};
+use sqlparser_derive::{Visit, VisitMut};
 
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq, Visit, VisitMut)]
 pub struct AlterTable {
     table_name: ObjectName,
     alter_operation: AlterTableOperation,
@@ -38,7 +39,7 @@ impl AlterTable {
     }
 }
 
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq, Visit, VisitMut)]
 pub enum AlterTableOperation {
     /// `ADD <table_constraint>`
     AddConstraint(TableConstraint),
diff --git a/src/sql/src/statements/copy.rs b/src/sql/src/statements/copy.rs
index 494c07c707..cd3893300b 100644
--- a/src/sql/src/statements/copy.rs
+++ b/src/sql/src/statements/copy.rs
@@ -12,35 +12,36 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::collections::HashMap;
-
 use sqlparser::ast::ObjectName;
+use sqlparser_derive::{Visit, VisitMut};
 
-#[derive(Debug, Clone, PartialEq, Eq)]
+use crate::statements::OptionMap;
+
+#[derive(Debug, Clone, PartialEq, Eq, Visit, VisitMut)]
 pub enum Copy {
     CopyTable(CopyTable),
     CopyDatabase(CopyDatabaseArgument),
 }
 
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq, Visit, VisitMut)]
 pub enum CopyTable {
     To(CopyTableArgument),
     From(CopyTableArgument),
 }
 
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq, Visit, VisitMut)]
 pub struct CopyDatabaseArgument {
     pub database_name: ObjectName,
-    pub with: HashMap<String, String>,
-    pub connection: HashMap<String, String>,
+    pub with: OptionMap,
+    pub connection: OptionMap,
     pub location: String,
 }
 
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq, Visit, VisitMut)]
 pub struct CopyTableArgument {
     pub table_name: ObjectName,
-    pub with: HashMap<String, String>,
-    pub connection: HashMap<String, String>,
+    pub with: OptionMap,
+    pub connection: OptionMap,
     /// Copy tbl [To|From] 'location'.
     pub location: String,
 }
diff --git a/src/sql/src/statements/create.rs b/src/sql/src/statements/create.rs
index f2075a15f5..47b511be56 100644
--- a/src/sql/src/statements/create.rs
+++ b/src/sql/src/statements/create.rs
@@ -12,13 +12,14 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::collections::HashMap;
 use std::fmt::{Display, Formatter};
 
 use common_catalog::consts::FILE_ENGINE;
 use itertools::Itertools;
+use sqlparser_derive::{Visit, VisitMut};
 
 use crate::ast::{ColumnDef, Ident, ObjectName, SqlOption, TableConstraint, Value as SqlValue};
+use crate::statements::OptionMap;
 
 const LINE_SEP: &str = ",\n";
 const COMMA_SEP: &str = ", ";
@@ -57,7 +58,7 @@ pub fn is_time_index(constraint: &TableConstraint) -> bool {
     }  if name.value == TIME_INDEX)
 }
 
-#[derive(Debug, PartialEq, Eq, Clone)]
+#[derive(Debug, PartialEq, Eq, Clone, Visit, VisitMut)]
 pub struct CreateTable {
     /// Create if not exists
     pub if_not_exists: bool,
@@ -124,7 +125,7 @@ impl CreateTable {
     }
 }
 
-#[derive(Debug, PartialEq, Eq, Clone)]
+#[derive(Debug, PartialEq, Eq, Clone, Visit, VisitMut)]
 pub struct Partitions {
     pub column_list: Vec<Ident>,
     pub entries: Vec<PartitionEntry>,
@@ -139,7 +140,7 @@ impl Partitions {
     }
 }
 
-#[derive(Debug, PartialEq, Eq, Clone)]
+#[derive(Debug, PartialEq, Eq, Clone, Visit, VisitMut)]
 pub struct PartitionEntry {
     pub name: Ident,
     pub value_list: Vec<SqlValue>,
@@ -197,14 +198,14 @@ impl Display for CreateTable {
     }
 }
 
-#[derive(Debug, PartialEq, Eq, Clone)]
+#[derive(Debug, PartialEq, Eq, Clone, Visit, VisitMut)]
 pub struct CreateDatabase {
     pub name: ObjectName,
     /// Create if not exists
     pub if_not_exists: bool,
 }
 
-#[derive(Debug, PartialEq, Eq, Clone)]
+#[derive(Debug, PartialEq, Eq, Clone, Visit, VisitMut)]
 pub struct CreateExternalTable {
     /// Table name
     pub name: ObjectName,
@@ -212,7 +213,7 @@ pub struct CreateExternalTable {
     pub constraints: Vec<TableConstraint>,
     /// Table options in `WITH`.
     /// All keys are lowercase.
-    pub options: HashMap<String, String>,
+    pub options: OptionMap,
     pub if_not_exists: bool,
     pub engine: String,
 }
diff --git a/src/sql/src/statements/delete.rs b/src/sql/src/statements/delete.rs
index 47ba87a011..4346610b7d 100644
--- a/src/sql/src/statements/delete.rs
+++ b/src/sql/src/statements/delete.rs
@@ -13,8 +13,9 @@
 // limitations under the License.
 
 use sqlparser::ast::Statement;
+use sqlparser_derive::{Visit, VisitMut};
 
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq, Visit, VisitMut)]
 pub struct Delete {
     pub inner: Statement,
 }
diff --git a/src/sql/src/statements/describe.rs b/src/sql/src/statements/describe.rs
index 006dd83986..ee86509305 100644
--- a/src/sql/src/statements/describe.rs
+++ b/src/sql/src/statements/describe.rs
@@ -13,9 +13,10 @@
 // limitations under the License.
 
 use sqlparser::ast::ObjectName;
+use sqlparser_derive::{Visit, VisitMut};
 
 /// SQL structure for `DESCRIBE TABLE`.
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq, Visit, VisitMut)]
 pub struct DescribeTable {
     name: ObjectName,
 }
diff --git a/src/sql/src/statements/drop.rs b/src/sql/src/statements/drop.rs
index b8fc0401fa..b8e46ac3f2 100644
--- a/src/sql/src/statements/drop.rs
+++ b/src/sql/src/statements/drop.rs
@@ -13,9 +13,10 @@
 // limitations under the License.
 
 use sqlparser::ast::ObjectName;
+use sqlparser_derive::{Visit, VisitMut};
 
 /// DROP TABLE statement.
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq, Visit, VisitMut)]
 pub struct DropTable {
     table_name: ObjectName,
 }
diff --git a/src/sql/src/statements/explain.rs b/src/sql/src/statements/explain.rs
index 82b757d193..fe953d6618 100644
--- a/src/sql/src/statements/explain.rs
+++ b/src/sql/src/statements/explain.rs
@@ -13,11 +13,12 @@
 // limitations under the License.
 
 use sqlparser::ast::Statement as SpStatement;
+use sqlparser_derive::{Visit, VisitMut};
 
 use crate::error::Error;
 
 /// Explain statement.
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq, Visit, VisitMut)]
 pub struct Explain {
     pub inner: SpStatement,
 }
diff --git a/src/sql/src/statements/insert.rs b/src/sql/src/statements/insert.rs
index b5bdf55b09..96af1a8de9 100644
--- a/src/sql/src/statements/insert.rs
+++ b/src/sql/src/statements/insert.rs
@@ -13,12 +13,13 @@
 // limitations under the License.
 use sqlparser::ast::{ObjectName, Query, SetExpr, Statement, UnaryOperator, Values};
 use sqlparser::parser::ParserError;
+use sqlparser_derive::{Visit, VisitMut};
 
 use crate::ast::{Expr, Value};
 use crate::error::Result;
 use crate::statements::query::Query as GtQuery;
 
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq, Visit, VisitMut)]
 pub struct Insert {
     // Can only be sqlparser::ast::Statement::Insert variant
     pub inner: Statement,
diff --git a/src/sql/src/statements/option_map.rs b/src/sql/src/statements/option_map.rs
new file mode 100644
index 0000000000..63069f6fc0
--- /dev/null
+++ b/src/sql/src/statements/option_map.rs
@@ -0,0 +1,63 @@
+// Copyright 2023 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.
+
+mod visit;
+mod visit_mut;
+
+use std::borrow::Borrow;
+use std::collections::HashMap;
+use std::iter::FromIterator;
+
+/// Options hashmap.
+/// Because the trait `Visit` and `VisitMut` is not implemented for `HashMap<String, String>`, we have to wrap it and implement them by ourself.
+#[derive(Clone, Eq, PartialEq, Debug)]
+pub struct OptionMap {
+    pub map: HashMap<String, String>,
+}
+
+impl OptionMap {
+    pub fn insert(&mut self, k: String, v: String) {
+        self.map.insert(k, v);
+    }
+
+    pub fn get(&self, k: &str) -> Option<&String> {
+        self.map.get(k)
+    }
+}
+
+impl From<HashMap<String, String>> for OptionMap {
+    fn from(map: HashMap<String, String>) -> Self {
+        Self { map }
+    }
+}
+
+impl AsRef<HashMap<String, String>> for OptionMap {
+    fn as_ref(&self) -> &HashMap<String, String> {
+        &self.map
+    }
+}
+
+impl Borrow<HashMap<String, String>> for OptionMap {
+    fn borrow(&self) -> &HashMap<String, String> {
+        &self.map
+    }
+}
+
+impl FromIterator<(String, String)> for OptionMap {
+    fn from_iter<I: IntoIterator<Item = (String, String)>>(iter: I) -> Self {
+        Self {
+            map: iter.into_iter().collect(),
+        }
+    }
+}
diff --git a/src/sql/src/statements/option_map/visit.rs b/src/sql/src/statements/option_map/visit.rs
new file mode 100644
index 0000000000..1242ae69d6
--- /dev/null
+++ b/src/sql/src/statements/option_map/visit.rs
@@ -0,0 +1,29 @@
+// Copyright 2023 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.
+
+use std::ops::ControlFlow;
+
+use sqlparser::ast::{Visit, Visitor};
+
+use crate::statements::OptionMap;
+
+impl Visit for OptionMap {
+    fn visit<V: Visitor>(&self, visitor: &mut V) -> ControlFlow<V::Break> {
+        for (k, v) in &self.map {
+            k.visit(visitor)?;
+            v.visit(visitor)?;
+        }
+        ControlFlow::Continue(())
+    }
+}
diff --git a/src/sql/src/statements/option_map/visit_mut.rs b/src/sql/src/statements/option_map/visit_mut.rs
new file mode 100644
index 0000000000..0c61430560
--- /dev/null
+++ b/src/sql/src/statements/option_map/visit_mut.rs
@@ -0,0 +1,28 @@
+// Copyright 2023 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.
+
+use std::ops::ControlFlow;
+
+use sqlparser::ast::{VisitMut, VisitorMut};
+
+use crate::statements::OptionMap;
+
+impl VisitMut for OptionMap {
+    fn visit<V: VisitorMut>(&mut self, visitor: &mut V) -> ControlFlow<V::Break> {
+        for (_, v) in self.map.iter_mut() {
+            v.visit(visitor)?;
+        }
+        ControlFlow::Continue(())
+    }
+}
diff --git a/src/sql/src/statements/query.rs b/src/sql/src/statements/query.rs
index c2b720ce15..e03372d268 100644
--- a/src/sql/src/statements/query.rs
+++ b/src/sql/src/statements/query.rs
@@ -15,11 +15,12 @@
 use std::fmt;
 
 use sqlparser::ast::Query as SpQuery;
+use sqlparser_derive::{Visit, VisitMut};
 
 use crate::error::Error;
 
 /// Query statement instance.
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq, Visit, VisitMut)]
 pub struct Query {
     pub inner: SpQuery,
 }
diff --git a/src/sql/src/statements/show.rs b/src/sql/src/statements/show.rs
index 4093982279..eed546fba9 100644
--- a/src/sql/src/statements/show.rs
+++ b/src/sql/src/statements/show.rs
@@ -14,10 +14,12 @@
 
 use std::fmt;
 
+use sqlparser_derive::{Visit, VisitMut};
+
 use crate::ast::{Expr, Ident, ObjectName};
 
 /// Show kind for SQL expressions like `SHOW DATABASE` or `SHOW TABLE`
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq, Visit, VisitMut)]
 pub enum ShowKind {
     All,
     Like(Ident),
@@ -35,7 +37,7 @@ impl fmt::Display for ShowKind {
 }
 
 /// SQL structure for `SHOW DATABASES`.
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq, Visit, VisitMut)]
 pub struct ShowDatabases {
     pub kind: ShowKind,
 }
@@ -48,14 +50,14 @@ impl ShowDatabases {
 }
 
 /// SQL structure for `SHOW TABLES`.
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq, Visit, VisitMut)]
 pub struct ShowTables {
     pub kind: ShowKind,
     pub database: Option<String>,
 }
 
 /// SQL structure for `SHOW CREATE TABLE`.
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq, Visit, VisitMut)]
 pub struct ShowCreateTable {
     pub table_name: ObjectName,
 }
diff --git a/src/sql/src/statements/statement.rs b/src/sql/src/statements/statement.rs
index 5263f5558d..462535be4c 100644
--- a/src/sql/src/statements/statement.rs
+++ b/src/sql/src/statements/statement.rs
@@ -14,6 +14,7 @@
 
 use datafusion_sql::parser::Statement as DfStatement;
 use sqlparser::ast::Statement as SpStatement;
+use sqlparser_derive::{Visit, VisitMut};
 
 use crate::error::{ConvertToDfStatementSnafu, Error};
 use crate::statements::alter::AlterTable;
@@ -30,7 +31,7 @@ use crate::statements::truncate::TruncateTable;
 
 /// Tokens parsed by `DFParser` are converted into these values.
 #[allow(clippy::large_enum_variant)]
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq, Visit, VisitMut)]
 pub enum Statement {
     // Query
     Query(Box<Query>),
diff --git a/src/sql/src/statements/tql.rs b/src/sql/src/statements/tql.rs
index d9b7187bf7..24bfeb1870 100644
--- a/src/sql/src/statements/tql.rs
+++ b/src/sql/src/statements/tql.rs
@@ -11,14 +11,16 @@
 // 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.
-#[derive(Debug, Clone, PartialEq, Eq)]
+use sqlparser_derive::{Visit, VisitMut};
+
+#[derive(Debug, Clone, PartialEq, Eq, Visit, VisitMut)]
 pub enum Tql {
     Eval(TqlEval),
     Explain(TqlExplain),
     Analyze(TqlAnalyze),
 }
 
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq, Visit, VisitMut)]
 pub struct TqlEval {
     pub start: String,
     pub end: String,
@@ -27,7 +29,7 @@ pub struct TqlEval {
 }
 
 /// TQL EXPLAIN (like SQL EXPLAIN): doesn't execute the query but tells how the query would be executed.
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq, Visit, VisitMut)]
 pub struct TqlExplain {
     pub start: String,
     pub end: String,
@@ -36,7 +38,7 @@ pub struct TqlExplain {
 }
 
 /// TQL ANALYZE (like SQL ANALYZE): executes the plan and tells the detailed per-step execution time.
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq, Visit, VisitMut)]
 pub struct TqlAnalyze {
     pub start: String,
     pub end: String,
diff --git a/src/sql/src/statements/transform.rs b/src/sql/src/statements/transform.rs
new file mode 100644
index 0000000000..e48cb29f1b
--- /dev/null
+++ b/src/sql/src/statements/transform.rs
@@ -0,0 +1,64 @@
+// Copyright 2023 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.
+
+use std::ops::ControlFlow;
+use std::sync::Arc;
+
+use lazy_static::lazy_static;
+use sqlparser::ast::{visit_expressions_mut, Expr};
+
+use crate::error::Result;
+use crate::statements::statement::Statement;
+mod type_alias;
+pub use type_alias::get_data_type_by_alias_name;
+use type_alias::TypeAliasTransformRule;
+
+lazy_static! {
+    /// [TransformRule] registry
+    static ref RULES: Vec<Arc<dyn TransformRule>> = vec![
+        Arc::new(TypeAliasTransformRule{}),
+    ];
+}
+
+/// Transform rule to transform statement or expr
+pub(crate) trait TransformRule: Send + Sync {
+    /// Visit a [Statement]
+    fn visit_statement(&self, _stmt: &mut Statement) -> Result<()> {
+        Ok(())
+    }
+
+    /// Visit an [Expr]
+    fn visit_expr(&self, _expr: &mut Expr) -> ControlFlow<()> {
+        ControlFlow::<()>::Continue(())
+    }
+}
+
+/// Transform statements by rules
+pub fn transform_statements(stmts: &mut Vec<Statement>) -> Result<()> {
+    for stmt in &mut *stmts {
+        for rule in RULES.iter() {
+            rule.visit_statement(stmt)?;
+        }
+    }
+
+    visit_expressions_mut(stmts, |expr| {
+        for rule in RULES.iter() {
+            rule.visit_expr(expr)?;
+        }
+
+        ControlFlow::<()>::Continue(())
+    });
+
+    Ok(())
+}
diff --git a/src/sql/src/statements/transform/type_alias.rs b/src/sql/src/statements/transform/type_alias.rs
new file mode 100644
index 0000000000..871b38a81f
--- /dev/null
+++ b/src/sql/src/statements/transform/type_alias.rs
@@ -0,0 +1,353 @@
+// Copyright 2023 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.
+use std::ops::ControlFlow;
+
+use datatypes::data_type::DataType as GreptimeDataType;
+use sqlparser::ast::{
+    ColumnDef, DataType, Expr, Function, FunctionArg, FunctionArgExpr, Ident, ObjectName, Value,
+};
+
+use crate::error::Result;
+use crate::statements::create::{CreateExternalTable, CreateTable};
+use crate::statements::statement::Statement;
+use crate::statements::transform::TransformRule;
+use crate::statements::{sql_data_type_to_concrete_data_type, TimezoneInfo};
+
+/// SQL data type alias transformer:
+///  - `TimestampSecond`, `Timestamp_s`, `Timestamp_sec` for `Timestamp(0)`.
+///  - `TimestampMillisecond`, `Timestamp_ms` for `Timestamp(3)`.
+///  - `TimestampMicrosecond`, `Timestamp_us` for `Timestamp(6)`.
+///  - `TimestampNanosecond`, `Timestamp_ns` for `Timestamp(9)`.
+///  - `INT8` for `tinyint`
+///  - `INT16` for `smallint`
+///  - `INT32` for `int`
+///  - `INT32` for `bigint`
+///  -  And `UINT8`, `UINT16` etc. for `UnsignedTinyint` etc.
+pub(crate) struct TypeAliasTransformRule;
+
+impl TransformRule for TypeAliasTransformRule {
+    fn visit_statement(&self, stmt: &mut Statement) -> Result<()> {
+        match stmt {
+            Statement::CreateTable(CreateTable { columns, .. }) => {
+                columns
+                    .iter_mut()
+                    .for_each(|ColumnDef { data_type, .. }| replace_type_alias(data_type));
+            }
+            Statement::CreateExternalTable(CreateExternalTable { columns, .. }) => {
+                columns
+                    .iter_mut()
+                    .for_each(|ColumnDef { data_type, .. }| replace_type_alias(data_type));
+            }
+            _ => {}
+        }
+
+        Ok(())
+    }
+
+    fn visit_expr(&self, expr: &mut Expr) -> ControlFlow<()> {
+        match expr {
+            // Type alias
+            Expr::Cast {
+                data_type: DataType::Custom(name, tokens),
+                expr: cast_expr,
+            } if name.0.len() == 1 && tokens.is_empty() => {
+                if let Some(new_type) = get_data_type_by_alias_name(name.0[0].value.as_str()) {
+                    if let Ok(concrete_type) = sql_data_type_to_concrete_data_type(&new_type) {
+                        let new_type = concrete_type.as_arrow_type();
+                        *expr = Expr::Function(Function {
+                            name: ObjectName(vec![Ident::new("arrow_cast")]),
+                            args: vec![
+                                FunctionArg::Unnamed(FunctionArgExpr::Expr((**cast_expr).clone())),
+                                FunctionArg::Unnamed(FunctionArgExpr::Expr(Expr::Value(
+                                    Value::SingleQuotedString(new_type.to_string()),
+                                ))),
+                            ],
+                            over: None,
+                            distinct: false,
+                            special: false,
+                            order_by: vec![],
+                        });
+                    }
+                }
+            }
+
+            // Timestamp(precision) in cast, datafusion doesn't support Timestamp(9) etc.
+            // We have to transform it into arrow_cast(expr, type).
+            Expr::Cast {
+                data_type: DataType::Timestamp(precision, zone),
+                expr: cast_expr,
+            } => {
+                if let Ok(concrete_type) =
+                    sql_data_type_to_concrete_data_type(&DataType::Timestamp(*precision, *zone))
+                {
+                    let new_type = concrete_type.as_arrow_type();
+                    *expr = Expr::Function(Function {
+                        name: ObjectName(vec![Ident::new("arrow_cast")]),
+                        args: vec![
+                            FunctionArg::Unnamed(FunctionArgExpr::Expr((**cast_expr).clone())),
+                            FunctionArg::Unnamed(FunctionArgExpr::Expr(Expr::Value(
+                                Value::SingleQuotedString(new_type.to_string()),
+                            ))),
+                        ],
+                        over: None,
+                        distinct: false,
+                        special: false,
+                        order_by: vec![],
+                    });
+                }
+            }
+
+            // TODO(dennis): supports try_cast
+            _ => {}
+        }
+
+        ControlFlow::<()>::Continue(())
+    }
+}
+
+fn replace_type_alias(data_type: &mut DataType) {
+    match data_type {
+        // TODO(dennis): The sqlparser latest version contains the Int8 alias for postres Bigint.
+        // Which means 8 bytes in postgres (not 8 bits). If we upgrade the sqlparser, need to process it.
+        // See https://docs.rs/sqlparser/latest/sqlparser/ast/enum.DataType.html#variant.Int8
+        DataType::Custom(name, tokens) if name.0.len() == 1 && tokens.is_empty() => {
+            if let Some(new_type) = get_data_type_by_alias_name(name.0[0].value.as_str()) {
+                *data_type = new_type;
+            }
+        }
+        _ => {}
+    }
+}
+
+pub fn get_data_type_by_alias_name(name: &str) -> Option<DataType> {
+    match name.to_uppercase().as_ref() {
+        // Timestamp type alias
+        "TIMESTAMP_S" | "TIMESTAMP_SEC" | "TIMESTAMPSECOND" => {
+            Some(DataType::Timestamp(Some(0), TimezoneInfo::None))
+        }
+
+        "TIMESTAMP_MS" | "TIMESTAMPMILLISECOND" => {
+            Some(DataType::Timestamp(Some(3), TimezoneInfo::None))
+        }
+        "TIMESTAMP_US" | "TIMESTAMPMICROSECOND" => {
+            Some(DataType::Timestamp(Some(6), TimezoneInfo::None))
+        }
+        "TIMESTAMP_NS" | "TIMESTAMPNANOSECOND" => {
+            Some(DataType::Timestamp(Some(9), TimezoneInfo::None))
+        }
+        // Number type alias
+        "INT8" => Some(DataType::TinyInt(None)),
+        "INT16" => Some(DataType::SmallInt(None)),
+        "INT32" => Some(DataType::Int(None)),
+        "INT64" => Some(DataType::BigInt(None)),
+        "UINT8" => Some(DataType::UnsignedTinyInt(None)),
+        "UINT16" => Some(DataType::UnsignedSmallInt(None)),
+        "UINT32" => Some(DataType::UnsignedInt(None)),
+        "UINT64" => Some(DataType::UnsignedBigInt(None)),
+        "FLOAT32" => Some(DataType::Float(None)),
+        "FLOAT64" => Some(DataType::Double),
+        _ => None,
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use sqlparser::dialect::GenericDialect;
+
+    use super::*;
+    use crate::parser::ParserContext;
+    use crate::statements::transform_statements;
+
+    #[test]
+    fn test_get_data_type_by_alias_name() {
+        assert_eq!(
+            get_data_type_by_alias_name("float64"),
+            Some(DataType::Double)
+        );
+        assert_eq!(
+            get_data_type_by_alias_name("Float64"),
+            Some(DataType::Double)
+        );
+        assert_eq!(
+            get_data_type_by_alias_name("FLOAT64"),
+            Some(DataType::Double)
+        );
+
+        assert_eq!(
+            get_data_type_by_alias_name("float32"),
+            Some(DataType::Float(None))
+        );
+        assert_eq!(
+            get_data_type_by_alias_name("int8"),
+            Some(DataType::TinyInt(None))
+        );
+        assert_eq!(
+            get_data_type_by_alias_name("INT16"),
+            Some(DataType::SmallInt(None))
+        );
+        assert_eq!(
+            get_data_type_by_alias_name("INT32"),
+            Some(DataType::Int(None))
+        );
+        assert_eq!(
+            get_data_type_by_alias_name("INT64"),
+            Some(DataType::BigInt(None))
+        );
+        assert_eq!(
+            get_data_type_by_alias_name("Uint8"),
+            Some(DataType::UnsignedTinyInt(None))
+        );
+        assert_eq!(
+            get_data_type_by_alias_name("UINT16"),
+            Some(DataType::UnsignedSmallInt(None))
+        );
+        assert_eq!(
+            get_data_type_by_alias_name("UINT32"),
+            Some(DataType::UnsignedInt(None))
+        );
+        assert_eq!(
+            get_data_type_by_alias_name("uint64"),
+            Some(DataType::UnsignedBigInt(None))
+        );
+
+        assert_eq!(
+            get_data_type_by_alias_name("TimestampSecond"),
+            Some(DataType::Timestamp(Some(0), TimezoneInfo::None))
+        );
+        assert_eq!(
+            get_data_type_by_alias_name("Timestamp_s"),
+            Some(DataType::Timestamp(Some(0), TimezoneInfo::None))
+        );
+        assert_eq!(
+            get_data_type_by_alias_name("Timestamp_sec"),
+            Some(DataType::Timestamp(Some(0), TimezoneInfo::None))
+        );
+
+        assert_eq!(
+            get_data_type_by_alias_name("TimestampMilliSecond"),
+            Some(DataType::Timestamp(Some(3), TimezoneInfo::None))
+        );
+        assert_eq!(
+            get_data_type_by_alias_name("Timestamp_ms"),
+            Some(DataType::Timestamp(Some(3), TimezoneInfo::None))
+        );
+
+        assert_eq!(
+            get_data_type_by_alias_name("TimestampMicroSecond"),
+            Some(DataType::Timestamp(Some(6), TimezoneInfo::None))
+        );
+        assert_eq!(
+            get_data_type_by_alias_name("Timestamp_us"),
+            Some(DataType::Timestamp(Some(6), TimezoneInfo::None))
+        );
+
+        assert_eq!(
+            get_data_type_by_alias_name("TimestampNanoSecond"),
+            Some(DataType::Timestamp(Some(9), TimezoneInfo::None))
+        );
+        assert_eq!(
+            get_data_type_by_alias_name("Timestamp_ns"),
+            Some(DataType::Timestamp(Some(9), TimezoneInfo::None))
+        );
+    }
+
+    fn test_timestamp_alias(alias: &str, expected: &str) {
+        let sql = format!("SELECT TIMESTAMP '2020-01-01 01:23:45.12345678'::{alias}");
+        let mut stmts = ParserContext::create_with_dialect(&sql, &GenericDialect {}).unwrap();
+        transform_statements(&mut stmts).unwrap();
+
+        match &stmts[0] {
+            Statement::Query(q) => assert_eq!(format!("SELECT arrow_cast(TIMESTAMP '2020-01-01 01:23:45.12345678', 'Timestamp({expected}, None)')"), q.to_string()),
+            _ => unreachable!(),
+        }
+    }
+
+    fn test_timestamp_precision_type(precision: i32, expected: &str) {
+        test_timestamp_alias(&format!("Timestamp({precision})"), expected);
+    }
+
+    #[test]
+    fn test_transform_timestamp_alias() {
+        // Timestamp[Second | Millisecond | Microsecond | Nanosecond]
+        test_timestamp_alias("TimestampSecond", "Second");
+        test_timestamp_alias("Timestamp_s", "Second");
+        test_timestamp_alias("TimestampMillisecond", "Millisecond");
+        test_timestamp_alias("Timestamp_ms", "Millisecond");
+        test_timestamp_alias("TimestampMicrosecond", "Microsecond");
+        test_timestamp_alias("Timestamp_us", "Microsecond");
+        test_timestamp_alias("TimestampNanosecond", "Nanosecond");
+        test_timestamp_alias("Timestamp_ns", "Nanosecond");
+        // Timestamp(precision)
+        test_timestamp_precision_type(0, "Second");
+        test_timestamp_precision_type(3, "Millisecond");
+        test_timestamp_precision_type(6, "Microsecond");
+        test_timestamp_precision_type(9, "Nanosecond");
+    }
+
+    #[test]
+    fn test_create_sql_with_type_alias() {
+        let sql = r#"
+CREATE TABLE data_types (
+  s string,
+  tint int8,
+  sint int16,
+  i int32,
+  bint int64,
+  v varchar,
+  f float32,
+  d float64,
+  b boolean,
+  vb varbinary,
+  dt date,
+  dtt datetime,
+  ts0 TimestampSecond,
+  ts3 TimestampMillisecond,
+  ts6 TimestampMicrosecond,
+  ts9 TimestampNanosecond DEFAULT CURRENT_TIMESTAMP TIME INDEX,
+  PRIMARY KEY(s));"#;
+
+        let mut stmts = ParserContext::create_with_dialect(sql, &GenericDialect {}).unwrap();
+        transform_statements(&mut stmts).unwrap();
+
+        match &stmts[0] {
+            Statement::CreateTable(c) => {
+                let expected = r#"CREATE TABLE  data_types (
+  s STRING,
+  tint TINYINT,
+  sint SMALLINT,
+  i INT,
+  bint BIGINT,
+  v VARCHAR,
+  f FLOAT,
+  d DOUBLE,
+  b BOOLEAN,
+  vb VARBINARY,
+  dt DATE,
+  dtt DATETIME,
+  ts0 TIMESTAMP(0),
+  ts3 TIMESTAMP(3),
+  ts6 TIMESTAMP(6),
+  ts9 TIMESTAMP(9) DEFAULT CURRENT_TIMESTAMP() NOT NULL,
+  TIME INDEX (ts9),
+  PRIMARY KEY (s)
+)
+ENGINE=mito
+"#;
+
+                assert_eq!(expected, c.to_string());
+            }
+            _ => unreachable!(),
+        }
+    }
+}
diff --git a/src/sql/src/statements/truncate.rs b/src/sql/src/statements/truncate.rs
index e7eae4e185..aa08cde559 100644
--- a/src/sql/src/statements/truncate.rs
+++ b/src/sql/src/statements/truncate.rs
@@ -13,9 +13,10 @@
 // limitations under the License.
 
 use sqlparser::ast::ObjectName;
+use sqlparser_derive::{Visit, VisitMut};
 
 /// TRUNCATE TABLE statement.
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq, Visit, VisitMut)]
 pub struct TruncateTable {
     table_name: ObjectName,
 }
diff --git a/tests/cases/standalone/common/create/create_type_alias.result b/tests/cases/standalone/common/create/create_type_alias.result
new file mode 100644
index 0000000000..6e192a4e5f
--- /dev/null
+++ b/tests/cases/standalone/common/create/create_type_alias.result
@@ -0,0 +1,80 @@
+CREATE TABLE data_types (
+  s string,
+  tint int8,
+  sint int16,
+  i INT32,
+  bint INT64,
+  v varchar,
+  f FLOAT32,
+  d FLOAT64,
+  b boolean,
+  vb varbinary,
+  dt date,
+  dtt datetime,
+  ts0 TimestampSecond,
+  ts3 Timestamp_MS,
+  ts6 Timestamp_US,
+  ts9 TimestampNanosecond DEFAULT CURRENT_TIMESTAMP TIME INDEX,
+  PRIMARY KEY(s));
+
+Affected Rows: 0
+
+SHOW CREATE TABLE data_types;
+
++------------+------------------------------------------------------------+
+| Table      | Create Table                                               |
++------------+------------------------------------------------------------+
+| data_types | CREATE TABLE IF NOT EXISTS "data_types" (                  |
+|            |   "s" STRING NULL,                                         |
+|            |   "tint" TINYINT NULL,                                     |
+|            |   "sint" SMALLINT NULL,                                    |
+|            |   "i" INT NULL,                                            |
+|            |   "bint" BIGINT NULL,                                      |
+|            |   "v" STRING NULL,                                         |
+|            |   "f" FLOAT NULL,                                          |
+|            |   "d" DOUBLE NULL,                                         |
+|            |   "b" BOOLEAN NULL,                                        |
+|            |   "vb" VARBINARY NULL,                                     |
+|            |   "dt" DATE NULL,                                          |
+|            |   "dtt" DATETIME NULL,                                     |
+|            |   "ts0" TIMESTAMP(0) NULL,                                 |
+|            |   "ts3" TIMESTAMP(3) NULL,                                 |
+|            |   "ts6" TIMESTAMP(6) NULL,                                 |
+|            |   "ts9" TIMESTAMP(9) NOT NULL DEFAULT current_timestamp(), |
+|            |   TIME INDEX ("ts9"),                                      |
+|            |   PRIMARY KEY ("s")                                        |
+|            | )                                                          |
+|            |                                                            |
+|            | ENGINE=mito                                                |
+|            | WITH(                                                      |
+|            |   regions = 1                                              |
+|            | )                                                          |
++------------+------------------------------------------------------------+
+
+DESC TABLE data_types;
+
++--------+----------------------+-----+------+---------------------+---------------+
+| Column | Type                 | Key | Null | Default             | Semantic Type |
++--------+----------------------+-----+------+---------------------+---------------+
+| s      | String               | PRI | YES  |                     | TAG           |
+| tint   | Int8                 |     | YES  |                     | FIELD         |
+| sint   | Int16                |     | YES  |                     | FIELD         |
+| i      | Int32                |     | YES  |                     | FIELD         |
+| bint   | Int64                |     | YES  |                     | FIELD         |
+| v      | String               |     | YES  |                     | FIELD         |
+| f      | Float32              |     | YES  |                     | FIELD         |
+| d      | Float64              |     | YES  |                     | FIELD         |
+| b      | Boolean              |     | YES  |                     | FIELD         |
+| vb     | Binary               |     | YES  |                     | FIELD         |
+| dt     | Date                 |     | YES  |                     | FIELD         |
+| dtt    | DateTime             |     | YES  |                     | FIELD         |
+| ts0    | TimestampSecond      |     | YES  |                     | FIELD         |
+| ts3    | TimestampMillisecond |     | YES  |                     | FIELD         |
+| ts6    | TimestampMicrosecond |     | YES  |                     | FIELD         |
+| ts9    | TimestampNanosecond  | PRI | NO   | current_timestamp() | TIMESTAMP     |
++--------+----------------------+-----+------+---------------------+---------------+
+
+DROP TABLE data_types;
+
+Affected Rows: 1
+
diff --git a/tests/cases/standalone/common/create/create_type_alias.sql b/tests/cases/standalone/common/create/create_type_alias.sql
new file mode 100644
index 0000000000..e4ef338245
--- /dev/null
+++ b/tests/cases/standalone/common/create/create_type_alias.sql
@@ -0,0 +1,24 @@
+CREATE TABLE data_types (
+  s string,
+  tint int8,
+  sint int16,
+  i INT32,
+  bint INT64,
+  v varchar,
+  f FLOAT32,
+  d FLOAT64,
+  b boolean,
+  vb varbinary,
+  dt date,
+  dtt datetime,
+  ts0 TimestampSecond,
+  ts3 Timestamp_MS,
+  ts6 Timestamp_US,
+  ts9 TimestampNanosecond DEFAULT CURRENT_TIMESTAMP TIME INDEX,
+  PRIMARY KEY(s));
+
+SHOW CREATE TABLE data_types;
+
+DESC TABLE data_types;
+
+DROP TABLE data_types;
diff --git a/tests/cases/standalone/common/delete/delete.result b/tests/cases/standalone/common/delete/delete.result
index 4c517017d1..0a593f3f5c 100644
--- a/tests/cases/standalone/common/delete/delete.result
+++ b/tests/cases/standalone/common/delete/delete.result
@@ -31,7 +31,7 @@ SELECT ts, host, cpu, memory FROM monitor ORDER BY ts;
 | 2022-06-15T07:02:39 | host3 | 88.8 | 4096.0 |
 +---------------------+-------+------+--------+
 
-DELETE FROM monitor WHERE host = 'host1' AND ts = 1655276557000000000::timestamp;
+DELETE FROM monitor WHERE host = 'host1' AND ts = 1655276557000::timestamp;
 
 Affected Rows: 1
 
diff --git a/tests/cases/standalone/common/delete/delete.sql b/tests/cases/standalone/common/delete/delete.sql
index 8cb9a2559c..f59eb73df8 100644
--- a/tests/cases/standalone/common/delete/delete.sql
+++ b/tests/cases/standalone/common/delete/delete.sql
@@ -13,7 +13,7 @@ INSERT INTO monitor(ts, host, cpu, memory) VALUES
 
 SELECT ts, host, cpu, memory FROM monitor ORDER BY ts;
 
-DELETE FROM monitor WHERE host = 'host1' AND ts = 1655276557000000000::timestamp;
+DELETE FROM monitor WHERE host = 'host1' AND ts = 1655276557000::timestamp;
 
 DELETE FROM monitor WHERE host = 'host2';
 
diff --git a/tests/cases/standalone/common/types/timestamp/timestamp_precision.result b/tests/cases/standalone/common/types/timestamp/timestamp_precision.result
index 441c9192bc..6c68770200 100644
--- a/tests/cases/standalone/common/types/timestamp/timestamp_precision.result
+++ b/tests/cases/standalone/common/types/timestamp/timestamp_precision.result
@@ -29,31 +29,59 @@ Error: 1003(Internal), Execution error: Date part 'MICROSECONDS' not supported
 -- any other precision is rounded up (e.g. 1/2 -> 3, 4/5 -> 6, 7/8 -> 9)
 SELECT TIMESTAMP '2020-01-01 01:23:45.123456789'::TIMESTAMP(0);
 
-Error: 3000(PlanQuery), This feature is not implemented: Unsupported SQL type Timestamp(Some(0), None)
++---------------------------------------+
+| Utf8("2020-01-01 01:23:45.123456789") |
++---------------------------------------+
+| 2020-01-01T01:23:45                   |
++---------------------------------------+
 
 SELECT TIMESTAMP '2020-01-01 01:23:45.123456789'::TIMESTAMP(3);
 
-Error: 3000(PlanQuery), This feature is not implemented: Unsupported SQL type Timestamp(Some(3), None)
++---------------------------------------+
+| Utf8("2020-01-01 01:23:45.123456789") |
++---------------------------------------+
+| 2020-01-01T01:23:45.123               |
++---------------------------------------+
 
 SELECT TIMESTAMP '2020-01-01 01:23:45.123456789'::TIMESTAMP(6);
 
-Error: 3000(PlanQuery), This feature is not implemented: Unsupported SQL type Timestamp(Some(6), None)
++---------------------------------------+
+| Utf8("2020-01-01 01:23:45.123456789") |
++---------------------------------------+
+| 2020-01-01T01:23:45.123456            |
++---------------------------------------+
 
 SELECT TIMESTAMP '2020-01-01 01:23:45.123456789'::TIMESTAMP(9);
 
-Error: 3000(PlanQuery), This feature is not implemented: Unsupported SQL type Timestamp(Some(9), None)
++---------------------------------------+
+| Utf8("2020-01-01 01:23:45.123456789") |
++---------------------------------------+
+| 2020-01-01T01:23:45.123456789         |
++---------------------------------------+
 
 SELECT TIMESTAMP '2020-01-01 01:23:45.12'::TIMESTAMP(3);
 
-Error: 3000(PlanQuery), This feature is not implemented: Unsupported SQL type Timestamp(Some(3), None)
++--------------------------------+
+| Utf8("2020-01-01 01:23:45.12") |
++--------------------------------+
+| 2020-01-01T01:23:45.120        |
++--------------------------------+
 
 SELECT TIMESTAMP '2020-01-01 01:23:45.12345'::TIMESTAMP(6);
 
-Error: 3000(PlanQuery), This feature is not implemented: Unsupported SQL type Timestamp(Some(6), None)
++-----------------------------------+
+| Utf8("2020-01-01 01:23:45.12345") |
++-----------------------------------+
+| 2020-01-01T01:23:45.123450        |
++-----------------------------------+
 
 SELECT TIMESTAMP '2020-01-01 01:23:45.12345678'::TIMESTAMP(9);
 
-Error: 3000(PlanQuery), This feature is not implemented: Unsupported SQL type Timestamp(Some(9), None)
++--------------------------------------+
+| Utf8("2020-01-01 01:23:45.12345678") |
++--------------------------------------+
+| 2020-01-01T01:23:45.123456780        |
++--------------------------------------+
 
 DROP TABLE ts_precision;
 
diff --git a/tests/cases/standalone/common/types/timestamp/timestamp_types.result b/tests/cases/standalone/common/types/timestamp/timestamp_types.result
new file mode 100644
index 0000000000..349caedcb0
--- /dev/null
+++ b/tests/cases/standalone/common/types/timestamp/timestamp_types.result
@@ -0,0 +1,380 @@
+--description: Test TIMESTAMP types
+CREATE TABLE IF NOT EXISTS timestamp (sec TIMESTAMP_S, milli TIMESTAMP_MS,micro TIMESTAMP_US, nano TIMESTAMP_NS, ts TIMESTAMP TIME INDEX);
+
+Affected Rows: 0
+
+INSERT INTO timestamp VALUES ('2008-01-01 00:00:01','2008-01-01 00:00:01.594','2008-01-01 00:00:01.88926','2008-01-01 00:00:01.889268321', 1);
+
+Affected Rows: 1
+
+SELECT * from timestamp;
+
++---------------------+-------------------------+----------------------------+-------------------------------+-------------------------+
+| sec                 | milli                   | micro                      | nano                          | ts                      |
++---------------------+-------------------------+----------------------------+-------------------------------+-------------------------+
+| 2008-01-01T00:00:01 | 2008-01-01T00:00:01.594 | 2008-01-01T00:00:01.889260 | 2008-01-01T00:00:01.889268321 | 1970-01-01T00:00:00.001 |
++---------------------+-------------------------+----------------------------+-------------------------------+-------------------------+
+
+SELECT extract(YEAR from sec),extract( YEAR from milli),extract(YEAR from nano) from timestamp;
+
++---------------------------------------+-----------------------------------------+----------------------------------------+
+| date_part(Utf8("YEAR"),timestamp.sec) | date_part(Utf8("YEAR"),timestamp.milli) | date_part(Utf8("YEAR"),timestamp.nano) |
++---------------------------------------+-----------------------------------------+----------------------------------------+
+| 2008.0                                | 2008.0                                  | 2008.0                                 |
++---------------------------------------+-----------------------------------------+----------------------------------------+
+
+SELECT nano::TIMESTAMP, milli::TIMESTAMP,sec::TIMESTAMP from timestamp;
+
++-------------------------+-------------------------+---------------------+
+| timestamp.nano          | milli                   | timestamp.sec       |
++-------------------------+-------------------------+---------------------+
+| 2008-01-01T00:00:01.889 | 2008-01-01T00:00:01.594 | 2008-01-01T00:00:01 |
++-------------------------+-------------------------+---------------------+
+
+SELECT micro::TIMESTAMP_S as m1, micro::TIMESTAMP_MS as m2,micro::TIMESTAMP_NS as m3 from timestamp;
+
++---------------------+-------------------------+----------------------------+
+| m1                  | m2                      | m3                         |
++---------------------+-------------------------+----------------------------+
+| 2008-01-01T00:00:01 | 2008-01-01T00:00:01.889 | 2008-01-01T00:00:01.889260 |
++---------------------+-------------------------+----------------------------+
+
+INSERT INTO timestamp VALUES ('2008-01-01 00:00:51','2008-01-01 00:00:01.894','2008-01-01 00:00:01.99926','2008-01-01 00:00:01.999268321', 2);
+
+Affected Rows: 1
+
+INSERT INTO timestamp VALUES ('2008-01-01 00:00:11','2008-01-01 00:00:01.794','2008-01-01 00:00:01.98926','2008-01-01 00:00:01.899268321', 3);
+
+Affected Rows: 1
+
+select '90000-01-19 03:14:07.999999'::TIMESTAMP_US::TIMESTAMP_NS;
+
+Error: 3001(EngineExecuteQuery), Parser error: Error parsing timestamp from '90000-01-19 03:14:07.999999': error parsing date
+
+select sec::DATE from timestamp;
+
++---------------+
+| timestamp.sec |
++---------------+
+| 2008-01-01    |
+| 2008-01-01    |
+| 2008-01-01    |
++---------------+
+
+select milli::DATE from timestamp;
+
++-----------------+
+| timestamp.milli |
++-----------------+
+| 2008-01-01      |
+| 2008-01-01      |
+| 2008-01-01      |
++-----------------+
+
+select nano::DATE from timestamp;
+
++----------------+
+| timestamp.nano |
++----------------+
+| 2008-01-01     |
+| 2008-01-01     |
+| 2008-01-01     |
++----------------+
+
+select sec::TIME from timestamp;
+
++---------------+
+| timestamp.sec |
++---------------+
+| 00:00:01      |
+| 00:00:51      |
+| 00:00:11      |
++---------------+
+
+select milli::TIME from timestamp;
+
++-----------------+
+| timestamp.milli |
++-----------------+
+| 00:00:01.594    |
+| 00:00:01.894    |
+| 00:00:01.794    |
++-----------------+
+
+select nano::TIME from timestamp;
+
++--------------------+
+| timestamp.nano     |
++--------------------+
+| 00:00:01.889268321 |
+| 00:00:01.999268321 |
+| 00:00:01.899268321 |
++--------------------+
+
+select sec::TIMESTAMP_MS from timestamp;
+
++---------------------+
+| timestamp.sec       |
++---------------------+
+| 2008-01-01T00:00:01 |
+| 2008-01-01T00:00:51 |
+| 2008-01-01T00:00:11 |
++---------------------+
+
+select sec::TIMESTAMP_NS from timestamp;
+
++---------------------+
+| timestamp.sec       |
++---------------------+
+| 2008-01-01T00:00:01 |
+| 2008-01-01T00:00:51 |
+| 2008-01-01T00:00:11 |
++---------------------+
+
+select milli::TIMESTAMP_SEC from timestamp;
+
++---------------------+
+| timestamp.milli     |
++---------------------+
+| 2008-01-01T00:00:01 |
+| 2008-01-01T00:00:01 |
+| 2008-01-01T00:00:01 |
++---------------------+
+
+select milli::TIMESTAMP_NS from timestamp;
+
++-------------------------+
+| timestamp.milli         |
++-------------------------+
+| 2008-01-01T00:00:01.594 |
+| 2008-01-01T00:00:01.894 |
+| 2008-01-01T00:00:01.794 |
++-------------------------+
+
+select nano::TIMESTAMP_SEC from timestamp;
+
++---------------------+
+| timestamp.nano      |
++---------------------+
+| 2008-01-01T00:00:01 |
+| 2008-01-01T00:00:01 |
+| 2008-01-01T00:00:01 |
++---------------------+
+
+select nano::TIMESTAMP_MS from timestamp;
+
++-------------------------+
+| timestamp.nano          |
++-------------------------+
+| 2008-01-01T00:00:01.889 |
+| 2008-01-01T00:00:01.999 |
+| 2008-01-01T00:00:01.899 |
++-------------------------+
+
+select sec from timestamp order by sec;
+
++---------------------+
+| sec                 |
++---------------------+
+| 2008-01-01T00:00:01 |
+| 2008-01-01T00:00:11 |
+| 2008-01-01T00:00:51 |
++---------------------+
+
+select milli from timestamp order by milli;
+
++-------------------------+
+| milli                   |
++-------------------------+
+| 2008-01-01T00:00:01.594 |
+| 2008-01-01T00:00:01.794 |
+| 2008-01-01T00:00:01.894 |
++-------------------------+
+
+select nano from timestamp order by nano;
+
++-------------------------------+
+| nano                          |
++-------------------------------+
+| 2008-01-01T00:00:01.889268321 |
+| 2008-01-01T00:00:01.899268321 |
+| 2008-01-01T00:00:01.999268321 |
++-------------------------------+
+
+INSERT INTO timestamp VALUES ('2008-01-01 00:00:51','2008-01-01 00:00:01.894','2008-01-01 00:00:01.99926','2008-01-01 00:00:01.999268321', 4);
+
+Affected Rows: 1
+
+INSERT INTO timestamp VALUES ('2008-01-01 00:00:11','2008-01-01 00:00:01.794','2008-01-01 00:00:01.98926','2008-01-01 00:00:01.899268321', 5);
+
+Affected Rows: 1
+
+select count(*), nano from timestamp group by nano order by nano;
+
++-----------------+-------------------------------+
+| COUNT(UInt8(1)) | nano                          |
++-----------------+-------------------------------+
+| 1               | 2008-01-01T00:00:01.889268321 |
+| 2               | 2008-01-01T00:00:01.899268321 |
+| 2               | 2008-01-01T00:00:01.999268321 |
++-----------------+-------------------------------+
+
+select count(*), sec from timestamp group by sec order by sec;
+
++-----------------+---------------------+
+| COUNT(UInt8(1)) | sec                 |
++-----------------+---------------------+
+| 1               | 2008-01-01T00:00:01 |
+| 2               | 2008-01-01T00:00:11 |
+| 2               | 2008-01-01T00:00:51 |
++-----------------+---------------------+
+
+select count(*), milli from timestamp group by milli order by milli;
+
++-----------------+-------------------------+
+| COUNT(UInt8(1)) | milli                   |
++-----------------+-------------------------+
+| 1               | 2008-01-01T00:00:01.594 |
+| 2               | 2008-01-01T00:00:01.794 |
+| 2               | 2008-01-01T00:00:01.894 |
++-----------------+-------------------------+
+
+CREATE TABLE IF NOT EXISTS timestamp_two (sec TIMESTAMP_S, milli TIMESTAMP_MS,micro TIMESTAMP_US, nano TIMESTAMP_NS TIME INDEX);
+
+Affected Rows: 0
+
+INSERT INTO timestamp_two VALUES ('2008-01-01 00:00:11','2008-01-01 00:00:01.794','2008-01-01 00:00:01.98926','2008-01-01 00:00:01.899268321');
+
+Affected Rows: 1
+
+select timestamp.sec from timestamp inner join  timestamp_two on (timestamp.sec = timestamp_two.sec);
+
++---------------------+
+| sec                 |
++---------------------+
+| 2008-01-01T00:00:11 |
+| 2008-01-01T00:00:11 |
++---------------------+
+
+select timestamp.milli from timestamp inner join  timestamp_two on (timestamp.milli = timestamp_two.milli);
+
++-------------------------+
+| milli                   |
++-------------------------+
+| 2008-01-01T00:00:01.794 |
+| 2008-01-01T00:00:01.794 |
++-------------------------+
+
+select timestamp.nano from timestamp inner join  timestamp_two on (timestamp.nano = timestamp_two.nano);
+
++-------------------------------+
+| nano                          |
++-------------------------------+
+| 2008-01-01T00:00:01.899268321 |
+| 2008-01-01T00:00:01.899268321 |
++-------------------------------+
+
+select '2008-01-01 00:00:11'::TIMESTAMP_US = '2008-01-01 00:00:11'::TIMESTAMP_MS;
+
++-----------------------------------------------------------+
+| Utf8("2008-01-01 00:00:11") = Utf8("2008-01-01 00:00:11") |
++-----------------------------------------------------------+
+| true                                                      |
++-----------------------------------------------------------+
+
+select '2008-01-01 00:00:11'::TIMESTAMP_US = '2008-01-01 00:00:11'::TIMESTAMP_NS;
+
++-----------------------------------------------------------+
+| Utf8("2008-01-01 00:00:11") = Utf8("2008-01-01 00:00:11") |
++-----------------------------------------------------------+
+| true                                                      |
++-----------------------------------------------------------+
+
+select '2008-01-01 00:00:11'::TIMESTAMP_US = '2008-01-01 00:00:11'::TIMESTAMP_S;
+
++-----------------------------------------------------------+
+| Utf8("2008-01-01 00:00:11") = Utf8("2008-01-01 00:00:11") |
++-----------------------------------------------------------+
+| true                                                      |
++-----------------------------------------------------------+
+
+select '2008-01-01 00:00:11.1'::TIMESTAMP_US = '2008-01-01 00:00:11'::TIMESTAMP_MS;
+
++-------------------------------------------------------------+
+| Utf8("2008-01-01 00:00:11.1") = Utf8("2008-01-01 00:00:11") |
++-------------------------------------------------------------+
+| false                                                       |
++-------------------------------------------------------------+
+
+select '2008-01-01 00:00:11.1'::TIMESTAMP_US = '2008-01-01 00:00:11'::TIMESTAMP_NS;
+
++-------------------------------------------------------------+
+| Utf8("2008-01-01 00:00:11.1") = Utf8("2008-01-01 00:00:11") |
++-------------------------------------------------------------+
+| false                                                       |
++-------------------------------------------------------------+
+
+select '2008-01-01 00:00:11.1'::TIMESTAMP_US = '2008-01-01 00:00:11.1'::TIMESTAMP_S;
+
++---------------------------------------------------------------+
+| Utf8("2008-01-01 00:00:11.1") = Utf8("2008-01-01 00:00:11.1") |
++---------------------------------------------------------------+
+| true                                                          |
++---------------------------------------------------------------+
+
+select '2008-01-01 00:00:11.1'::TIMESTAMP_MS = '2008-01-01 00:00:11'::TIMESTAMP_NS;
+
++-------------------------------------------------------------+
+| Utf8("2008-01-01 00:00:11.1") = Utf8("2008-01-01 00:00:11") |
++-------------------------------------------------------------+
+| false                                                       |
++-------------------------------------------------------------+
+
+select '2008-01-01 00:00:11.1'::TIMESTAMP_MS = '2008-01-01 00:00:11'::TIMESTAMP_S;
+
++-------------------------------------------------------------+
+| Utf8("2008-01-01 00:00:11.1") = Utf8("2008-01-01 00:00:11") |
++-------------------------------------------------------------+
+| true                                                        |
++-------------------------------------------------------------+
+
+select '2008-01-01 00:00:11.1'::TIMESTAMP_NS = '2008-01-01 00:00:11'::TIMESTAMP_S;
+
++-------------------------------------------------------------+
+| Utf8("2008-01-01 00:00:11.1") = Utf8("2008-01-01 00:00:11") |
++-------------------------------------------------------------+
+| true                                                        |
++-------------------------------------------------------------+
+
+select '2008-01-01 00:00:11'::TIMESTAMP_MS = '2008-01-01 00:00:11'::TIMESTAMP_NS;
+
++-----------------------------------------------------------+
+| Utf8("2008-01-01 00:00:11") = Utf8("2008-01-01 00:00:11") |
++-----------------------------------------------------------+
+| true                                                      |
++-----------------------------------------------------------+
+
+select '2008-01-01 00:00:11'::TIMESTAMP_MS = '2008-01-01 00:00:11'::TIMESTAMP_S;
+
++-----------------------------------------------------------+
+| Utf8("2008-01-01 00:00:11") = Utf8("2008-01-01 00:00:11") |
++-----------------------------------------------------------+
+| true                                                      |
++-----------------------------------------------------------+
+
+select '2008-01-01 00:00:11'::TIMESTAMP_NS = '2008-01-01 00:00:11'::TIMESTAMP_S;
+
++-----------------------------------------------------------+
+| Utf8("2008-01-01 00:00:11") = Utf8("2008-01-01 00:00:11") |
++-----------------------------------------------------------+
+| true                                                      |
++-----------------------------------------------------------+
+
+DROP TABLE timestamp;
+
+Affected Rows: 1
+
+DROP TABLE timestamp_two;
+
+Affected Rows: 1
+
diff --git a/tests/cases/standalone/common/types/timestamp/timestamp_types.sql b/tests/cases/standalone/common/types/timestamp/timestamp_types.sql
new file mode 100644
index 0000000000..6d73a453d1
--- /dev/null
+++ b/tests/cases/standalone/common/types/timestamp/timestamp_types.sql
@@ -0,0 +1,99 @@
+--description: Test TIMESTAMP types
+
+CREATE TABLE IF NOT EXISTS timestamp (sec TIMESTAMP_S, milli TIMESTAMP_MS,micro TIMESTAMP_US, nano TIMESTAMP_NS, ts TIMESTAMP TIME INDEX);
+
+INSERT INTO timestamp VALUES ('2008-01-01 00:00:01','2008-01-01 00:00:01.594','2008-01-01 00:00:01.88926','2008-01-01 00:00:01.889268321', 1);
+
+SELECT * from timestamp;
+
+SELECT extract(YEAR from sec),extract( YEAR from milli),extract(YEAR from nano) from timestamp;
+
+SELECT nano::TIMESTAMP, milli::TIMESTAMP,sec::TIMESTAMP from timestamp;
+
+SELECT micro::TIMESTAMP_S as m1, micro::TIMESTAMP_MS as m2,micro::TIMESTAMP_NS as m3 from timestamp;
+
+
+INSERT INTO timestamp VALUES ('2008-01-01 00:00:51','2008-01-01 00:00:01.894','2008-01-01 00:00:01.99926','2008-01-01 00:00:01.999268321', 2);
+
+INSERT INTO timestamp VALUES ('2008-01-01 00:00:11','2008-01-01 00:00:01.794','2008-01-01 00:00:01.98926','2008-01-01 00:00:01.899268321', 3);
+
+
+select '90000-01-19 03:14:07.999999'::TIMESTAMP_US::TIMESTAMP_NS;
+
+select sec::DATE from timestamp;
+
+select milli::DATE from timestamp;
+
+select nano::DATE from timestamp;
+
+select sec::TIME from timestamp;
+
+select milli::TIME from timestamp;
+
+select nano::TIME from timestamp;
+
+select sec::TIMESTAMP_MS from timestamp;
+
+select sec::TIMESTAMP_NS from timestamp;
+
+select milli::TIMESTAMP_SEC from timestamp;
+
+select milli::TIMESTAMP_NS from timestamp;
+
+select nano::TIMESTAMP_SEC from timestamp;
+
+select nano::TIMESTAMP_MS from timestamp;
+
+select sec from timestamp order by sec;
+
+select milli from timestamp order by milli;
+
+select nano from timestamp order by nano;
+
+INSERT INTO timestamp VALUES ('2008-01-01 00:00:51','2008-01-01 00:00:01.894','2008-01-01 00:00:01.99926','2008-01-01 00:00:01.999268321', 4);
+
+INSERT INTO timestamp VALUES ('2008-01-01 00:00:11','2008-01-01 00:00:01.794','2008-01-01 00:00:01.98926','2008-01-01 00:00:01.899268321', 5);
+
+select count(*), nano from timestamp group by nano order by nano;
+
+select count(*), sec from timestamp group by sec order by sec;
+
+select count(*), milli from timestamp group by milli order by milli;
+
+CREATE TABLE IF NOT EXISTS timestamp_two (sec TIMESTAMP_S, milli TIMESTAMP_MS,micro TIMESTAMP_US, nano TIMESTAMP_NS TIME INDEX);
+
+INSERT INTO timestamp_two VALUES ('2008-01-01 00:00:11','2008-01-01 00:00:01.794','2008-01-01 00:00:01.98926','2008-01-01 00:00:01.899268321');
+
+select timestamp.sec from timestamp inner join  timestamp_two on (timestamp.sec = timestamp_two.sec);
+
+select timestamp.milli from timestamp inner join  timestamp_two on (timestamp.milli = timestamp_two.milli);
+
+select timestamp.nano from timestamp inner join  timestamp_two on (timestamp.nano = timestamp_two.nano);
+
+select '2008-01-01 00:00:11'::TIMESTAMP_US = '2008-01-01 00:00:11'::TIMESTAMP_MS;
+
+select '2008-01-01 00:00:11'::TIMESTAMP_US = '2008-01-01 00:00:11'::TIMESTAMP_NS;
+
+select '2008-01-01 00:00:11'::TIMESTAMP_US = '2008-01-01 00:00:11'::TIMESTAMP_S;
+
+select '2008-01-01 00:00:11.1'::TIMESTAMP_US = '2008-01-01 00:00:11'::TIMESTAMP_MS;
+
+select '2008-01-01 00:00:11.1'::TIMESTAMP_US = '2008-01-01 00:00:11'::TIMESTAMP_NS;
+
+select '2008-01-01 00:00:11.1'::TIMESTAMP_US = '2008-01-01 00:00:11.1'::TIMESTAMP_S;
+
+select '2008-01-01 00:00:11.1'::TIMESTAMP_MS = '2008-01-01 00:00:11'::TIMESTAMP_NS;
+
+select '2008-01-01 00:00:11.1'::TIMESTAMP_MS = '2008-01-01 00:00:11'::TIMESTAMP_S;
+
+select '2008-01-01 00:00:11.1'::TIMESTAMP_NS = '2008-01-01 00:00:11'::TIMESTAMP_S;
+
+select '2008-01-01 00:00:11'::TIMESTAMP_MS = '2008-01-01 00:00:11'::TIMESTAMP_NS;
+
+select '2008-01-01 00:00:11'::TIMESTAMP_MS = '2008-01-01 00:00:11'::TIMESTAMP_S;
+
+select '2008-01-01 00:00:11'::TIMESTAMP_NS = '2008-01-01 00:00:11'::TIMESTAMP_S;
+
+DROP TABLE timestamp;
+
+DROP TABLE timestamp_two;
diff --git a/tests/cases/standalone/common/types/timestamp/timestamp_tz.result b/tests/cases/standalone/common/types/timestamp/timestamp_tz.result
index f863d1e02a..8084c46457 100644
--- a/tests/cases/standalone/common/types/timestamp/timestamp_tz.result
+++ b/tests/cases/standalone/common/types/timestamp/timestamp_tz.result
@@ -11,7 +11,7 @@ select '2021-11-15 02:30:00'::TIMESTAMP::TIMESTAMPTZ;
 +-----------------------------+
 | Utf8("2021-11-15 02:30:00") |
 +-----------------------------+
-| 2021-11-15T02:30:00Z        |
+| 2021-11-15T02:30:00         |
 +-----------------------------+
 
 SELECT '2021-04-29 10:50:09-05'::TIMESTAMPTZ::DATE;