This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new f8a28cdef6 [core] Allow primary-key-only deletes for deduplicate
tables (#9978)
f8a28cdef6 is described below
commit f8a28cdef6a82c5d7fef4ee581969213ab01d002
Author: Akash Reddy Jammula <[email protected]>
AuthorDate: Sat Sep 19 22:47:20 2026 -0700
[core] Allow primary-key-only deletes for deduplicate tables (#9978)
---
.../paimon/table/PrimaryKeyFileStoreTable.java | 35 +++-
.../apache/paimon/table/sink/TableWriteImpl.java | 62 ++++--
.../apache/paimon/table/sink/TableWriteTest.java | 210 +++++++++++++++++++++
3 files changed, 290 insertions(+), 17 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyFileStoreTable.java
b/paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyFileStoreTable.java
index f521f2e9b1..a478d17bfc 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyFileStoreTable.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyFileStoreTable.java
@@ -45,8 +45,10 @@ import org.apache.paimon.utils.RowKindFilter;
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;
@@ -81,11 +83,24 @@ public class PrimaryKeyFileStoreTable extends
AbstractFileStoreTable {
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 =
@@ -97,7 +112,7 @@ public class PrimaryKeyFileStoreTable extends
AbstractFileStoreTable {
options,
tableSchema.logicalPartitionType(),
keyType,
- rowType,
+ valueType,
extractor,
mfFactory,
name(),
@@ -189,6 +204,7 @@ public class PrimaryKeyFileStoreTable extends
AbstractFileStoreTable {
private TableWriteImpl<KeyValue> newWrite(AbstractFileStoreWrite<KeyValue>
storeWrite) {
KeyValue kv = new KeyValue();
+ List<String> deleteNotNullFields = deleteNotNullFields(coreOptions());
return new TableWriteImpl<>(
rowType(),
storeWrite,
@@ -200,7 +216,20 @@ public class PrimaryKeyFileStoreTable extends
AbstractFileStoreTable {
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
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/sink/TableWriteImpl.java
b/paimon-core/src/main/java/org/apache/paimon/table/sink/TableWriteImpl.java
index cb26d47110..3d728aeed7 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/sink/TableWriteImpl.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/sink/TableWriteImpl.java
@@ -41,7 +41,9 @@ import org.apache.paimon.utils.RowKindFilter;
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;
@@ -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,
@@ -73,19 +77,33 @@ public class TableWriteImpl<T> implements InnerTableWrite,
Restorable<List<State
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);
}
@@ -115,13 +133,26 @@ public class TableWriteImpl<T> implements
InnerTableWrite, Restorable<List<State
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
@@ -203,13 +234,14 @@ public class TableWriteImpl<T> implements
InnerTableWrite, Restorable<List<State
@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);
@@ -219,8 +251,10 @@ public class TableWriteImpl<T> implements InnerTableWrite,
Restorable<List<State
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(
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/sink/TableWriteTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/sink/TableWriteTest.java
index afef435380..3441dadecf 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/sink/TableWriteTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/sink/TableWriteTest.java
@@ -21,6 +21,7 @@ package org.apache.paimon.table.sink;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.BinaryRowWriter;
+import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.disk.IOManagerImpl;
@@ -316,6 +317,173 @@ public class TableWriteTest {
write.close();
}
+ @Test
+ public void testDeduplicateDeleteWithPrimaryKeyOnly() throws Exception {
+ FileStoreTable table = createNotNullFileStoreTable(new Options());
+ TableWriteImpl<?> write = table.newWrite(commitUser);
+ StreamTableCommit commit = table.newCommit(commitUser);
+
+ write.write(GenericRow.of(1, 1, 10L));
+ commit.commit(0, write.prepareCommit(false, 0));
+
+ assertThatThrownBy(() -> write.write(GenericRow.ofKind(RowKind.DELETE,
1, null, null)))
+ .hasMessageContaining("non-null column(k)");
+ assertThatThrownBy(() -> write.write(GenericRow.of(1, 1, null)))
+ .hasMessageContaining("non-null column(v)");
+ assertThatThrownBy(() ->
write.write(GenericRow.ofKind(RowKind.UPDATE_BEFORE, 1, 1, null)))
+ .hasMessageContaining("non-null column(v)");
+ assertThatThrownBy(() ->
write.write(GenericRow.ofKind(RowKind.UPDATE_AFTER, 1, 1, null)))
+ .hasMessageContaining("non-null column(v)");
+
+ write.write(GenericRow.ofKind(RowKind.DELETE, 1, 1, null));
+ commit.commit(1, write.prepareCommit(false, 1));
+
+ assertThat(readRows(table)).isEmpty();
+ write.close();
+ commit.close();
+ }
+
+ @Test
+ public void testLookupDeleteWithPrimaryKeyOnly() throws Exception {
+ Options options = new Options();
+ options.set(CoreOptions.CHANGELOG_PRODUCER,
CoreOptions.ChangelogProducer.LOOKUP);
+ options.set(CoreOptions.LOOKUP_MERGE_RECORDS_THRESHOLD, 1);
+ FileStoreTable table = createNotNullFileStoreTable(options);
+ TableWriteImpl<?> write =
+ table.newWrite(commitUser)
+ .withIOManager(
+ new IOManagerImpl(
+
tempDir.resolve(UUID.randomUUID().toString()).toString()));
+ StreamTableCommit commit = table.newCommit(commitUser);
+
+ write.write(GenericRow.of(1, 1, 10L));
+ commit.commit(0, write.prepareCommit(false, 0));
+ write.write(GenericRow.ofKind(RowKind.DELETE, 1, 1, null));
+ commit.commit(1, write.prepareCommit(false, 1));
+
+ assertThat(readRows(table)).isEmpty();
+ write.close();
+ commit.close();
+ }
+
+ @Test
+ public void testPartialUpdateDeleteRequiresNonKeyFields() throws Exception
{
+ Options options = new Options();
+ options.set(CoreOptions.MERGE_ENGINE,
CoreOptions.MergeEngine.PARTIAL_UPDATE);
+ FileStoreTable table = createNotNullFileStoreTable(options);
+ TableWriteImpl<?> write = table.newWrite(commitUser);
+
+ assertThatThrownBy(() -> write.write(GenericRow.ofKind(RowKind.DELETE,
1, 1, null)))
+ .hasMessageContaining("non-null column(v)");
+
+ write.close();
+ }
+
+ @Test
+ public void testDeleteRequiresSequenceField() throws Exception {
+ Options options = new Options();
+ options.set(CoreOptions.SEQUENCE_FIELD, "seq");
+ RowType rowType =
+ RowType.of(
+ new DataType[] {
+ DataTypes.INT().notNull(),
+ DataTypes.INT().notNull(),
+ DataTypes.INT().notNull(),
+ DataTypes.BIGINT().notNull()
+ },
+ new String[] {"pt", "k", "seq", "v"});
+ FileStoreTable table =
+ createFileStoreTable(
+ rowType,
+ Collections.singletonList("pt"),
+ Arrays.asList("pt", "k"),
+ options);
+ TableWriteImpl<?> write = table.newWrite(commitUser);
+ StreamTableCommit commit = table.newCommit(commitUser);
+
+ write.write(GenericRow.of(1, 1, 10, 100L));
+ commit.commit(0, write.prepareCommit(false, 0));
+
+ assertThatThrownBy(() -> write.write(GenericRow.ofKind(RowKind.DELETE,
1, 1, null, null)))
+ .hasMessageContaining("non-null column(seq)");
+
+ write.write(GenericRow.ofKind(RowKind.DELETE, 1, 1, 9, null));
+ commit.commit(1, write.prepareCommit(false, 1));
+ List<InternalRow> rows = readRows(table);
+ assertThat(rows).hasSize(1);
+ assertThat(rows.get(0).getInt(2)).isEqualTo(10);
+ assertThat(rows.get(0).getLong(3)).isEqualTo(100L);
+
+ write.write(GenericRow.ofKind(RowKind.DELETE, 1, 1, 11, null));
+ commit.commit(2, write.prepareCommit(false, 2));
+ assertThat(readRows(table)).isEmpty();
+
+ write.close();
+ commit.close();
+ }
+
+ @Test
+ public void testGeneratedDeleteWithPrimaryKeyOnly() throws Exception {
+ Options options = new Options();
+ options.set(CoreOptions.ROWKIND_FIELD, "kind");
+ RowType rowType =
+ RowType.of(
+ new DataType[] {
+ DataTypes.INT().notNull(),
+ DataTypes.INT().notNull(),
+ DataTypes.BIGINT().notNull(),
+ DataTypes.STRING().notNull()
+ },
+ new String[] {"pt", "k", "v", "kind"});
+ FileStoreTable table =
+ createFileStoreTable(
+ rowType,
+ Collections.singletonList("pt"),
+ Arrays.asList("pt", "k"),
+ options);
+ TableWriteImpl<?> write = table.newWrite(commitUser);
+ StreamTableCommit commit = table.newCommit(commitUser);
+
+ write.write(GenericRow.of(1, 1, 10L, BinaryString.fromString("+I")));
+ commit.commit(0, write.prepareCommit(false, 0));
+ assertThatThrownBy(
+ () -> write.write(GenericRow.of(1, 1, null,
BinaryString.fromString("-U"))))
+ .hasMessageContaining("non-null column(v)");
+ assertThatThrownBy(
+ () -> write.write(GenericRow.of(1, 1, null,
BinaryString.fromString("+U"))))
+ .hasMessageContaining("non-null column(v)");
+ write.write(GenericRow.of(1, 1, null, BinaryString.fromString("-D")));
+ commit.commit(1, write.prepareCommit(false, 1));
+
+ assertThat(readRows(table)).isEmpty();
+ write.close();
+ commit.close();
+ }
+
+ @Test
+ public void testCrossPartitionDeleteRequiresPartitionField() throws
Exception {
+ RowType rowType =
+ RowType.of(
+ new DataType[] {
+ DataTypes.INT().notNull(),
+ DataTypes.INT().notNull(),
+ DataTypes.BIGINT().notNull()
+ },
+ new String[] {"pt", "k", "v"});
+ FileStoreTable table =
+ createFileStoreTable(
+ rowType,
+ Collections.singletonList("pt"),
+ Collections.singletonList("k"),
+ new Options());
+ TableWriteImpl<?> write = table.newWrite(commitUser);
+
+ assertThatThrownBy(() -> write.write(GenericRow.ofKind(RowKind.DELETE,
null, 1, null)))
+ .hasMessageContaining("non-null column(pt)");
+
+ write.close();
+ }
+
private BinaryRow partition(int x) {
BinaryRow partition = new BinaryRow(1);
BinaryRowWriter writer = new BinaryRowWriter(partition);
@@ -335,6 +503,48 @@ public class TableWriteTest {
return actual;
}
+ private List<InternalRow> readRows(FileStoreTable table) throws Exception {
+ List<InternalRow> rows = new ArrayList<>();
+ TableScan.Plan plan = table.newScan().plan();
+ try (RecordReaderIterator<InternalRow> iterator =
+ new
RecordReaderIterator<>(table.newRead().createReader(plan))) {
+ while (iterator.hasNext()) {
+ rows.add(iterator.next());
+ }
+ }
+ return rows;
+ }
+
+ private FileStoreTable createNotNullFileStoreTable(Options options) throws
Exception {
+ RowType rowType =
+ RowType.of(
+ new DataType[] {
+ DataTypes.INT().notNull(),
+ DataTypes.INT().notNull(),
+ DataTypes.BIGINT().notNull()
+ },
+ new String[] {"pt", "k", "v"});
+ return createFileStoreTable(
+ rowType, Collections.singletonList("pt"), Arrays.asList("pt",
"k"), options);
+ }
+
+ private FileStoreTable createFileStoreTable(
+ RowType rowType, List<String> partitionKeys, List<String>
primaryKeys, Options options)
+ throws Exception {
+ options.set(CoreOptions.BUCKET, 1);
+ Path path = new
Path(tempDir.resolve(UUID.randomUUID().toString()).toUri());
+ TableSchema tableSchema =
+ SchemaUtils.forceCommit(
+ new FileSystemSchemaManager(LocalFileIO.create(),
path),
+ new Schema(
+ rowType.getFields(),
+ partitionKeys,
+ primaryKeys,
+ options.toMap(),
+ ""));
+ return FileStoreTableFactory.create(LocalFileIO.create(), path,
tableSchema);
+ }
+
private FileStoreTable createFileStoreTable(Options conf) throws Exception
{
TableSchema tableSchema =
SchemaUtils.forceCommit(
