[FLINK-37475] Drop ChangelogNormalize for piping from upsert source to sink

* Add information about DELETE to ChangelogMode
* Adapt FlinkChangelogModeInferenceProgram to remove ChangelogNormalize
  if possible

Author: dawidwys

flink-table/flink-table-common/src/main/java/org/apache/flink/table/connector/ChangelogMode.java

@@ -43,6 +43,7 @@ public final class ChangelogMode {
                     .addContainedKind(RowKind.INSERT)
                     .addContainedKind(RowKind.UPDATE_AFTER)
                     .addContainedKind(RowKind.DELETE)
+                    .deletesOnKey(true)
                     .build();
 
     private static final ChangelogMode ALL =
@@ -54,11 +55,13 @@ public final class ChangelogMode {
                     .build();
 
     private final Set<RowKind> kinds;
+    private final boolean deletesOnKey;
 
-    private ChangelogMode(Set<RowKind> kinds) {
+    private ChangelogMode(Set<RowKind> kinds, boolean deletesOnKey) {
         Preconditions.checkArgument(
                 kinds.size() > 0, "At least one kind of row should be contained in a changelog.");
         this.kinds = Collections.unmodifiableSet(kinds);
+        this.deletesOnKey = deletesOnKey;
     }
 
     /** Shortcut for a simple {@link RowKind#INSERT}-only changelog. */
@@ -96,6 +99,10 @@ public boolean containsOnly(RowKind kind) {
         return kinds.size() == 1 && kinds.contains(kind);
     }
 
+    public boolean deletesOnKey() {
+        return deletesOnKey;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) {
@@ -125,6 +132,7 @@ public String toString() {
     public static class Builder {
 
         private final Set<RowKind> kinds = EnumSet.noneOf(RowKind.class);
+        private boolean deletesOnKey = false;
 
         private Builder() {
             // default constructor to allow a fluent definition
@@ -135,8 +143,13 @@ public Builder addContainedKind(RowKind kind) {
             return this;
         }
 
+        public Builder deletesOnKey(boolean flag) {
+            this.deletesOnKey = flag;
+            return this;
+        }
+
         public ChangelogMode build() {
-            return new ChangelogMode(kinds);
+            return new ChangelogMode(kinds, deletesOnKey);
         }
     }
 }

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/trait/DeleteKind.java

@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.plan.trait;
+
+/** Lists all kinds of {@link ModifyKind#DELETE} operation. */
+public enum DeleteKind {
+
+    /** This kind indicates that operators do not emit {@link ModifyKind#DELETE} operation. */
+    NONE,
+
+    /**
+     * This kind indicates that operators can emit deletes with the key only. The rest of the row
+     * may be not present.
+     */
+    DELETE_ON_KEY,
+
+    /** This kind indicates that operators should emit deletes with the full row. */
+    FULL_DELETE
+}

flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala

@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.table.planner.plan.`trait`
+
+import org.apache.flink.table.connector.ChangelogMode
+import org.apache.flink.types.RowKind
+
+import org.apache.calcite.plan.{RelOptPlanner, RelTrait, RelTraitDef}
+
+/** DeleteKindTrait is used to describe the kind of delete operation. */
+class DeleteKindTrait(val deleteKind: DeleteKind) extends RelTrait {
+
+  override def satisfies(relTrait: RelTrait): Boolean = relTrait match {
+    case other: DeleteKindTrait =>
+      // should totally match
+      other.deleteKind == this.deleteKind
+    case _ => false
+  }
+
+  override def getTraitDef: RelTraitDef[_ <: RelTrait] = DeleteKindTraitDef.INSTANCE
+
+  override def register(planner: RelOptPlanner): Unit = {}
+
+  override def hashCode(): Int = deleteKind.hashCode()
+
+  override def equals(obj: Any): Boolean = obj match {
+    case t: DeleteKindTrait => this.deleteKind.equals(t.deleteKind)
+    case _ => false
+  }
+
+  override def toString: String = s"[${deleteKind.toString}]"
+}
+
+object DeleteKindTrait {
+
+  /** An [[DeleteKindTrait]] that describes the node does not support delete operation. */
+  val NONE = new DeleteKindTrait(DeleteKind.NONE)
+
+  /** An [[DeleteKindTrait]] that describes the node supports deletes on key only. */
+  val DELETE_ON_KEY = new DeleteKindTrait(DeleteKind.DELETE_ON_KEY)
+
+  /** An [[DeleteKindTrait]] that describes the node produces requires deletes by full records. */
+  val FULL_DELETE = new DeleteKindTrait(DeleteKind.FULL_DELETE)
+
+  /**
+   * Returns DELETE_ON_KEY [[DeleteKindTrait]] if there is delete changes. Otherwise, returns NONE
+   * [[DeleteKindTrait]].
+   */
+  def deleteOnKeyOrNone(modifyKindSet: ModifyKindSet): DeleteKindTrait = {
+    val deleteKind = if (modifyKindSet.contains(ModifyKind.DELETE)) {
+      DeleteKind.DELETE_ON_KEY
+    } else {
+      DeleteKind.NONE
+    }
+    new DeleteKindTrait(deleteKind)
+  }
+
+  /**
+   * Returns FULL_DELETE [[DeleteKindTrait]] if there is delete changes. Otherwise, returns NONE
+   * [[DeleteKindTrait]].
+   */
+  def fullDeleteOrNone(modifyKindSet: ModifyKindSet): DeleteKindTrait = {
+    val deleteKind = if (modifyKindSet.contains(ModifyKind.DELETE)) {
+      DeleteKind.FULL_DELETE
+    } else {
+      DeleteKind.NONE
+    }
+    new DeleteKindTrait(deleteKind)
+  }
+
+  /** Creates an instance of [[DeleteKindTrait]] from the given [[ChangelogMode]]. */
+  def fromChangelogMode(changelogMode: ChangelogMode): DeleteKindTrait = {
+    val hasDelete = changelogMode.contains(RowKind.DELETE)
+    if (!hasDelete) {
+      DeleteKindTrait.NONE
+    } else {
+      val hasDeleteOnKey = changelogMode.deletesOnKey()
+      if (hasDeleteOnKey) {
+        DeleteKindTrait.DELETE_ON_KEY
+      } else {
+        DeleteKindTrait.FULL_DELETE
+      }
+    }
+  }
+}

flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/trait/DeleteKindTrait.scala

@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.table.planner.plan.`trait`
+
+import org.apache.calcite.plan.{RelOptPlanner, RelTraitDef}
+import org.apache.calcite.rel.RelNode
+
+class DeleteKindTraitDef extends RelTraitDef[DeleteKindTrait] {
+
+  override def getTraitClass: Class[DeleteKindTrait] = classOf[DeleteKindTrait]
+
+  override def getSimpleName: String = this.getClass.getSimpleName
+
+  override def convert(
+      planner: RelOptPlanner,
+      rel: RelNode,
+      toTrait: DeleteKindTrait,
+      allowInfiniteCostConverters: Boolean): RelNode = {
+    rel.copy(rel.getTraitSet.plus(toTrait), rel.getInputs)
+  }
+
+  override def canConvert(
+      planner: RelOptPlanner,
+      fromTrait: DeleteKindTrait,
+      toTrait: DeleteKindTrait): Boolean = {
+    throw new UnsupportedOperationException("DeleteKindTrait conversion is not supported for now.")
+  }
+
+  override def getDefault: DeleteKindTrait = DeleteKindTrait.FULL_DELETE
+}
+
+object DeleteKindTraitDef {
+  val INSTANCE = new DeleteKindTraitDef()
+}

flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/trait/DeleteKindTraitDef.scala

@@ -457,6 +457,22 @@ private static RowKind parseRowKind(String rowKindShortString) {
                     .withDescription(
                             "Option to determine whether or not to require the distribution bucket count");
 
+    private static final ConfigOption<Boolean> SINK_SUPPORTS_DELETE_ON_KEY =
+            ConfigOptions.key("sink.supports-delete-on-key")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            "Option to determine whether or not to require deletes to have the"
+                                    + " entire row or is a delete by key sufficient.");
+
+    private static final ConfigOption<Boolean> SOURCE_PRODUCES_DELETE_ON_KEY =
+            ConfigOptions.key("source.produces-delete-on-key")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            "Option to determine whether or not to require deletes to have the"
+                                    + " entire row or is a delete by key sufficient.");
+
     private static final ConfigOption<Integer> SOURCE_NUM_ELEMENT_TO_SKIP =
             ConfigOptions.key("source.num-element-to-skip")
                     .intType()
@@ -504,7 +520,10 @@ public DynamicTableSource createDynamicTableSource(Context context) {
 
         helper.validate();
 
-        ChangelogMode changelogMode = parseChangelogMode(helper.getOptions().get(CHANGELOG_MODE));
+        ChangelogMode changelogMode =
+                parseChangelogMode(
+                        helper.getOptions().get(CHANGELOG_MODE),
+                        helper.getOptions().get(SOURCE_PRODUCES_DELETE_ON_KEY));
         String runtimeSource = helper.getOptions().get(RUNTIME_SOURCE);
         boolean isBounded = helper.getOptions().get(BOUNDED);
         boolean isFinite = helper.getOptions().get(TERMINATING);
@@ -749,6 +768,7 @@ public DynamicTableSink createDynamicTableSink(Context context) {
                 TableSchemaUtils.getPhysicalSchema(context.getCatalogTable().getSchema());
 
         boolean requireBucketCount = helper.getOptions().get(SINK_BUCKET_COUNT_REQUIRED);
+        boolean supportsDeleteByKey = helper.getOptions().get(SINK_SUPPORTS_DELETE_ON_KEY);
         if (sinkClass.equals("DEFAULT")) {
             int rowTimeIndex =
                     validateAndExtractRowtimeIndex(
@@ -765,7 +785,8 @@ public DynamicTableSink createDynamicTableSink(Context context) {
                     changelogMode,
                     rowTimeIndex,
                     tableSchema,
-                    requireBucketCount);
+                    requireBucketCount,
+                    supportsDeleteByKey);
         } else {
             try {
                 return InstantiationUtil.instantiate(
@@ -816,6 +837,8 @@ public Set<ConfigOption<?>> optionalOptions() {
                         ENABLE_WATERMARK_PUSH_DOWN,
                         SINK_DROP_LATE_EVENT,
                         SINK_BUCKET_COUNT_REQUIRED,
+                        SINK_SUPPORTS_DELETE_ON_KEY,
+                        SOURCE_PRODUCES_DELETE_ON_KEY,
                         SOURCE_NUM_ELEMENT_TO_SKIP,
                         SOURCE_SLEEP_AFTER_ELEMENTS,
                         SOURCE_SLEEP_TIME,
@@ -916,6 +939,10 @@ private static Map<Map<String, String>, Collection<Row>> mapPartitionToRow(
     }
 
     private ChangelogMode parseChangelogMode(String string) {
+        return parseChangelogMode(string, false);
+    }
+
+    private ChangelogMode parseChangelogMode(String string, boolean producesDeleteByKey) {
         ChangelogMode.Builder builder = ChangelogMode.newBuilder();
         for (String split : string.split(",")) {
             switch (split.trim()) {
@@ -935,6 +962,7 @@ private ChangelogMode parseChangelogMode(String string) {
                     throw new IllegalArgumentException("Invalid ChangelogMode string: " + string);
             }
         }
+        builder.deletesOnKey(producesDeleteByKey);
         return builder.build();
     }
 
@@ -1621,7 +1649,7 @@ private static class TestValuesScanTableSourceWithWatermarkPushDown
             implements SupportsWatermarkPushDown, SupportsSourceWatermark {
         private final String tableName;
 
-        private WatermarkStrategy<RowData> watermarkStrategy;
+        private WatermarkStrategy<RowData> watermarkStrategy = WatermarkStrategy.noWatermarks();
 
         private TestValuesScanTableSourceWithWatermarkPushDown(
                 DataType producedDataType,
@@ -2207,6 +2235,7 @@ private static class TestValuesTableSink
         private final int rowtimeIndex;
         private final TableSchema tableSchema;
         private final boolean requireBucketCount;
+        private final boolean supportsDeleteByKey;
 
         private TestValuesTableSink(
                 DataType consumedDataType,
@@ -2220,7 +2249,8 @@ private TestValuesTableSink(
                 @Nullable ChangelogMode changelogModeEnforced,
                 int rowtimeIndex,
                 TableSchema tableSchema,
-                boolean requireBucketCount) {
+                boolean requireBucketCount,
+                boolean supportsDeleteByKey) {
             this.consumedDataType = consumedDataType;
             this.primaryKeyIndices = primaryKeyIndices;
             this.tableName = tableName;
@@ -2233,10 +2263,19 @@ private TestValuesTableSink(
             this.rowtimeIndex = rowtimeIndex;
             this.tableSchema = tableSchema;
             this.requireBucketCount = requireBucketCount;
+            this.supportsDeleteByKey = supportsDeleteByKey;
         }
 
         @Override
         public ChangelogMode getChangelogMode(ChangelogMode requestedMode) {
+            final ChangelogMode mode = getMode(requestedMode);
+            final ChangelogMode.Builder builder = ChangelogMode.newBuilder();
+            mode.getContainedKinds().forEach(builder::addContainedKind);
+            builder.deletesOnKey(supportsDeleteByKey);
+            return builder.build();
+        }
+
+        private ChangelogMode getMode(ChangelogMode requestedMode) {
             // if param [changelogModeEnforced] is passed in, return it directly
             if (changelogModeEnforced != null) {
                 return changelogModeEnforced;
@@ -2376,7 +2415,8 @@ public DynamicTableSink copy() {
                     changelogModeEnforced,
                     rowtimeIndex,
                     tableSchema,
-                    requireBucketCount);
+                    requireBucketCount,
+                    supportsDeleteByKey);
         }
 
         @Override