From c041a67a6c33ff1d251ab1eecec9ef89d592260d Mon Sep 17 00:00:00 2001 From: Timo Walther Date: Fri, 4 Sep 2026 16:16:50 +0200 Subject: [PATCH] [FLINK-40528][table-planner] Don't evaluate non-key expressions for partial delete StreamExecCalc --- .../plan/nodes/exec/batch/BatchExecCalc.java | 2 + .../nodes/exec/common/CommonExecCalc.java | 18 +- .../nodes/exec/stream/StreamExecCalc.java | 4 + .../planner/codegen/CalcCodeGenerator.scala | 88 ++++++- .../physical/stream/StreamPhysicalCalc.scala | 37 +++ .../FlinkChangelogModeInferenceProgram.scala | 17 +- .../nodes/exec/common/CalcTestPrograms.java | 40 +++ .../nodes/exec/stream/CalcRestoreTest.java | 1 + .../exec/stream/DeletesByKeyPrograms.java | 232 +++++++++++++++--- .../stream/DeletesByKeySemanticTests.java | 14 +- ...ial-delete-with-expression-and-filter.json | 169 +++++++++++++ .../savepoint/_metadata | Bin 0 -> 7176 bytes 12 files changed, 578 insertions(+), 44 deletions(-) create mode 100644 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 create mode 100644 flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-calc_1/calc-partial-delete-with-expression-and-filter/savepoint/_metadata 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 ab9dbca87f1c14..5bec1df741ed60 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 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/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 e1ddbcbc46e6ff..91eed908ad5970 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 @@ 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,6 +59,7 @@ public abstract class CommonExecCalc extends ExecNodeBase 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 projection; @@ -64,6 +67,16 @@ public abstract class CommonExecCalc extends ExecNodeBase @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 projection, @Nullable RexNode condition, + @Nullable int[] partialDeleteKeys, Class operatorBaseClass, boolean retainHeader, List 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 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/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 72b4be1a871605..fc1492acbc26ed 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 StreamExecCalc( ReadableConfig tableConfig, List 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 projection, @JsonProperty(FIELD_NAME_CONDITION) @Nullable RexNode condition, + @JsonProperty(FIELD_NAME_PARTIAL_DELETE_KEYS) @Nullable int[] partialDeleteKeys, @JsonProperty(FIELD_NAME_INPUT_PROPERTIES) List 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/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 966aca0abec22c..4a02a97c576f38 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,17 +152,17 @@ 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 { "" } @@ -164,10 +170,53 @@ object CalcCodeGenerator { 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/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 520a9505d25680..a13938bf78618d 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 8d22466c3eec02..300691a6982a69 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 e7abd8de9407f1..2a1110b3aa2134 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.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 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> 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 c4edc11e8c7352..3460e5973d6535 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 CalcRestoreTest() { public List 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 a66837d1c45421..19e129bbc21602 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 constrains 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 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> 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 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> 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 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 eba93c8bba9cea..7fb7202addd4fa 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 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 00000000000000..9ce21b4ebfb630 --- /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 NOT NULL" + } ], + "primaryKey" : { + "name" : "PK_id", + "type" : "PRIMARY_KEY", + "columns" : [ "id" ] + } + } + } + }, + "abilities" : [ { + "type" : "FilterPushDown", + "predicates" : [ ] + } ] + }, + "outputType" : "ROW<`id` INT NOT NULL, `arr` ARRAY 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 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 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 NOT NULL" + } ], + "type" : "ROW<`EXPR$0` INT NOT NULL, `EXPR$1` ARRAY 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 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> 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 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 0000000000000000000000000000000000000000..6d9f7145db1ecf34c71d273950e29fbdfb603e62 GIT binary patch literal 7176 zcmeGhZHybmb=G$mF66^ba-2v=&?#If74fb=_wI6{jn8qws_~s{hmV}VjCaSrP1d{H z-SNeDQqx381gTP?6{tcLZ4^-<+C-pA75&rxX$3|6DhTxlQVaY+;^Rl#s7Mu{zM0)U z@132CkET&2GScqO$D23rYu=lAn}4hkMF?$zhZVv7pdl14YsSe#V-CElbUyV6+LWrY z#-Y)d|MlnR_FjJVr|r)d-N{ZK$M2;!(s60vJXlf!= zi`KBLM(ZRNug7CJo~&UY@O>ZPA57x!K|`}1lD2?eyh&c^&9hoVT~1YX1vBTFi14xMh> z=)k(z{mHLA_0N%?g&uE)P^f^yifJr@&Nb7>>evCqJ6yy%HX0J|nl7L%y4lbatTVDG zN)%un$(5JHNt_GGgOv)MIn!-x(MEf^$>VI7%u^sLRt7?75LV;7&MXyi6IKfc z7`-1`)AK0ojnA2mMuBjOcFhNfo~@lA3I(bMPilrr(O3_1*}4*}`yGG)$nBPg%3pMM zV9^2?1X1t9&H`J$ZGjpOc5}iF##N&y01@yrfw`<9f$MP-r``E3I*kO-SgbEZ0ZDd{ zg^r_}HK*9pb+&W?4YS2drb#K1dvMHvuB;$74_HH|ZIs$5No=l17TSpraJ8M8rpPQv z%o(j_aL}S?pfvj+$nZ`M#Oznhnz{}u164ZX$Bd?_YITh>1$O;;_(1$OpZqYwr8lTU zk1xA}S4Yr6yn6F*!B(qKk2^Y=GnwJe2=mn?a2m=V*m_aC5~cimTS~I12$czwq`V#O&hQ& zm;`QvI@36-54J6kYe-Z3HqE?2SDPkkGeq0aY4(XwbW^@q$=#bPaZB9;*ae$cw;~3x zlh;T1`__OI*cb=KeCT$8b&T;k@2&xXVqbFNa$ij%c)=}1Cr#qg&QK4!R0}@ULzz-3 z^Kfu&4sa)b@9rY^YW_Gw&8DXKesj;0wGIE^SB0(m|x8LF}C$oPXi7 zv*#wk(DG<#5$i1iJ0jA7E4G?7Vn1>5*?n7{dG7}dndK79CF=@{SGVx)egw`$aaWLQ zUf*M}{=0eoAlHN#gyB3)^PaUTOt6VEd))&Hlh##OXYcrO+MjxB->3S8SNT>8dm8fx zDf-6x&dCRyUXk&L>)c&`;z@b>%b#32FFgLiZCrkg(r?vbH# z1va{aJnT(IMIgp^^#npZ-~V2C?=LR&*nD5lNZ>n}4N1@QpdH5ve!+{Xy-!6qV(g6= z%dBPNF*f4G*splxWlp2oE5m9AJCEuZB>u z%dcWf6BV=BG>yQ`sfHaQg69i()m5#4M;$;ne05*fD(}=TT#r-x8ee^_PH|stYyIDa z%j=>7Z#(vT{p@kfUhwz^e*UNmmt5e8v`p5c=+Xrx7?cq5XQ=3{)U&=Mr(2S{XPyK=YT(P z_4lSM)|L4LWO)-+$h0YE-g)uK&oA#A9e(E{{f~neo&acu;S8JQqhj@7xg`r36l!D5 zpgE&%+6u8*-gq=sg)E;p5QQtLrebQ^+4pXp|J^;`8(FqKl;8M!>JNbFsw?a&n=L>K zS!w=-ymb5#G#szG&A0!t_gAld8UEYipML+N`FHnU@G(NZR!!CuaRt{D84|s9wLT#$ zlQl9GPuJ91B7qYCy43@R4=+BKUM_owyL5?Q$WfOmrfy>Gx+kzF)d`}=ll7>qBqpM% zbS)m8lG9VsSW-?UVe*H9OeSJ&ze0sN6@-OR< zsaH<&pA2Glr8;Mu7O|;D_^%FFCKN)$$5BMp90l8I78?qIM5B5fZS%U0V)*v~?Xch0 z49!^x_LN&Sr`2RVKALjNv529vu1$Q8L1lBxP;?Bx3RK7?cW7g7spkR5V?*kIO1?UwCNbw>+nKGtk(?JljW6x;yuxm^P6}n<$Wp zH4SM6(b1J@e%6tqo*Bsz$eq)U5;8oGOVNGZOZDP9NcT7-^t&*DEZ}&X>=FW_TU?=A z>_ImdA-wv@NpH?(inGOhHdBBfAfATDR&W9VAdWS^h_0rEp4