Skip to content

[FLINK-40303][table] Skip non-key NOT NULL checks for key-only deletes #28883

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Au-Miner wants to merge 2 commits into
base: master
Choose a base branch
from
Open

[FLINK-40303][table] Skip non-key NOT NULL checks for key-only deletes #28883

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
f1a03f5
[FLINK-40303][table] Skip non-key NOT NULL checks for key-only delete…
Au-Miner Jul 30, 2026
ee4dfb9
trigger ci
Au-Miner Aug 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 6 additions & 3 deletions ...r/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecSink.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -188,7 +188,7 @@ protected Transformation<Object> createSinkTransformation(
TableLineageUtils.extractLineageDataset(outputObject);

Transformation<RowData> sinkTransform =
applyConstraintValidations(inputTransform, config, persistedRowType);
applyConstraintValidations(inputTransform, config, persistedRowType, primaryKeys);

if (hasPk) {
sinkTransform =
Expand Down Expand Up @@ -252,14 +252,17 @@ protected Transformation<Object> createSinkTransformation(
private Transformation<RowData> applyConstraintValidations(
Transformation<RowData> inputTransform,
ExecNodeConfig config,
RowType physicalRowType) {
RowType physicalRowType,
int[] primaryKeys) {
final Optional<ConstraintEnforcerExecutor> enforcerExecutor =
ConstraintEnforcerExecutor.create(
physicalRowType,
config.get(ExecutionConfigOptions.TABLE_EXEC_SINK_NOT_NULL_ENFORCER),
config.get(ExecutionConfigOptions.TABLE_EXEC_SINK_TYPE_LENGTH_ENFORCER),
config.get(
ExecutionConfigOptions.TABLE_EXEC_SINK_NESTED_CONSTRAINT_ENFORCER));
ExecutionConfigOptions.TABLE_EXEC_SINK_NESTED_CONSTRAINT_ENFORCER),
inputChangelogMode.keyOnlyDeletes(),
primaryKeys);

return enforcerExecutor
.map(
Expand Down
29 changes: 29 additions & 0 deletions ...test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeletesByKeyPrograms.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -75,6 +75,35 @@ public final class DeletesByKeyPrograms {
.runSql("INSERT INTO sink_t SELECT id, name, `value` FROM source_t")
.build();

public static final TableTestProgram
INSERT_SELECT_DELETE_BY_KEY_DELETE_BY_KEY_WITH_NOT_NULL_SINK =
TableTestProgram.of(
"select-delete-on-key-to-not-null-sink",
"No ChangelogNormalize: validates that key-only delete can contain null"
+ " non-key fields when writing to a sink with NOT NULL constraints")
.setupTableSource(
SourceTestStep.newBuilder("source_t")
.addSchema(
"id INT PRIMARY KEY NOT ENFORCED",
"name STRING")
.addOption("changelog-mode", "I,UA,D")
.addOption("source.produces-delete-by-key", "true")
.producedValues(
Row.ofKind(RowKind.INSERT, 1, "Alice"),
Row.ofKind(RowKind.DELETE, 1, null))
.build())
.setupTableSink(
SinkTestStep.newBuilder("sink_t")
.addSchema(
"id INT PRIMARY KEY NOT ENFORCED",
"name STRING NOT NULL")
.addOption("changelog-mode", "I,UA,D")
.addOption("sink.supports-delete-by-key", "true")
.consumedValues("+I[1, Alice]", "-D[1, null]")
.build())
.runSql("INSERT INTO sink_t SELECT id, name FROM source_t")
.build();

public static final TableTestProgram INSERT_SELECT_DELETE_BY_KEY_DELETE_BY_KEY_WITH_PROJECTION =
TableTestProgram.of(
"select-delete-on-key-to-delete-on-key-with-projection",
Expand Down
1 change: 1 addition & 0 deletions ...java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeletesByKeySemanticTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@ public class DeletesByKeySemanticTests extends SemanticTestBase {
public List<TableTestProgram> programs() {
return List.of(
DeletesByKeyPrograms.INSERT_SELECT_DELETE_BY_KEY_DELETE_BY_KEY,
DeletesByKeyPrograms.INSERT_SELECT_DELETE_BY_KEY_DELETE_BY_KEY_WITH_NOT_NULL_SINK,
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,
Expand Down
23 changes: 20 additions & 3 deletions .../org/apache/flink/table/runtime/operators/sink/constraint/ConstraintEnforcerExecutor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,10 +70,16 @@ public static Optional<ConstraintEnforcerExecutor> create(
final RowType physicalType,
final NotNullEnforcer notNullEnforcer,
final TypeLengthEnforcer typeLengthEnforcer,
final NestedEnforcer nestedConstraints) {
final NestedEnforcer nestedConstraints,
final boolean keyOnlyDeletes,
final int[] primaryKeyIndices) {
final Constraint[] topLevelConstraints =
createConstraints(
physicalType, notNullEnforcer, typeLengthEnforcer, nestedConstraints);
physicalType,
notNullEnforcer,
typeLengthEnforcer,
nestedConstraints,
keyOnlyDeletes ? primaryKeyIndices : null);

return create(topLevelConstraints);
}
Expand All @@ -83,6 +89,16 @@ private static Constraint[] createConstraints(
final NotNullEnforcer notNullEnforcer,
final TypeLengthEnforcer typeLengthEnforcer,
final NestedEnforcer nestedEnforcer) {
return createConstraints(
physicalType, notNullEnforcer, typeLengthEnforcer, nestedEnforcer, null);
}

private static Constraint[] createConstraints(
final RowType physicalType,
final NotNullEnforcer notNullEnforcer,
final TypeLengthEnforcer typeLengthEnforcer,
final NestedEnforcer nestedEnforcer,
final @Nullable int[] keyOnlyDeleteFieldIndices) {
final String[] fieldNames = physicalType.getFieldNames().toArray(new String[0]);
final List<Constraint> constraints = new ArrayList<>();

Expand All @@ -98,7 +114,8 @@ private static Constraint[] createConstraints(
new NotNullConstraint(
NotNullEnforcementStrategy.of(notNullEnforcer),
notNullFieldIndices,
notNullFieldNames));
notNullFieldNames,
keyOnlyDeleteFieldIndices));
}

if (typeLengthEnforcer != TypeLengthEnforcer.IGNORE) {
Expand Down
14 changes: 13 additions & 1 deletion ...main/java/org/apache/flink/table/runtime/operators/sink/constraint/NotNullConstraint.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,9 @@
import org.apache.flink.annotation.Internal;
import org.apache.flink.table.api.config.ExecutionConfigOptions;
import org.apache.flink.table.data.RowData;
import org.apache.flink.types.RowKind;

import org.apache.commons.lang3.ArrayUtils;

import javax.annotation.Nullable;

Expand All @@ -32,21 +35,30 @@ final class NotNullConstraint implements Constraint {
private final NotNullEnforcementStrategy enforcementStrategy;
private final int[] notNullFieldIndices;
private final String[] notNullFieldNames;
private final @Nullable int[] keyOnlyDeleteFieldIndices;

NotNullConstraint(
NotNullEnforcementStrategy enforcementStrategy,
int[] notNullFieldIndices,
String[] notNullFieldNames) {
String[] notNullFieldNames,
@Nullable int[] keyOnlyDeleteFieldIndices) {
this.enforcementStrategy = enforcementStrategy;
this.notNullFieldIndices = notNullFieldIndices;
this.notNullFieldNames = notNullFieldNames;
this.keyOnlyDeleteFieldIndices = keyOnlyDeleteFieldIndices;
}

@Nullable
@Override
public RowData enforce(RowData input) {
for (int i = 0; i < notNullFieldIndices.length; i++) {
final int index = notNullFieldIndices[i];
// Non-key fields have no value semantics in a key-only DELETE.
if (input.getRowKind() == RowKind.DELETE
&& keyOnlyDeleteFieldIndices != null
&& !ArrayUtils.contains(keyOnlyDeleteFieldIndices, index)) {
continue;
}
if (input.isNullAt(index)) {
switch (enforcementStrategy) {
case ERROR:
Expand Down