[core] Add delete-row.thin-mode for primary-key tables

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:

docs/generated/core_configuration.html

@@ -650,6 +650,12 @@
             <td>Boolean</td>
             <td>Enable data file thin mode to avoid duplicate columns storage.</td>
         </tr>
+        <tr>
+            <td><h5>delete-row.thin-mode</h5></td>
+            <td style="word-wrap: break-word;">false</td>
+            <td>Boolean</td>
+            <td>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.</td>
+        </tr>
         <tr>
             <td><h5>delete.force-produce-changelog</h5></td>
             <td style="word-wrap: break-word;">false</td>

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<Boolean> 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<Duration> 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<Long> commitLastSafeSnapshot() {
         return options.getOptional(COMMIT_LAST_SAFE_SNAPSHOT);
     }

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<String> 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());
+    }
+}

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<Integer, Integer> 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) {

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<String> managedBlobFields) {
+            Set<String> 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) {

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<String> managedBlobFields) {
+            Set<String> managedBlobFields,
+            @Nullable DeleteRowFieldsProjector deleteRowFieldsProjector) {
         super(
                 fileIO,
                 context,
@@ -64,7 +67,8 @@ public KeyValueDataFileWriterImpl(
                 fileSource,
                 fileIndexOptions,
                 isExternalPath,
-                managedBlobFields);
+                managedBlobFields,
+                deleteRowFieldsProjector);
     }
 
     @Override