feat(flink): implement suggestion support for table properties in CREATE TABLE

Author: ekazakas

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',
 }
 
 /**

src/parser/flink/index.ts

@@ -41,6 +41,8 @@ export class FlinkSQL extends BasicSQL<FlinkSqlLexer, ProgramContext, FlinkSqlPa
         FlinkSqlParser.RULE_functionNameCreate, // functionName that will be created
         FlinkSqlParser.RULE_columnName,
         FlinkSqlParser.RULE_columnNameCreate,
+        FlinkSqlParser.RULE_tablePropertyKey,
+        FlinkSqlParser.RULE_tablePropertyValue,
     ]);
 
     protected get splitListener() {
@@ -127,6 +129,14 @@ export class FlinkSQL extends BasicSQL<FlinkSqlLexer, ProgramContext, FlinkSqlPa
                     syntaxContextType = EntityContextType.COLUMN_CREATE;
                     break;
                 }
+                case FlinkSqlParser.RULE_tablePropertyKey: {
+                    syntaxContextType = EntityContextType.TABLE_PROPERTY_KEY;
+                    break;
+                }
+                case FlinkSqlParser.RULE_tablePropertyValue: {
+                    syntaxContextType = EntityContextType.TABLE_PROPERTY_VALUE;
+                    break;
+                }
                 default:
                     break;
             }

test/parser/flink/suggestion/tokenSuggestion.test.ts

@@ -1,7 +1,7 @@
 import fs from 'fs';
 import path from 'path';
 import { FlinkSQL } from 'src/parser/flink';
-import { CaretPosition } from 'src/parser/common/types';
+import { CaretPosition, EntityContextType } from 'src/parser/common/types';
 import { commentOtherLine } from 'test/helper';
 
 const tokenSql = fs.readFileSync(path.join(__dirname, 'fixtures', 'tokenSuggestion.sql'), 'utf-8');
@@ -93,4 +93,33 @@ describe('Flink SQL Token Suggestion', () => {
         expect(suggestion).toContain('NOT');
         expect(suggestion).toContain('NOT EXISTS');
     });
+
+    test('Create Statement table properties', () => {
+        const tokenSql = `CREATE TABLE tmp_table (col INT) WITH ('connector'='kafka');`;
+        const scenarios = [
+            {
+                caretPosition: {
+                    lineNumber: 1,
+                    column: 45,
+                },
+                entityContextType: EntityContextType.TABLE_PROPERTY_KEY,
+            },
+            {
+                caretPosition: {
+                    lineNumber: 1,
+                    column: 55,
+                },
+                entityContextType: EntityContextType.TABLE_PROPERTY_VALUE,
+            },
+        ];
+
+        scenarios.forEach((scenario) => {
+            const suggestion = flink.getSuggestionAtCaretPosition(
+                tokenSql,
+                scenario.caretPosition
+            )?.syntax;
+
+            expect(suggestion[0].syntaxContextType).toBe(scenario.entityContextType);
+        });
+    });
 });