[core] support decouple the delta files lifecycle

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

@@ -28,6 +28,7 @@
 import org.apache.paimon.manifest.ManifestList;
 import org.apache.paimon.metastore.AddPartitionTagCallback;
 import org.apache.paimon.metastore.MetastoreClient;
+import org.apache.paimon.operation.ChangelogDeletion;
 import org.apache.paimon.operation.FileStoreCommitImpl;
 import org.apache.paimon.operation.PartitionExpire;
 import org.apache.paimon.operation.SnapshotDeletion;
@@ -201,8 +202,21 @@ public FileStoreCommitImpl newCommit(String commitUser, String branchName) {
     }
 
     @Override
-    public SnapshotDeletion newSnapshotDeletion() {
+    public SnapshotDeletion newSnapshotDeletion(CoreOptions options) {
         return new SnapshotDeletion(
+                fileIO,
+                pathFactory(),
+                manifestFileFactory().create(),
+                manifestListFactory().create(),
+                newIndexFileHandler(),
+                newStatsFileHandler(),
+                options.changelogLifecycleDecoupled(),
+                options.changelogProducer() != CoreOptions.ChangelogProducer.NONE);
+    }
+
+    @Override
+    public ChangelogDeletion newChangelogDeletion(CoreOptions options) {
+        return new ChangelogDeletion(
                 fileIO,
                 pathFactory(),
                 manifestFileFactory().create(),

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

@@ -22,6 +22,7 @@
 import org.apache.paimon.manifest.ManifestCacheFilter;
 import org.apache.paimon.manifest.ManifestFile;
 import org.apache.paimon.manifest.ManifestList;
+import org.apache.paimon.operation.ChangelogDeletion;
 import org.apache.paimon.operation.FileStoreCommit;
 import org.apache.paimon.operation.FileStoreScan;
 import org.apache.paimon.operation.FileStoreWrite;
@@ -83,7 +84,9 @@ public interface FileStore<T> extends Serializable {
 
     FileStoreCommit newCommit(String commitUser, String branchName);
 
-    SnapshotDeletion newSnapshotDeletion();
+    SnapshotDeletion newSnapshotDeletion(CoreOptions options);
+
+    ChangelogDeletion newChangelogDeletion(CoreOptions options);
 
     TagManager newTagManager();
 

paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyWriter.java

@@ -251,7 +251,8 @@ private RowDataRollingFileWriter createRollingRowWriter() {
                 seqNumCounter,
                 fileCompression,
                 statsCollectors,
-                fileIndexOptions);
+                fileIndexOptions,
+                false);
     }
 
     private void trySyncLatestCompaction(boolean blocking)

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

@@ -23,6 +23,7 @@
 import org.apache.paimon.data.BinaryRow;
 import org.apache.paimon.data.Timestamp;
 import org.apache.paimon.fs.Path;
+import org.apache.paimon.manifest.FileSource;
 import org.apache.paimon.stats.BinaryTableStats;
 import org.apache.paimon.stats.FieldStatsArraySerializer;
 import org.apache.paimon.types.ArrayType;
@@ -31,12 +32,14 @@
 import org.apache.paimon.types.DataTypes;
 import org.apache.paimon.types.IntType;
 import org.apache.paimon.types.RowType;
+import org.apache.paimon.types.TinyIntType;
 
 import javax.annotation.Nullable;
 
 import java.time.LocalDateTime;
 import java.time.ZoneId;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
 import java.util.Objects;
@@ -85,6 +88,7 @@ public class DataFileMeta {
 
     // file index filter bytes, if it is small, store in data file meta
     private final @Nullable byte[] embeddedIndex;
+    private final @Nullable FileSource fileSource;
 
     public static DataFileMeta forAppend(
             String fileName,
@@ -93,7 +97,8 @@ public static DataFileMeta forAppend(
             BinaryTableStats rowStats,
             long minSequenceNumber,
             long maxSequenceNumber,
-            long schemaId) {
+            long schemaId,
+            FileSource fileSource) {
         return forAppend(
                 fileName,
                 fileSize,
@@ -103,7 +108,8 @@ public static DataFileMeta forAppend(
                 maxSequenceNumber,
                 schemaId,
                 Collections.emptyList(),
-                null);
+                null,
+                fileSource);
     }
 
     public static DataFileMeta forAppend(
@@ -115,7 +121,8 @@ public static DataFileMeta forAppend(
             long maxSequenceNumber,
             long schemaId,
             List<String> extraFiles,
-            @Nullable byte[] embeddedIndex) {
+            @Nullable byte[] embeddedIndex,
+            @Nullable FileSource fileSource) {
         return new DataFileMeta(
                 fileName,
                 fileSize,
@@ -131,7 +138,8 @@ public static DataFileMeta forAppend(
                 extraFiles,
                 Timestamp.fromLocalDateTime(LocalDateTime.now()).toMillisTimestamp(),
                 0L,
-                embeddedIndex);
+                embeddedIndex,
+                fileSource);
     }
 
     public DataFileMeta(
@@ -147,7 +155,8 @@ public DataFileMeta(
             long schemaId,
             int level,
             @Nullable Long deleteRowCount,
-            @Nullable byte[] embeddedIndex) {
+            @Nullable byte[] embeddedIndex,
+            @Nullable FileSource fileSource) {
         this(
                 fileName,
                 fileSize,
@@ -163,7 +172,8 @@ public DataFileMeta(
                 Collections.emptyList(),
                 Timestamp.fromLocalDateTime(LocalDateTime.now()).toMillisTimestamp(),
                 deleteRowCount,
-                embeddedIndex);
+                embeddedIndex,
+                fileSource);
     }
 
     public DataFileMeta(
@@ -181,7 +191,8 @@ public DataFileMeta(
             List<String> extraFiles,
             Timestamp creationTime,
             @Nullable Long deleteRowCount,
-            @Nullable byte[] embeddedIndex) {
+            @Nullable byte[] embeddedIndex,
+            @Nullable FileSource fileSource) {
         this.fileName = fileName;
         this.fileSize = fileSize;
 
@@ -201,6 +212,7 @@ public DataFileMeta(
         this.creationTime = creationTime;
 
         this.deleteRowCount = deleteRowCount;
+        this.fileSource = fileSource;
     }
 
     public String fileName() {
@@ -223,6 +235,10 @@ public Optional<Long> deleteRowCount() {
         return Optional.ofNullable(deleteRowCount);
     }
 
+    public Optional<FileSource> fileSource() {
+        return Optional.ofNullable(fileSource);
+    }
+
     public byte[] embeddedIndex() {
         return embeddedIndex;
     }
@@ -313,7 +329,8 @@ public DataFileMeta upgrade(int newLevel) {
                 extraFiles,
                 creationTime,
                 deleteRowCount,
-                embeddedIndex);
+                embeddedIndex,
+                fileSource);
     }
 
     public List<Path> collectFiles(DataFilePathFactory pathFactory) {
@@ -339,7 +356,8 @@ public DataFileMeta copy(List<String> newExtraFiles) {
                 newExtraFiles,
                 creationTime,
                 deleteRowCount,
-                embeddedIndex);
+                embeddedIndex,
+                fileSource);
     }
 
     @Override
@@ -365,7 +383,8 @@ public boolean equals(Object o) {
                 && level == that.level
                 && Objects.equals(extraFiles, that.extraFiles)
                 && Objects.equals(creationTime, that.creationTime)
-                && Objects.equals(deleteRowCount, that.deleteRowCount);
+                && Objects.equals(deleteRowCount, that.deleteRowCount)
+                && Objects.equals(fileSource, that.fileSource);
     }
 
     @Override
@@ -374,7 +393,7 @@ public int hashCode() {
                 fileName,
                 fileSize,
                 rowCount,
-                embeddedIndex,
+                Arrays.hashCode(embeddedIndex),
                 minKey,
                 maxKey,
                 keyStats,
@@ -385,7 +404,8 @@ public int hashCode() {
                 level,
                 extraFiles,
                 creationTime,
-                deleteRowCount);
+                deleteRowCount,
+                fileSource);
     }
 
     @Override
@@ -394,11 +414,10 @@ public String toString() {
                 "{fileName: %s, fileSize: %d, rowCount: %d, "
                         + "minKey: %s, maxKey: %s, keyStats: %s, valueStats: %s, "
                         + "minSequenceNumber: %d, maxSequenceNumber: %d, "
-                        + "schemaId: %d, level: %d, extraFiles: %s, creationTime: %s, deleteRowCount: %d}",
+                        + "schemaId: %d, level: %d, extraFiles: %s, creationTime: %s, deleteRowCount: %d, fileSource: %s}",
                 fileName,
                 fileSize,
                 rowCount,
-                embeddedIndex,
                 minKey,
                 maxKey,
                 keyStats,
@@ -409,7 +428,8 @@ public String toString() {
                 level,
                 extraFiles,
                 creationTime,
-                deleteRowCount);
+                deleteRowCount,
+                fileSource);
     }
 
     public static RowType schema() {
@@ -429,6 +449,7 @@ public static RowType schema() {
         fields.add(new DataField(12, "_CREATION_TIME", DataTypes.TIMESTAMP_MILLIS()));
         fields.add(new DataField(13, "_DELETE_ROW_COUNT", new BigIntType(true)));
         fields.add(new DataField(14, "_EMBEDDED_FILE_INDEX", newBytesType(true)));
+        fields.add(new DataField(15, "_FILE_SOURCE", new TinyIntType(true)));
         return new RowType(fields);
     }
 

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

@@ -21,6 +21,7 @@
 import org.apache.paimon.data.BinaryString;
 import org.apache.paimon.data.GenericRow;
 import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.manifest.FileSource;
 import org.apache.paimon.stats.BinaryTableStats;
 import org.apache.paimon.utils.ObjectSerializer;
 
@@ -55,7 +56,8 @@ public InternalRow toRow(DataFileMeta meta) {
                 toStringArrayData(meta.extraFiles()),
                 meta.creationTime(),
                 meta.deleteRowCount().orElse(null),
-                meta.embeddedIndex());
+                meta.embeddedIndex(),
+                meta.fileSource().map(FileSource::toByteValue).orElse(null));
     }
 
     @Override
@@ -75,6 +77,7 @@ public DataFileMeta fromRow(InternalRow row) {
                 fromStringArrayData(row.getArray(11)),
                 row.getTimestamp(12, 3),
                 row.isNullAt(13) ? null : row.getLong(13),
-                row.isNullAt(14) ? null : row.getBinary(14));
+                row.isNullAt(14) ? null : row.getBinary(14),
+                row.isNullAt(15) ? null : FileSource.fromByteValue(row.getByte(15)));
     }
 }

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

