diff --git a/src/datanode/src/sql/create.rs b/src/datanode/src/sql/create.rs
index 4c29d3c2e4b..ced3ef89ac4 100644
--- a/src/datanode/src/sql/create.rs
+++ b/src/datanode/src/sql/create.rs
@@ -349,7 +349,12 @@ mod tests {
 
     #[tokio::test]
     async fn test_create_table_with_options() {
-        let sql = r#"CREATE TABLE demo_table (timestamp BIGINT TIME INDEX, value DOUBLE, host STRING PRIMARY KEY) engine=mito with(regions=1, ttl='7days',write_buffer_size='32MB',some='other');"#;
+        let sql = r#"
+            CREATE TABLE demo_table (
+                "timestamp" BIGINT TIME INDEX, 
+                "value" DOUBLE,
+                host STRING PRIMARY KEY
+            ) engine=mito with(regions=1, ttl='7days',write_buffer_size='32MB',some='other');"#;
         let parsed_stmt = sql_to_statement(sql);
         let handler = create_mock_sql_handler().await;
         let c = handler
@@ -368,7 +373,12 @@ mod tests {
     pub async fn test_create_with_inline_primary_key() {
         let handler = create_mock_sql_handler().await;
         let parsed_stmt = sql_to_statement(
-            r#"CREATE TABLE demo_table (timestamp BIGINT TIME INDEX, value DOUBLE, host STRING PRIMARY KEY) engine=mito with(regions=1);"#,
+            r#"
+            CREATE TABLE demo_table(
+                "timestamp" BIGINT TIME INDEX, 
+                "value" DOUBLE,
+                host STRING PRIMARY KEY
+            ) engine=mito with(regions=1);"#,
         );
         let c = handler
             .create_to_request(42, parsed_stmt, &TableReference::bare("demo_table"))
@@ -407,8 +417,8 @@ mod tests {
         let handler = create_mock_sql_handler().await;
         let parsed_stmt = sql_to_statement(
             r#"create table demo_table (
-                      timestamp BIGINT TIME INDEX,
-                      value DOUBLE,
+                      "timestamp" BIGINT TIME INDEX,
+                      "value" DOUBLE,
                       host STRING PRIMARY KEY,
                       PRIMARY KEY(host)) engine=mito with(regions=1);"#,
         );
@@ -423,8 +433,8 @@ mod tests {
         let handler = create_mock_sql_handler().await;
         let parsed_stmt = sql_to_statement(
             r#"create table demo_table (
-                      timestamp BIGINT TIME INDEX,
-                      value DOUBLE PRIMARY KEY,
+                      "timestamp" BIGINT TIME INDEX,
+                      "value" DOUBLE PRIMARY KEY,
                       host STRING PRIMARY KEY) engine=mito with(regions=1);"#,
         );
         let error = handler
diff --git a/src/frontend/src/tests/promql_test.rs b/src/frontend/src/tests/promql_test.rs
index ab359de37ab..34db1c28a37 100644
--- a/src/frontend/src/tests/promql_test.rs
+++ b/src/frontend/src/tests/promql_test.rs
@@ -176,8 +176,8 @@ async fn sql_insert_promql_query_ceil(instance: Arc<dyn MockInstance>) {
 const AGGREGATORS_CREATE_TABLE: &str = r#"create table http_requests (
     job string,
     instance string,
-    group string,
-    value double,
+    "group" string,
+    "value" double,
     ts timestamp TIME INDEX,
     PRIMARY KEY (job, instance, group),
 );"#;
diff --git a/src/sql/src/parsers/create_parser.rs b/src/sql/src/parsers/create_parser.rs
index 8ed3361b6d3..eddd23e1caf 100644
--- a/src/sql/src/parsers/create_parser.rs
+++ b/src/sql/src/parsers/create_parser.rs
@@ -20,6 +20,7 @@ use once_cell::sync::Lazy;
 use snafu::{ensure, OptionExt, ResultExt};
 use sqlparser::ast::{ColumnOption, ColumnOptionDef, DataType, Value};
 use sqlparser::dialect::keywords::Keyword;
+use sqlparser::keywords::ALL_KEYWORDS;
 use sqlparser::parser::IsOptional::Mandatory;
 use sqlparser::parser::{Parser, ParserError};
 use sqlparser::tokenizer::{Token, TokenWithLocation, Word};
@@ -386,6 +387,16 @@ impl<'a> ParserContext<'a> {
         let parser = &mut self.parser;
 
         let name = parser.parse_identifier()?;
+        if name.quote_style.is_none() &&
+            // "ALL_KEYWORDS" are sorted.
+            ALL_KEYWORDS.binary_search(&name.value.to_uppercase().as_str()).is_ok()
+        {
+            return Err(ParserError::ParserError(format!(
+                "Cannot use keyword '{}' as column name. Hint: add quotes to the name.",
+                &name.value
+            )));
+        }
+
         let data_type = parser.parse_data_type()?;
         let collation = if parser.parse_keyword(Keyword::COLLATE) {
             Some(parser.parse_object_name()?)
@@ -1441,4 +1452,21 @@ ENGINE=mito";
         assert!(result.is_err());
         assert_matches!(result, Err(crate::error::Error::InvalidTimeIndex { .. }));
     }
+
+    #[test]
+    fn test_invalid_column_name() {
+        let sql = "create table foo(user string, i bigint time index)";
+        let result = ParserContext::create_with_dialect(sql, &GenericDialect {});
+        assert!(result
+            .unwrap_err()
+            .to_string()
+            .contains("Cannot use keyword 'user' as column name"));
+
+        // If column name is quoted, it's valid even same with keyword.
+        let sql = r#"
+            create table foo("user" string, i bigint time index)
+        "#;
+        let result = ParserContext::create_with_dialect(sql, &GenericDialect {});
+        assert!(result.is_ok());
+    }
 }
diff --git a/tests/cases/standalone/common/create/create.result b/tests/cases/standalone/common/create/create.result
index 700fb41caa3..a818ae22cc7 100644
--- a/tests/cases/standalone/common/create/create.result
+++ b/tests/cases/standalone/common/create/create.result
@@ -88,7 +88,7 @@ DROP TABLE test2;
 
 Affected Rows: 1
 
-CREATE TABLE test_pk (timestamp BIGINT TIME INDEX, host STRING PRIMARY KEY, value DOUBLE);
+CREATE TABLE test_pk ("timestamp" BIGINT TIME INDEX, host STRING PRIMARY KEY, "value" DOUBLE);
 
 Affected Rows: 0
 
@@ -106,15 +106,15 @@ DROP TABLE test_pk;
 
 Affected Rows: 1
 
-CREATE TABLE test_multiple_pk_definitions (timestamp BIGINT TIME INDEX, host STRING PRIMARY KEY, value DOUBLE, PRIMARY KEY(host));
+CREATE TABLE test_multiple_pk_definitions ("timestamp" BIGINT TIME INDEX, host STRING PRIMARY KEY, "value" DOUBLE, PRIMARY KEY(host));
 
 Error: 1004(InvalidArguments), Illegal primary keys definition: found definitions of primary keys in multiple places
 
-CREATE TABLE test_multiple_pk_definitions (timestamp BIGINT TIME INDEX, host STRING PRIMARY KEY, value DOUBLE, PRIMARY KEY(host), PRIMARY KEY(host));
+CREATE TABLE test_multiple_pk_definitions ("timestamp" BIGINT TIME INDEX, host STRING PRIMARY KEY, "value" DOUBLE, PRIMARY KEY(host), PRIMARY KEY(host));
 
 Error: 1004(InvalidArguments), Illegal primary keys definition: found definitions of primary keys in multiple places
 
-CREATE TABLE test_multiple_inline_pk_definitions (timestamp BIGINT TIME INDEX, host STRING PRIMARY KEY, value DOUBLE PRIMARY KEY);
+CREATE TABLE test_multiple_inline_pk_definitions ("timestamp" BIGINT TIME INDEX, host STRING PRIMARY KEY, "value" DOUBLE PRIMARY KEY);
 
 Error: 1004(InvalidArguments), Illegal primary keys definition: not allowed to inline multiple primary keys in columns options
 
diff --git a/tests/cases/standalone/common/create/create.sql b/tests/cases/standalone/common/create/create.sql
index f9bcbfea110..00350c82c58 100644
--- a/tests/cases/standalone/common/create/create.sql
+++ b/tests/cases/standalone/common/create/create.sql
@@ -36,15 +36,15 @@ DROP TABLE test1;
 
 DROP TABLE test2;
 
-CREATE TABLE test_pk (timestamp BIGINT TIME INDEX, host STRING PRIMARY KEY, value DOUBLE);
+CREATE TABLE test_pk ("timestamp" BIGINT TIME INDEX, host STRING PRIMARY KEY, "value" DOUBLE);
 
 DESC TABLE test_pk;
 
 DROP TABLE test_pk;
 
-CREATE TABLE test_multiple_pk_definitions (timestamp BIGINT TIME INDEX, host STRING PRIMARY KEY, value DOUBLE, PRIMARY KEY(host));
+CREATE TABLE test_multiple_pk_definitions ("timestamp" BIGINT TIME INDEX, host STRING PRIMARY KEY, "value" DOUBLE, PRIMARY KEY(host));
 
-CREATE TABLE test_multiple_pk_definitions (timestamp BIGINT TIME INDEX, host STRING PRIMARY KEY, value DOUBLE, PRIMARY KEY(host), PRIMARY KEY(host));
+CREATE TABLE test_multiple_pk_definitions ("timestamp" BIGINT TIME INDEX, host STRING PRIMARY KEY, "value" DOUBLE, PRIMARY KEY(host), PRIMARY KEY(host));
 
-CREATE TABLE test_multiple_inline_pk_definitions (timestamp BIGINT TIME INDEX, host STRING PRIMARY KEY, value DOUBLE PRIMARY KEY);
+CREATE TABLE test_multiple_inline_pk_definitions ("timestamp" BIGINT TIME INDEX, host STRING PRIMARY KEY, "value" DOUBLE PRIMARY KEY);
 
diff --git a/tests/cases/standalone/order/order_variable_size_payload.result b/tests/cases/standalone/order/order_variable_size_payload.result
index e7724aa22cd..9eab1bc784f 100644
--- a/tests/cases/standalone/order/order_variable_size_payload.result
+++ b/tests/cases/standalone/order/order_variable_size_payload.result
@@ -352,8 +352,8 @@ CREATE TABLE DirectReports
     Name varchar NOT NULL,
     Title varchar NOT NULL,
     EmployeeLevel int NOT NULL,
-    Sort varchar NOT NULL,
-    Timestamp BIGINT TIME INDEX,
+    "Sort" varchar NOT NULL,
+    "Timestamp" BIGINT TIME INDEX,
 );
 
 Affected Rows: 0
diff --git a/tests/cases/standalone/order/order_variable_size_payload.sql b/tests/cases/standalone/order/order_variable_size_payload.sql
index 4b33a44d07d..bc57713b14b 100644
--- a/tests/cases/standalone/order/order_variable_size_payload.sql
+++ b/tests/cases/standalone/order/order_variable_size_payload.sql
@@ -87,8 +87,8 @@ CREATE TABLE DirectReports
     Name varchar NOT NULL,
     Title varchar NOT NULL,
     EmployeeLevel int NOT NULL,
-    Sort varchar NOT NULL,
-    Timestamp BIGINT TIME INDEX,
+    "Sort" varchar NOT NULL,
+    "Timestamp" BIGINT TIME INDEX,
 );
 
 INSERT INTO DirectReports VALUES