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

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

* fix(suggestion): move SQL to fixtures

* chore(sql): add EOL

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

# Conflicts:
#	test/parser/flink/suggestion/tokenSuggestion.test.ts

* fix(suggestion): move SQL to fixtures

* fix(syntaxSuggestion): resolve conflicts and move tests

* fix: revert old changes

* fix: revert old changes

* fix: merge issues

* fix: unit tests

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

@@ -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;
             }

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;
+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');

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);
+        });
+    });
 });