fixup! [FLINK-37856] Ensure sink option hints are present in compiled plan

Move merging to planner

Author: AHeise

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/SinkModifyOperation.java

@@ -29,8 +29,6 @@
 import java.util.LinkedHashMap;
 import java.util.Map;
 
-import static org.apache.flink.table.operations.utils.HintUtils.mergeTableOptions;
-
 /**
  * DML operation that tells to write to a sink.
  *
@@ -100,7 +98,7 @@ public SinkModifyOperation(
             boolean overwrite,
             Map<String, String> dynamicOptions,
             ModifyType modifyType) {
-        this.contextResolvedTable = withDynamicOptions(contextResolvedTable, dynamicOptions);
+        this.contextResolvedTable = contextResolvedTable;
         this.child = child;
         this.staticPartitions = staticPartitions;
         this.targetColumns = targetColumns;
@@ -109,16 +107,6 @@ public SinkModifyOperation(
         this.modifyType = modifyType;
     }
 
-    private ContextResolvedTable withDynamicOptions(
-            ContextResolvedTable contextResolvedTable, Map<String, String> dynamicOptions) {
-        if (dynamicOptions.isEmpty()) {
-            return contextResolvedTable;
-        }
-        return contextResolvedTable.copy(
-                mergeTableOptions(
-                        dynamicOptions, contextResolvedTable.getResolvedTable().getOptions()));
-    }
-
     public ContextResolvedTable getContextResolvedTable() {
         return contextResolvedTable;
     }

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/utils/HintUtils.java

@@ -107,6 +107,7 @@
 import java.util.stream.IntStream;
 import java.util.stream.Stream;
 
+import static org.apache.flink.table.planner.hint.FlinkHints.mergeTableOptions;
 import static org.apache.flink.table.planner.utils.ShortcutUtils.unwrapContext;
 import static org.apache.flink.table.planner.utils.ShortcutUtils.unwrapTypeFactory;
 import static org.apache.flink.table.types.logical.utils.LogicalTypeCasts.supportsAvoidingCast;
@@ -258,6 +259,14 @@ private static RelNode convertSinkToRel(
             int[][] targetColumns,
             boolean isOverwrite,
             DynamicTableSink sink) {
+        if (!dynamicOptions.isEmpty()) {
+            contextResolvedTable =
+                    contextResolvedTable.copy(
+                            mergeTableOptions(
+                                    dynamicOptions,
+                                    contextResolvedTable.getResolvedTable().getOptions()));
+        }
+
         final DataTypeFactory dataTypeFactory =
                 unwrapContext(relBuilder).getCatalogManager().getDataTypeFactory();
         final FlinkTypeFactory typeFactory = unwrapTypeFactory(relBuilder);

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/connectors/DynamicSinkUtils.java

@@ -37,6 +37,7 @@
 import org.apache.commons.lang3.StringUtils;
 
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -76,6 +77,29 @@ public static Map<String, String> getHintedOptions(List<RelHint> tableHints) {
                 .orElse(Collections.emptyMap());
     }
 
+    /**
+     * Merges the dynamic table options from {@code hints} and static table options from table
+     * definition {@code props}.
+     *
+     * <p>The options in {@code hints} would override the ones in {@code props} if they have the
+     * same option key.
+     *
+     * @param hints Dynamic table options, usually from the OPTIONS hint
+     * @param props Static table options defined in DDL or connect API
+     * @return New options with merged dynamic table options, or the old {@code props} if there is
+     *     no dynamic table options
+     */
+    public static Map<String, String> mergeTableOptions(
+            Map<String, String> hints, Map<String, String> props) {
+        if (hints.size() == 0) {
+            return props;
+        }
+        Map<String, String> newProps = new HashMap<>();
+        newProps.putAll(props);
+        newProps.putAll(hints);
+        return Collections.unmodifiableMap(newProps);
+    }
+
     public static Optional<String> getTableAlias(RelNode node) {
         if (node instanceof Hintable) {
             Hintable aliasNode = (Hintable) node;

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/hint/FlinkHints.java

@@ -29,7 +29,6 @@
 import org.apache.flink.table.factories.DynamicTableSourceFactory;
 import org.apache.flink.table.factories.FactoryUtil;
 import org.apache.flink.table.module.Module;
-import org.apache.flink.table.operations.utils.HintUtils;
 import org.apache.flink.table.planner.calcite.FlinkContext;
 import org.apache.flink.table.planner.calcite.FlinkRelBuilder;
 import org.apache.flink.table.planner.catalog.CatalogSchemaTable;
@@ -148,7 +147,7 @@ private ContextResolvedTable computeContextResolvedTable(
                             contextResolvedTable.getIdentifier()));
         }
         return contextResolvedTable.copy(
-                HintUtils.mergeTableOptions(
+                FlinkHints.mergeTableOptions(
                         hintedOptions, contextResolvedTable.getResolvedTable().getOptions()));
     }
 

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/schema/CatalogSourceTable.java

