[FLINK-38489][table] Avoid unnecessary DELETE inference for ChangelogNormalize

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

@@ -467,8 +467,21 @@ class FlinkChangelogModeInferenceProgram extends FlinkOptimizeProgram[StreamOpti
       case normalize: StreamPhysicalChangelogNormalize =>
         // changelog normalize support update&delete input
         val children = visitChildren(normalize, ModifyKindSetTrait.ALL_CHANGES)
-        // changelog normalize will output all changes
-        val providedTrait = ModifyKindSetTrait.ALL_CHANGES
+        // A filter can turn an update into a delete when the updated row stops matching.
+        val inputModifyKindSet = getModifyKindSet(children.head)
+        val providedTrait =
+          if (
+            normalize.filterCondition == null && !inputModifyKindSet.contains(ModifyKind.DELETE)
+          ) {
+            new ModifyKindSetTrait(
+              ModifyKindSet
+                .newBuilder()
+                .addContainedKind(ModifyKind.INSERT)
+                .addContainedKind(ModifyKind.UPDATE)
+                .build())
+          } else {
+            ModifyKindSetTrait.ALL_CHANGES
+          }
         createNewNode(normalize, children, providedTrait, requiredTrait, requester)
 
       case ts: StreamPhysicalTableSourceScan =>

flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/physical/stream/ChangelogModeInferenceTest.xml

@@ -16,6 +16,47 @@ See the License for the specific language governing permissions and
 limitations under the License.
 -->
 <Root>
+  <TestCase name="testChangelogNormalizeDoesNotInferDeleteWithoutFilterOrInputDelete">
+    <Resource name="sql">
+      <![CDATA[INSERT INTO changelogSink SELECT * FROM upsertSource]]>
+    </Resource>
+    <Resource name="ast">
+      <![CDATA[
+LogicalSink(table=[default_catalog.default_database.changelogSink], fields=[id, payload])
++- LogicalProject(id=[$0], payload=[$1])
+   +- LogicalTableScan(table=[[default_catalog, default_database, upsertSource]])
+]]>
+    </Resource>
+    <Resource name="optimized rel plan">
+      <![CDATA[
+Sink(table=[default_catalog.default_database.changelogSink], fields=[id, payload], changelogMode=[NONE])
++- ChangelogNormalize(key=[id], changelogMode=[I,UB,UA])
+   +- Exchange(distribution=[hash[id]], changelogMode=[I,UA])
+      +- TableSourceScan(table=[[default_catalog, default_database, upsertSource]], fields=[id, payload], changelogMode=[I,UA])
+]]>
+    </Resource>
+  </TestCase>
+  <TestCase name="testChangelogNormalizeInfersDeleteWithFilterWithoutInputDelete">
+    <Resource name="sql">
+      <![CDATA[INSERT INTO changelogSink SELECT * FROM upsertSource WHERE payload <> 'x']]>
+    </Resource>
+    <Resource name="ast">
+      <![CDATA[
+LogicalSink(table=[default_catalog.default_database.changelogSink], fields=[id, payload])
++- LogicalProject(id=[$0], payload=[$1])
+   +- LogicalFilter(condition=[<>($1, _UTF-16LE'x')])
+      +- LogicalTableScan(table=[[default_catalog, default_database, upsertSource]])
+]]>
+    </Resource>
+    <Resource name="optimized rel plan">
+      <![CDATA[
+Sink(table=[default_catalog.default_database.changelogSink], fields=[id, payload], changelogMode=[NONE])
++- ChangelogNormalize(key=[id], condition=[<>(payload, 'x')], changelogMode=[I,UB,UA,D])
+   +- Exchange(distribution=[hash[id]], changelogMode=[I,UA])
+      +- TableSourceScan(table=[[default_catalog, default_database, upsertSource, filter=[]]], fields=[id, payload], changelogMode=[I,UA])
+]]>
+    </Resource>
+  </TestCase>
   <TestCase name="testFilterNotContainedByAnyUpsertKey">
     <Resource name="sql">
       <![CDATA[

flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/DeltaJoinTest.xml

@@ -1717,15 +1717,11 @@ LogicalSink(table=[default_catalog.default_database.snk_for_cdc_src], fields=[a0
     <Resource name="optimized rel plan">
       <![CDATA[
 Sink(table=[default_catalog.default_database.snk_for_cdc_src], fields=[a0, a1, a2, a3, b0, b2, b1], changelogMode=[NONE])
-+- Join(joinType=[InnerJoin], where=[AND(=(a1, b1), =(a2, b2))], select=[a0, a1, a2, a3, b0, b2, b1], leftInputSpec=[HasUniqueKey], rightInputSpec=[JoinKeyContainsUniqueKey], changelogMode=[I,UA,D])
-   :- Exchange(distribution=[hash[a1, a2]], changelogMode=[I,UA,D])
-   :  +- ChangelogNormalize(key=[a0, a1, a2], changelogMode=[I,UA,D])
-   :     +- Exchange(distribution=[hash[a0, a1, a2]], changelogMode=[I,UA])
-   :        +- TableSourceScan(table=[[default_catalog, default_database, no_delete_and_update_before_src1]], fields=[a0, a1, a2, a3], changelogMode=[I,UA])
-   +- Exchange(distribution=[hash[b1, b2]], changelogMode=[I,UA,D])
-      +- ChangelogNormalize(key=[b1, b2], changelogMode=[I,UA,D])
-         +- Exchange(distribution=[hash[b1, b2]], changelogMode=[I,UA])
-            +- TableSourceScan(table=[[default_catalog, default_database, no_delete_and_update_before_src2]], fields=[b0, b2, b1], changelogMode=[I,UA])
++- DeltaJoin(joinType=[InnerJoin], where=[AND(=(a1, b1), =(a2, b2))], leftToRight=[Binary], rightToLeft=[Binary], select=[a0, a1, a2, a3, b0, b2, b1], changelogMode=[I,UA])
+   :- Exchange(distribution=[hash[a1, a2]], changelogMode=[I,UA])
+   :  +- TableSourceScan(table=[[default_catalog, default_database, no_delete_and_update_before_src1]], fields=[a0, a1, a2, a3], changelogMode=[I,UA])
+   +- Exchange(distribution=[hash[b1, b2]], changelogMode=[I,UA])
+      +- TableSourceScan(table=[[default_catalog, default_database, no_delete_and_update_before_src2]], fields=[b0, b2, b1], changelogMode=[I,UA])
 ]]>
     </Resource>
   </TestCase>

flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/TableScanTest.xml

@@ -391,12 +391,12 @@ LogicalProject(currency_name=[$2], amount=[$0], rate=[$5], EXPR$3=[*($0, $5)])
     </Resource>
     <Resource name="optimized rel plan">
       <![CDATA[
-Calc(select=[currency_name, amount, rate, *(amount, rate) AS EXPR$3], changelogMode=[I,UB,UA,D])
-+- Join(joinType=[InnerJoin], where=[AND(=(currency_id, currency_id0), =(currency_name, currency_name0))], select=[amount, currency_id, currency_name, currency_id0, currency_name0, rate], leftInputSpec=[NoUniqueKey], rightInputSpec=[JoinKeyContainsUniqueKey], changelogMode=[I,UB,UA,D])
+Calc(select=[currency_name, amount, rate, *(amount, rate) AS EXPR$3], changelogMode=[I,UB,UA])
++- Join(joinType=[InnerJoin], where=[AND(=(currency_id, currency_id0), =(currency_name, currency_name0))], select=[amount, currency_id, currency_name, currency_id0, currency_name0, rate], leftInputSpec=[NoUniqueKey], rightInputSpec=[JoinKeyContainsUniqueKey], changelogMode=[I,UB,UA])
    :- Exchange(distribution=[hash[currency_id, currency_name]], changelogMode=[I])
    :  +- TableSourceScan(table=[[default_catalog, default_database, orders]], fields=[amount, currency_id, currency_name], changelogMode=[I])
-   +- Exchange(distribution=[hash[currency_id, currency_name]], changelogMode=[I,UB,UA,D])
-      +- ChangelogNormalize(key=[currency_id], changelogMode=[I,UB,UA,D])
+   +- Exchange(distribution=[hash[currency_id, currency_name]], changelogMode=[I,UB,UA])
+      +- ChangelogNormalize(key=[currency_id], changelogMode=[I,UB,UA])
          +- Exchange(distribution=[hash[currency_id]], changelogMode=[I,UA])
             +- DropUpdateBefore(changelogMode=[I,UA])
                +- TableSourceScan(table=[[default_catalog, default_database, rates_history]], fields=[currency_id, currency_name, rate], changelogMode=[I,UB,UA])
@@ -560,10 +560,8 @@ LogicalProject(id1=[$1], a=[$2], b=[$4])
     </Resource>
     <Resource name="optimized rel plan">
       <![CDATA[
-Calc(select=[id1, a, b], changelogMode=[I,UA,D])
-+- ChangelogNormalize(key=[id2, id1], changelogMode=[I,UA,D])
-   +- Exchange(distribution=[hash[id2, id1]], changelogMode=[UA])
-      +- TableSourceScan(table=[[default_catalog, default_database, src, project=[id1, a, b, id2], metadata=[]]], fields=[id1, a, b, id2], changelogMode=[UA])
+Calc(select=[id1, a, b], changelogMode=[I,UA])
++- TableSourceScan(table=[[default_catalog, default_database, src, project=[id1, a, b, id2], metadata=[]]], fields=[id1, a, b, id2], changelogMode=[UA])
 ]]>
     </Resource>
   </TestCase>

flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/physical/stream/ChangelogModeInferenceTest.scala

@@ -88,6 +88,26 @@ class ChangelogModeInferenceTest extends TableTestBase {
                     |  'changelog-mode' = 'I,UA,UB,D'
                     |)
       """.stripMargin)
+
+    util.addTable("""
+                    |CREATE TABLE upsertSource (
+                    |  id INT PRIMARY KEY NOT ENFORCED,
+                    |  payload STRING
+                    |) WITH (
+                    |  'connector' = 'values',
+                    |  'changelog-mode' = 'I,UA'
+                    |)
+                    |""".stripMargin)
+
+    util.addTable("""
+                    |CREATE TABLE changelogSink (
+                    |  id INT,
+                    |  payload STRING
+                    |) WITH (
+                    |  'connector' = 'values',
+                    |  'sink-insert-only' = 'false'
+                    |)
+                    |""".stripMargin)
   }
 
   @Test
@@ -583,4 +603,18 @@ class ChangelogModeInferenceTest extends TableTestBase {
       "INSERT INTO keyless_upsert_sink_no_key SELECT rate FROM DeduplicatedView",
       ExplainDetail.CHANGELOG_MODE)
   }
+
+  @Test
+  def testChangelogNormalizeDoesNotInferDeleteWithoutFilterOrInputDelete(): Unit = {
+    util.verifyRelPlanInsert(
+      "INSERT INTO changelogSink SELECT * FROM upsertSource",
+      ExplainDetail.CHANGELOG_MODE)
+  }
+
+  @Test
+  def testChangelogNormalizeInfersDeleteWithFilterWithoutInputDelete(): Unit = {
+    util.verifyRelPlanInsert(
+      "INSERT INTO changelogSink SELECT * FROM upsertSource WHERE payload <> 'x'",
+      ExplainDetail.CHANGELOG_MODE)
+  }
 }

flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/DeltaJoinTest.scala

@@ -852,8 +852,6 @@ class DeltaJoinTest extends TableTestBase {
 
   @Test
   def testPKContainsJoinKeyAndSourceNoUBAndD(): Unit = {
-    // FLINK-38489 Currently, ChangelogNormalize will always generate changelog mode with D,
-    // and Join with D cannot be optimized into Delta Join
     replaceTable(
       "no_delete_src1",
       "no_delete_and_update_before_src1",