Skip to content

[core] Allow primary-key-only deletes for deduplicate tables #9978

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

Merged
JingsongLi merged 2 commits into from
Sep 20, 2026
Merged

[core] Allow primary-key-only deletes for deduplicate tables #9978

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
d6d51fe
[core] Allow primary-key-only deletes for deduplicate tables
Akash3121 Sep 19, 2026
6f264de
[core] Test primary-key-only delete writes
Akash3121 Sep 19, 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
35 changes: 32 additions & 3 deletions paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyFileStoreTable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,8 +45,10 @@

import javax.annotation.Nullable;

import java.util.ArrayList;
import java.util.List;
import java.util.function.BiConsumer;
import java.util.stream.Collectors;

import static org.apache.paimon.predicate.PredicateBuilder.and;
import static org.apache.paimon.predicate.PredicateBuilder.pickTransformFieldMapping;
Expand Down Expand Up @@ -81,11 +83,24 @@ public KeyValueFileStore store() {
PrimaryKeyTableUtils.PrimaryKeyFieldsExtractor.EXTRACTOR;

RowType keyType = new RowType(extractor.keyFields(tableSchema));
List<String> deleteNotNullFields = deleteNotNullFields(options);
RowType valueType =
deleteNotNullFields == null
? rowType
: rowType.copy(
rowType.getFields().stream()
.map(
field ->
deleteNotNullFields.contains(
field.name())
? field
: field.copy(true))
.collect(Collectors.toList()));

MergeFunctionFactory<KeyValue> mfFactory =
PrimaryKeyTableUtils.createMergeFunctionFactory(tableSchema);
if (options.needLookup()) {
mfFactory = LookupMergeFunction.wrap(mfFactory, options, keyType, rowType);
mfFactory = LookupMergeFunction.wrap(mfFactory, options, keyType, valueType);
}

lazyStore =
Expand All @@ -97,7 +112,7 @@ public KeyValueFileStore store() {
options,
tableSchema.logicalPartitionType(),
keyType,
rowType,
valueType,
extractor,
mfFactory,
name(),
Expand Down Expand Up @@ -189,6 +204,7 @@ public TableWriteImpl<KeyValue> newPostponeFixedBucketWrite(

private TableWriteImpl<KeyValue> newWrite(AbstractFileStoreWrite<KeyValue> storeWrite) {
KeyValue kv = new KeyValue();
List<String> deleteNotNullFields = deleteNotNullFields(coreOptions());
return new TableWriteImpl<>(
rowType(),
storeWrite,
Expand All @@ -200,7 +216,20 @@ private TableWriteImpl<KeyValue> newWrite(AbstractFileStoreWrite<KeyValue> store
rowKind,
record.row()),
rowKindGenerator(),
RowKindFilter.of(coreOptions()));
RowKindFilter.of(coreOptions()),
deleteNotNullFields);
}

@Nullable
private List<String> deleteNotNullFields(CoreOptions options) {
if (options.mergeEngine() != CoreOptions.MergeEngine.DEDUPLICATE
|| schema().crossPartitionUpdate()) {
return null;
}

List<String> fields = new ArrayList<>(schema().primaryKeys());
fields.addAll(options.sequenceField());
return fields;
}

@Override
Expand Down
62 changes: 48 additions & 14 deletions paimon-core/src/main/java/org/apache/paimon/table/sink/TableWriteImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,9 @@

import javax.annotation.Nullable;

import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.stream.Collectors;

Expand All @@ -63,8 +65,10 @@ public class TableWriteImpl<T> implements InnerTableWrite, Restorable<List<State
private boolean batchCommitted = false;
private RowType writeType;
private int[] notNullFieldIndex;
private int[] deleteNotNullFieldIndex;

private final @Nullable DefaultValueRow defaultValueRow;
private final @Nullable Set<String> deleteNotNullFieldNames;

public TableWriteImpl(
RowType rowType,
Expand All @@ -73,19 +77,33 @@ public TableWriteImpl(
RecordExtractor<T> recordExtractor,
@Nullable RowKindGenerator rowKindGenerator,
@Nullable RowKindFilter rowKindFilter) {
this(
rowType,
write,
keyAndBucketExtractor,
recordExtractor,
rowKindGenerator,
rowKindFilter,
null);
}

public TableWriteImpl(
RowType rowType,
FileStoreWrite<T> write,
KeyAndBucketExtractor<InternalRow> keyAndBucketExtractor,
RecordExtractor<T> recordExtractor,
@Nullable RowKindGenerator rowKindGenerator,
@Nullable RowKindFilter rowKindFilter,
@Nullable List<String> deleteNotNullFieldNames) {
this.writeType = rowType;
this.write = write;
this.keyAndBucketExtractor = keyAndBucketExtractor;
this.recordExtractor = recordExtractor;
this.rowKindGenerator = rowKindGenerator;
this.rowKindFilter = rowKindFilter;

List<String> notNullColumnNames =
rowType.getFields().stream()
.filter(field -> !field.type().isNullable())
.map(DataField::name)
.collect(Collectors.toList());
this.notNullFieldIndex = rowType.getFieldIndices(notNullColumnNames);
this.deleteNotNullFieldNames =
deleteNotNullFieldNames == null ? null : new HashSet<>(deleteNotNullFieldNames);
updateNotNullFieldIndexes();
this.defaultValueRow = DefaultValueRow.create(rowType);
}

Expand Down Expand Up @@ -115,13 +133,26 @@ public TableWriteImpl<T> withIOManager(IOManager ioManager) {
public TableWriteImpl<T> withWriteType(RowType writeType) {
write.withWriteType(writeType);
this.writeType = writeType;
updateNotNullFieldIndexes();
return this;
}

private void updateNotNullFieldIndexes() {
List<String> notNullColumnNames =
writeType.getFields().stream()
.filter(field -> !field.type().isNullable())
.map(DataField::name)
.collect(Collectors.toList());
this.notNullFieldIndex = writeType.getFieldIndices(notNullColumnNames);
return this;
if (deleteNotNullFieldNames == null) {
this.deleteNotNullFieldIndex = notNullFieldIndex;
} else {
List<String> deleteNotNullColumnNames =
notNullColumnNames.stream()
.filter(deleteNotNullFieldNames::contains)
.collect(Collectors.toList());
this.deleteNotNullFieldIndex = writeType.getFieldIndices(deleteNotNullColumnNames);
}
}

@Override
Expand Down Expand Up @@ -203,13 +234,14 @@ public SinkRecord writeAndReturn(InternalRow row, int bucket, int totalBuckets)
@Nullable
private SinkRecord writeAndReturn(InternalRow row, int bucket, @Nullable Integer totalBuckets)
throws Exception {
checkNullability(row);
row = wrapDefaultValue(row);
RowKind rowKind = RowKindGenerator.getRowKind(rowKindGenerator, row);
InternalRow wrappedRow = wrapDefaultValue(row);
RowKind rowKind = RowKindGenerator.getRowKind(rowKindGenerator, wrappedRow);
checkNullability(row, rowKind);
if (rowKindFilter != null && !rowKindFilter.test(rowKind)) {
return null;
}
SinkRecord record = bucket == -1 ? toSinkRecord(row) : toSinkRecord(row, bucket);
SinkRecord record =
bucket == -1 ? toSinkRecord(wrappedRow) : toSinkRecord(wrappedRow, bucket);
T extracted = recordExtractor.extract(record, rowKind);
if (totalBuckets == null) {
write.write(record.partition(), record.bucket(), extracted);
Expand All @@ -219,8 +251,10 @@ private SinkRecord writeAndReturn(InternalRow row, int bucket, @Nullable Integer
return record;
}

private void checkNullability(InternalRow row) {
for (int idx : notNullFieldIndex) {
private void checkNullability(InternalRow row, RowKind rowKind) {
int[] fieldIndexes =
rowKind == RowKind.DELETE ? deleteNotNullFieldIndex : notNullFieldIndex;
for (int idx : fieldIndexes) {
if (row.isNullAt(idx)) {
String columnName = writeType.getFields().get(idx).name();
throw new RuntimeException(
Expand Down
Loading
Loading