From 68bbc9b0643b96ae8047a30ff1b1128b1ffd634f Mon Sep 17 00:00:00 2001 From: umi Date: Sun, 20 Sep 2026 15:20:13 +0800 Subject: [PATCH 1/6] proto # Conflicts: # paimon-api/src/main/java/org/apache/paimon/CoreOptions.java --- docs/docs/concepts/spec/datafile.md | 7 + .../java/org/apache/paimon/CoreOptions.java | 4 + .../paimon/io/DeleteRowFieldsProjector.java | 97 ++++++++++++ .../io/KeyValueClusteringFileWriter.java | 9 +- .../paimon/io/KeyValueDataFileWriter.java | 9 +- .../paimon/io/KeyValueDataFileWriterImpl.java | 8 +- .../paimon/io/KeyValueFileWriterFactory.java | 53 ++++++- .../io/KeyValueThinDataFileWriterImpl.java | 8 +- .../operation/KeyValueFileStoreWrite.java | 2 +- .../PostponeBucketFileStoreWrite.java | 2 +- .../io/DeleteRowFieldsProjectorTest.java | 135 ++++++++++++++++ .../paimon/io/KeyValueFileReadWriteTest.java | 141 +++++++++++++++-- .../io/PrimaryKeyBlobFileWriterTest.java | 14 +- .../paimon/table/DeleteRowFieldsTest.java | 149 ++++++++++++++++++ 14 files changed, 610 insertions(+), 28 deletions(-) create mode 100644 paimon-core/src/main/java/org/apache/paimon/io/DeleteRowFieldsProjector.java create mode 100644 paimon-core/src/test/java/org/apache/paimon/io/DeleteRowFieldsProjectorTest.java create mode 100644 paimon-core/src/test/java/org/apache/paimon/table/DeleteRowFieldsTest.java diff --git a/docs/docs/concepts/spec/datafile.md b/docs/docs/concepts/spec/datafile.md index 25408c53d1b1..31375578cc8e 100644 --- a/docs/docs/concepts/spec/datafile.md +++ b/docs/docs/concepts/spec/datafile.md @@ -137,6 +137,13 @@ contains five columns: _SEQUENCE_NUMBER | _VALUE_KIND | a | b | c ``` +For a primary-key merge engine where `DELETE` removes the whole record, Paimon keeps the +primary-key, sequence, record-level time, clustering, indexed, and non-null fields in data-file +DELETE rows, and writes null for other nullable value fields. This applies to deduplicate tables, +partial-update tables with `partial-update.remove-record-on-delete`, and aggregation tables with +`aggregation.remove-record-on-delete`. The physical schema is unchanged. Changelog files are not +projected by this optimization. + ### Table w/o Primary key Data File An ordinary append table stores its declared value columns without the primary-key merge fields: diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java index c3202188dc09..283aae1dbabf 100644 --- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java +++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java @@ -4785,6 +4785,10 @@ public boolean aggregationRemoveRecordOnDelete() { return options.get(AGGREGATION_REMOVE_RECORD_ON_DELETE); } + public boolean partialUpdateRemoveRecordOnDelete() { + return options.get(PARTIAL_UPDATE_REMOVE_RECORD_ON_DELETE); + } + public Optional commitLastSafeSnapshot() { return options.getOptional(COMMIT_LAST_SAFE_SNAPSHOT); } diff --git a/paimon-core/src/main/java/org/apache/paimon/io/DeleteRowFieldsProjector.java b/paimon-core/src/main/java/org/apache/paimon/io/DeleteRowFieldsProjector.java new file mode 100644 index 000000000000..34e54e7be1f9 --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/io/DeleteRowFieldsProjector.java @@ -0,0 +1,97 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.io; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.KeyValue; +import org.apache.paimon.schema.TableSchema; +import org.apache.paimon.types.DataField; +import org.apache.paimon.utils.ProjectedRow; + +import javax.annotation.Nullable; + +import java.util.Arrays; +import java.util.HashSet; +import java.util.Set; + +/** Projects whole-record DELETE rows to fields required by data files. */ +final class DeleteRowFieldsProjector { + + private final ProjectedRow projectedValue; + private final KeyValue reused = new KeyValue(); + + static @Nullable int[] createProjection(TableSchema schema, CoreOptions options) { + if (schema.primaryKeys().isEmpty() || !removeWholeRecordOnDelete(options)) { + return null; + } + + Set requiredFields = new HashSet<>(schema.primaryKeys()); + requiredFields.addAll(options.sequenceField()); + if (options.recordLevelTimeField() != null) { + requiredFields.add(options.recordLevelTimeField()); + } + requiredFields.addAll(options.clusteringColumns()); + requiredFields.addAll(options.primaryKeyVectorIndexColumns()); + requiredFields.addAll(options.primaryKeyBTreeIndexColumns()); + requiredFields.addAll(options.primaryKeyBitmapIndexColumns()); + requiredFields.addAll(options.primaryKeyMultiValueIndexColumns()); + requiredFields.addAll(options.primaryKeyFullTextIndexColumns()); + requiredFields.addAll(options.primaryKeyFMIndexColumns()); + for (DataField field : schema.fields()) { + if (!field.type().isNullable()) { + requiredFields.add(field.name()); + } + } + + int[] projection = new int[schema.fields().size()]; + Arrays.fill(projection, -1); + for (int i = 0; i < schema.fields().size(); i++) { + if (requiredFields.contains(schema.fields().get(i).name())) { + projection[i] = i; + } + } + return Arrays.stream(projection).anyMatch(field -> field < 0) ? projection : null; + } + + private static boolean removeWholeRecordOnDelete(CoreOptions options) { + switch (options.mergeEngine()) { + case DEDUPLICATE: + return true; + case PARTIAL_UPDATE: + return options.partialUpdateRemoveRecordOnDelete(); + case AGGREGATE: + return options.aggregationRemoveRecordOnDelete(); + default: + return false; + } + } + + DeleteRowFieldsProjector(int[] projection) { + this.projectedValue = ProjectedRow.from(projection); + } + + KeyValue project(KeyValue kv) { + return reused.replace( + kv.key(), + kv.sequenceNumber(), + kv.valueKind(), + projectedValue.replaceRow(kv.value())) + .setLevel(kv.level()); + } +} diff --git a/paimon-core/src/main/java/org/apache/paimon/io/KeyValueClusteringFileWriter.java b/paimon-core/src/main/java/org/apache/paimon/io/KeyValueClusteringFileWriter.java index 57548e831153..876c35a04dbb 100644 --- a/paimon-core/src/main/java/org/apache/paimon/io/KeyValueClusteringFileWriter.java +++ b/paimon-core/src/main/java/org/apache/paimon/io/KeyValueClusteringFileWriter.java @@ -34,6 +34,7 @@ import org.apache.paimon.stats.SimpleStats; import org.apache.paimon.stats.SimpleStatsConverter; import org.apache.paimon.table.SpecialFields; +import org.apache.paimon.types.RowKind; import org.apache.paimon.types.RowType; import org.apache.paimon.utils.Pair; @@ -72,6 +73,7 @@ public class KeyValueClusteringFileWriter private final Projection clusteringProjection; @Nullable private final DataFileIndexWriter dataFileIndexWriter; private final int[] keyStatMapping; + @Nullable private final DeleteRowFieldsProjector deleteRowFieldsProjector; private BinaryRow minClusteringFields = null; private BinaryRow currentClusteringFields = null; @@ -90,7 +92,8 @@ public KeyValueClusteringFileWriter( boolean thinMode, CoreOptions options, FileIndexOptions fileIndexOptions, - boolean isExternalPath) { + boolean isExternalPath, + @Nullable DeleteRowFieldsProjector deleteRowFieldsProjector) { super( fileIO, context, @@ -116,6 +119,7 @@ public KeyValueClusteringFileWriter( this.dataFileIndexWriter = DataFileIndexWriter.create( fileIO, dataFileToFileIndexPath(path), valueType, fileIndexOptions); + this.deleteRowFieldsProjector = deleteRowFieldsProjector; Map idToIndex = new HashMap<>(valueType.getFieldCount()); for (int i = 0; i < valueType.getFieldCount(); i++) { @@ -131,6 +135,9 @@ public KeyValueClusteringFileWriter( @Override public void write(KeyValue kv) throws IOException { + if (deleteRowFieldsProjector != null && kv.valueKind() == RowKind.DELETE) { + kv = deleteRowFieldsProjector.project(kv); + } super.write(kv); if (dataFileIndexWriter != null) { diff --git a/paimon-core/src/main/java/org/apache/paimon/io/KeyValueDataFileWriter.java b/paimon-core/src/main/java/org/apache/paimon/io/KeyValueDataFileWriter.java index 133a7af27225..ca62041e4972 100644 --- a/paimon-core/src/main/java/org/apache/paimon/io/KeyValueDataFileWriter.java +++ b/paimon-core/src/main/java/org/apache/paimon/io/KeyValueDataFileWriter.java @@ -32,6 +32,7 @@ import org.apache.paimon.manifest.FileSource; import org.apache.paimon.stats.SimpleStats; import org.apache.paimon.stats.SimpleStatsConverter; +import org.apache.paimon.types.RowKind; import org.apache.paimon.types.RowType; import org.apache.paimon.utils.Pair; @@ -74,6 +75,7 @@ public abstract class KeyValueDataFileWriter private final FileSource fileSource; @Nullable private final DataFileIndexWriter dataFileIndexWriter; @Nullable private final ManagedBlobReferenceCollector blobReferenceCollector; + @Nullable private final DeleteRowFieldsProjector deleteRowFieldsProjector; private BinaryRow minKey = null; private long minSeqNumber = Long.MAX_VALUE; @@ -94,7 +96,8 @@ public KeyValueDataFileWriter( FileSource fileSource, FileIndexOptions fileIndexOptions, boolean isExternalPath, - Set managedBlobFields) { + Set managedBlobFields, + @Nullable DeleteRowFieldsProjector deleteRowFieldsProjector) { super(fileIO, context, path, converter, writeRowType, options.asyncFileWrite()); this.keyType = keyType; @@ -115,10 +118,14 @@ public KeyValueDataFileWriter( ? null : new ManagedBlobReferenceCollector( fileIO, path, valueType, managedBlobFields); + this.deleteRowFieldsProjector = deleteRowFieldsProjector; } @Override public void write(KeyValue kv) throws IOException { + if (deleteRowFieldsProjector != null && kv.valueKind() == RowKind.DELETE) { + kv = deleteRowFieldsProjector.project(kv); + } super.write(kv); if (dataFileIndexWriter != null) { diff --git a/paimon-core/src/main/java/org/apache/paimon/io/KeyValueDataFileWriterImpl.java b/paimon-core/src/main/java/org/apache/paimon/io/KeyValueDataFileWriterImpl.java index 628fdf5cfed1..fe3a8a4219a7 100644 --- a/paimon-core/src/main/java/org/apache/paimon/io/KeyValueDataFileWriterImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/io/KeyValueDataFileWriterImpl.java @@ -29,6 +29,8 @@ import org.apache.paimon.types.RowType; import org.apache.paimon.utils.Pair; +import javax.annotation.Nullable; + import java.util.Arrays; import java.util.Set; import java.util.function.Function; @@ -49,7 +51,8 @@ public KeyValueDataFileWriterImpl( FileSource fileSource, FileIndexOptions fileIndexOptions, boolean isExternalPath, - Set managedBlobFields) { + Set managedBlobFields, + @Nullable DeleteRowFieldsProjector deleteRowFieldsProjector) { super( fileIO, context, @@ -64,7 +67,8 @@ public KeyValueDataFileWriterImpl( fileSource, fileIndexOptions, isExternalPath, - managedBlobFields); + managedBlobFields, + deleteRowFieldsProjector); } @Override diff --git a/paimon-core/src/main/java/org/apache/paimon/io/KeyValueFileWriterFactory.java b/paimon-core/src/main/java/org/apache/paimon/io/KeyValueFileWriterFactory.java index 7c55360fef4c..4c7d39556a4d 100644 --- a/paimon-core/src/main/java/org/apache/paimon/io/KeyValueFileWriterFactory.java +++ b/paimon-core/src/main/java/org/apache/paimon/io/KeyValueFileWriterFactory.java @@ -34,6 +34,7 @@ import org.apache.paimon.fs.FileIO; import org.apache.paimon.fs.Path; import org.apache.paimon.manifest.FileSource; +import org.apache.paimon.schema.TableSchema; import org.apache.paimon.statistics.NoneSimpleColStatsCollector; import org.apache.paimon.statistics.SimpleColStatsCollector; import org.apache.paimon.table.SpecialFields; @@ -72,13 +73,15 @@ public class KeyValueFileWriterFactory { private final FileIndexOptions fileIndexOptions; private final Set managedBlobFields; @Nullable private final PrimaryKeyBlobExternalizer blobExternalizer; + @Nullable private final int[] deleteRowFieldsProjection; private KeyValueFileWriterFactory( FileIO fileIO, long schemaId, FileWriterContextFactory formatContext, long suggestedFileSize, - CoreOptions options) { + CoreOptions options, + @Nullable TableSchema tableSchema) { this.fileIO = fileIO; this.schemaId = schemaId; this.keyType = formatContext.keyType; @@ -87,6 +90,10 @@ private KeyValueFileWriterFactory( this.suggestedFileSize = suggestedFileSize; this.options = options; this.fileIndexOptions = options.indexColumnsOptions(); + this.deleteRowFieldsProjection = + tableSchema == null + ? null + : DeleteRowFieldsProjector.createProjection(tableSchema, options); this.managedBlobFields = managedBlobFields(); this.blobExternalizer = managedBlobFields.isEmpty() @@ -203,7 +210,10 @@ private KeyValueClusteringFileWriter createKvSeparatedFileWriter( formatContext.thinModeEnabled, options, fileIndexOptions, - isExternalPath); + isExternalPath, + deleteRowFieldsProjection == null + ? null + : new DeleteRowFieldsProjector(deleteRowFieldsProjection)); } private KeyValueDataFileWriter createDataFileWriter( @@ -216,6 +226,10 @@ private KeyValueDataFileWriter createDataFileWriter( FileIndexOptions indexOptions = key.isChangelog ? new FileIndexOptions() : fileIndexOptions; Set dataFileManagedBlobFields = key.isChangelog ? Collections.emptySet() : managedBlobFields; + DeleteRowFieldsProjector deleteRowFieldsProjector = + key.isChangelog || deleteRowFieldsProjection == null + ? null + : new DeleteRowFieldsProjector(deleteRowFieldsProjection); return formatContext.thinModeEnabled ? new KeyValueThinDataFileWriterImpl( fileIO, @@ -230,7 +244,8 @@ private KeyValueDataFileWriter createDataFileWriter( fileSource, indexOptions, isExternalPath, - dataFileManagedBlobFields) + dataFileManagedBlobFields, + deleteRowFieldsProjector) : new KeyValueDataFileWriterImpl( fileIO, formatContext.fileWriterContext(key, writerFactory), @@ -244,7 +259,8 @@ private KeyValueDataFileWriter createDataFileWriter( fileSource, indexOptions, isExternalPath, - dataFileManagedBlobFields); + dataFileManagedBlobFields, + deleteRowFieldsProjector); } private Set managedBlobFields() { @@ -267,6 +283,25 @@ public String newChangelogFileName(int level) { return formatContext.pathFactory(new WriteFormatKey(level, true)).newChangelogFileName(); } + public static Builder builder( + FileIO fileIO, + TableSchema schema, + RowType keyType, + RowType valueType, + FileFormat fileFormat, + Function format2PathFactory, + long suggestedFileSize) { + return new Builder( + fileIO, + schema.id(), + keyType, + valueType, + fileFormat, + format2PathFactory, + suggestedFileSize, + schema); + } + public static Builder builder( FileIO fileIO, long schemaId, @@ -282,7 +317,8 @@ public static Builder builder( valueType, fileFormat, format2PathFactory, - suggestedFileSize); + suggestedFileSize, + null); } /** Builder of {@link KeyValueFileWriterFactory}. */ @@ -295,6 +331,7 @@ public static class Builder { private final FileFormat fileFormat; private final Function format2PathFactory; private final long suggestedFileSize; + @Nullable private final TableSchema tableSchema; private Builder( FileIO fileIO, @@ -303,7 +340,8 @@ private Builder( RowType valueType, FileFormat fileFormat, Function format2PathFactory, - long suggestedFileSize) { + long suggestedFileSize, + @Nullable TableSchema tableSchema) { this.fileIO = fileIO; this.schemaId = schemaId; this.keyType = keyType; @@ -311,6 +349,7 @@ private Builder( this.fileFormat = fileFormat; this.format2PathFactory = format2PathFactory; this.suggestedFileSize = suggestedFileSize; + this.tableSchema = tableSchema; } public KeyValueFileWriterFactory build( @@ -325,7 +364,7 @@ public KeyValueFileWriterFactory build( format2PathFactory, options); return new KeyValueFileWriterFactory( - fileIO, schemaId, context, suggestedFileSize, options); + fileIO, schemaId, context, suggestedFileSize, options, tableSchema); } } diff --git a/paimon-core/src/main/java/org/apache/paimon/io/KeyValueThinDataFileWriterImpl.java b/paimon-core/src/main/java/org/apache/paimon/io/KeyValueThinDataFileWriterImpl.java index fd05478533e1..3888426180ef 100644 --- a/paimon-core/src/main/java/org/apache/paimon/io/KeyValueThinDataFileWriterImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/io/KeyValueThinDataFileWriterImpl.java @@ -30,6 +30,8 @@ import org.apache.paimon.types.RowType; import org.apache.paimon.utils.Pair; +import javax.annotation.Nullable; + import java.util.Arrays; import java.util.HashMap; import java.util.Map; @@ -57,7 +59,8 @@ public KeyValueThinDataFileWriterImpl( FileSource fileSource, FileIndexOptions fileIndexOptions, boolean isExternalPath, - Set managedBlobFields) { + Set managedBlobFields, + @Nullable DeleteRowFieldsProjector deleteRowFieldsProjector) { super( fileIO, context, @@ -72,7 +75,8 @@ public KeyValueThinDataFileWriterImpl( fileSource, fileIndexOptions, isExternalPath, - managedBlobFields); + managedBlobFields, + deleteRowFieldsProjector); Map idToIndex = new HashMap<>(valueType.getFieldCount()); for (int i = 0; i < valueType.getFieldCount(); i++) { idToIndex.put(valueType.getFields().get(i).id(), i); diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java b/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java index b685a152c7c1..2c01875896fb 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java @@ -131,7 +131,7 @@ public KeyValueFileStoreWrite( this.writerFactoryBuilder = KeyValueFileWriterFactory.builder( fileIO, - schema.id(), + schema, keyType, valueType, fileFormat(options), diff --git a/paimon-core/src/main/java/org/apache/paimon/postpone/PostponeBucketFileStoreWrite.java b/paimon-core/src/main/java/org/apache/paimon/postpone/PostponeBucketFileStoreWrite.java index f75be9a2b5a6..9e95451cea16 100644 --- a/paimon-core/src/main/java/org/apache/paimon/postpone/PostponeBucketFileStoreWrite.java +++ b/paimon-core/src/main/java/org/apache/paimon/postpone/PostponeBucketFileStoreWrite.java @@ -130,7 +130,7 @@ public PostponeBucketFileStoreWrite( this.writerFactoryBuilder = KeyValueFileWriterFactory.builder( fileIO, - schema.id(), + schema, keyType, valueType, fileFormat, diff --git a/paimon-core/src/test/java/org/apache/paimon/io/DeleteRowFieldsProjectorTest.java b/paimon-core/src/test/java/org/apache/paimon/io/DeleteRowFieldsProjectorTest.java new file mode 100644 index 000000000000..925a4aad66ca --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/io/DeleteRowFieldsProjectorTest.java @@ -0,0 +1,135 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.io; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.KeyValue; +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.options.Options; +import org.apache.paimon.schema.TableSchema; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.types.RowKind; + +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.Collections; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for {@link DeleteRowFieldsProjector}. */ +class DeleteRowFieldsProjectorTest { + + @Test + void testRequiredFieldProjection() { + Options options = new Options(); + options.set(CoreOptions.SEQUENCE_FIELD, "seq"); + options.set(CoreOptions.RECORD_LEVEL_TIME_FIELD, "expire"); + options.set(CoreOptions.CLUSTERING_COLUMNS, "cluster"); + options.set(CoreOptions.PK_VECTOR_INDEX_COLUMNS, "vector_index"); + options.set(CoreOptions.PK_BTREE_INDEX_COLUMNS, "btree_index"); + options.set(CoreOptions.PK_BITMAP_INDEX_COLUMNS, "bitmap_index"); + options.set(CoreOptions.PK_MULTIVALUE_INDEX_COLUMNS, "multivalue_index"); + options.set(CoreOptions.PK_FULL_TEXT_INDEX_COLUMNS, "fulltext_index"); + options.set(CoreOptions.PK_FM_INDEX_COLUMNS, "fm_index"); + + assertThat( + DeleteRowFieldsProjector.createProjection( + tableSchema(), new CoreOptions(options))) + .containsExactly(0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, -1); + } + + @Test + void testOnlyWholeRecordDeletesAreProjected() { + assertThat(createProjection(new Options())).isNotNull(); + + Options partialUpdate = new Options(); + partialUpdate.set(CoreOptions.MERGE_ENGINE, CoreOptions.MergeEngine.PARTIAL_UPDATE); + assertThat(createProjection(partialUpdate)).isNull(); + partialUpdate.set(CoreOptions.PARTIAL_UPDATE_REMOVE_RECORD_ON_DELETE, true); + assertThat(createProjection(partialUpdate)).isNotNull(); + + Options aggregation = new Options(); + aggregation.set(CoreOptions.MERGE_ENGINE, CoreOptions.MergeEngine.AGGREGATE); + assertThat(createProjection(aggregation)).isNull(); + aggregation.set(CoreOptions.AGGREGATION_REMOVE_RECORD_ON_DELETE, true); + assertThat(createProjection(aggregation)).isNotNull(); + + Options firstRow = new Options(); + firstRow.set(CoreOptions.MERGE_ENGINE, CoreOptions.MergeEngine.FIRST_ROW); + assertThat(createProjection(firstRow)).isNull(); + } + + @Test + void testProjectionDoesNotChangeInput() { + GenericRow value = + GenericRow.of( + 1, + "required", + 2L, + null, + null, + null, + null, + null, + null, + null, + null, + "payload"); + KeyValue input = new KeyValue().replace(GenericRow.of(1), 2L, RowKind.DELETE, value); + + KeyValue projected = + new DeleteRowFieldsProjector(createProjection(new Options())).project(input); + + assertThat(projected.key()).isSameAs(input.key()); + assertThat(projected.sequenceNumber()).isEqualTo(2L); + assertThat(projected.valueKind()).isEqualTo(RowKind.DELETE); + assertThat(projected.value().isNullAt(1)).isFalse(); + assertThat(projected.value().isNullAt(11)).isTrue(); + assertThat(value.isNullAt(11)).isFalse(); + } + + private int[] createProjection(Options options) { + return DeleteRowFieldsProjector.createProjection(tableSchema(), new CoreOptions(options)); + } + + private TableSchema tableSchema() { + return new TableSchema( + 0, + Arrays.asList( + new DataField(0, "key", DataTypes.INT()), + new DataField(1, "not_null", DataTypes.STRING().notNull()), + new DataField(2, "seq", DataTypes.BIGINT()), + new DataField(3, "expire", DataTypes.BIGINT()), + new DataField(4, "cluster", DataTypes.STRING()), + new DataField(5, "vector_index", DataTypes.STRING()), + new DataField(6, "btree_index", DataTypes.STRING()), + new DataField(7, "bitmap_index", DataTypes.STRING()), + new DataField(8, "multivalue_index", DataTypes.STRING()), + new DataField(9, "fulltext_index", DataTypes.STRING()), + new DataField(10, "fm_index", DataTypes.STRING()), + new DataField(11, "payload", DataTypes.STRING())), + 11, + Collections.emptyList(), + Collections.singletonList("key"), + Collections.emptyMap(), + ""); + } +} diff --git a/paimon-core/src/test/java/org/apache/paimon/io/KeyValueFileReadWriteTest.java b/paimon-core/src/test/java/org/apache/paimon/io/KeyValueFileReadWriteTest.java index 333d61302b48..f9ec29b0ce24 100644 --- a/paimon-core/src/test/java/org/apache/paimon/io/KeyValueFileReadWriteTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/io/KeyValueFileReadWriteTest.java @@ -54,6 +54,7 @@ import org.apache.paimon.table.SpecialFields; import org.apache.paimon.types.DataType; import org.apache.paimon.types.IntType; +import org.apache.paimon.types.RowKind; import org.apache.paimon.types.RowType; import org.apache.paimon.types.VarCharType; import org.apache.paimon.utils.CloseableIterator; @@ -66,6 +67,7 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; import org.junit.jupiter.params.provider.ValueSource; import java.io.IOException; @@ -407,6 +409,11 @@ protected KeyValueFileWriterFactory createWriterFactory(String pathStr, String f protected KeyValueFileWriterFactory createWriterFactory( String pathStr, String format, Options options) { + return createWriterFactory(pathStr, format, options, false); + } + + private KeyValueFileWriterFactory createWriterFactory( + String pathStr, String format, Options options, boolean useTableSchema) { Path path = new Path(pathStr); int suggestedFileSize = ThreadLocalRandom.current().nextInt(8192) + 1024; FileIO fileIO = FileIOFinder.find(path); @@ -430,18 +437,26 @@ protected KeyValueFileWriterFactory createWriterFactory( false, null); - return KeyValueFileWriterFactory.builder( - fileIO, - 0, - KEY_TYPE, - DEFAULT_ROW_TYPE, - // normal format will buffer changes in memory and we can't determine - // if the written file size is really larger than suggested, so we use a - // special format which flushes for every added element - new FlushingFileFormat(format), - pathFactoryMap, - suggestedFileSize) - .build(BinaryRow.EMPTY_ROW, 0, new CoreOptions(options)); + FileFormat fileFormat = new FlushingFileFormat(format); + KeyValueFileWriterFactory.Builder builder = + useTableSchema + ? KeyValueFileWriterFactory.builder( + fileIO, + createTestSchemaManager(path).schema(0), + KEY_TYPE, + DEFAULT_ROW_TYPE, + fileFormat, + pathFactoryMap, + suggestedFileSize) + : KeyValueFileWriterFactory.builder( + fileIO, + 0, + KEY_TYPE, + DEFAULT_ROW_TYPE, + fileFormat, + pathFactoryMap, + suggestedFileSize); + return builder.build(BinaryRow.EMPTY_ROW, 0, new CoreOptions(options)); } private KeyValueFileReaderFactory createReaderFactory( @@ -604,6 +619,108 @@ void testChangelogFile() throws Exception { } } + @ParameterizedTest(name = "format={0}, thinMode={1}") + @CsvSource({ + "avro, false", + "avro, true", + "orc, false", + "orc, true", + "parquet, false", + "parquet, true" + }) + void testDeleteRowFieldsProjection(String format, boolean thinMode) throws Exception { + Options options = new Options(); + options.set(CoreOptions.METADATA_STATS_MODE, "FULL"); + options.set(CoreOptions.METADATA_STATS_DENSE_STORE, false); + options.set(CoreOptions.DATA_FILE_THIN_MODE, thinMode); + options.set(CoreOptions.SEQUENCE_FIELD, "itemId"); + + KeyValueFileWriterFactory writerFactory = + createWriterFactory(tempDir.toString(), format, options, true); + TestKeyValueGenerator generator = new TestKeyValueGenerator(); + KeyValue insert = + generator.nextInsert("20201110", 9, 42L, new int[] {10, 2}, "delete-before-image"); + KeyValue delete = + new KeyValue() + .replace( + insert.key(), + insert.sequenceNumber(), + RowKind.DELETE, + insert.value()); + + RollingFileWriter dataWriter = + writerFactory.createRollingMergeTreeFileWriter(0, FileSource.APPEND); + dataWriter.write(delete); + dataWriter.close(); + DataFileMeta dataFile = dataWriter.result().get(0); + + KeyValueFileReaderFactory readerFactory = + createReaderFactory(tempDir.toString(), format, null, null); + KeyValue dataDelete = readSingle(readerFactory, dataFile); + assertThat(dataDelete.valueKind()).isEqualTo(RowKind.DELETE); + assertThat(dataDelete.value().getLong(4)).isEqualTo(42L); + assertThat(dataDelete.value().isNullAt(5)).isTrue(); + assertThat(dataDelete.value().isNullAt(6)).isTrue(); + assertThat(dataFile.valueStats().nullCounts().getLong(6)).isEqualTo(1L); + + RollingFileWriter changelogWriter = + writerFactory.createRollingChangelogFileWriter(0); + changelogWriter.write(delete); + changelogWriter.close(); + + KeyValue changelogDelete = readSingle(readerFactory, changelogWriter.result().get(0)); + assertThat(changelogDelete.valueKind()).isEqualTo(RowKind.DELETE); + assertThat(changelogDelete.value().isNullAt(5)).isFalse(); + assertThat(changelogDelete.value().getString(6).toString()) + .isEqualTo("delete-before-image"); + } + + @Test + void testDeleteRowFieldsProjectionInClusteringFile() throws Exception { + Options options = new Options(); + options.set(CoreOptions.CLUSTERING_COLUMNS, "comment"); + options.set(CoreOptions.SEQUENCE_FIELD, "itemId"); + + KeyValueFileWriterFactory writerFactory = + createWriterFactory(tempDir.toString(), "avro", options, true); + KeyValue insert = + new TestKeyValueGenerator() + .nextInsert("20201110", 9, 42L, new int[] {10, 2}, "clustering-field"); + KeyValue delete = + new KeyValue() + .replace( + insert.key(), + insert.sequenceNumber(), + RowKind.DELETE, + insert.value()); + + RollingFileWriter writer = + writerFactory.createRollingClusteringFileWriter(); + writer.write(delete); + writer.close(); + + KeyValueFileReaderFactory readerFactory = + createReaderFactory(tempDir.toString(), "avro", null, null); + KeyValue actual = readSingle(readerFactory, writer.result().get(0)); + assertThat(actual.value().isNullAt(5)).isTrue(); + assertThat(actual.value().getString(6).toString()).isEqualTo("clustering-field"); + } + + private KeyValue readSingle(KeyValueFileReaderFactory readerFactory, DataFileMeta file) + throws Exception { + try (RecordReaderIterator iterator = + new RecordReaderIterator<>(readerFactory.createRecordReader(file))) { + assertThat(iterator.hasNext()).isTrue(); + KeyValue result = + iterator.next() + .copy( + TestKeyValueGenerator.KEY_SERIALIZER, + TestKeyValueGenerator.DEFAULT_ROW_SERIALIZER); + assertThat(iterator.hasNext()).isFalse(); + return result; + } + } + @ParameterizedTest @ValueSource(strings = {"parquet", "orc", "avro"}) public void testReaderUseFileSizeFromMetadata(String format) throws Exception { diff --git a/paimon-core/src/test/java/org/apache/paimon/io/PrimaryKeyBlobFileWriterTest.java b/paimon-core/src/test/java/org/apache/paimon/io/PrimaryKeyBlobFileWriterTest.java index d7f4d2ee0e54..f706139c81b8 100644 --- a/paimon-core/src/test/java/org/apache/paimon/io/PrimaryKeyBlobFileWriterTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/io/PrimaryKeyBlobFileWriterTest.java @@ -33,6 +33,7 @@ import org.apache.paimon.fs.local.LocalFileIO; import org.apache.paimon.manifest.FileSource; import org.apache.paimon.options.Options; +import org.apache.paimon.schema.TableSchema; import org.apache.paimon.table.SpecialFields; import org.apache.paimon.types.DataField; import org.apache.paimon.types.DataTypes; @@ -91,10 +92,19 @@ void testAttachReferenceSidecarToDataFile() throws Exception { Options options = new Options(); options.set(CoreOptions.BLOB_FIELD, "payload"); options.set(CoreOptions.BLOB_DESCRIPTOR_FIELD, "descriptor"); + TableSchema schema = + new TableSchema( + 0, + valueType.getFields(), + 2, + Collections.emptyList(), + Collections.singletonList("id"), + options.toMap(), + ""); KeyValueFileWriterFactory factory = KeyValueFileWriterFactory.builder( fileIO, - 0, + schema, keyType, valueType, new FlushingFileFormat("avro"), @@ -117,10 +127,12 @@ void testAttachReferenceSidecarToDataFile() throws Exception { factory.createRollingMergeTreeFileWriter(0, FileSource.APPEND); writer.write(new KeyValue().replace(GenericRow.of(1), 0, RowKind.INSERT, externalized)); + writer.write(new KeyValue().replace(GenericRow.of(1), 1, RowKind.DELETE, externalized)); writer.close(); factory.prepareCommit(); DataFileMeta meta = writer.result().get(0); + assertThat(meta.deleteRowCount()).hasValue(1L); assertThat(meta.extraFiles()).singleElement().asString().endsWith(".blobref"); Path sidecar = pathFactory.toAlignedPath(meta.extraFiles().get(0), meta); assertThat(ManagedBlobReferenceFile.read(fileIO, sidecar)) diff --git a/paimon-core/src/test/java/org/apache/paimon/table/DeleteRowFieldsTest.java b/paimon-core/src/test/java/org/apache/paimon/table/DeleteRowFieldsTest.java new file mode 100644 index 000000000000..f7c6680a1666 --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/table/DeleteRowFieldsTest.java @@ -0,0 +1,149 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.table; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.data.BinaryString; +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.data.serializer.InternalRowSerializer; +import org.apache.paimon.predicate.Predicate; +import org.apache.paimon.predicate.PredicateBuilder; +import org.apache.paimon.reader.RecordReader; +import org.apache.paimon.schema.Schema; +import org.apache.paimon.table.source.InnerTableRead; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.types.RowKind; +import org.apache.paimon.types.RowType; + +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for storing only required fields in primary-key DELETE rows. */ +public class DeleteRowFieldsTest extends TableTestBase { + + @ParameterizedTest + @ValueSource(strings = {"deduplicate", "partial-update", "aggregation"}) + public void testDeleteReadFilterCompactionAndReinsert(String mergeEngine) throws Exception { + FileStoreTable table = createTable(mergeEngine); + write(table, row(RowKind.INSERT, 1L, "old-payload")); + write(table, row(RowKind.DELETE, 2L, "old-payload")); + + assertThat(read(table)).isEmpty(); + + List deletes = readKeepingDeletes(table); + assertThat(deletes).hasSize(1); + InternalRow delete = deletes.get(0); + assertThat(delete.getRowKind()).isEqualTo(RowKind.DELETE); + assertThat(delete.getInt(0)).isEqualTo(1); + assertThat(delete.getInt(1)).isEqualTo(1); + assertThat(delete.getLong(2)).isEqualTo(2L); + assertThat(delete.getString(3).toString()).isEqualTo("required-value"); + assertThat(delete.isNullAt(4)).isTrue(); + + Predicate payloadFilter = + new PredicateBuilder(table.rowType()) + .equal(4, BinaryString.fromString("old-payload")); + assertThat(readWithFilter(table, payloadFilter)).isEmpty(); + + FileStoreTable compactionTable = + (FileStoreTable) + table.copy(Collections.singletonMap(CoreOptions.WRITE_ONLY.key(), "false")); + compact(compactionTable, partition(1), 0); + assertThat(read(compactionTable)).isEmpty(); + + write(compactionTable, row(RowKind.INSERT, 3L, "new-payload")); + List rows = read(compactionTable); + assertThat(rows).hasSize(1); + assertThat(rows.get(0).getLong(2)).isEqualTo(3L); + assertThat(rows.get(0).getString(4).toString()).isEqualTo("new-payload"); + } + + private FileStoreTable createTable(String mergeEngine) throws Exception { + Schema.Builder builder = + Schema.newBuilder() + .column("pt", DataTypes.INT()) + .column("k", DataTypes.INT()) + .column("seq", DataTypes.BIGINT()) + .column("required_value", DataTypes.STRING().notNull()) + .column("payload", DataTypes.STRING()) + .partitionKeys("pt") + .primaryKey("pt", "k") + .option(CoreOptions.BUCKET.key(), "1") + .option(CoreOptions.WRITE_ONLY.key(), "true") + .option(CoreOptions.SEQUENCE_FIELD.key(), "seq") + .option(CoreOptions.DATA_FILE_THIN_MODE.key(), "true") + .option(CoreOptions.MERGE_ENGINE.key(), mergeEngine); + if (mergeEngine.equals("partial-update")) { + builder.option(CoreOptions.PARTIAL_UPDATE_REMOVE_RECORD_ON_DELETE.key(), "true"); + } else if (mergeEngine.equals("aggregation")) { + builder.option(CoreOptions.AGGREGATION_REMOVE_RECORD_ON_DELETE.key(), "true"); + } + catalog.createTable(identifier(), builder.build(), false); + return (FileStoreTable) catalog.getTable(identifier()); + } + + private GenericRow row(RowKind kind, long sequence, String payload) { + return GenericRow.ofKind( + kind, + 1, + 1, + sequence, + BinaryString.fromString("required-value"), + BinaryString.fromString(payload)); + } + + private BinaryRow partition(int value) { + InternalRowSerializer serializer = + new InternalRowSerializer(RowType.of(DataTypes.INT().notNull())); + return serializer.toBinaryRow(GenericRow.of(value)).copy(); + } + + private List readKeepingDeletes(FileStoreTable table) throws Exception { + InnerTableRead read = table.newRead().forceKeepDelete(); + return read(table, read, null); + } + + private List readWithFilter(FileStoreTable table, Predicate predicate) + throws Exception { + InnerTableRead read = table.newRead().withFilter(predicate); + return read(table, read, predicate); + } + + private List read(FileStoreTable table, InnerTableRead read, Predicate predicate) + throws Exception { + List rows = new ArrayList<>(); + InternalRowSerializer serializer = new InternalRowSerializer(table.rowType()); + try (RecordReader reader = + read.createReader( + predicate == null + ? table.newScan().plan() + : table.newScan().withFilter(predicate).plan())) { + reader.forEachRemaining(row -> rows.add(serializer.copy(row))); + } + return rows; + } +} From 778a7d21c1a5060c4787e21f1318fc449d1653c6 Mon Sep 17 00:00:00 2001 From: umi Date: Tue, 27 Oct 2026 14:31:18 +0800 Subject: [PATCH 2/6] [core] Stop retaining indexed fields in whole-record deletes --- .../paimon/io/DeleteRowFieldsProjector.java | 6 ---- .../io/DeleteRowFieldsProjectorTest.java | 5 ++-- .../source/PrimaryKeySortedIndexReadTest.java | 29 ++++++++++++++++--- 3 files changed, 28 insertions(+), 12 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/io/DeleteRowFieldsProjector.java b/paimon-core/src/main/java/org/apache/paimon/io/DeleteRowFieldsProjector.java index 34e54e7be1f9..7f96392070c1 100644 --- a/paimon-core/src/main/java/org/apache/paimon/io/DeleteRowFieldsProjector.java +++ b/paimon-core/src/main/java/org/apache/paimon/io/DeleteRowFieldsProjector.java @@ -47,12 +47,6 @@ final class DeleteRowFieldsProjector { requiredFields.add(options.recordLevelTimeField()); } requiredFields.addAll(options.clusteringColumns()); - requiredFields.addAll(options.primaryKeyVectorIndexColumns()); - requiredFields.addAll(options.primaryKeyBTreeIndexColumns()); - requiredFields.addAll(options.primaryKeyBitmapIndexColumns()); - requiredFields.addAll(options.primaryKeyMultiValueIndexColumns()); - requiredFields.addAll(options.primaryKeyFullTextIndexColumns()); - requiredFields.addAll(options.primaryKeyFMIndexColumns()); for (DataField field : schema.fields()) { if (!field.type().isNullable()) { requiredFields.add(field.name()); diff --git a/paimon-core/src/test/java/org/apache/paimon/io/DeleteRowFieldsProjectorTest.java b/paimon-core/src/test/java/org/apache/paimon/io/DeleteRowFieldsProjectorTest.java index 925a4aad66ca..ec3633b52f5d 100644 --- a/paimon-core/src/test/java/org/apache/paimon/io/DeleteRowFieldsProjectorTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/io/DeleteRowFieldsProjectorTest.java @@ -44,7 +44,8 @@ void testRequiredFieldProjection() { options.set(CoreOptions.RECORD_LEVEL_TIME_FIELD, "expire"); options.set(CoreOptions.CLUSTERING_COLUMNS, "cluster"); options.set(CoreOptions.PK_VECTOR_INDEX_COLUMNS, "vector_index"); - options.set(CoreOptions.PK_BTREE_INDEX_COLUMNS, "btree_index"); + options.set( + CoreOptions.PK_BTREE_INDEX_COLUMNS, "btree_index,key,not_null,seq,expire,cluster"); options.set(CoreOptions.PK_BITMAP_INDEX_COLUMNS, "bitmap_index"); options.set(CoreOptions.PK_MULTIVALUE_INDEX_COLUMNS, "multivalue_index"); options.set(CoreOptions.PK_FULL_TEXT_INDEX_COLUMNS, "fulltext_index"); @@ -53,7 +54,7 @@ void testRequiredFieldProjection() { assertThat( DeleteRowFieldsProjector.createProjection( tableSchema(), new CoreOptions(options))) - .containsExactly(0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, -1); + .containsExactly(0, 1, 2, 3, 4, -1, -1, -1, -1, -1, -1, -1); } @Test diff --git a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexReadTest.java b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexReadTest.java index c76ba9ddb7b6..39f40d31fcfa 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexReadTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexReadTest.java @@ -42,6 +42,7 @@ import org.apache.paimon.table.sink.TableCommitImpl; import org.apache.paimon.table.sink.TableWriteImpl; import org.apache.paimon.types.DataTypes; +import org.apache.paimon.types.RowKind; import org.apache.paimon.shade.guava30.com.google.common.util.concurrent.MoreExecutors; @@ -89,10 +90,30 @@ void testDeletionVectorAndResidualPredicateRemainActive() throws Exception { .newIndexFileHandler() .scanSourceIndexes(compactedSnapshot, BinaryRow.EMPTY_ROW, 0)) .isNotEmpty(); - write( - table, - ioManager, - GenericRow.ofKind(org.apache.paimon.types.RowKind.DELETE, 3, 10, keep)); + FileStoreTable writeOnlyTable = + table.copy(Collections.singletonMap(CoreOptions.WRITE_ONLY.key(), "true")); + write(writeOnlyTable, ioManager, GenericRow.ofKind(RowKind.DELETE, 3, 10, keep)); + + // Verify that the indexed value is cleared in the persisted DELETE before compaction. + List deletedIds = new ArrayList<>(); + try (RecordReader reader = + writeOnlyTable + .newRead() + .forceKeepDelete() + .createReader( + writeOnlyTable + .newScan() + .withLevelFilter(level -> level == 0) + .plan())) { + reader.forEachRemaining( + row -> { + assertThat(row.getRowKind()).isEqualTo(RowKind.DELETE); + assertThat(row.isNullAt(1)).isTrue(); + deletedIds.add(row.getInt(0)); + }); + } + assertThat(deletedIds).containsExactly(3); + compact(table, BinaryRow.EMPTY_ROW, 0, ioManager, false); Snapshot snapshot = table.store().snapshotManager().latestSnapshot(); List payloads = From 7fe1dabfeeded7e457c69ddea2925dc739a68fbd Mon Sep 17 00:00:00 2001 From: umi Date: Tue, 27 Oct 2026 14:54:17 +0800 Subject: [PATCH 3/6] [core] Make delete row thin mode opt-in --- docs/docs/concepts/spec/datafile.md | 16 +++--- docs/generated/core_configuration.html | 6 +++ .../java/org/apache/paimon/CoreOptions.java | 14 ++++++ .../paimon/io/DeleteRowFieldsProjector.java | 4 +- .../io/DeleteRowFieldsProjectorTest.java | 49 ++++++++++++------- .../paimon/io/KeyValueFileReadWriteTest.java | 2 + .../paimon/table/DeleteRowFieldsTest.java | 27 +++++++--- .../source/PrimaryKeySortedIndexReadTest.java | 6 ++- 8 files changed, 91 insertions(+), 33 deletions(-) diff --git a/docs/docs/concepts/spec/datafile.md b/docs/docs/concepts/spec/datafile.md index 31375578cc8e..78b323336990 100644 --- a/docs/docs/concepts/spec/datafile.md +++ b/docs/docs/concepts/spec/datafile.md @@ -137,12 +137,16 @@ contains five columns: _SEQUENCE_NUMBER | _VALUE_KIND | a | b | c ``` -For a primary-key merge engine where `DELETE` removes the whole record, Paimon keeps the -primary-key, sequence, record-level time, clustering, indexed, and non-null fields in data-file -DELETE rows, and writes null for other nullable value fields. This applies to deduplicate tables, -partial-update tables with `partial-update.remove-record-on-delete`, and aggregation tables with -`aggregation.remove-record-on-delete`. The physical schema is unchanged. Changelog files are not -projected by this optimization. +With `delete-row.thin-mode = true`, for a primary-key merge engine where `DELETE` removes the +whole record, Paimon keeps the primary-key, sequence, record-level time, clustering, and non-null +fields in data-file DELETE rows, and writes null for other nullable value fields. This applies to +deduplicate tables, partial-update tables with `partial-update.remove-record-on-delete`, and +aggregation tables with `aggregation.remove-record-on-delete`. The physical schema is unchanged. +Changelog files are not projected by this optimization. + +This option defaults to `false` and is independent of `data-file.thin-mode`. When enabled, readers +that expose DELETE rows from data files, such as audit-log and delta readers, may observe nulls in +these fields. Disabling the option does not restore values already written as null. ### Table w/o Primary key Data File diff --git a/docs/generated/core_configuration.html b/docs/generated/core_configuration.html index 0fc8c42fd261..d6a4d7ac67f4 100644 --- a/docs/generated/core_configuration.html +++ b/docs/generated/core_configuration.html @@ -650,6 +650,12 @@ Boolean Enable data file thin mode to avoid duplicate columns storage. + +
delete-row.thin-mode
+ false + Boolean + Enable thin mode for whole-record DELETE rows in primary-key data files. Non-essential nullable value fields are stored as null. Only applies to deduplicate, or partial-update and aggregation with remove-record-on-delete enabled. Changelog files are unaffected. +
delete.force-produce-changelog
false diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java index 283aae1dbabf..9982a041165c 100644 --- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java +++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java @@ -2427,6 +2427,16 @@ public String toString() { .withDescription( "Enable data file thin mode to avoid duplicate columns storage."); + public static final ConfigOption DELETE_ROW_THIN_MODE = + key("delete-row.thin-mode") + .booleanType() + .defaultValue(false) + .withDescription( + "Enable thin mode for whole-record DELETE rows in primary-key data files. " + + "Non-essential nullable value fields are stored as null. " + + "Only applies to deduplicate, or partial-update and aggregation " + + "with remove-record-on-delete enabled. Changelog files are unaffected."); + public static final ConfigOption PARTITION_IDLE_TIME_TO_REPORT_STATISTIC = key("partition.idle-time-to-report-statistic") .durationType() @@ -4781,6 +4791,10 @@ public boolean dataFileThinMode() { return options.get(DATA_FILE_THIN_MODE); } + public boolean deleteRowThinMode() { + return options.get(DELETE_ROW_THIN_MODE); + } + public boolean aggregationRemoveRecordOnDelete() { return options.get(AGGREGATION_REMOVE_RECORD_ON_DELETE); } diff --git a/paimon-core/src/main/java/org/apache/paimon/io/DeleteRowFieldsProjector.java b/paimon-core/src/main/java/org/apache/paimon/io/DeleteRowFieldsProjector.java index 7f96392070c1..de5b17cb0cbb 100644 --- a/paimon-core/src/main/java/org/apache/paimon/io/DeleteRowFieldsProjector.java +++ b/paimon-core/src/main/java/org/apache/paimon/io/DeleteRowFieldsProjector.java @@ -37,7 +37,9 @@ final class DeleteRowFieldsProjector { private final KeyValue reused = new KeyValue(); static @Nullable int[] createProjection(TableSchema schema, CoreOptions options) { - if (schema.primaryKeys().isEmpty() || !removeWholeRecordOnDelete(options)) { + if (!options.deleteRowThinMode() + || schema.primaryKeys().isEmpty() + || !removeWholeRecordOnDelete(options)) { return null; } diff --git a/paimon-core/src/test/java/org/apache/paimon/io/DeleteRowFieldsProjectorTest.java b/paimon-core/src/test/java/org/apache/paimon/io/DeleteRowFieldsProjectorTest.java index ec3633b52f5d..da15148549a8 100644 --- a/paimon-core/src/test/java/org/apache/paimon/io/DeleteRowFieldsProjectorTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/io/DeleteRowFieldsProjectorTest.java @@ -37,9 +37,20 @@ /** Tests for {@link DeleteRowFieldsProjector}. */ class DeleteRowFieldsProjectorTest { + @Test + void testDisabledByDefault() { + Options options = new Options(); + assertThat(createProjection(options)).isNull(); + options.set(CoreOptions.DELETE_ROW_THIN_MODE, true); + assertThat(createProjection(options)).isNotNull(); + options.set(CoreOptions.DELETE_ROW_THIN_MODE, false); + assertThat(createProjection(options)).isNull(); + } + @Test void testRequiredFieldProjection() { Options options = new Options(); + options.set(CoreOptions.DELETE_ROW_THIN_MODE, true); options.set(CoreOptions.SEQUENCE_FIELD, "seq"); options.set(CoreOptions.RECORD_LEVEL_TIME_FIELD, "expire"); options.set(CoreOptions.CLUSTERING_COLUMNS, "cluster"); @@ -59,23 +70,22 @@ void testRequiredFieldProjection() { @Test void testOnlyWholeRecordDeletesAreProjected() { - assertThat(createProjection(new Options())).isNotNull(); - - Options partialUpdate = new Options(); - partialUpdate.set(CoreOptions.MERGE_ENGINE, CoreOptions.MergeEngine.PARTIAL_UPDATE); - assertThat(createProjection(partialUpdate)).isNull(); - partialUpdate.set(CoreOptions.PARTIAL_UPDATE_REMOVE_RECORD_ON_DELETE, true); - assertThat(createProjection(partialUpdate)).isNotNull(); - - Options aggregation = new Options(); - aggregation.set(CoreOptions.MERGE_ENGINE, CoreOptions.MergeEngine.AGGREGATE); - assertThat(createProjection(aggregation)).isNull(); - aggregation.set(CoreOptions.AGGREGATION_REMOVE_RECORD_ON_DELETE, true); - assertThat(createProjection(aggregation)).isNotNull(); - - Options firstRow = new Options(); - firstRow.set(CoreOptions.MERGE_ENGINE, CoreOptions.MergeEngine.FIRST_ROW); - assertThat(createProjection(firstRow)).isNull(); + Options options = new Options(); + options.set(CoreOptions.DELETE_ROW_THIN_MODE, true); + assertThat(createProjection(options)).isNotNull(); + + options.set(CoreOptions.MERGE_ENGINE, CoreOptions.MergeEngine.PARTIAL_UPDATE); + assertThat(createProjection(options)).isNull(); + options.set(CoreOptions.PARTIAL_UPDATE_REMOVE_RECORD_ON_DELETE, true); + assertThat(createProjection(options)).isNotNull(); + + options.set(CoreOptions.MERGE_ENGINE, CoreOptions.MergeEngine.AGGREGATE); + assertThat(createProjection(options)).isNull(); + options.set(CoreOptions.AGGREGATION_REMOVE_RECORD_ON_DELETE, true); + assertThat(createProjection(options)).isNotNull(); + + options.set(CoreOptions.MERGE_ENGINE, CoreOptions.MergeEngine.FIRST_ROW); + assertThat(createProjection(options)).isNull(); } @Test @@ -96,8 +106,9 @@ void testProjectionDoesNotChangeInput() { "payload"); KeyValue input = new KeyValue().replace(GenericRow.of(1), 2L, RowKind.DELETE, value); - KeyValue projected = - new DeleteRowFieldsProjector(createProjection(new Options())).project(input); + Options options = new Options(); + options.set(CoreOptions.DELETE_ROW_THIN_MODE, true); + KeyValue projected = new DeleteRowFieldsProjector(createProjection(options)).project(input); assertThat(projected.key()).isSameAs(input.key()); assertThat(projected.sequenceNumber()).isEqualTo(2L); diff --git a/paimon-core/src/test/java/org/apache/paimon/io/KeyValueFileReadWriteTest.java b/paimon-core/src/test/java/org/apache/paimon/io/KeyValueFileReadWriteTest.java index f9ec29b0ce24..2934354314fd 100644 --- a/paimon-core/src/test/java/org/apache/paimon/io/KeyValueFileReadWriteTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/io/KeyValueFileReadWriteTest.java @@ -630,6 +630,7 @@ void testChangelogFile() throws Exception { }) void testDeleteRowFieldsProjection(String format, boolean thinMode) throws Exception { Options options = new Options(); + options.set(CoreOptions.DELETE_ROW_THIN_MODE, true); options.set(CoreOptions.METADATA_STATS_MODE, "FULL"); options.set(CoreOptions.METADATA_STATS_DENSE_STORE, false); options.set(CoreOptions.DATA_FILE_THIN_MODE, thinMode); @@ -678,6 +679,7 @@ void testDeleteRowFieldsProjection(String format, boolean thinMode) throws Excep @Test void testDeleteRowFieldsProjectionInClusteringFile() throws Exception { Options options = new Options(); + options.set(CoreOptions.DELETE_ROW_THIN_MODE, true); options.set(CoreOptions.CLUSTERING_COLUMNS, "comment"); options.set(CoreOptions.SEQUENCE_FIELD, "itemId"); diff --git a/paimon-core/src/test/java/org/apache/paimon/table/DeleteRowFieldsTest.java b/paimon-core/src/test/java/org/apache/paimon/table/DeleteRowFieldsTest.java index f7c6680a1666..dcf029b37552 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/DeleteRowFieldsTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/DeleteRowFieldsTest.java @@ -34,7 +34,7 @@ import org.apache.paimon.types.RowType; import org.junit.jupiter.params.ParameterizedTest; -import org.junit.jupiter.params.provider.ValueSource; +import org.junit.jupiter.params.provider.CsvSource; import java.util.ArrayList; import java.util.Collections; @@ -46,9 +46,17 @@ public class DeleteRowFieldsTest extends TableTestBase { @ParameterizedTest - @ValueSource(strings = {"deduplicate", "partial-update", "aggregation"}) - public void testDeleteReadFilterCompactionAndReinsert(String mergeEngine) throws Exception { - FileStoreTable table = createTable(mergeEngine); + @CsvSource({ + "deduplicate, false", + "deduplicate, true", + "partial-update, false", + "partial-update, true", + "aggregation, false", + "aggregation, true" + }) + public void testDeleteReadFilterCompactionAndReinsert(String mergeEngine, boolean thinMode) + throws Exception { + FileStoreTable table = createTable(mergeEngine, thinMode); write(table, row(RowKind.INSERT, 1L, "old-payload")); write(table, row(RowKind.DELETE, 2L, "old-payload")); @@ -62,7 +70,11 @@ public void testDeleteReadFilterCompactionAndReinsert(String mergeEngine) throws assertThat(delete.getInt(1)).isEqualTo(1); assertThat(delete.getLong(2)).isEqualTo(2L); assertThat(delete.getString(3).toString()).isEqualTo("required-value"); - assertThat(delete.isNullAt(4)).isTrue(); + if (thinMode) { + assertThat(delete.isNullAt(4)).isTrue(); + } else { + assertThat(delete.getString(4).toString()).isEqualTo("old-payload"); + } Predicate payloadFilter = new PredicateBuilder(table.rowType()) @@ -82,7 +94,7 @@ public void testDeleteReadFilterCompactionAndReinsert(String mergeEngine) throws assertThat(rows.get(0).getString(4).toString()).isEqualTo("new-payload"); } - private FileStoreTable createTable(String mergeEngine) throws Exception { + private FileStoreTable createTable(String mergeEngine, boolean thinMode) throws Exception { Schema.Builder builder = Schema.newBuilder() .column("pt", DataTypes.INT()) @@ -97,6 +109,9 @@ private FileStoreTable createTable(String mergeEngine) throws Exception { .option(CoreOptions.SEQUENCE_FIELD.key(), "seq") .option(CoreOptions.DATA_FILE_THIN_MODE.key(), "true") .option(CoreOptions.MERGE_ENGINE.key(), mergeEngine); + if (thinMode) { + builder.option(CoreOptions.DELETE_ROW_THIN_MODE.key(), "true"); + } if (mergeEngine.equals("partial-update")) { builder.option(CoreOptions.PARTIAL_UPDATE_REMOVE_RECORD_ON_DELETE.key(), "true"); } else if (mergeEngine.equals("aggregation")) { diff --git a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexReadTest.java b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexReadTest.java index 39f40d31fcfa..bde6879c054c 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexReadTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexReadTest.java @@ -77,7 +77,11 @@ protected Schema schemaDefault() { @Test void testDeletionVectorAndResidualPredicateRemainActive() throws Exception { createTableDefault(); - FileStoreTable table = getTableDefault(); + FileStoreTable table = + getTableDefault() + .copy( + Collections.singletonMap( + CoreOptions.DELETE_ROW_THIN_MODE.key(), "true")); BinaryString keep = BinaryString.fromString("keep"); BinaryString drop = BinaryString.fromString("drop"); write(table, ioManager, GenericRow.of(1, 10, keep), GenericRow.of(2, 10, drop)); From 948a8a327391b6f197984dcf3150b588bf76495d Mon Sep 17 00:00:00 2001 From: umi Date: Tue, 27 Oct 2026 17:10:59 +0800 Subject: [PATCH 4/6] [core] Validate delete row thin mode compatibility in schema --- docs/docs/concepts/spec/datafile.md | 9 +- docs/generated/core_configuration.html | 2 +- .../java/org/apache/paimon/CoreOptions.java | 6 +- .../paimon/io/DeleteRowFieldsProjector.java | 17 +-- .../paimon/schema/SchemaValidation.java | 46 ++++++++ .../io/DeleteRowFieldsProjectorTest.java | 29 +++-- .../paimon/schema/SchemaValidationTest.java | 103 ++++++++++++++++++ .../paimon/table/DeleteRowFieldsTest.java | 46 ++++++++ 8 files changed, 221 insertions(+), 37 deletions(-) diff --git a/docs/docs/concepts/spec/datafile.md b/docs/docs/concepts/spec/datafile.md index 78b323336990..0afd8f3b93af 100644 --- a/docs/docs/concepts/spec/datafile.md +++ b/docs/docs/concepts/spec/datafile.md @@ -139,9 +139,12 @@ _SEQUENCE_NUMBER | _VALUE_KIND | a | b | c With `delete-row.thin-mode = true`, for a primary-key merge engine where `DELETE` removes the whole record, Paimon keeps the primary-key, sequence, record-level time, clustering, and non-null -fields in data-file DELETE rows, and writes null for other nullable value fields. This applies to -deduplicate tables, partial-update tables with `partial-update.remove-record-on-delete`, and -aggregation tables with `aggregation.remove-record-on-delete`. The physical schema is unchanged. +fields in data-file DELETE rows, and writes null for other nullable value fields. Enabling this option +requires a primary-key table using deduplicate, partial-update with +`partial-update.remove-record-on-delete = true`, or aggregation with +`aggregation.remove-record-on-delete = true`. Postpone bucket tables (`bucket = -2`) are not supported. +Unsupported configurations are rejected during schema validation. If all fields must be retained, +the configuration is valid but no fields are projected. The physical schema is unchanged. Changelog files are not projected by this optimization. This option defaults to `false` and is independent of `data-file.thin-mode`. When enabled, readers diff --git a/docs/generated/core_configuration.html b/docs/generated/core_configuration.html index d6a4d7ac67f4..eda7d42dba64 100644 --- a/docs/generated/core_configuration.html +++ b/docs/generated/core_configuration.html @@ -654,7 +654,7 @@
delete-row.thin-mode
false Boolean - Enable thin mode for whole-record DELETE rows in primary-key data files. Non-essential nullable value fields are stored as null. Only applies to deduplicate, or partial-update and aggregation with remove-record-on-delete enabled. Changelog files are unaffected. + Enable thin mode for whole-record DELETE rows in primary-key data files. Non-essential nullable value fields are stored as null. Requires a primary-key table using deduplicate, or partial-update and aggregation with remove-record-on-delete enabled. Postpone bucket tables (bucket = -2) are not supported. Unsupported configurations are rejected. Changelog files are unaffected.
delete.force-produce-changelog
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java index 9982a041165c..7349439fe012 100644 --- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java +++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java @@ -2434,8 +2434,10 @@ public String toString() { .withDescription( "Enable thin mode for whole-record DELETE rows in primary-key data files. " + "Non-essential nullable value fields are stored as null. " - + "Only applies to deduplicate, or partial-update and aggregation " - + "with remove-record-on-delete enabled. Changelog files are unaffected."); + + "Requires a primary-key table using deduplicate, or partial-update " + + "and aggregation with remove-record-on-delete enabled. " + + "Postpone bucket tables (bucket = -2) are not supported. " + + "Unsupported configurations are rejected. Changelog files are unaffected."); public static final ConfigOption PARTITION_IDLE_TIME_TO_REPORT_STATISTIC = key("partition.idle-time-to-report-statistic") diff --git a/paimon-core/src/main/java/org/apache/paimon/io/DeleteRowFieldsProjector.java b/paimon-core/src/main/java/org/apache/paimon/io/DeleteRowFieldsProjector.java index de5b17cb0cbb..a331f2435e37 100644 --- a/paimon-core/src/main/java/org/apache/paimon/io/DeleteRowFieldsProjector.java +++ b/paimon-core/src/main/java/org/apache/paimon/io/DeleteRowFieldsProjector.java @@ -37,9 +37,7 @@ final class DeleteRowFieldsProjector { private final KeyValue reused = new KeyValue(); static @Nullable int[] createProjection(TableSchema schema, CoreOptions options) { - if (!options.deleteRowThinMode() - || schema.primaryKeys().isEmpty() - || !removeWholeRecordOnDelete(options)) { + if (!options.deleteRowThinMode()) { return null; } @@ -65,19 +63,6 @@ final class DeleteRowFieldsProjector { return Arrays.stream(projection).anyMatch(field -> field < 0) ? projection : null; } - private static boolean removeWholeRecordOnDelete(CoreOptions options) { - switch (options.mergeEngine()) { - case DEDUPLICATE: - return true; - case PARTIAL_UPDATE: - return options.partialUpdateRemoveRecordOnDelete(); - case AGGREGATE: - return options.aggregationRemoveRecordOnDelete(); - default: - return false; - } - } - DeleteRowFieldsProjector(int[] projection) { this.projectedValue = ProjectedRow.from(projection); } diff --git a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java index e279d6479587..c5522fa4863c 100644 --- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java +++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java @@ -187,6 +187,8 @@ public static void validateTableSchema(TableSchema schema, Set dynamicOp validateBucket(schema, options); + validateDeleteRowThinMode(schema, options); + validateStartupMode(options); validateFieldsPrefix(schema, options); @@ -879,6 +881,50 @@ private static void validateSequenceGroupSortOrder(TableSchema schema, CoreOptio } } + private static void validateDeleteRowThinMode(TableSchema schema, CoreOptions options) { + if (!options.deleteRowThinMode()) { + return; + } + + String option = CoreOptions.DELETE_ROW_THIN_MODE.key(); + checkArgument( + !schema.primaryKeys().isEmpty(), + "'%s' can only be enabled for a table with primary keys.", + option); + checkArgument( + options.bucket() != BucketMode.POSTPONE_BUCKET, + "'%s' does not support postpone bucket tables (bucket = -2).", + option); + + switch (options.mergeEngine()) { + case DEDUPLICATE: + return; + case PARTIAL_UPDATE: + checkArgument( + options.partialUpdateRemoveRecordOnDelete(), + "'%s' requires '%s' = true when '%s' = '%s'.", + option, + CoreOptions.PARTIAL_UPDATE_REMOVE_RECORD_ON_DELETE.key(), + CoreOptions.MERGE_ENGINE.key(), + options.mergeEngine()); + return; + case AGGREGATE: + checkArgument( + options.aggregationRemoveRecordOnDelete(), + "'%s' requires '%s' = true when '%s' = '%s'.", + option, + CoreOptions.AGGREGATION_REMOVE_RECORD_ON_DELETE.key(), + CoreOptions.MERGE_ENGINE.key(), + options.mergeEngine()); + return; + default: + throw new IllegalArgumentException( + String.format( + "'%s' does not support merge engine '%s'.", + option, options.mergeEngine())); + } + } + private static void validateMergeFunction(TableSchema schema) { if (schema.primaryKeys().isEmpty()) { return; diff --git a/paimon-core/src/test/java/org/apache/paimon/io/DeleteRowFieldsProjectorTest.java b/paimon-core/src/test/java/org/apache/paimon/io/DeleteRowFieldsProjectorTest.java index da15148549a8..7abcd985128d 100644 --- a/paimon-core/src/test/java/org/apache/paimon/io/DeleteRowFieldsProjectorTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/io/DeleteRowFieldsProjectorTest.java @@ -69,23 +69,22 @@ void testRequiredFieldProjection() { } @Test - void testOnlyWholeRecordDeletesAreProjected() { + void testNoFieldsToProject() { Options options = new Options(); options.set(CoreOptions.DELETE_ROW_THIN_MODE, true); - assertThat(createProjection(options)).isNotNull(); - - options.set(CoreOptions.MERGE_ENGINE, CoreOptions.MergeEngine.PARTIAL_UPDATE); - assertThat(createProjection(options)).isNull(); - options.set(CoreOptions.PARTIAL_UPDATE_REMOVE_RECORD_ON_DELETE, true); - assertThat(createProjection(options)).isNotNull(); - - options.set(CoreOptions.MERGE_ENGINE, CoreOptions.MergeEngine.AGGREGATE); - assertThat(createProjection(options)).isNull(); - options.set(CoreOptions.AGGREGATION_REMOVE_RECORD_ON_DELETE, true); - assertThat(createProjection(options)).isNotNull(); - - options.set(CoreOptions.MERGE_ENGINE, CoreOptions.MergeEngine.FIRST_ROW); - assertThat(createProjection(options)).isNull(); + TableSchema schema = + new TableSchema( + 0, + Arrays.asList( + new DataField(0, "key", DataTypes.INT().notNull()), + new DataField(1, "value", DataTypes.STRING().notNull())), + 1, + Collections.emptyList(), + Collections.singletonList("key"), + options.toMap(), + ""); + assertThat(DeleteRowFieldsProjector.createProjection(schema, new CoreOptions(options))) + .isNull(); } @Test diff --git a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java index 54a7088950bc..b3fa68fa81c9 100644 --- a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java @@ -28,6 +28,9 @@ import org.assertj.core.api.ThrowableAssert; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.CsvSource; +import org.junit.jupiter.params.provider.MethodSource; import org.junit.jupiter.params.provider.CsvSource; import org.junit.jupiter.params.provider.ValueSource; @@ -37,6 +40,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.stream.Stream; import static java.util.Collections.emptyList; import static java.util.Collections.singletonList; @@ -55,6 +59,105 @@ class SchemaValidationTest { + @ParameterizedTest + @CsvSource({ + "deduplicate, 1", + "deduplicate, -1", + "partial-update, 1", + "partial-update, -1", + "aggregation, 1", + "aggregation, -1" + }) + void testDeleteRowThinModeSupportedMergeEngines(String mergeEngine, int bucket) { + Map options = new HashMap<>(); + options.put(CoreOptions.DELETE_ROW_THIN_MODE.key(), "true"); + options.put(CoreOptions.MERGE_ENGINE.key(), mergeEngine); + options.put(BUCKET.key(), String.valueOf(bucket)); + if (mergeEngine.equals("partial-update")) { + options.put(CoreOptions.PARTIAL_UPDATE_REMOVE_RECORD_ON_DELETE.key(), "true"); + } else if (mergeEngine.equals("aggregation")) { + options.put(CoreOptions.AGGREGATION_REMOVE_RECORD_ON_DELETE.key(), "true"); + } + + // All fields are required, so enabling the option is valid even without fields to project. + assertThatCode(() -> validateTableSchema(deleteRowThinModeSchema(options, true))) + .doesNotThrowAnyException(); + } + + @ParameterizedTest + @MethodSource("unsupportedDeleteRowThinModeMergeEngines") + void testDeleteRowThinModeRequiresWholeRecordDeletes(String mergeEngine, String message) { + Map options = new HashMap<>(); + options.put(BUCKET.key(), "1"); + options.put(CoreOptions.MERGE_ENGINE.key(), mergeEngine); + assertThatCode(() -> validateTableSchema(deleteRowThinModeSchema(options, true))) + .doesNotThrowAnyException(); + options.put(CoreOptions.DELETE_ROW_THIN_MODE.key(), "false"); + assertThatCode(() -> validateTableSchema(deleteRowThinModeSchema(options, true))) + .doesNotThrowAnyException(); + + options.put(CoreOptions.DELETE_ROW_THIN_MODE.key(), "true"); + assertThatThrownBy(() -> validateTableSchema(deleteRowThinModeSchema(options, true))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage(message); + } + + private static Stream unsupportedDeleteRowThinModeMergeEngines() { + return Stream.of( + Arguments.of( + "partial-update", + "'delete-row.thin-mode' requires 'partial-update.remove-record-on-delete' = true when 'merge-engine' = 'partial-update'."), + Arguments.of( + "aggregation", + "'delete-row.thin-mode' requires 'aggregation.remove-record-on-delete' = true when 'merge-engine' = 'aggregation'."), + Arguments.of( + "first-row", + "'delete-row.thin-mode' does not support merge engine 'first-row'.")); + } + + @ParameterizedTest + @ValueSource(strings = {"none", "input"}) + void testDeleteRowThinModeRejectsPostponeBucket(String changelogProducer) { + Map options = new HashMap<>(); + options.put(BUCKET.key(), "-2"); + options.put(CoreOptions.CHANGELOG_PRODUCER.key(), changelogProducer); + assertThatCode(() -> validateTableSchema(deleteRowThinModeSchema(options, true))) + .doesNotThrowAnyException(); + + options.put(CoreOptions.DELETE_ROW_THIN_MODE.key(), "true"); + assertThatThrownBy(() -> validateTableSchema(deleteRowThinModeSchema(options, true))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage( + "'delete-row.thin-mode' does not support postpone bucket tables (bucket = -2)."); + } + + @Test + void testDeleteRowThinModeRequiresPrimaryKey() { + Map options = new HashMap<>(); + options.put(BUCKET.key(), "-1"); + assertThatCode(() -> validateTableSchema(deleteRowThinModeSchema(options, false))) + .doesNotThrowAnyException(); + + options.put(CoreOptions.DELETE_ROW_THIN_MODE.key(), "true"); + assertThatThrownBy(() -> validateTableSchema(deleteRowThinModeSchema(options, false))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage( + "'delete-row.thin-mode' can only be enabled for a table with primary keys."); + } + + private TableSchema deleteRowThinModeSchema(Map options, boolean primaryKey) { + return new TableSchema( + 0, + Arrays.asList( + new DataField(0, "id", DataTypes.INT().notNull()), + new DataField(1, "value", DataTypes.INT().notNull())), + 1, + emptyList(), + primaryKey ? singletonList("id") : emptyList(), + options, + ""); + } + @Test void testNullablePrimaryKeyRequiresPrimaryKeyTable() { Map options = new HashMap<>(); diff --git a/paimon-core/src/test/java/org/apache/paimon/table/DeleteRowFieldsTest.java b/paimon-core/src/test/java/org/apache/paimon/table/DeleteRowFieldsTest.java index dcf029b37552..9ec3aa0259c0 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/DeleteRowFieldsTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/DeleteRowFieldsTest.java @@ -19,6 +19,7 @@ package org.apache.paimon.table; import org.apache.paimon.CoreOptions; +import org.apache.paimon.catalog.Catalog; import org.apache.paimon.data.BinaryRow; import org.apache.paimon.data.BinaryString; import org.apache.paimon.data.GenericRow; @@ -28,11 +29,13 @@ import org.apache.paimon.predicate.PredicateBuilder; import org.apache.paimon.reader.RecordReader; import org.apache.paimon.schema.Schema; +import org.apache.paimon.schema.SchemaChange; import org.apache.paimon.table.source.InnerTableRead; import org.apache.paimon.types.DataTypes; import org.apache.paimon.types.RowKind; import org.apache.paimon.types.RowType; +import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.CsvSource; @@ -41,10 +44,53 @@ import java.util.List; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; /** Tests for storing only required fields in primary-key DELETE rows. */ public class DeleteRowFieldsTest extends TableTestBase { + @Test + public void testPostponeBucketRejectsDeleteRowThinMode() throws Exception { + Schema.Builder builder = + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column("value", DataTypes.INT()) + .primaryKey("id") + .option(CoreOptions.BUCKET.key(), "-2") + .option(CoreOptions.CHANGELOG_PRODUCER.key(), "input") + .option(CoreOptions.DELETE_ROW_THIN_MODE.key(), "true"); + String message = + "'delete-row.thin-mode' does not support postpone bucket tables (bucket = -2)."; + assertThatThrownBy(() -> catalog.createTable(identifier(), builder.build(), false)) + .hasRootCauseInstanceOf(IllegalArgumentException.class) + .hasRootCauseMessage(message); + assertThatThrownBy(() -> catalog.getTable(identifier())) + .isInstanceOf(Catalog.TableNotExistException.class); + + builder.option(CoreOptions.DELETE_ROW_THIN_MODE.key(), "false"); + catalog.createTable(identifier(), builder.build(), false); + FileStoreTable table = (FileStoreTable) catalog.getTable(identifier()); + assertThatThrownBy( + () -> + catalog.alterTable( + identifier(), + SchemaChange.setOption( + CoreOptions.DELETE_ROW_THIN_MODE.key(), "true"), + false)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage(message); + assertThat(((FileStoreTable) catalog.getTable(identifier())).schema()) + .isEqualTo(table.schema()); + + assertThatThrownBy( + () -> + table.copy( + Collections.singletonMap( + CoreOptions.DELETE_ROW_THIN_MODE.key(), "true"))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage(message); + } + @ParameterizedTest @CsvSource({ "deduplicate, false", From 81828c9feacdfc40ca279fcd9402be960a8ac4f0 Mon Sep 17 00:00:00 2001 From: umi Date: Thu, 29 Oct 2026 17:03:21 +0800 Subject: [PATCH 5/6] fix --- .../test/java/org/apache/paimon/schema/SchemaValidationTest.java | 1 - 1 file changed, 1 deletion(-) diff --git a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java index b3fa68fa81c9..63df74f1e31a 100644 --- a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java @@ -31,7 +31,6 @@ import org.junit.jupiter.params.provider.Arguments; import org.junit.jupiter.params.provider.CsvSource; import org.junit.jupiter.params.provider.MethodSource; -import org.junit.jupiter.params.provider.CsvSource; import org.junit.jupiter.params.provider.ValueSource; import java.time.LocalDateTime; From a900ede209b8c3bdfcb3d5f48436e7a5fe0498b3 Mon Sep 17 00:00:00 2001 From: umi Date: Thu, 29 Oct 2026 17:58:44 +0800 Subject: [PATCH 6/6] fix --- .../java/org/apache/paimon/io/DeleteRowFieldsProjectorTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/paimon-core/src/test/java/org/apache/paimon/io/DeleteRowFieldsProjectorTest.java b/paimon-core/src/test/java/org/apache/paimon/io/DeleteRowFieldsProjectorTest.java index 7abcd985128d..29cf951eb6bb 100644 --- a/paimon-core/src/test/java/org/apache/paimon/io/DeleteRowFieldsProjectorTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/io/DeleteRowFieldsProjectorTest.java @@ -102,7 +102,7 @@ void testProjectionDoesNotChangeInput() { null, null, null, - "payload"); + "payloads"); KeyValue input = new KeyValue().replace(GenericRow.of(1), 2L, RowKind.DELETE, value); Options options = new Options();