[FLINK-36000][table-planner] Fix DynamicTableSink#Context's getTarget…

…Columns should return an Optional#empty instead of int[0] for insert without column list
lincoln-lil · Aug 7, 2024 · a15f9c1 · a15f9c1
1 parent 8ab0112
commit a15f9c1
Show file tree

Hide file tree

Showing 3 changed files with 11 additions and 2 deletions.
diff --git a/...src/main/java/org/apache/flink/table/planner/operations/SqlNodeToOperationConversion.java b/...src/main/java/org/apache/flink/table/planner/operations/SqlNodeToOperationConversion.java
@@ -1389,8 +1389,11 @@ private Operation convertUpdate(SqlUpdate sqlUpdate) {
     private int[][] getTargetColumnIndices(
             @Nonnull ContextResolvedTable contextResolvedTable,
             @Nullable SqlNodeList targetColumns) {
+        if (targetColumns == null) {
+            return null;
+        }
         List<String> allColumns = contextResolvedTable.getResolvedSchema().getColumnNames();
-        return Optional.ofNullable(targetColumns).orElse(SqlNodeList.EMPTY).stream()
+        return targetColumns.stream()
                 .mapToInt(c -> allColumns.indexOf(((SqlIdentifier) c).getSimple()))
                 .mapToObj(idx -> new int[] {idx})
                 .toArray(int[][]::new);

diff --git a/...table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/calcite/Sink.scala b/...table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/calcite/Sink.scala
@@ -73,7 +73,7 @@ abstract class Sink(
           .getOrElse(Array.empty[Array[Int]])
           .map(_.mkString("[", ",", "]"))
           .mkString(","),
-        targetColumns != null && targetColumns.length > 0
+        targetColumns != null
       )
       .item("fields", getRowType.getFieldNames.mkString(", "))
       .itemIf("hints", RelExplainUtil.hintsToString(hints), !hints.isEmpty)

diff --git a/...lanner/src/test/java/org/apache/flink/table/planner/factories/TestValuesTableFactory.java b/...lanner/src/test/java/org/apache/flink/table/planner/factories/TestValuesTableFactory.java
@@ -2105,6 +2105,12 @@ public Optional<Integer> getParallelism() {
             } else {
                 // we don't support OutputFormat for updating query in the TestValues connector
                 assertThat(runtimeSink.equals("SinkFunction")).isTrue();
+                // check the contract of the context.getTargetColumns method returns the expected
+                // empty Option or non-empty Option with a non-empty array
+                assertThat(
+                                !context.getTargetColumns().isPresent()
+                                        || context.getTargetColumns().get().length > 0)
+                        .isTrue();
                 SinkFunction<RowData> sinkFunction;
                 if (primaryKeyIndices.length > 0) {
                     // TODO FLINK-31301 currently partial-insert composite columns are not supported