@@ -299,7 +299,7 @@ abstract class PlannerBase(
       case catalogSink: SinkModifyOperation =>
         val input = createRelBuilder.queryOperation(modifyOperation.getChild).build()
         val dynamicOptions = catalogSink.getDynamicOptions
-        getTableSink(catalogSink.getContextResolvedTable).map {
+        getTableSink(catalogSink.getContextResolvedTable, dynamicOptions).map {
           case (table, sink: TableSink[_]) =>
             // Legacy tables can't be anonymous
             val identifier = catalogSink.getContextResolvedTable.getIdentifier
@@ -451,7 +451,8 @@ abstract class PlannerBase(
   }
 
   private def getTableSink(
-      contextResolvedTable: ContextResolvedTable): Option[(ResolvedCatalogTable, Any)] = {
+      contextResolvedTable: ContextResolvedTable,
+      dynamicOptions: JMap[String, String]): Option[(ResolvedCatalogTable, Any)] = {
     contextResolvedTable.getTable[CatalogBaseTable] match {
       case connectorTable: ConnectorCatalogTable[_, _] =>
         val resolvedTable = contextResolvedTable.getResolvedTable[ResolvedCatalogTable]
@@ -461,7 +462,15 @@ abstract class PlannerBase(
         }
 
       case regularTable: CatalogTable =>
-        val resolvedTable = contextResolvedTable.getResolvedTable[ResolvedCatalogTable]
+        val resolvedTable = {
+          val resolvedTable = contextResolvedTable.getResolvedTable[ResolvedCatalogTable]
+          if (dynamicOptions.nonEmpty) {
+            resolvedTable.copy(
+              FlinkHints.mergeTableOptions(dynamicOptions, resolvedTable.getOptions))
+          } else {
+            resolvedTable
+          }
+        }
         val catalog = toScala(contextResolvedTable.getCatalog)
         val objectIdentifier = contextResolvedTable.getIdentifier
         val isTemporary = contextResolvedTable.isTemporary

flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/delegation/PlannerBase.scala

@@ -28,7 +28,7 @@ import org.apache.flink.table.legacy.sources.TableSource
 import org.apache.flink.table.planner.JMap
 import org.apache.flink.table.planner.calcite.{FlinkRelBuilder, FlinkTypeFactory}
 import org.apache.flink.table.planner.catalog.CatalogSchemaTable
-import org.apache.flink.table.operations.utils.HintUtils
+import org.apache.flink.table.planner.hint.FlinkHints
 import org.apache.flink.table.planner.utils.ShortcutUtils.unwrapContext
 import org.apache.flink.table.sources.TableSourceValidation
 import org.apache.flink.table.types.logical.{LocalZonedTimestampType, TimestampKind, TimestampType}
@@ -171,7 +171,7 @@ class LegacyCatalogSourceTable[T](
       hintedOptions: JMap[String, String],
       conf: ReadableConfig): TableSource[T] = {
     val tableToFind = if (hintedOptions.nonEmpty) {
-      catalogTable.copy(HintUtils.mergeTableOptions(hintedOptions, catalogTable.getOptions))
+      catalogTable.copy(FlinkHints.mergeTableOptions(hintedOptions, catalogTable.getOptions))
     } else {
       catalogTable
     }