apache · twalthr · Sep 4, 2026
diff --git a/...ner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecCalc.java b/...ner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecCalc.java
@@ -69,6 +69,7 @@ public BatchExecCalc(
                 ExecNodeContext.newPersistedConfig(BatchExecCalc.class, tableConfig),
                 projection,
                 condition,
+                null, // partialDeleteKeys: not applicable in batch mode
                 TableStreamOperator.class,
                 false, // retainHeader
                 Collections.singletonList(inputProperty),
@@ -92,6 +93,7 @@ public BatchExecCalc(
                 persistedConfig,
                 projection,
                 condition,
+                null, // partialDeleteKeys: not applicable in batch mode
                 TableStreamOperator.class,
                 false, // retainHeader
                 inputProperties,

diff --git a/...r/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecCalc.java b/...r/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecCalc.java
@@ -20,6 +20,7 @@
 
 import org.apache.flink.api.dag.Transformation;
 import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.connector.ChangelogMode;
 import org.apache.flink.table.data.RowData;
 import org.apache.flink.table.planner.codegen.CalcCodeGenerator;
 import org.apache.flink.table.planner.codegen.CodeGeneratorContext;
@@ -37,6 +38,7 @@
 import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
 import org.apache.flink.table.types.logical.RowType;
 
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
 
 import org.apache.calcite.rex.RexNode;
@@ -57,13 +59,24 @@ public abstract class CommonExecCalc extends ExecNodeBase<RowData>
 
     public static final String FIELD_NAME_PROJECTION = "projection";
     public static final String FIELD_NAME_CONDITION = "condition";
+    public static final String FIELD_NAME_PARTIAL_DELETE_KEYS = "partialDeleteKeys";
 
     @JsonProperty(FIELD_NAME_PROJECTION)
     protected final List<RexNode> projection;
 
     @JsonProperty(FIELD_NAME_CONDITION)
     protected final @Nullable RexNode condition;
 
+    /**
+     * Output column indices that serve as keys for partial deletes. Only set when this calc deals
+     * with {@link ChangelogMode#keyOnlyDeletes()}. When a {@code -D} row enters, only these output
+     * columns are evaluated (from {@link #projection}); every other column becomes a typed {@code
+     * NULL} instead of evaluating its (potentially unsafe) expression.
+     */
+    @JsonProperty(FIELD_NAME_PARTIAL_DELETE_KEYS)
+    @JsonInclude(JsonInclude.Include.NON_NULL)
+    protected final @Nullable int[] partialDeleteKeys;
+
     private final Class<?> operatorBaseClass;
     private final boolean retainHeader;
 
@@ -73,6 +86,7 @@ protected CommonExecCalc(
             ReadableConfig persistedConfig,
             List<RexNode> projection,
             @Nullable RexNode condition,
+            @Nullable int[] partialDeleteKeys,
             Class<?> operatorBaseClass,
             boolean retainHeader,
             List<InputProperty> inputProperties,
@@ -82,6 +96,7 @@ protected CommonExecCalc(
         checkArgument(inputProperties.size() == 1);
         this.projection = checkNotNull(projection);
         this.condition = condition;
+        this.partialDeleteKeys = partialDeleteKeys;
         this.operatorBaseClass = checkNotNull(operatorBaseClass);
         this.retainHeader = retainHeader;
     }
@@ -103,7 +118,8 @@ protected Transformation<RowData> translateToPlanInternal(
                         (RowType) inputEdge.getOutputType(),
                         (RowType) getOutputType(),
                         JavaScalaConversionUtil.toScala(projection),
-                        JavaScalaConversionUtil.toScala(Optional.ofNullable(this.condition)),
+                        JavaScalaConversionUtil.toScala(Optional.ofNullable(condition)),
+                        partialDeleteKeys,
                         ShortcutUtils.unwrapTypeFactory(planner),
                         retainHeader,
                         getClass().getSimpleName());

diff --git a/...r/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecCalc.java b/...r/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecCalc.java
@@ -52,6 +52,7 @@ public StreamExecCalc(
             ReadableConfig tableConfig,
             List<RexNode> projection,
             @Nullable RexNode condition,
+            @Nullable int[] partialDeleteKeys,
             InputProperty inputProperty,
             RowType outputType,
             String description) {
@@ -61,6 +62,7 @@ public StreamExecCalc(
                 ExecNodeContext.newPersistedConfig(StreamExecCalc.class, tableConfig),
                 projection,
                 condition,
+                partialDeleteKeys,
                 Collections.singletonList(inputProperty),
                 outputType,
                 description);
@@ -73,6 +75,7 @@ public StreamExecCalc(
             @JsonProperty(FIELD_NAME_CONFIGURATION) ReadableConfig persistedConfig,
             @JsonProperty(FIELD_NAME_PROJECTION) List<RexNode> projection,
             @JsonProperty(FIELD_NAME_CONDITION) @Nullable RexNode condition,
+            @JsonProperty(FIELD_NAME_PARTIAL_DELETE_KEYS) @Nullable int[] partialDeleteKeys,
             @JsonProperty(FIELD_NAME_INPUT_PROPERTIES) List<InputProperty> inputProperties,
             @JsonProperty(FIELD_NAME_OUTPUT_TYPE) RowType outputType,
             @JsonProperty(FIELD_NAME_DESCRIPTION) String description) {
@@ -82,6 +85,7 @@ public StreamExecCalc(
                 persistedConfig,
                 projection,
                 condition,
+                partialDeleteKeys,
                 TableStreamOperator.class,
                 true, // retainHeader
                 inputProperties,

diff --git a/...ble-planner/src/main/scala/org/apache/flink/table/planner/codegen/CalcCodeGenerator.scala b/...ble-planner/src/main/scala/org/apache/flink/table/planner/codegen/CalcCodeGenerator.scala
@@ -23,10 +23,12 @@ import org.apache.flink.table.api.{TableException, ValidationException}
 import org.apache.flink.table.data.{BoxedWrapperRowData, RowData}
 import org.apache.flink.table.functions.FunctionKind
 import org.apache.flink.table.planner.calcite.{FlinkRexBuilder, FlinkTypeFactory}
+import org.apache.flink.table.planner.codegen.CodeGenUtils.className
 import org.apache.flink.table.planner.functions.bridging.BridgingSqlFunction
 import org.apache.flink.table.runtime.generated.GeneratedFunction
 import org.apache.flink.table.runtime.operators.CodeGenOperatorFactory
 import org.apache.flink.table.types.logical.RowType
+import org.apache.flink.types.RowKind
 
 import org.apache.calcite.rex._
 
@@ -40,6 +42,7 @@ object CalcCodeGenerator {
       outputType: RowType,
       projection: Seq[RexNode],
       condition: Option[RexNode],
+      partialDeleteKeys: Array[Int],
       typeFactory: FlinkTypeFactory,
       retainHeader: Boolean = false,
       opName: String): CodeGenOperatorFactory[RowData] = {
@@ -52,6 +55,7 @@ object CalcCodeGenerator {
       classOf[BoxedWrapperRowData],
       projection,
       condition,
+      Option(partialDeleteKeys),
       typeFactory,
       inputTerm,
       CodeGenUtils.DEFAULT_OPERATOR_COLLECTOR_TERM,
@@ -92,6 +96,7 @@ object CalcCodeGenerator {
       outRowClass,
       calcProjection,
       calcCondition,
+      None,
       typeFactory,
       inputTerm,
       collectorTerm = collectorTerm,
@@ -117,6 +122,7 @@ object CalcCodeGenerator {
       outRowClass: Class[_ <: RowData],
       projection: Seq[RexNode],
       condition: Option[RexNode],
+      partialDeleteKeys: Option[Array[Int]],
       typeFactory: FlinkTypeFactory,
       inputTerm: String = CodeGenUtils.DEFAULT_INPUT1_TERM,
       collectorTerm: String = CodeGenUtils.DEFAULT_OPERATOR_COLLECTOR_TERM,
@@ -146,28 +152,71 @@ object CalcCodeGenerator {
       s"${OperatorCodeGenerator.generateCollect(resultTerm)}"
     }
 
-    def produceProjectionCode: String = {
-      val projection = rexProgram.getProjectList.asScala
+    def produceFullProjectionCode: String = {
+      val fullProjectList = rexProgram.getProjectList.asScala
 
-      val projectionExprs = projection.map(exprGenerator.generateExpression)
-      val projectionExpression =
-        exprGenerator.generateResultExpression(projectionExprs, outRowType, outRowClass)
+      val expressions = fullProjectList.map(exprGenerator.generateExpression)
+      val resultExpression =
+        exprGenerator.generateResultExpression(expressions, outRowType, outRowClass)
 
-      val projectionExpressionCode = projectionExpression.code
+      val projectionExpressionCode = resultExpression.code
 
       val header = if (retainHeader) {
-        s"${projectionExpression.resultTerm}.setRowKind($inputTerm.getRowKind());"
+        s"${resultExpression.resultTerm}.setRowKind($inputTerm.getRowKind());"
       } else {
         ""
       }
 
       s"""
          |$header
          |$projectionExpressionCode
-         |${produceOutputCode(projectionExpression.resultTerm)}
+         |${produceOutputCode(resultExpression.resultTerm)}
          |""".stripMargin
     }
 
+    def produceProjectionCode: String = partialDeleteKeys match {
+      case Some(keys) =>
+        // In case of partial deletes, the calc must only forward the key columns, non-key
+        // expressions must not be evaluated.
+        //
+        // Any RexLocalRef sub-expression evaluated while generating either branch (e.g. the
+        // BinaryRowWriter code backing a ROW(...) constructor) would otherwise be hoisted, by
+        // default, to the bottom (unconditional) local-ref cache scope and run for *every* row
+        // regardless of which branch is actually taken. Each branch is therefore generated
+        // inside its own pushed local-ref scope (mirroring
+        // ExprCodeGenerator.visitOperandInScopedCache, used for CASE/AND/OR short-circuiting) so
+        // that such code is folded into that branch only and never runs unconditionally.
+        ctx.pushLocalRefScope()
+        val keyProjection = buildKeyProjections(exprGenerator, rexProgram, outRowType, keys)
+        val resultExpression =
+          exprGenerator.generateResultExpression(keyProjection, outRowType, outRowClass)
+        val keyScopedCode = ctx.popLocalRefScope().values.map(_.code).mkString("\n")
+
+        val keyHeader = if (retainHeader) {
+          s"${resultExpression.resultTerm}.setRowKind($inputTerm.getRowKind());"
+        } else {
+          ""
+        }
+
+        ctx.pushLocalRefScope()
+        val fullProjectionCode = produceFullProjectionCode
+        val fullScopedCode = ctx.popLocalRefScope().values.map(_.code).mkString("\n")
+
+        s"""
+           |if ($inputTerm.getRowKind() == ${className[RowKind]}.DELETE) {
+           |  $keyScopedCode
+           |  ${resultExpression.code}
+           |  $keyHeader
+           |  ${produceOutputCode(resultExpression.resultTerm)}
+           |} else {
+           |  $fullScopedCode
+           |  $fullProjectionCode
+           |}
+           |""".stripMargin
+      case None =>
+        produceFullProjectionCode
+    }
+
     if (condition.isEmpty && onlyFilter) {
       throw new TableException(
         "This calc has no useful projection and no filter. " +
@@ -233,6 +282,29 @@ object CalcCodeGenerator {
     }
   }
 
+  /**
+   * Builds the key-only projection's per-output-column expressions for a Calc forwarding partial
+   * delete changes. Output columns listed in `partialDeleteKeys` are evaluated from the regular
+   * projection; every other column is generated directly as a typed `NULL` so that its (potentially
+   * unsafe) expression is never evaluated on a delete-by-key tombstone, whose non-key columns may
+   * not be present.
+   */
+  private def buildKeyProjections(
+      exprGenerator: ExprCodeGenerator,
+      rexProgram: RexProgram,
+      outRowType: RowType,
+      partialDeleteKeys: Array[Int]): Seq[GeneratedExpression] = {
+    val keyIndices = partialDeleteKeys.toSet
+    rexProgram.getProjectList.asScala.zipWithIndex.map {
+      case (projectRef, idx) =>
+        if (keyIndices.contains(idx)) {
+          exprGenerator.generateExpression(projectRef)
+        } else {
+          GenerateUtils.generateNullLiteral(outRowType.getTypeAt(idx).copy(true))
+        }
+    }
+  }
+
   private def buildRexProgram(
       typeFactory: FlinkTypeFactory,
       inputType: RowType,

diff --git a/.../scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalCalc.scala b/.../scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalCalc.scala
@@ -18,6 +18,8 @@
 package org.apache.flink.table.planner.plan.nodes.physical.stream
 
 import org.apache.flink.table.planner.calcite.FlinkTypeFactory
+import org.apache.flink.table.planner.plan.`trait`.{DeleteKind, DeleteKindTraitDef}
+import org.apache.flink.table.planner.plan.metadata.FlinkRelMetadataQuery
 import org.apache.flink.table.planner.plan.nodes.exec.{ExecNode, InputProperty}
 import org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecCalc
 import org.apache.flink.table.planner.utils.ShortcutUtils.unwrapTableConfig
@@ -55,8 +57,43 @@ class StreamPhysicalCalc(
       unwrapTableConfig(this),
       projection,
       condition,
+      partialDeleteKeys,
       InputProperty.DEFAULT,
       FlinkTypeFactory.toLogicalRowType(getRowType),
       getRelDetailedDescription)
   }
+
+  /**
+   * If this Calc forwards DELETE_BY_KEY changes, the rest of a delete-by-key tombstone's row may
+   * not be present (see DeleteKind.DELETE_BY_KEY). Returns the output column indices to keep
+   * (evaluated from the regular projection); every other column is handled separately in code
+   * generation as a typed NULL. Returns `null` when this Calc does not need such handling, or no
+   * output key column could be identified (in which case the full projection is always evaluated).
+   */
+  private def partialDeleteKeys: Array[Int] = {
+    val deleteKind = Option(getTraitSet.getTrait(DeleteKindTraitDef.INSTANCE))
+      .map(_.deleteKind)
+      .getOrElse(DeleteKind.NONE)
+    if (deleteKind != DeleteKind.DELETE_BY_KEY) {
+      return null
+    }
+
+    val outputUpsertKeys = FlinkRelMetadataQuery
+      .reuseOrCreate(cluster.getMetadataQuery)
+      .getUpsertKeys(this)
+    if (outputUpsertKeys == null || outputUpsertKeys.isEmpty) {
+      // no identifiable output key column: fall back to always evaluating the full projection
+      return null
+    }
+
+    // Every column in every candidate is, by construction of
+    // FlinkRelMdUniqueKeys.getProjectUniqueKeys, guaranteed to be a trivial
+    // pass-through of an input field - never a risky expression to evaluate.
+    val keyIndices = outputUpsertKeys.flatMap(bitSet => bitSet.map(_.intValue())).toSet.toArray
+    if (keyIndices.nonEmpty) {
+      keyIndices
+    } else {
+      null
+    }
+  }
 }
diff --git a/...apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala b/...apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala
@@ -1433,7 +1433,7 @@ class FlinkChangelogModeInferenceProgram extends FlinkOptimizeProgram[StreamOpti
 
         // if the condition is applied on the upsert key, we can emit whatever the requiredTrait
         // is, because we will filter all records based on the condition that applies to that key
-        case calc: StreamPhysicalCalcBase =>
+        case calc: StreamPhysicalCalc =>
           if (
             requiredTrait == DeleteKindTrait.DELETE_BY_KEY &&
             isNonUpsertKeyCondition(calc)
@@ -1449,6 +1449,21 @@ class FlinkChangelogModeInferenceProgram extends FlinkOptimizeProgram[StreamOpti
             }
           }
 
+        // Unlike StreamPhysicalCalc, other Calc nodes do not skip evaluating non-key expressions
+        // for a delete-by-key tombstone. We are conservative by default and never forward
+        // DELETE_BY_KEY.
+        case _: StreamPhysicalCalcBase =>
+          if (requiredTrait == DeleteKindTrait.DELETE_BY_KEY) {
+            None
+          } else {
+            visitChildren(rel, requiredTrait) match {
+              case None => None
+              case Some(children) =>
+                val childTrait = children.head.getTraitSet.getTrait(DeleteKindTraitDef.INSTANCE)
+                createNewNode(rel, Some(children), childTrait)
+            }
+          }
+
         case _: StreamPhysicalExchange | _: StreamPhysicalExpand |
             _: StreamPhysicalMiniBatchAssigner | _: StreamPhysicalDropUpdateBefore =>
           // transparent forward requiredTrait to children