[FLINK-38802][table] DDL with duplicated keys in table options should not fail

Author: snuyanzin

flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/SqlParseUtils.java

@@ -24,9 +24,12 @@
 import org.apache.calcite.sql.SqlNode;
 import org.apache.calcite.sql.SqlNodeList;
 import org.apache.calcite.util.NlsString;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import javax.annotation.Nullable;
 
+import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
@@ -36,6 +39,7 @@
 
 /** Utils methods for parsing DDLs. */
 public class SqlParseUtils {
+    private static final Logger LOG = LoggerFactory.getLogger(SqlParseUtils.class);
 
     private SqlParseUtils() {}
 
@@ -85,9 +89,15 @@ public static Map<String, String> extractMap(@Nullable SqlNodeList propList) {
         if (propList == null) {
             return Map.of();
         }
-        return propList.getList().stream()
-                .map(p -> (SqlTableOption) p)
-                .collect(Collectors.toMap(k -> k.getKeyString(), SqlTableOption::getValueString));
+        final Map<String, String> result = new HashMap<>();
+        for (SqlNode node : propList) {
+            final SqlTableOption tableOption = (SqlTableOption) node;
+            final String key = tableOption.getKeyString();
+            if (result.put(key, tableOption.getValueString()) != null) {
+                LOG.warn("There are duplicated values for the same key {}", key);
+            }
+        }
+        return result;
     }
 
     public static List<String> extractList(

flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/catalog/SqlAlterCatalogOptions.java

@@ -18,8 +18,8 @@
 
 package org.apache.flink.sql.parser.ddl.catalog;
 
+import org.apache.flink.sql.parser.SqlParseUtils;
 import org.apache.flink.sql.parser.SqlUnparseUtils;
-import org.apache.flink.sql.parser.ddl.SqlTableOption;
 
 import org.apache.calcite.sql.SqlIdentifier;
 import org.apache.calcite.sql.SqlNode;
@@ -30,7 +30,6 @@
 
 import java.util.List;
 import java.util.Map;
-import java.util.stream.Collectors;
 
 import static java.util.Objects.requireNonNull;
 
@@ -55,13 +54,7 @@ public SqlNodeList getPropertyList() {
     }
 
     public Map<String, String> getProperties() {
-        return propertyList.stream()
-                .map(p -> (SqlTableOption) p)
-                .collect(
-                        Collectors.toMap(
-                                SqlTableOption::getKeyString,
-                                SqlTableOption::getValueString,
-                                (option1, option2) -> option2));
+        return SqlParseUtils.extractMap(propertyList);
     }
 
     @Override

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlDdlToOperationConverterTest.java

@@ -1117,28 +1117,33 @@ void testAlterTable() throws Exception {
         // test alter table options
         checkAlterNonExistTable("alter table %s nonexistent set ('k1' = 'v1', 'K2' = 'V2')");
         Operation operation =
-                parse("alter table if exists cat1.db1.tb1 set ('k1' = 'v1', 'K2' = 'V2')");
+                parse(
+                        "alter table if exists cat1.db1.tb1 set ('k1' = 'v1', 'k2' = 'v2', 'K2' = 'V1', 'K2' = 'V2')");
         Map<String, String> expectedOptions = new HashMap<>();
         expectedOptions.put("connector", "dummy");
         expectedOptions.put("k", "v");
         expectedOptions.put("k1", "v1");
+        expectedOptions.put("k2", "v2");
         expectedOptions.put("K2", "V2");
 
         assertAlterTableOptions(
                 operation,
                 expectedIdentifier,
                 expectedOptions,
-                Arrays.asList(TableChange.set("k1", "v1"), TableChange.set("K2", "V2")),
-                "ALTER TABLE IF EXISTS cat1.db1.tb1\n  SET 'k1' = 'v1',\n  SET 'K2' = 'V2'");
+                List.of(
+                        TableChange.set("k1", "v1"),
+                        TableChange.set("k2", "v2"),
+                        TableChange.set("K2", "V2")),
+                "ALTER TABLE IF EXISTS cat1.db1.tb1\n  SET 'k1' = 'v1',\n  SET 'k2' = 'v2',\n  SET 'K2' = 'V2'");
 
         // test alter table reset
         checkAlterNonExistTable("alter table %s nonexistent reset ('k')");
         operation = parse("alter table if exists cat1.db1.tb1 reset ('k')");
         assertAlterTableOptions(
                 operation,
                 expectedIdentifier,
-                Collections.singletonMap("connector", "dummy"),
-                Collections.singletonList(TableChange.reset("k")),
+                Map.of("connector", "dummy"),
+                List.of(TableChange.reset("k")),
                 "ALTER TABLE IF EXISTS cat1.db1.tb1\n  RESET 'k'");
         assertThatThrownBy(() -> parse("alter table cat1.db1.tb1 reset ('connector')"))
                 .isInstanceOf(ValidationException.class)
@@ -2885,7 +2890,8 @@ private void assertAlterTableOptions(
         assertThat(alterTableOptionsOperation.getTableIdentifier()).isEqualTo(expectedIdentifier);
         assertThat(alterTableOptionsOperation.getNewTable().getOptions())
                 .isEqualTo(expectedOptions);
-        assertThat(expectedChanges).isEqualTo(alterTableOptionsOperation.getTableChanges());
+        assertThat(expectedChanges)
+                .containsExactlyInAnyOrderElementsOf(alterTableOptionsOperation.getTableChanges());
         assertThat(alterTableOptionsOperation.asSummaryString()).isEqualTo(expectedSummary);
     }