diff --git a/docs/docs/concepts/spec/datafile.md b/docs/docs/concepts/spec/datafile.md
index 25408c53d1b1..0afd8f3b93af 100644
--- a/docs/docs/concepts/spec/datafile.md
+++ b/docs/docs/concepts/spec/datafile.md
@@ -137,6 +137,20 @@ contains five columns:
_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. 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
+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
An ordinary append table stores its declared value columns without the primary-key merge fields:
diff --git a/docs/generated/core_configuration.html b/docs/generated/core_configuration.html
index 0fc8c42fd261..eda7d42dba64 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. 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 |
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 c3202188dc09..7349439fe012 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,18 @@ 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. "
+ + "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")
.durationType()
@@ -4781,10 +4793,18 @@ 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);
}
+ 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..a331f2435e37
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/io/DeleteRowFieldsProjector.java
@@ -0,0 +1,78 @@
+/*
+ * 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 (!options.deleteRowThinMode()) {
+ return null;
+ }
+
+ Set requiredFields = new HashSet<>(schema.primaryKeys());
+ requiredFields.addAll(options.sequenceField());
+ if (options.recordLevelTimeField() != null) {
+ requiredFields.add(options.recordLevelTimeField());
+ }
+ requiredFields.addAll(options.clusteringColumns());
+ 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;
+ }
+
+ 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/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
new file mode 100644
index 000000000000..29cf951eb6bb
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/io/DeleteRowFieldsProjectorTest.java
@@ -0,0 +1,146 @@
+/*
+ * 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 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");
+ options.set(CoreOptions.PK_VECTOR_INDEX_COLUMNS, "vector_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");
+ options.set(CoreOptions.PK_FM_INDEX_COLUMNS, "fm_index");
+
+ assertThat(
+ DeleteRowFieldsProjector.createProjection(
+ tableSchema(), new CoreOptions(options)))
+ .containsExactly(0, 1, 2, 3, 4, -1, -1, -1, -1, -1, -1, -1);
+ }
+
+ @Test
+ void testNoFieldsToProject() {
+ Options options = new Options();
+ options.set(CoreOptions.DELETE_ROW_THIN_MODE, true);
+ 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
+ void testProjectionDoesNotChangeInput() {
+ GenericRow value =
+ GenericRow.of(
+ 1,
+ "required",
+ 2L,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ "payloads");
+ KeyValue input = new KeyValue().replace(GenericRow.of(1), 2L, RowKind.DELETE, value);
+
+ 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);
+ 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..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
@@ -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,110 @@ 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.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);
+ 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.DELETE_ROW_THIN_MODE, true);
+ 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/schema/SchemaValidationTest.java b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
index 54a7088950bc..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
@@ -28,7 +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.ValueSource;
import java.time.LocalDateTime;
@@ -37,6 +39,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 +58,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
new file mode 100644
index 000000000000..9ec3aa0259c0
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/table/DeleteRowFieldsTest.java
@@ -0,0 +1,210 @@
+/*
+ * 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.catalog.Catalog;
+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.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;
+
+import java.util.ArrayList;
+import java.util.Collections;
+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",
+ "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"));
+
+ 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");
+ if (thinMode) {
+ assertThat(delete.isNullAt(4)).isTrue();
+ } else {
+ assertThat(delete.getString(4).toString()).isEqualTo("old-payload");
+ }
+
+ 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, boolean thinMode) 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 (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")) {
+ 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;
+ }
+}
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..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
@@ -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;
@@ -76,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));
@@ -89,10 +94,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 =