[flink] create table validate options

apache · Aug 6, 2024 · edda6b9 · edda6b9
1 parent 1a80dfa
commit edda6b9
Showing 1 changed file with 9 additions and 7 deletions.
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalog.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalog.java
@@ -18,6 +18,8 @@
 
 package org.apache.paimon.flink;
 
+import java.util.Arrays;
+import java.util.HashSet;
 import org.apache.paimon.CoreOptions;
 import org.apache.paimon.catalog.Catalog;
 import org.apache.paimon.catalog.Identifier;
@@ -348,13 +350,13 @@ private void validateOptions(Map<String, String> options) {
                                 })
                         .collect(Collectors.toSet());
 
-        options.keySet()
-                .forEach(
-                        optionKey -> {
-                            if (!validOptions.contains(optionKey)) {
-                                throw new IllegalArgumentException("Invalid option: " + optionKey);
-                            }
-                        });
+        Set<String> additionalValidOptions = new HashSet<>(Arrays.asList("is_streaming", "connector", "log.system"));
+
+        options.keySet().forEach(optionKey -> {
+            if (!validOptions.contains(optionKey) && !additionalValidOptions.contains(optionKey)) {
+                throw new IllegalArgumentException("Invalid option: " + optionKey);
+            }
+        });
     }
 
     protected Schema buildPaimonSchema(