@@ -28,6 +28,7 @@
 import org.apache.paimon.format.TableStatsExtractor;
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.fs.Path;
+import org.apache.paimon.manifest.FileSource;
 import org.apache.paimon.stats.BinaryTableStats;
 import org.apache.paimon.stats.FieldStatsArraySerializer;
 import org.apache.paimon.types.RowType;
@@ -68,6 +69,7 @@ public class KeyValueDataFileWriter
     private long minSeqNumber = Long.MAX_VALUE;
     private long maxSeqNumber = Long.MIN_VALUE;
     private long deleteRecordCount = 0;
+    private final boolean isCompact;
 
     public KeyValueDataFileWriter(
             FileIO fileIO,
@@ -80,7 +82,8 @@ public KeyValueDataFileWriter(
             long schemaId,
             int level,
             String compression,
-            CoreOptions options) {
+            CoreOptions options,
+            boolean isCompact) {
         super(
                 fileIO,
                 factory,
@@ -100,6 +103,7 @@ public KeyValueDataFileWriter(
         this.keyStatsConverter = new FieldStatsArraySerializer(keyType);
         this.valueStatsConverter = new FieldStatsArraySerializer(valueType);
         this.keySerializer = new InternalRowSerializer(keyType);
+        this.isCompact = isCompact;
     }
 
     @Override
@@ -170,6 +174,7 @@ public DataFileMeta result() throws IOException {
                 level,
                 deleteRecordCount,
                 // TODO: enable file filter for primary key table (e.g. deletion table).
-                null);
+                null,
+                isCompact ? FileSource.COMPACT : FileSource.APPEND);
     }
 }

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

