This is an automated email from the ASF dual-hosted git repository.
fhueske 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 d554162df59 [FLINK-40477][table] Fix constraint enforcer for partial
deletes (#29067)
d554162df59 is described below
commit d554162df59e643b256ee1a2de0e12a8e045a317
Author: Fabian Hueske <[email protected]>
AuthorDate: Thu Sep 3 12:52:05 2026 +0200
[FLINK-40477][table] Fix constraint enforcer for partial deletes (#29067)
The sink NOT NULL enforcer inspected every row regardless of RowKind,
so a by-key (partial) delete tombstone, whose non-key columns are
legitimately null,
was rejected (ERROR) or silently dropped (DROP), losing the delete.
Relax NOT NULL for the non-key columns of DELETE rows in key-only-delete
pipelines;
key columns and all other row kinds stay enforced.
Co-Generated: Claude Opus 4.8 (1M context)
---
.../plan/nodes/exec/common/CommonExecSink.java | 13 +-
.../stream/ConstraintEnforcerSemanticTests.java | 8 +-
.../stream/ConstraintEnforcerTestPrograms.java | 200 +++++++++++++++++++++
.../constraint/ConstraintEnforcerExecutor.java | 48 ++++-
.../sink/constraint/NotNullConstraint.java | 18 +-
5 files changed, 278 insertions(+), 9 deletions(-)
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecSink.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecSink.java
index 3738840ab27..ebfc58b91b6 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecSink.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecSink.java
@@ -198,7 +198,12 @@ public abstract class CommonExecSink extends
ExecNodeBase<Object>
upsertMaterialize && (!inputInsertOnly ||
detectsDuplicateKeys(conflictStrategy));
Transformation<RowData> sinkTransform =
- applyConstraintValidations(inputTransform, config,
persistedRowType);
+ applyConstraintValidations(
+ inputTransform,
+ config,
+ persistedRowType,
+ primaryKeys,
+ inputChangelogMode.keyOnlyDeletes());
if (hasPk) {
sinkTransform =
@@ -270,10 +275,14 @@ public abstract class CommonExecSink extends
ExecNodeBase<Object>
private Transformation<RowData> applyConstraintValidations(
Transformation<RowData> inputTransform,
ExecNodeConfig config,
- RowType physicalRowType) {
+ RowType physicalRowType,
+ int[] primaryKeys,
+ boolean keyOnlyDeletes) {
final Optional<ConstraintEnforcerExecutor> enforcerExecutor =
ConstraintEnforcerExecutor.create(
physicalRowType,
+ primaryKeys,
+ keyOnlyDeletes,
config.get(ExecutionConfigOptions.TABLE_EXEC_SINK_NOT_NULL_ENFORCER),
config.get(ExecutionConfigOptions.TABLE_EXEC_SINK_TYPE_LENGTH_ENFORCER),
config.get(
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ConstraintEnforcerSemanticTests.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ConstraintEnforcerSemanticTests.java
index 5f08ddea816..7e0cac96748 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ConstraintEnforcerSemanticTests.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ConstraintEnforcerSemanticTests.java
@@ -44,6 +44,12 @@ public class ConstraintEnforcerSemanticTests extends
SemanticTestBase {
ConstraintEnforcerTestPrograms.CHAR_LENGTH_TRIM_PAD_WITH_NULLABLE_COLUMNS,
ConstraintEnforcerTestPrograms.BINARY_LENGTH_TRIM_PAD_WITH_NULLABLE_COLUMNS,
ConstraintEnforcerTestPrograms.CHAR_LENGTH_ERROR_WITH_NULLABLE_COLUMNS,
-
ConstraintEnforcerTestPrograms.BINARY_LENGTH_ERROR_WITH_NULLABLE_COLUMNS);
+
ConstraintEnforcerTestPrograms.BINARY_LENGTH_ERROR_WITH_NULLABLE_COLUMNS,
+ ConstraintEnforcerTestPrograms.NOT_NULL_ERROR_DELETE_BY_KEY,
+ ConstraintEnforcerTestPrograms.NOT_NULL_DROP_DELETE_BY_KEY,
+
ConstraintEnforcerTestPrograms.NOT_NULL_ERROR_INSERT_IN_DELETE_BY_KEY,
+
ConstraintEnforcerTestPrograms.NOT_NULL_ERROR_DELETE_BY_KEY_NULL_KEY,
+
ConstraintEnforcerTestPrograms.NOT_NULL_ERROR_DELETE_BY_KEY_NESTED_ROW,
+
ConstraintEnforcerTestPrograms.NOT_NULL_ERROR_DELETE_BY_KEY_COMPOSITE_PK);
}
}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ConstraintEnforcerTestPrograms.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ConstraintEnforcerTestPrograms.java
index f777486c221..5f03fb5f815 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ConstraintEnforcerTestPrograms.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ConstraintEnforcerTestPrograms.java
@@ -28,6 +28,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 javax.annotation.Nullable;
@@ -937,6 +938,205 @@ public class ConstraintEnforcerTestPrograms {
.runSql("INSERT INTO sink_t SELECT * FROM source_t")
.build();
+ //
------------------------------------------------------------------------------------------
+ // Delete-by-key: a by-key delete might carry null in its non-key columns,
regardless
+ // of nullability constraints. The constraint enforcer should only check
key columns for
+ // by-key delete messages and ignores all value columns.
+ //
------------------------------------------------------------------------------------------
+
+ public static final String SCHEMA_DELETE_BY_KEY_NOT_NULL =
+ "id INT PRIMARY KEY NOT ENFORCED, v INT NOT NULL";
+
+ static final TableTestProgram NOT_NULL_ERROR_DELETE_BY_KEY =
+ TableTestProgram.of(
+ "constraint-enforcer-error-not-null-delete-by-key",
+ "validates that a by-key delete carrying null in a
scalar NOT NULL"
+ + " non-key column is applied rather than
rejected by the NOT"
+ + " NULL enforcer (ERROR strategy)")
+ .setupConfig(TABLE_EXEC_SINK_NOT_NULL_ENFORCER,
NotNullEnforcer.ERROR)
+ .setupTableSource(
+ SourceTestStep.newBuilder("source_t")
+ .addSchema(SCHEMA_DELETE_BY_KEY_NOT_NULL)
+ .addOption("changelog-mode", "I,UA,D")
+
.addOption("source.produces-delete-by-key", "true")
+ .producedValues(
+ Row.ofKind(RowKind.INSERT, 1, 10),
+ Row.ofKind(RowKind.INSERT, 2, 20),
+ // tombstone: key only, NOT NULL
value column is null
+ Row.ofKind(RowKind.DELETE, 1,
null))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink_t")
+ .addSchema(SCHEMA_DELETE_BY_KEY_NOT_NULL)
+ .addOption("changelog-mode", "I,UA,D")
+ .addOption("sink.supports-delete-by-key",
"true")
+ .testMaterializedData()
+ .consumedValues("+I[2, 20]")
+ .build())
+ .runSql("INSERT INTO sink_t SELECT id, v FROM source_t")
+ .build();
+
+ static final TableTestProgram NOT_NULL_DROP_DELETE_BY_KEY =
+ TableTestProgram.of(
+ "constraint-enforcer-drop-not-null-delete-by-key",
+ "validates that a by-key delete carrying null in a
scalar NOT NULL"
+ + " non-key column is applied rather than
silently dropped by the"
+ + " NOT NULL enforcer (DROP strategy),
which would leave a stale"
+ + " row")
+ .setupConfig(TABLE_EXEC_SINK_NOT_NULL_ENFORCER,
NotNullEnforcer.DROP)
+ .setupTableSource(
+ SourceTestStep.newBuilder("source_t")
+ .addSchema(SCHEMA_DELETE_BY_KEY_NOT_NULL)
+ .addOption("changelog-mode", "I,UA,D")
+
.addOption("source.produces-delete-by-key", "true")
+ .producedValues(
+ Row.ofKind(RowKind.INSERT, 1, 10),
+ Row.ofKind(RowKind.INSERT, 2, 20),
+ // tombstone: key only, NOT NULL
value column is null
+ Row.ofKind(RowKind.DELETE, 1,
null))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink_t")
+ .addSchema(SCHEMA_DELETE_BY_KEY_NOT_NULL)
+ .addOption("changelog-mode", "I,UA,D")
+ .addOption("sink.supports-delete-by-key",
"true")
+ .testMaterializedData()
+ .consumedValues("+I[2, 20]")
+ .build())
+ .runSql("INSERT INTO sink_t SELECT id, v FROM source_t")
+ .build();
+
+ static final TableTestProgram NOT_NULL_ERROR_INSERT_IN_DELETE_BY_KEY =
+ TableTestProgram.of(
+
"constraint-enforcer-error-not-null-insert-in-delete-by-key",
+ "validates that in a key-only-delete pipeline an
INSERT carrying null in"
+ + " a NOT NULL column is still rejected by
the NOT NULL enforcer")
+ .setupConfig(TABLE_EXEC_SINK_NOT_NULL_ENFORCER,
NotNullEnforcer.ERROR)
+ .setupTableSource(
+ SourceTestStep.newBuilder("source_t")
+ .addSchema(SCHEMA_DELETE_BY_KEY_NOT_NULL)
+ .addOption("changelog-mode", "I,UA,D")
+
.addOption("source.produces-delete-by-key", "true")
+ .producedValues(
+ // genuine violation: an INSERT
with a null NOT NULL
+ // value
+ Row.ofKind(RowKind.INSERT, 1,
null))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink_t")
+ .addSchema(SCHEMA_DELETE_BY_KEY_NOT_NULL)
+ .addOption("changelog-mode", "I,UA,D")
+ .addOption("sink.supports-delete-by-key",
"true")
+ .consumedValues(new Row[0])
+ .build())
+ .runFailingSql(
+ "INSERT INTO sink_t SELECT id, v FROM source_t",
+ TableRuntimeException.class,
+ "Column 'v' is NOT NULL, however, a null value is
being written into it."
+ + " You can set job configuration"
+ + "
'table.exec.sink.not-null-enforcer'='DROP' to suppress this"
+ + " exception and drop such records
silently.")
+ .build();
+
+ static final TableTestProgram NOT_NULL_ERROR_DELETE_BY_KEY_NULL_KEY =
+ TableTestProgram.of(
+
"constraint-enforcer-error-not-null-delete-by-key-null-key",
+ "validates that a by-key delete carrying null in
its NOT NULL key column"
+ + " is still rejected by the NOT NULL
enforcer")
+ .setupConfig(TABLE_EXEC_SINK_NOT_NULL_ENFORCER,
NotNullEnforcer.ERROR)
+ .setupTableSource(
+ SourceTestStep.newBuilder("source_t")
+ .addSchema(SCHEMA_DELETE_BY_KEY_NOT_NULL)
+ .addOption("changelog-mode", "I,UA,D")
+
.addOption("source.produces-delete-by-key", "true")
+ .producedValues(
+ // genuine violation: the key
column is null on a delete
+ Row.ofKind(RowKind.DELETE, null,
null))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink_t")
+ .addSchema(SCHEMA_DELETE_BY_KEY_NOT_NULL)
+ .addOption("changelog-mode", "I,UA,D")
+ .addOption("sink.supports-delete-by-key",
"true")
+ .consumedValues(new Row[0])
+ .build())
+ .runFailingSql(
+ "INSERT INTO sink_t SELECT id, v FROM source_t",
+ TableRuntimeException.class,
+ "Column 'id' is NOT NULL, however, a null value is
being written into it."
+ + " You can set job configuration"
+ + "
'table.exec.sink.not-null-enforcer'='DROP' to suppress this"
+ + " exception and drop such records
silently.")
+ .build();
+
+ public static final String SCHEMA_DELETE_BY_KEY_NESTED_NOT_NULL =
+ "id INT PRIMARY KEY NOT ENFORCED, v ROW<a INT NOT NULL, b INT> NOT
NULL";
+
+ static final TableTestProgram NOT_NULL_ERROR_DELETE_BY_KEY_NESTED_ROW =
+ TableTestProgram.of(
+
"constraint-enforcer-error-not-null-delete-by-key-nested-row",
+ "validates that a by-key delete carrying a null
nested NOT NULL ROW"
+ + " non-key column is applied rather than
rejected, with nested"
+ + " constraint checking enabled")
+ .setupConfig(TABLE_EXEC_SINK_NESTED_CONSTRAINT_ENFORCER,
NestedEnforcer.ROWS)
+ .setupConfig(TABLE_EXEC_SINK_NOT_NULL_ENFORCER,
NotNullEnforcer.ERROR)
+ .setupTableSource(
+ SourceTestStep.newBuilder("source_t")
+
.addSchema(SCHEMA_DELETE_BY_KEY_NESTED_NOT_NULL)
+ .addOption("changelog-mode", "I,UA,D")
+
.addOption("source.produces-delete-by-key", "true")
+ .producedValues(
+ Row.ofKind(RowKind.INSERT, 1,
Row.of(10, 100)),
+ Row.ofKind(RowKind.INSERT, 2,
Row.of(20, 200)),
+ // tombstone: key only, the nested
NOT NULL ROW is null
+ Row.ofKind(RowKind.DELETE, 1,
null))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink_t")
+
.addSchema(SCHEMA_DELETE_BY_KEY_NESTED_NOT_NULL)
+ .addOption("changelog-mode", "I,UA,D")
+ .addOption("sink.supports-delete-by-key",
"true")
+ .testMaterializedData()
+ .consumedValues("+I[2, +I[20, 200]]")
+ .build())
+ .runSql("INSERT INTO sink_t SELECT id, v FROM source_t")
+ .build();
+
+ // Multi-column, non-leading primary key: the key mask is index-based (not
positional), so on a
+ // by-key delete the NOT NULL value columns on both sides of the key (v1,
v2) are relaxed while
+ // both key columns (id1, id2) are still enforced. Guards against a
positional regression.
+ public static final String SCHEMA_DELETE_BY_KEY_COMPOSITE_PK =
+ "v1 INT NOT NULL, id1 INT, id2 INT, v2 INT NOT NULL, PRIMARY KEY
(id1, id2) NOT ENFORCED";
+
+ static final TableTestProgram NOT_NULL_ERROR_DELETE_BY_KEY_COMPOSITE_PK =
+ TableTestProgram.of(
+
"constraint-enforcer-error-not-null-delete-by-key-composite-pk",
+ "validates that a by-key delete with a
multi-column, non-leading primary"
+ + " key is applied: value columns around
the key are relaxed"
+ + " while both key columns remain
enforced")
+ .setupConfig(TABLE_EXEC_SINK_NOT_NULL_ENFORCER,
NotNullEnforcer.ERROR)
+ .setupTableSource(
+ SourceTestStep.newBuilder("source_t")
+
.addSchema(SCHEMA_DELETE_BY_KEY_COMPOSITE_PK)
+ .addOption("changelog-mode", "I,UA,D")
+
.addOption("source.produces-delete-by-key", "true")
+ .producedValues(
+ Row.ofKind(RowKind.INSERT, 10, 1,
2, 20),
+ Row.ofKind(RowKind.INSERT, 11, 3,
4, 21),
+ // tombstone: only the key columns
id1, id2 are set
+ Row.ofKind(RowKind.DELETE, null,
1, 2, null))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink_t")
+
.addSchema(SCHEMA_DELETE_BY_KEY_COMPOSITE_PK)
+ .addOption("changelog-mode", "I,UA,D")
+ .addOption("sink.supports-delete-by-key",
"true")
+ .testMaterializedData()
+ .consumedValues("+I[11, 3, 4, 21]")
+ .build())
+ .runSql("INSERT INTO sink_t SELECT v1, id1, id2, v2 FROM
source_t")
+ .build();
+
private static Map<Long, Long> mapOfNullable(@Nullable Long key, @Nullable
Long value) {
final Map<Long, Long> map = new HashMap<>();
map.put(key, value);
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/sink/constraint/ConstraintEnforcerExecutor.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/sink/constraint/ConstraintEnforcerExecutor.java
index b8377af468e..8c64b4a71b5 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/sink/constraint/ConstraintEnforcerExecutor.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/sink/constraint/ConstraintEnforcerExecutor.java
@@ -56,6 +56,7 @@ import java.util.stream.Stream;
public class ConstraintEnforcerExecutor implements Serializable {
private static final long serialVersionUID = 1L;
+ private static final int[] NO_PRIMARY_KEYS = new int[0];
private final Constraint[] constraints;
private ConstraintEnforcerExecutor(Constraint[] constraints) {
@@ -68,18 +69,44 @@ public class ConstraintEnforcerExecutor implements
Serializable {
public static Optional<ConstraintEnforcerExecutor> create(
final RowType physicalType,
+ final int[] primaryKeys,
+ final boolean keyOnlyDeletes,
final NotNullEnforcer notNullEnforcer,
final TypeLengthEnforcer typeLengthEnforcer,
final NestedEnforcer nestedConstraints) {
final Constraint[] topLevelConstraints =
createConstraints(
- physicalType, notNullEnforcer, typeLengthEnforcer,
nestedConstraints);
+ physicalType,
+ keyOnlyDeletes,
+ primaryKeys,
+ notNullEnforcer,
+ typeLengthEnforcer,
+ nestedConstraints);
return create(topLevelConstraints);
}
+ private static Constraint[] createNestedConstraints(
+ final RowType physicalType,
+ final NotNullEnforcer notNullEnforcer,
+ final TypeLengthEnforcer typeLengthEnforcer,
+ final NestedEnforcer nestedEnforcer) {
+ // Nested constraints are never relaxed for partial deletes: a by-key
delete nulls the whole
+ // top-level field, so the nested constraint skips it (it only
descends into non-null
+ // fields).
+ return createConstraints(
+ physicalType,
+ false,
+ NO_PRIMARY_KEYS,
+ notNullEnforcer,
+ typeLengthEnforcer,
+ nestedEnforcer);
+ }
+
private static Constraint[] createConstraints(
final RowType physicalType,
+ final boolean keyOnlyDeletes,
+ final int[] primaryKeys,
final NotNullEnforcer notNullEnforcer,
final TypeLengthEnforcer typeLengthEnforcer,
final NestedEnforcer nestedEnforcer) {
@@ -94,11 +121,22 @@ public class ConstraintEnforcerExecutor implements
Serializable {
.mapToObj(idx -> fieldNames[idx])
.toArray(String[]::new);
+ final BitSet primaryKeySet = new BitSet();
+ for (int pk : primaryKeys) {
+ primaryKeySet.set(pk);
+ }
+ final boolean[] keyField = new boolean[notNullFieldIndices.length];
+ for (int i = 0; i < notNullFieldIndices.length; i++) {
+ keyField[i] = primaryKeySet.get(notNullFieldIndices[i]);
+ }
+
constraints.add(
new NotNullConstraint(
NotNullEnforcementStrategy.of(notNullEnforcer),
notNullFieldIndices,
- notNullFieldNames));
+ notNullFieldNames,
+ keyOnlyDeletes,
+ keyField));
}
if (typeLengthEnforcer != TypeLengthEnforcer.IGNORE) {
@@ -144,7 +182,7 @@ public class ConstraintEnforcerExecutor implements
Serializable {
nestedRowInfo.stream()
.map(
r ->
- createConstraints(
+ createNestedConstraints(
r.getFieldType(),
notNullEnforcer,
typeLengthEnforcer,
@@ -184,7 +222,7 @@ public class ConstraintEnforcerExecutor implements
Serializable {
nestedArrayInfos.stream()
.map(
r1 ->
- createConstraints(
+ createNestedConstraints(
new RowType(
List.of(
new
RowType.RowField(
@@ -223,7 +261,7 @@ public class ConstraintEnforcerExecutor implements
Serializable {
nestedMapInfos.stream()
.map(
r1 ->
- createConstraints(
+ createNestedConstraints(
new RowType(
List.of(
new
RowType.RowField(
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/sink/constraint/NotNullConstraint.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/sink/constraint/NotNullConstraint.java
index f2ae509c5b5..984273fc341 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/sink/constraint/NotNullConstraint.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/sink/constraint/NotNullConstraint.java
@@ -21,6 +21,7 @@ package
org.apache.flink.table.runtime.operators.sink.constraint;
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 javax.annotation.Nullable;
@@ -32,20 +33,35 @@ final class NotNullConstraint implements Constraint {
private final NotNullEnforcementStrategy enforcementStrategy;
private final int[] notNullFieldIndices;
private final String[] notNullFieldNames;
+ // Whether the input is a key-only-delete pipeline, in which a DELETE
tombstone carries only its
+ // key and sets all non-key fields to null regardless of nullability
constraints.
+ private final boolean keyOnlyDeletes;
+ // Aligned with notNullFieldIndices: true if the field is part of the
primary key.
+ private final boolean[] keyField;
NotNullConstraint(
NotNullEnforcementStrategy enforcementStrategy,
int[] notNullFieldIndices,
- String[] notNullFieldNames) {
+ String[] notNullFieldNames,
+ boolean keyOnlyDeletes,
+ boolean[] keyField) {
this.enforcementStrategy = enforcementStrategy;
this.notNullFieldIndices = notNullFieldIndices;
this.notNullFieldNames = notNullFieldNames;
+ this.keyOnlyDeletes = keyOnlyDeletes;
+ this.keyField = keyField;
}
@Nullable
@Override
public RowData enforce(RowData input) {
+ // A key-only DELETE legitimately carries null in its non-key columns,
so skip those.
+ // Key columns and all columns of other row kinds are always enforced.
+ final boolean partialDelete = keyOnlyDeletes && input.getRowKind() ==
RowKind.DELETE;
for (int i = 0; i < notNullFieldIndices.length; i++) {
+ if (partialDelete && !keyField[i]) {
+ continue;
+ }
final int index = notNullFieldIndices[i];
if (input.isNullAt(index)) {
switch (enforcementStrategy) {
