This is an automated email from the ASF dual-hosted git repository.
twalthr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 13f9da4b191 [FLINK-40528][table-planner] Don't evaluate non-key
expressions for partial delete StreamExecCalc
13f9da4b191 is described below
commit 13f9da4b191b122a9f1f6a3b8ec3e88ce4a1a637
Author: Timo Walther <[email protected]>
AuthorDate: Tue Sep 15 09:51:24 2026 +0200
[FLINK-40528][table-planner] Don't evaluate non-key expressions for partial
delete StreamExecCalc
This closes #29106.
---
.../plan/nodes/exec/batch/BatchExecCalc.java | 2 +
.../plan/nodes/exec/common/CommonExecCalc.java | 18 +-
.../plan/nodes/exec/stream/StreamExecCalc.java | 4 +
.../table/planner/codegen/CalcCodeGenerator.scala | 90 +++++++-
.../nodes/physical/stream/StreamPhysicalCalc.scala | 37 ++++
.../FlinkChangelogModeInferenceProgram.scala | 17 +-
.../plan/nodes/exec/common/CalcTestPrograms.java | 40 ++++
.../plan/nodes/exec/stream/CalcRestoreTest.java | 1 +
.../nodes/exec/stream/DeletesByKeyPrograms.java | 232 ++++++++++++++++++---
.../exec/stream/DeletesByKeySemanticTests.java | 14 +-
...-partial-delete-with-expression-and-filter.json | 169 +++++++++++++++
.../savepoint/_metadata | Bin 0 -> 7176 bytes
12 files changed, 579 insertions(+), 45 deletions(-)
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecCalc.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecCalc.java
index ab9dbca87f1..5bec1df741e 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecCalc.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecCalc.java
@@ -69,6 +69,7 @@ public class BatchExecCalc extends CommonExecCalc implements
BatchExecNode<RowDa
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 class BatchExecCalc extends CommonExecCalc implements
BatchExecNode<RowDa
persistedConfig,
projection,
condition,
+ null, // partialDeleteKeys: not applicable in batch mode
TableStreamOperator.class,
false, // retainHeader
inputProperties,
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecCalc.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecCalc.java
index e1ddbcbc46e..91eed908ad5 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecCalc.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecCalc.java
@@ -20,6 +20,7 @@ package org.apache.flink.table.planner.plan.nodes.exec.common;
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.operators.CodeGenOperatorFactory;
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,6 +59,7 @@ 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;
@@ -64,6 +67,16 @@ public abstract class CommonExecCalc extends
ExecNodeBase<RowData>
@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 @@ public abstract class CommonExecCalc extends
ExecNodeBase<RowData>
ReadableConfig persistedConfig,
List<RexNode> projection,
@Nullable RexNode condition,
+ @Nullable int[] partialDeleteKeys,
Class<?> operatorBaseClass,
boolean retainHeader,
List<InputProperty> inputProperties,
@@ -82,6 +96,7 @@ public abstract class CommonExecCalc extends
ExecNodeBase<RowData>
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 @@ public abstract class CommonExecCalc extends
ExecNodeBase<RowData>
(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/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecCalc.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecCalc.java
index 72b4be1a871..fc1492acbc2 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecCalc.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecCalc.java
@@ -52,6 +52,7 @@ public class StreamExecCalc extends CommonExecCalc implements
StreamExecNode<Row
ReadableConfig tableConfig,
List<RexNode> projection,
@Nullable RexNode condition,
+ @Nullable int[] partialDeleteKeys,
InputProperty inputProperty,
RowType outputType,
String description) {
@@ -61,6 +62,7 @@ public class StreamExecCalc extends CommonExecCalc implements
StreamExecNode<Row
ExecNodeContext.newPersistedConfig(StreamExecCalc.class,
tableConfig),
projection,
condition,
+ partialDeleteKeys,
Collections.singletonList(inputProperty),
outputType,
description);
@@ -73,6 +75,7 @@ public class StreamExecCalc extends CommonExecCalc implements
StreamExecNode<Row
@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 class StreamExecCalc extends CommonExecCalc implements
StreamExecNode<Row
persistedConfig,
projection,
condition,
+ partialDeleteKeys,
TableStreamOperator.class,
true, // retainHeader
inputProperties,
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/CalcCodeGenerator.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/CalcCodeGenerator.scala
index 966aca0abec..0e2d438da4b 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/CalcCodeGenerator.scala
+++
b/flink-table/flink-table-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 resultExpressionCode = 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)}
+ |$resultExpressionCode
+ |${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/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalCalc.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalCalc.scala
index 520a9505d25..a13938bf786 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalCalc.scala
+++
b/flink-table/flink-table-planner/src/main/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/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala
index 8d22466c3ee..300691a6982 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/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
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/CalcTestPrograms.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/CalcTestPrograms.java
index e7abd8de940..2a1110b3aa2 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/CalcTestPrograms.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/CalcTestPrograms.java
@@ -29,6 +29,7 @@ import org.apache.flink.table.test.program.SinkTestStep;
import org.apache.flink.table.test.program.SourceTestStep;
import org.apache.flink.table.test.program.TableTestProgram;
import org.apache.flink.types.Row;
+import org.apache.flink.types.RowKind;
import org.apache.flink.types.variant.Variant;
import org.apache.flink.types.variant.VariantBuilder;
@@ -62,6 +63,45 @@ public class CalcTestPrograms {
.runSql("INSERT INTO sink_t SELECT a + 1, b FROM t")
.build();
+ public static final TableTestProgram
CALC_PARTIAL_DELETE_WITH_EXPRESSION_AND_FILTER =
+ TableTestProgram.of(
+ "calc-partial-delete-with-expression-and-filter",
+ "validates that a calc forwarding partial deletes
skips evaluating a"
+ + " non-key row constructor expression for
a delete-by-key row"
+ + " after being restored from a compiled
plan, with a key-safe"
+ + " filter present")
+ .setupTableSource(
+ SourceTestStep.newBuilder("source_t")
+ .addSchema(
+ "id INT PRIMARY KEY NOT ENFORCED",
+ "arr ARRAY<INT> NOT NULL")
+ .addOption("changelog-mode", "I,UA,D")
+
.addOption("source.produces-delete-by-key", "true")
+ .producedBeforeRestore(
+ // Filtered out by the WHERE
clause below
+ Row.ofKind(RowKind.INSERT, 0, new
Integer[] {99}),
+ Row.ofKind(RowKind.INSERT, 1, new
Integer[] {1, 2}),
+ Row.ofKind(RowKind.INSERT, 2, new
Integer[] {3}))
+ .producedAfterRestore(
+ // Delete by key: NOT NULL array
column is null
+ Row.ofKind(RowKind.DELETE, 1,
null),
+ Row.ofKind(
+ RowKind.UPDATE_AFTER, 2,
new Integer[] {3, 4}))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink_t")
+ .addSchema(
+ "id INT PRIMARY KEY NOT ENFORCED",
+ "r ROW<a INT, b ARRAY<INT>> NOT
NULL")
+ .addOption("changelog-mode", "I,UA,D")
+ .addOption("sink.supports-delete-by-key",
"true")
+ .consumedBeforeRestore(
+ "+I[1, +I[1, [1, 2]]]", "+I[2,
+I[2, [3]]]")
+ .consumedAfterRestore("-D[1, null]",
"+U[2, +I[2, [3, 4]]]")
+ .build())
+ .runSql("INSERT INTO sink_t SELECT id, ROW(id, arr) FROM
source_t WHERE id > 0")
+ .build();
+
public static final TableTestProgram CALC_PROJECT_PUSHDOWN =
TableTestProgram.of(
"calc-project-pushdown", "validates calc node with
project pushdown")
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/CalcRestoreTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/CalcRestoreTest.java
index c4edc11e8c7..3460e5973d6 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/CalcRestoreTest.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/CalcRestoreTest.java
@@ -36,6 +36,7 @@ public class CalcRestoreTest extends RestoreTestBase {
public List<TableTestProgram> programs() {
return Arrays.asList(
CalcTestPrograms.SIMPLE_CALC,
+
CalcTestPrograms.CALC_PARTIAL_DELETE_WITH_EXPRESSION_AND_FILTER,
CalcTestPrograms.CALC_FILTER,
CalcTestPrograms.CALC_FILTER_PUSHDOWN,
CalcTestPrograms.CALC_PROJECT_PUSHDOWN,
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeletesByKeyPrograms.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeletesByKeyPrograms.java
index a66837d1c45..c230b2fbfb1 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeletesByKeyPrograms.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeletesByKeyPrograms.java
@@ -18,12 +18,15 @@
package org.apache.flink.table.planner.plan.nodes.exec.stream;
+import org.apache.flink.table.functions.AsyncScalarFunction;
import org.apache.flink.table.test.program.SinkTestStep;
import org.apache.flink.table.test.program.SourceTestStep;
import org.apache.flink.table.test.program.TableTestProgram;
import org.apache.flink.types.Row;
import org.apache.flink.types.RowKind;
+import java.util.concurrent.CompletableFuture;
+
/**
* Tests for verifying semantic of operations when sources produce deletes by
key only and the sink
* can accept deletes by key only as well.
@@ -34,9 +37,9 @@ public final class DeletesByKeyPrograms {
* Tests a simple INSERT INTO SELECT scenario where ChangelogNormalize can
be eliminated since
* we don't need UPDATE_BEFORE, and we have key information for all
changes.
*/
- public static final TableTestProgram
INSERT_SELECT_DELETE_BY_KEY_DELETE_BY_KEY =
+ public static final TableTestProgram DELETE_BY_KEY_DELETE_BY_KEY =
TableTestProgram.of(
- "select-delete-on-key-to-delete-on-key",
+ "delete-by-key-delete-by-key",
"No ChangelogNormalize: validates results when
querying source with deletes by key"
+ " only, writing to sink supporting
deletes by key only, which"
+ " is a case where ChangelogNormalize can
be eliminated")
@@ -45,7 +48,7 @@ public final class DeletesByKeyPrograms {
.addSchema(
"id INT PRIMARY KEY NOT ENFORCED",
"name STRING",
- "`value` INT")
+ "v INT")
.addOption("changelog-mode", "I,UA,D")
.addOption("source.produces-delete-by-key", "true")
.producedValues(
@@ -61,7 +64,7 @@ public final class DeletesByKeyPrograms {
.addSchema(
"id INT PRIMARY KEY NOT ENFORCED",
"name STRING",
- "`value` INT")
+ "v INT")
.addOption(
"changelog-mode",
"I,UA,D") // Insert, UpdateAfter,
Delete
@@ -72,12 +75,12 @@ public final class DeletesByKeyPrograms {
"-D[1, null, null]",
"+U[2, Bob, 30]")
.build())
- .runSql("INSERT INTO sink_t SELECT id, name, `value` FROM
source_t")
+ .runSql("INSERT INTO sink_t SELECT id, name, v FROM
source_t")
.build();
- public static final TableTestProgram
INSERT_SELECT_DELETE_BY_KEY_DELETE_BY_KEY_WITH_PROJECTION =
+ public static final TableTestProgram
DELETE_BY_KEY_DELETE_BY_KEY_WITH_PROJECTION =
TableTestProgram.of(
-
"select-delete-on-key-to-delete-on-key-with-projection",
+ "delete-by-key-delete-by-key-with-projection",
"No ChangelogNormalize: validates results when
querying source with deletes by key"
+ " only, writing to sink supporting
deletes by key only with a"
+ "projection, which is a case where
ChangelogNormalize can be"
@@ -87,7 +90,7 @@ public final class DeletesByKeyPrograms {
.addSchema(
"id INT PRIMARY KEY NOT ENFORCED",
"name STRING NOT NULL",
- "`value` INT NOT NULL")
+ "v INT NOT NULL")
.addOption("changelog-mode", "I,UA,D")
.addOption("source.produces-delete-by-key", "true")
.producedValues(
@@ -103,7 +106,7 @@ public final class DeletesByKeyPrograms {
.addSchema(
"id INT PRIMARY KEY NOT ENFORCED",
"name STRING",
- "`value` INT")
+ "v INT")
.addOption(
"changelog-mode",
"I,UA,D") // Insert, UpdateAfter,
Delete
@@ -111,15 +114,15 @@ public final class DeletesByKeyPrograms {
.consumedValues(
"+I[1, Alice, 12]",
"+I[2, Bob, 22]",
- "-D[1, , -1]",
+ "-D[1, null, null]",
"+U[2, Bob, 32]")
.build())
- .runSql("INSERT INTO sink_t SELECT id, name, `value` + 2
FROM source_t")
+ .runSql("INSERT INTO sink_t SELECT id, name, v + 2 FROM
source_t")
.build();
- public static final TableTestProgram
INSERT_SELECT_DELETE_BY_KEY_FULL_DELETE =
+ public static final TableTestProgram DELETE_BY_KEY_FULL_DELETE =
TableTestProgram.of(
- "select-delete-on-key-to-full-delete",
+ "delete-by-key-full-delete",
"ChangelogNormalize: validates results when
querying source with deletes by key"
+ " only, writing to sink supporting
requiring full deletes, "
+ "which is a case where
ChangelogNormalize stays")
@@ -128,7 +131,7 @@ public final class DeletesByKeyPrograms {
.addSchema(
"id INT PRIMARY KEY NOT ENFORCED",
"name STRING",
- "`value` INT")
+ "v INT")
.addOption("changelog-mode", "I,UA,D")
.addOption("source.produces-delete-by-key", "true")
.producedValues(
@@ -144,7 +147,7 @@ public final class DeletesByKeyPrograms {
.addSchema(
"id INT PRIMARY KEY NOT ENFORCED",
"name STRING",
- "`value` INT")
+ "v INT")
.addOption("changelog-mode", "I,UA,D")
.addOption("sink.supports-delete-by-key",
"false")
.consumedValues(
@@ -153,12 +156,12 @@ public final class DeletesByKeyPrograms {
"-D[1, Alice, 10]",
"+U[2, Bob, 30]")
.build())
- .runSql("INSERT INTO sink_t SELECT id, name, `value` FROM
source_t")
+ .runSql("INSERT INTO sink_t SELECT id, name, v FROM
source_t")
.build();
- public static final TableTestProgram INSERT_SELECT_FULL_DELETE_FULL_DELETE
=
+ public static final TableTestProgram FULL_DELETE_FULL_DELETE =
TableTestProgram.of(
- "select-full-delete-to-full-delete",
+ "full-delete-full-delete",
"No ChangelogNormalize: validates results when
querying source with full deletes, "
+ "writing to sink requiring full deletes,
which is a case"
+ " where ChangelogNormalize can be
eliminated")
@@ -167,7 +170,7 @@ public final class DeletesByKeyPrograms {
.addSchema(
"id INT PRIMARY KEY NOT ENFORCED",
"name STRING",
- "`value` INT")
+ "v INT")
.addOption("changelog-mode", "I,UA,D")
.addOption("source.produces-delete-by-key", "false")
.producedValues(
@@ -183,7 +186,7 @@ public final class DeletesByKeyPrograms {
.addSchema(
"id INT PRIMARY KEY NOT ENFORCED",
"name STRING",
- "`value` INT")
+ "v INT")
.addOption("changelog-mode", "I,UA,D")
.addOption("sink.supports-delete-by-key",
"false")
.consumedValues(
@@ -192,18 +195,18 @@ public final class DeletesByKeyPrograms {
"-D[1, Alice, 10]",
"+U[2, Bob, 30]")
.build())
- .runSql("INSERT INTO sink_t SELECT id, name, `value` FROM
source_t")
+ .runSql("INSERT INTO sink_t SELECT id, name, v FROM
source_t")
.build();
public static final TableTestProgram JOIN_INTO_FULL_DELETES =
TableTestProgram.of(
- "join-to-full-delete",
+ "join-into-full-delete",
"ChangelogNormalize: validates results when
joining sources with deletes by key"
+ " only, writing to sink requiring full
deletes, which"
+ " is a case where ChangelogNormalize
stays")
.setupTableSource(
SourceTestStep.newBuilder("left_t")
- .addSchema("id INT PRIMARY KEY NOT
ENFORCED", "`value` INT")
+ .addSchema("id INT PRIMARY KEY NOT
ENFORCED", "v INT")
.addOption("changelog-mode", "I,UA,D")
.addOption("source.produces-delete-by-key", "true")
.producedValues(
@@ -234,25 +237,25 @@ public final class DeletesByKeyPrograms {
.addSchema(
"id INT PRIMARY KEY NOT ENFORCED",
"name STRING",
- "`value` INT")
+ "v INT")
.addOption("changelog-mode", "I,UA,D")
.addOption("sink.supports-delete-by-key",
"false")
.testMaterializedData()
.consumedValues("+I[3, Emily, 40]", "+I[2,
BOB, 20]")
.build())
.runSql(
- "INSERT INTO sink_t SELECT l.id, r.name, l.`value`
FROM left_t l JOIN right_t r ON l.id = r.id")
+ "INSERT INTO sink_t SELECT l.id, r.name, l.v FROM
left_t l JOIN right_t r ON l.id = r.id")
.build();
public static final TableTestProgram JOIN_INTO_DELETES_BY_KEY =
TableTestProgram.of(
- "join-to-delete-on-key",
+ "join-into-delete-by-key",
"No ChangelogNormalize: validates results when
joining sources with deletes by key"
+ " only, writing to sink supporting
deletes by key, which"
+ " is a case where ChangelogNormalize can
be removed")
.setupTableSource(
SourceTestStep.newBuilder("left_t")
- .addSchema("id INT PRIMARY KEY NOT
ENFORCED", "`value` INT")
+ .addSchema("id INT PRIMARY KEY NOT
ENFORCED", "v INT")
.addOption("changelog-mode", "I,UA,D")
.addOption("source.produces-delete-by-key", "true")
.producedValues(
@@ -283,15 +286,186 @@ public final class DeletesByKeyPrograms {
.addSchema(
"id INT PRIMARY KEY NOT ENFORCED",
"name STRING",
- "`value` INT")
+ "v INT")
.addOption("changelog-mode", "I,UA,D")
.addOption("sink.supports-delete-by-key",
"true")
.testMaterializedData()
.consumedValues("+I[2, BOB, 20]", "+I[3,
Emily, 40]")
.build())
.runSql(
- "INSERT INTO sink_t SELECT l.id, r.name, l.`value`
FROM left_t l JOIN right_t r ON l.id = r.id")
+ "INSERT INTO sink_t SELECT l.id, r.name, l.v FROM
left_t l JOIN right_t r ON l.id = r.id")
+ .build();
+
+ public static final TableTestProgram
DELETE_BY_KEY_DELETE_BY_KEY_WITH_EXPRESSION =
+ TableTestProgram.of(
+ "delete-by-key-delete-by-key-with-expression",
+ "NOT NULL constraints have no effect. The row
constructor expression"
+ + "is not evaluated for partial deletion.")
+ .setupTableSource(
+ SourceTestStep.newBuilder("source_t")
+ .addSchema(
+ "id INT PRIMARY KEY NOT ENFORCED",
+ "arr ARRAY<INT> NOT NULL")
+ .addOption("changelog-mode", "I,UA,D")
+
.addOption("source.produces-delete-by-key", "true")
+ .producedValues(
+ Row.ofKind(RowKind.INSERT, 1, new
Integer[] {1, 2}),
+ Row.ofKind(RowKind.INSERT, 2, new
Integer[] {3}),
+ // Delete by key: NOT NULL array
column is null
+ Row.ofKind(RowKind.DELETE, 1,
null),
+ // Update after only
+ Row.ofKind(
+ RowKind.UPDATE_AFTER, 2,
new Integer[] {3, 4}))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink_t")
+ .addSchema(
+ "id INT PRIMARY KEY NOT ENFORCED",
+ "r ROW<a INT, b ARRAY<INT>> NOT
NULL")
+ .addOption("changelog-mode", "I,UA,D")
+ .addOption("sink.supports-delete-by-key",
"true")
+ .consumedValues(
+ "+I[1, +I[1, [1, 2]]]",
+ "+I[2, +I[2, [3]]]",
+ "-D[1, null]",
+ "+U[2, +I[2, [3, 4]]]")
+ .build())
+ .runSql("INSERT INTO sink_t SELECT id, ROW(id, arr) FROM
source_t")
.build();
+ public static final TableTestProgram
DELETE_BY_KEY_DELETE_BY_KEY_WITH_EXPRESSION_AND_FILTER =
+ TableTestProgram.of(
+
"delete-by-key-delete-by-key-with-expression-and-filter",
+ "A key-safe filter combined with a non-key row
constructor expression."
+ + " Exercises the local-ref scope
isolation between the filter,"
+ + " the key-only projection and the full
projection.")
+ .setupTableSource(
+ SourceTestStep.newBuilder("source_t")
+ .addSchema(
+ "id INT PRIMARY KEY NOT ENFORCED",
+ "arr ARRAY<INT> NOT NULL")
+ .addOption("changelog-mode", "I,UA,D")
+
.addOption("source.produces-delete-by-key", "true")
+ .producedValues(
+ // Filtered out by the WHERE
clause below
+ Row.ofKind(RowKind.INSERT, 0, new
Integer[] {99}),
+ Row.ofKind(RowKind.INSERT, 1, new
Integer[] {1, 2}),
+ Row.ofKind(RowKind.INSERT, 2, new
Integer[] {3}),
+ // Delete by key: NOT NULL array
column is null
+ Row.ofKind(RowKind.DELETE, 1,
null),
+ // Update after only
+ Row.ofKind(
+ RowKind.UPDATE_AFTER, 2,
new Integer[] {3, 4}))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink_t")
+ .addSchema(
+ "id STRING PRIMARY KEY NOT
ENFORCED",
+ "r ROW<a INT, b ARRAY<INT>> NOT
NULL")
+ .addOption("changelog-mode", "I,UA,D")
+ .addOption("sink.supports-delete-by-key",
"true")
+ .consumedValues(
+ "+I[1, +I[1, [1, 2]]]",
+ "+I[2, +I[2, [3]]]",
+ "-D[1, null]",
+ "+U[2, +I[2, [3, 4]]]")
+ .build())
+ .runSql(
+ "INSERT INTO sink_t SELECT CAST(id AS STRING),
ROW(id, arr) "
+ + "FROM source_t WHERE CAST(id AS STRING)
<> '0'")
+ .build();
+
+ public static final TableTestProgram
DELETE_BY_KEY_DELETE_BY_KEY_WITH_DUPLICATE_KEY =
+ TableTestProgram.of(
+ "delete-by-key-delete-by-key-with-duplicate-key",
+ "The same key column is projected twice under
different names, so the"
+ + " calc has multiple candidate upsert
keys for its output. Both"
+ + " must be preserved on a partial delete,
not just one picked"
+ + " arbitrarily.")
+ .setupTableSource(
+ SourceTestStep.newBuilder("source_t")
+ .addSchema(
+ "id INT PRIMARY KEY NOT ENFORCED",
+ "name STRING NOT NULL",
+ "v INT NOT NULL")
+ .addOption("changelog-mode", "I,UA,D")
+
.addOption("source.produces-delete-by-key", "true")
+ .producedValues(
+ Row.ofKind(RowKind.INSERT, 1,
"Alice", 10),
+ Row.ofKind(RowKind.INSERT, 2,
"Bob", 20),
+ // Delete by key
+ Row.ofKind(RowKind.DELETE, 1,
null, null),
+ // Update after only
+ Row.ofKind(RowKind.UPDATE_AFTER,
2, "Bob", 30))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink_t")
+ .addSchema(
+ "id INT",
+ // Injective cast
+ "id2 STRING PRIMARY KEY NOT
ENFORCED",
+ "name STRING NOT NULL",
+ "v INT NOT NULL")
+ .addOption("changelog-mode", "I,UA,D")
+ .addOption("sink.supports-delete-by-key",
"true")
+ .consumedValues(
+ "+I[1, 1, Alice, 12]",
+ "+I[2, 2, Bob, 22]",
+ "-D[1, 1, null, null]",
+ "+U[2, 2, Bob, 32]")
+ .build())
+ .runSql(
+ "INSERT INTO sink_t SELECT id, CAST(id AS STRING)
AS id2, name, v + 2 FROM source_t")
+ .build();
+
+ public static final TableTestProgram
DELETE_BY_KEY_ASYNC_CALC_FALLS_BACK_TO_FULL_DELETE =
+ TableTestProgram.of(
+
"delete-by-key-async-calc-falls-back-to-full-delete",
+ "An async calc invokes a remote function per row,
so the planner is"
+ + " conservative and never lets a
delete-by-key tombstone reach"
+ + " it: a ChangelogNormalize is kept
upstream to materialize"
+ + " the full row instead, even though the
sink itself would"
+ + " accept delete-by-key. Without this,
the delete-by-key"
+ + " tombstone's null (for a NOT NULL
column) would silently"
+ + " leak through the async calc as a
partial delete, instead of"
+ + " the sink receiving the materialized
full row.")
+ .setupTemporaryCatalogFunction("udf1",
IncrementAsyncFunction.class)
+ .setupTableSource(
+ SourceTestStep.newBuilder("source_t")
+ .addSchema(
+ "id INT PRIMARY KEY NOT ENFORCED",
"v BIGINT NOT NULL")
+ .addOption("changelog-mode", "I,UA,D")
+
.addOption("source.produces-delete-by-key", "true")
+ .producedValues(
+ Row.ofKind(RowKind.INSERT, 1, 10L),
+ Row.ofKind(RowKind.INSERT, 2, 20L),
+ // Delete by key: NOT NULL non-key
column is null
+ Row.ofKind(RowKind.DELETE, 1,
null),
+ Row.ofKind(RowKind.UPDATE_AFTER,
2, 30L))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink_t")
+ .addSchema("id INT PRIMARY KEY NOT
ENFORCED", "v2 BIGINT")
+ .addOption("changelog-mode", "I,UA,D")
+ .addOption("sink.supports-delete-by-key",
"true")
+ .consumedValues(
+ "+I[1, 11]",
+ "+I[2, 21]",
+ // Full delete: the previous value
(10) is materialized
+ // by the ChangelogNormalize kept
upstream of the async
+ // calc, not the null carried by
the source's tombstone.
+ "-D[1, 11]",
+ "+U[2, 31]")
+ .build())
+ .runSql("INSERT INTO sink_t SELECT id, udf1(v) FROM
source_t")
+ .build();
+
+ /** Increments a {@code BIGINT} input asynchronously. */
+ public static class IncrementAsyncFunction extends AsyncScalarFunction {
+ public void eval(CompletableFuture<Long> future, Long l) {
+ future.complete(l + 1);
+ }
+ }
+
private DeletesByKeyPrograms() {}
}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeletesByKeySemanticTests.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeletesByKeySemanticTests.java
index eba93c8bba9..7fb7202addd 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeletesByKeySemanticTests.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeletesByKeySemanticTests.java
@@ -29,11 +29,15 @@ public class DeletesByKeySemanticTests extends
SemanticTestBase {
@Override
public List<TableTestProgram> programs() {
return List.of(
- DeletesByKeyPrograms.INSERT_SELECT_DELETE_BY_KEY_DELETE_BY_KEY,
- DeletesByKeyPrograms.INSERT_SELECT_DELETE_BY_KEY_FULL_DELETE,
- DeletesByKeyPrograms.INSERT_SELECT_FULL_DELETE_FULL_DELETE,
-
DeletesByKeyPrograms.INSERT_SELECT_DELETE_BY_KEY_DELETE_BY_KEY_WITH_PROJECTION,
+ DeletesByKeyPrograms.DELETE_BY_KEY_DELETE_BY_KEY,
+ DeletesByKeyPrograms.DELETE_BY_KEY_FULL_DELETE,
+ DeletesByKeyPrograms.FULL_DELETE_FULL_DELETE,
+
DeletesByKeyPrograms.DELETE_BY_KEY_DELETE_BY_KEY_WITH_PROJECTION,
DeletesByKeyPrograms.JOIN_INTO_FULL_DELETES,
- DeletesByKeyPrograms.JOIN_INTO_DELETES_BY_KEY);
+ DeletesByKeyPrograms.JOIN_INTO_DELETES_BY_KEY,
+
DeletesByKeyPrograms.DELETE_BY_KEY_DELETE_BY_KEY_WITH_EXPRESSION,
+
DeletesByKeyPrograms.DELETE_BY_KEY_DELETE_BY_KEY_WITH_EXPRESSION_AND_FILTER,
+
DeletesByKeyPrograms.DELETE_BY_KEY_DELETE_BY_KEY_WITH_DUPLICATE_KEY,
+
DeletesByKeyPrograms.DELETE_BY_KEY_ASYNC_CALC_FALLS_BACK_TO_FULL_DELETE);
}
}
diff --git
a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-calc_1/calc-partial-delete-with-expression-and-filter/plan/calc-partial-delete-with-expression-and-filter.json
b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-calc_1/calc-partial-delete-with-expression-and-filter/plan/calc-partial-delete-with-expression-and-filter.json
new file mode 100644
index 00000000000..9ce21b4ebfb
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-calc_1/calc-partial-delete-with-expression-and-filter/plan/calc-partial-delete-with-expression-and-filter.json
@@ -0,0 +1,169 @@
+{
+ "flinkVersion" : "2.4",
+ "nodes" : [ {
+ "id" : 1,
+ "type" : "stream-exec-table-source-scan_2",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`source_t`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "id",
+ "dataType" : "INT NOT NULL"
+ }, {
+ "name" : "arr",
+ "dataType" : "ARRAY<INT> NOT NULL"
+ } ],
+ "primaryKey" : {
+ "name" : "PK_id",
+ "type" : "PRIMARY_KEY",
+ "columns" : [ "id" ]
+ }
+ }
+ }
+ },
+ "abilities" : [ {
+ "type" : "FilterPushDown",
+ "predicates" : [ ]
+ } ]
+ },
+ "outputType" : "ROW<`id` INT NOT NULL, `arr` ARRAY<INT> NOT NULL>",
+ "description" : "TableSourceScan(table=[[default_catalog,
default_database, source_t, filter=[]]], fields=[id, arr])"
+ }, {
+ "id" : 2,
+ "type" : "stream-exec-calc_1",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "INT NOT NULL"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "ARRAY<INT> NOT NULL"
+ } ],
+ "condition" : {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$>$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "INT NOT NULL"
+ }, {
+ "kind" : "LITERAL",
+ "value" : 0,
+ "type" : "INT NOT NULL"
+ } ],
+ "type" : "BOOLEAN NOT NULL"
+ },
+ "partialDeleteKeys" : [ 0 ],
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`id` INT NOT NULL, `arr` ARRAY<INT> NOT NULL>",
+ "description" : "Calc(select=[id, arr], where=[(id > 0)])"
+ }, {
+ "id" : 3,
+ "type" : "stream-exec-calc_1",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "INT NOT NULL"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "SPECIAL",
+ "internalName" : "$ROW$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "INT NOT NULL"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "ARRAY<INT> NOT NULL"
+ } ],
+ "type" : "ROW<`EXPR$0` INT NOT NULL, `EXPR$1` ARRAY<INT> NOT NULL> NOT
NULL"
+ } ],
+ "condition" : null,
+ "partialDeleteKeys" : [ 0 ],
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`id` INT NOT NULL, `EXPR$1` ROW<`EXPR$0` INT NOT NULL,
`EXPR$1` ARRAY<INT> NOT NULL> NOT NULL>",
+ "description" : "Calc(select=[id, ROW(id, arr) AS EXPR$1])"
+ }, {
+ "id" : 4,
+ "type" : "stream-exec-sink_2",
+ "configuration" : {
+ "table.exec.sink.keyed-shuffle" : "AUTO",
+ "table.exec.sink.not-null-enforcer" : "ERROR",
+ "table.exec.sink.rowtime-inserter" : "ENABLED",
+ "table.exec.sink.type-length-enforcer" : "IGNORE",
+ "table.exec.sink.upsert-materialize" : "AUTO"
+ },
+ "dynamicTableSink" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`sink_t`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "id",
+ "dataType" : "INT NOT NULL"
+ }, {
+ "name" : "r",
+ "dataType" : "ROW<`a` INT, `b` ARRAY<INT>> NOT NULL"
+ } ],
+ "primaryKey" : {
+ "name" : "PK_id",
+ "type" : "PRIMARY_KEY",
+ "columns" : [ "id" ]
+ }
+ }
+ }
+ }
+ },
+ "inputChangelogMode" : [ "INSERT", "UPDATE_AFTER", "~DELETE" ],
+ "upsertMaterializeStrategy" : "VALUE",
+ "inputUpsertKey" : [ 0 ],
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`id` INT NOT NULL, `EXPR$1` ROW<`EXPR$0` INT NOT NULL,
`EXPR$1` ARRAY<INT> NOT NULL> NOT NULL>",
+ "description" : "Sink(table=[default_catalog.default_database.sink_t],
fields=[id, EXPR$1])"
+ } ],
+ "edges" : [ {
+ "source" : 1,
+ "target" : 2,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 2,
+ "target" : 3,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 3,
+ "target" : 4,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ } ]
+}
\ No newline at end of file
diff --git
a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-calc_1/calc-partial-delete-with-expression-and-filter/savepoint/_metadata
b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-calc_1/calc-partial-delete-with-expression-and-filter/savepoint/_metadata
new file mode 100644
index 00000000000..6d9f7145db1
Binary files /dev/null and
b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-calc_1/calc-partial-delete-with-expression-and-filter/savepoint/_metadata
differ