@@ -81,21 +81,24 @@ public DataFilePathFactory pathFactory(int level) {
         return formatContext.pathFactory(level);
     }
 
-    public RollingFileWriter<KeyValue, DataFileMeta> createRollingMergeTreeFileWriter(int level) {
+    public RollingFileWriter<KeyValue, DataFileMeta> createRollingMergeTreeFileWriter(
+            int level, boolean isCompact) {
         return new RollingFileWriter<>(
-                () -> createDataFileWriter(formatContext.pathFactory(level).newPath(), level),
+                () ->
+                        createDataFileWriter(
+                                formatContext.pathFactory(level).newPath(), level, isCompact),
                 suggestedFileSize);
     }
 
     public RollingFileWriter<KeyValue, DataFileMeta> createRollingChangelogFileWriter(int level) {
         return new RollingFileWriter<>(
                 () ->
                         createDataFileWriter(
-                                formatContext.pathFactory(level).newChangelogPath(), level),
+                                formatContext.pathFactory(level).newChangelogPath(), level, false),
                 suggestedFileSize);
     }
 
-    private KeyValueDataFileWriter createDataFileWriter(Path path, int level) {
+    private KeyValueDataFileWriter createDataFileWriter(Path path, int level, boolean isCompact) {
         KeyValueSerializer kvSerializer = new KeyValueSerializer(keyType, valueType);
         return new KeyValueDataFileWriter(
                 fileIO,
@@ -108,7 +111,8 @@ private KeyValueDataFileWriter createDataFileWriter(Path path, int level) {
                 schemaId,
                 level,
                 formatContext.compression(level),
-                options);
+                options,
+                isCompact);
     }
 
     public void deleteFile(String filename, int level) {