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 f521f2e9b17f..a478d17bfc05 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 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 KeyValueFileStore store() { PrimaryKeyTableUtils.PrimaryKeyFieldsExtractor.EXTRACTOR; RowType keyType = new RowType(extractor.keyFields(tableSchema)); + List 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 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 KeyValueFileStore store() { options, tableSchema.logicalPartitionType(), keyType, - rowType, + valueType, extractor, mfFactory, name(), @@ -189,6 +204,7 @@ public TableWriteImpl newPostponeFixedBucketWrite( private TableWriteImpl newWrite(AbstractFileStoreWrite storeWrite) { KeyValue kv = new KeyValue(); + List deleteNotNullFields = deleteNotNullFields(coreOptions()); return new TableWriteImpl<>( rowType(), storeWrite, @@ -200,7 +216,20 @@ private TableWriteImpl newWrite(AbstractFileStoreWrite store rowKind, record.row()), rowKindGenerator(), - RowKindFilter.of(coreOptions())); + RowKindFilter.of(coreOptions()), + deleteNotNullFields); + } + + @Nullable + private List deleteNotNullFields(CoreOptions options) { + if (options.mergeEngine() != CoreOptions.MergeEngine.DEDUPLICATE + || schema().crossPartitionUpdate()) { + return null; + } + + List 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 cb26d471106e..3d728aeed79a 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 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 implements InnerTableWrite, Restorable deleteNotNullFieldNames; public TableWriteImpl( RowType rowType, @@ -73,19 +77,33 @@ public TableWriteImpl( RecordExtractor recordExtractor, @Nullable RowKindGenerator rowKindGenerator, @Nullable RowKindFilter rowKindFilter) { + this( + rowType, + write, + keyAndBucketExtractor, + recordExtractor, + rowKindGenerator, + rowKindFilter, + null); + } + + public TableWriteImpl( + RowType rowType, + FileStoreWrite write, + KeyAndBucketExtractor keyAndBucketExtractor, + RecordExtractor recordExtractor, + @Nullable RowKindGenerator rowKindGenerator, + @Nullable RowKindFilter rowKindFilter, + @Nullable List deleteNotNullFieldNames) { this.writeType = rowType; this.write = write; this.keyAndBucketExtractor = keyAndBucketExtractor; this.recordExtractor = recordExtractor; this.rowKindGenerator = rowKindGenerator; this.rowKindFilter = rowKindFilter; - - List 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 TableWriteImpl withIOManager(IOManager ioManager) { public TableWriteImpl withWriteType(RowType writeType) { write.withWriteType(writeType); this.writeType = writeType; + updateNotNullFieldIndexes(); + return this; + } + + private void updateNotNullFieldIndexes() { List 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 deleteNotNullColumnNames = + notNullColumnNames.stream() + .filter(deleteNotNullFieldNames::contains) + .collect(Collectors.toList()); + this.deleteNotNullFieldIndex = writeType.getFieldIndices(deleteNotNullColumnNames); + } } @Override @@ -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); @@ -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( 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 afef43538009..3441dadecf89 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 @@ 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 void testPartitionTimestampValidation() throws Exception { 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 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 @@ List streamingRead( return actual; } + private List readRows(FileStoreTable table) throws Exception { + List rows = new ArrayList<>(); + TableScan.Plan plan = table.newScan().plan(); + try (RecordReaderIterator 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 partitionKeys, List 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(