fix: flink 1.18 set (#3061)

Signed-off-by: licho <[email protected]> Co-authored-by: leechor <[email protected]>
DataLinkDC · Jan 26, 2024 · 717214f · 717214f
1 parent b797a1a
commit 717214f
Showing 1 changed file with 3 additions and 37 deletions.
diff --git a/dinky-client/dinky-client-base/src/main/java/org/dinky/trans/ddl/CustomSetOperation.java b/dinky-client/dinky-client-base/src/main/java/org/dinky/trans/ddl/CustomSetOperation.java
@@ -29,8 +29,6 @@
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.table.api.TableConfig;
 import org.apache.flink.table.api.TableResult;
-import org.apache.flink.table.operations.Operation;
-import org.apache.flink.table.operations.command.ResetOperation;
 import org.apache.flink.table.operations.command.SetOperation;
 
 import java.util.Collections;
@@ -80,7 +78,9 @@ private void parseConfig() {
     public Optional<? extends TableResult> execute(CustomTableEnvironment tEnv) {
         try {
             if (null != Class.forName("org.apache.log4j.Logger")) {
-                parseAndLoadConfiguration(statement, tEnv);
+                if (this.isValid()) {
+                    callSet(new SetOperation(this.getKey(), this.getValue()), tEnv);
+                }
                 return Optional.of(TABLE_RESULT_OK);
             }
         } catch (ClassNotFoundException e) {
@@ -98,26 +98,6 @@ public Optional<? extends TableResult> execute(CustomTableEnvironment tEnv) {
         return Optional.of(TABLE_RESULT_OK);
     }
 
-    public boolean parseAndLoadConfiguration(String statement, CustomTableEnvironment tEnv) {
-        List<Operation> operations = tEnv.getParser().parse(statement);
-        for (Operation operation : operations) {
-            if (operation instanceof SetOperation) {
-                callSet((SetOperation) operation, tEnv);
-                return true;
-            } else if (operation instanceof ResetOperation) {
-                callReset((ResetOperation) operation, tEnv);
-                return true;
-            } else if (operation instanceof CustomSetOperation) {
-                CustomSetOperation customSetOperation = (CustomSetOperation) operation;
-                if (customSetOperation.isValid()) {
-                    callSet(new SetOperation(customSetOperation.getKey(), customSetOperation.getValue()), tEnv);
-                }
-                return true;
-            }
-        }
-        return false;
-    }
-
     private void callSet(SetOperation setOperation, CustomTableEnvironment environment) {
         if (!setOperation.getKey().isPresent() || !setOperation.getValue().isPresent()) {
             return;
@@ -132,20 +112,6 @@ private void callSet(SetOperation setOperation, CustomTableEnvironment environme
         setConfiguration(environment, Collections.singletonMap(key, value));
     }
 
-    private void callReset(ResetOperation resetOperation, CustomTableEnvironment environment) {
-        final Optional<String> keyOptional = resetOperation.getKey();
-        if (!keyOptional.isPresent()) {
-            return;
-        }
-
-        String key = keyOptional.get().trim();
-        if (Asserts.isNullString(key)) {
-            return;
-        }
-
-        setConfiguration(environment, Collections.singletonMap(key, null));
-    }
-
     private void setConfiguration(CustomTableEnvironment environment, Map<String, String> config) {
         Configuration configuration = Configuration.fromMap(config);
         environment.getStreamExecutionEnvironment().getConfig().configure(configuration, null);