[core] Introduce data-file.thin-mode in primary key table write

docs/layouts/shortcodes/generated/core_configuration.html

@@ -864,6 +864,12 @@
             <td>Integer</td>
             <td>Default spill compression zstd level. For higher compression rates, it can be configured to 9, but the read and write speed will significantly decrease.</td>
         </tr>
+        <tr>
+            <td><h5>data-file.thin-mode</h5></td>
+            <td style="word-wrap: break-word;">false</td>
+            <td>Boolean</td>
+            <td>Enable data file thin mode to avoid duplicate columns storage.</td>
+        </tr>
         <tr>
             <td><h5>streaming-read-mode</h5></td>
             <td style="word-wrap: break-word;">(none)</td>

paimon-common/src/main/java/org/apache/paimon/CoreOptions.java

@@ -1452,6 +1452,13 @@ public class CoreOptions implements Serializable {
                             "For DELETE manifest entry in manifest file, drop stats to reduce memory and storage."
                                     + " Default value is false only for compatibility of old reader.");
 
+    public static final ConfigOption<Boolean> DATA_FILE_THIN_MODE =
+            key("data-file.thin-mode")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            "Enable data file thin mode to avoid duplicate columns storage.");
+
     @ExcludeFromDocumentation("Only used internally to support materialized table")
     public static final ConfigOption<String> MATERIALIZED_TABLE_DEFINITION_QUERY =
             key("materialized-table.definition-query")
@@ -2356,6 +2363,10 @@ public boolean statsDenseStore() {
         return options.get(METADATA_STATS_DENSE_STORE);
     }
 
