diff --git a/src/parser/common/types.ts b/src/parser/common/types.ts
index 8d2d2e67..a248abf4 100644
--- a/src/parser/common/types.ts
+++ b/src/parser/common/types.ts
@@ -44,6 +44,10 @@ export enum EntityContextType {
     COLUMN = 'column',
     /** column name that will be created */
     COLUMN_CREATE = 'columnCreate',
+    /** table property key when creating table*/
+    TABLE_PROPERTY_KEY = 'tablePropertyKey',
+    /** table property value when creating table*/
+    TABLE_PROPERTY_VALUE = 'tablePropertyValue',
 }
 
 /**
diff --git a/src/parser/flink/index.ts b/src/parser/flink/index.ts
index 6bb80a28..53b2bdcc 100644
--- a/src/parser/flink/index.ts
+++ b/src/parser/flink/index.ts
@@ -45,6 +45,8 @@ export class FlinkSQL extends BasicSQL<FlinkSqlLexer, ProgramContext, FlinkSqlPa
         FlinkSqlParser.RULE_columnName,
         FlinkSqlParser.RULE_columnNamePath,
         FlinkSqlParser.RULE_columnNameCreate,
+        FlinkSqlParser.RULE_tablePropertyKey,
+        FlinkSqlParser.RULE_tablePropertyValue,
     ]);
 
     protected get splitListener() {
@@ -142,6 +144,14 @@ export class FlinkSQL extends BasicSQL<FlinkSqlLexer, ProgramContext, FlinkSqlPa
                     }
                     break;
                 }
+                case FlinkSqlParser.RULE_tablePropertyKey: {
+                    syntaxContextType = EntityContextType.TABLE_PROPERTY_KEY;
+                    break;
+                }
+                case FlinkSqlParser.RULE_tablePropertyValue: {
+                    syntaxContextType = EntityContextType.TABLE_PROPERTY_VALUE;
+                    break;
+                }
                 default:
                     break;
             }
diff --git a/test/parser/flink/suggestion/fixtures/syntaxSuggestion.sql b/test/parser/flink/suggestion/fixtures/syntaxSuggestion.sql
index 74cbb786..2bf8d560 100644
--- a/test/parser/flink/suggestion/fixtures/syntaxSuggestion.sql
+++ b/test/parser/flink/suggestion/fixtures/syntaxSuggestion.sql
@@ -44,4 +44,6 @@ SELECT SUM(amount) FROM Orders GROUP BY length(users) HAVING SUM(amount) > 50;
 
 SELECT * FROM Orders ORDER BY orderTime LIMIT length(order_id);
 
-SELECT age CASE WHEN age < 18 THEN 1  ELSE 0 END AS is_minor FROM dt_catalog.dt_db.subscriptions;
\ No newline at end of file
+SELECT age CASE WHEN age < 18 THEN 1  ELSE 0 END AS is_minor FROM dt_catalog.dt_db.subscriptions;
+
+CREATE TABLE tmp_table (col INT) WITH ('connector'='kafka');
\ No newline at end of file
diff --git a/test/parser/flink/suggestion/syntaxSuggestion.test.ts b/test/parser/flink/suggestion/syntaxSuggestion.test.ts
index ebe47c7f..7a6dd004 100644
--- a/test/parser/flink/suggestion/syntaxSuggestion.test.ts
+++ b/test/parser/flink/suggestion/syntaxSuggestion.test.ts
@@ -469,4 +469,32 @@ describe('Flink SQL Syntax Suggestion', () => {
         expect(suggestion).not.toBeUndefined();
         expect(suggestion?.wordRanges.map((token) => token.text)).toEqual(['age']);
     });
+
+    test('Create Statement table properties', () => {
+        const scenarios = [
+            {
+                caretPosition: {
+                    lineNumber: 49,
+                    column: 45,
+                },
+                entityContextType: EntityContextType.TABLE_PROPERTY_KEY,
+            },
+            {
+                caretPosition: {
+                    lineNumber: 49,
+                    column: 55,
+                },
+                entityContextType: EntityContextType.TABLE_PROPERTY_VALUE,
+            },
+        ];
+
+        scenarios.forEach((scenario) => {
+            const suggestion = flink.getSuggestionAtCaretPosition(
+                commentOtherLine(syntaxSql, scenario.caretPosition.lineNumber),
+                scenario.caretPosition
+            )?.syntax;
+
+            expect(suggestion[0].syntaxContextType).toBe(scenario.entityContextType);
+        });
+    });
 });