+    public boolean dataFileThinMode() {
+        return options.get(DATA_FILE_THIN_MODE);
+    }
+
     /** Specifies the merge engine for table with primary key. */
     public enum MergeEngine implements DescribedEnum {
         DEDUPLICATE("deduplicate", "De-duplicate and keep the last row."),

paimon-core/src/main/java/org/apache/paimon/KeyValueThinSerializer.java

@@ -0,0 +1,58 @@
+/*
+ * 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;
+
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.JoinedRow;
+import org.apache.paimon.types.RowKind;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.ObjectSerializer;
+
+/** Serialize KeyValue to InternalRow with ignorance of key. Only used to write KeyValue to disk. */
+public class KeyValueThinSerializer extends ObjectSerializer<KeyValue> {
+
+    private static final long serialVersionUID = 1L;
+
+    private final GenericRow reusedMeta;
+    private final JoinedRow reusedKeyWithMeta;
+
+    public KeyValueThinSerializer(RowType keyType, RowType valueType) {
+        super(KeyValue.schema(keyType, valueType));
+
+        this.reusedMeta = new GenericRow(2);
+        this.reusedKeyWithMeta = new JoinedRow();
+    }
+
+    public InternalRow toRow(KeyValue record) {
+        return toRow(record.sequenceNumber(), record.valueKind(), record.value());
+    }
+
+    public InternalRow toRow(long sequenceNumber, RowKind valueKind, InternalRow value) {
+        reusedMeta.setField(0, sequenceNumber);
+        reusedMeta.setField(1, valueKind.toByteValue());
+        return reusedKeyWithMeta.replace(reusedMeta, value);
+    }
+
+    @Override
+    public KeyValue fromRow(InternalRow row) {
+        throw new UnsupportedOperationException(
+                "KeyValue cannot be deserialized from InternalRow by this serializer.");
+    }
+}

paimon-core/src/main/java/org/apache/paimon/io/KeyValueDataFileWriter.java

@@ -42,7 +42,6 @@
 import javax.annotation.Nullable;
 
 import java.io.IOException;
-import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
 import java.util.function.Function;
@@ -56,13 +55,13 @@
  * <p>NOTE: records given to the writer must be sorted because it does not compare the min max keys
  * to produce {@link DataFileMeta}.
  */
-public class KeyValueDataFileWriter
+public abstract class KeyValueDataFileWriter
         extends StatsCollectingSingleFileWriter<KeyValue, DataFileMeta> {
 
     private static final Logger LOG = LoggerFactory.getLogger(KeyValueDataFileWriter.class);
 
-    private final RowType keyType;
-    private final RowType valueType;
+    protected final RowType keyType;
+    protected final RowType valueType;
     private final long schemaId;
     private final int level;
 
@@ -85,6 +84,7 @@ public KeyValueDataFileWriter(
             Function<KeyValue, InternalRow> converter,
             RowType keyType,
             RowType valueType,
+            RowType writeRowType,
             @Nullable SimpleStatsExtractor simpleStatsExtractor,
             long schemaId,
             int level,
@@ -97,11 +97,11 @@ public KeyValueDataFileWriter(
                 factory,
                 path,
                 converter,
-                KeyValue.schema(keyType, valueType),
+                writeRowType,
                 simpleStatsExtractor,
                 compression,
                 StatsCollectorFactories.createStatsFactories(
-                        options, KeyValue.schema(keyType, valueType).getFieldNames()),
+                        options, writeRowType.getFieldNames(), keyType.getFieldNames()),
                 options.asyncFileWrite());
 
         this.keyType = keyType;
@@ -166,17 +166,11 @@ public DataFileMeta result() throws IOException {
             return null;
         }
 
-        SimpleColStats[] rowStats = fieldStats();
-        int numKeyFields = keyType.getFieldCount();
-
-        SimpleColStats[] keyFieldStats = Arrays.copyOfRange(rowStats, 0, numKeyFields);
-        SimpleStats keyStats = keyStatsConverter.toBinaryAllMode(keyFieldStats);
-
-        SimpleColStats[] valFieldStats =
-                Arrays.copyOfRange(rowStats, numKeyFields + 2, rowStats.length);
+        Pair<SimpleColStats[], SimpleColStats[]> keyValueStats = fetchKeyValueStats(fieldStats());
 
+        SimpleStats keyStats = keyStatsConverter.toBinaryAllMode(keyValueStats.getKey());
         Pair<List<String>, SimpleStats> valueStatsPair =
-                valueStatsConverter.toBinary(valFieldStats);
+                valueStatsConverter.toBinary(keyValueStats.getValue());
 
         DataFileIndexWriter.FileIndexResult indexResult =
                 dataFileIndexWriter == null
@@ -204,6 +198,8 @@ public DataFileMeta result() throws IOException {
                 valueStatsPair.getKey());
     }
 
+    abstract Pair<SimpleColStats[], SimpleColStats[]> fetchKeyValueStats(SimpleColStats[] rowStats);
+
     @Override
     public void close() throws IOException {
         if (dataFileIndexWriter != null) {

paimon-core/src/main/java/org/apache/paimon/io/KeyValueDataFileWriterImpl.java

@@ -0,0 +1,80 @@
+/*
+ * 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.InternalRow;
+import org.apache.paimon.fileindex.FileIndexOptions;
+import org.apache.paimon.format.FormatWriterFactory;
+import org.apache.paimon.format.SimpleColStats;
+import org.apache.paimon.format.SimpleStatsExtractor;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.manifest.FileSource;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.Pair;
+
+import javax.annotation.Nullable;
+
+import java.util.Arrays;
+import java.util.function.Function;
+
+/** Write data files containing {@link KeyValue}s. */
+public class KeyValueDataFileWriterImpl extends KeyValueDataFileWriter {
+
+    public KeyValueDataFileWriterImpl(
+            FileIO fileIO,
+            FormatWriterFactory factory,
+            Path path,
+            Function<KeyValue, InternalRow> converter,
+            RowType keyType,
+            RowType valueType,
+            @Nullable SimpleStatsExtractor simpleStatsExtractor,
+            long schemaId,
+            int level,
+            String compression,
+            CoreOptions options,
+            FileSource fileSource,
+            FileIndexOptions fileIndexOptions) {
+        super(
+                fileIO,
+                factory,
+                path,
+                converter,
+                keyType,
+                valueType,
+                KeyValue.schema(keyType, valueType),
+                simpleStatsExtractor,
+                schemaId,
+                level,
+                compression,
+                options,
+                fileSource,
+                fileIndexOptions);
+    }
+
+    @Override
+    Pair<SimpleColStats[], SimpleColStats[]> fetchKeyValueStats(SimpleColStats[] rowStats) {
+        int numKeyFields = keyType.getFieldCount();
+        return Pair.of(
+                Arrays.copyOfRange(rowStats, 0, numKeyFields),
+                Arrays.copyOfRange(rowStats, numKeyFields + 2, rowStats.length));
+    }
+}