apache · JingsongLi · May 22, 2024 · May 21, 2024 · May 21, 2024 · May 22, 2024
diff --git a/paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyWriter.java b/paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyWriter.java
@@ -32,6 +32,7 @@
 import org.apache.paimon.io.DataFilePathFactory;
 import org.apache.paimon.io.DataIncrement;
 import org.apache.paimon.io.RowDataRollingFileWriter;
+import org.apache.paimon.manifest.FileSource;
 import org.apache.paimon.memory.MemoryOwner;
 import org.apache.paimon.memory.MemorySegmentPool;
 import org.apache.paimon.operation.AppendOnlyFileStoreWrite;
@@ -256,7 +257,8 @@ private RowDataRollingFileWriter createRollingRowWriter() {
                 seqNumCounter,
                 fileCompression,
                 statsCollectors,
-                fileIndexOptions);
+                fileIndexOptions,
+                FileSource.APPEND);
     }
 
     private void trySyncLatestCompaction(boolean blocking)

diff --git a/paimon-core/src/main/java/org/apache/paimon/io/DataFileMeta.java b/paimon-core/src/main/java/org/apache/paimon/io/DataFileMeta.java
@@ -22,6 +22,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.SimpleStats;
 import org.apache.paimon.stats.SimpleStatsConverter;
 import org.apache.paimon.types.ArrayType;
@@ -30,6 +31,7 @@
 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;
 
@@ -83,14 +85,17 @@ 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,
             long fileSize,
             long rowCount,
             SimpleStats rowStats,
             long minSequenceNumber,
             long maxSequenceNumber,
-            long schemaId) {
+            long schemaId,
+            @Nullable FileSource fileSource) {
         return forAppend(
                 fileName,
                 fileSize,
@@ -100,7 +105,8 @@ public static DataFileMeta forAppend(
                 maxSequenceNumber,
                 schemaId,
                 Collections.emptyList(),
-                null);
+                null,
+                fileSource);
     }
 
     public static DataFileMeta forAppend(
@@ -112,7 +118,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,
@@ -128,7 +135,8 @@ public static DataFileMeta forAppend(
                 extraFiles,
                 Timestamp.fromLocalDateTime(LocalDateTime.now()).toMillisTimestamp(),
                 0L,
-                embeddedIndex);
+                embeddedIndex,
+                fileSource);
     }
 
     public DataFileMeta(
@@ -144,7 +152,8 @@ public DataFileMeta(
             long schemaId,
             int level,
             @Nullable Long deleteRowCount,
-            @Nullable byte[] embeddedIndex) {
+            @Nullable byte[] embeddedIndex,
+            @Nullable FileSource fileSource) {
         this(
                 fileName,
                 fileSize,
@@ -160,7 +169,8 @@ public DataFileMeta(
                 Collections.emptyList(),
                 Timestamp.fromLocalDateTime(LocalDateTime.now()).toMillisTimestamp(),
                 deleteRowCount,
-                embeddedIndex);
+                embeddedIndex,
+                fileSource);
     }
 
     public DataFileMeta(
@@ -178,7 +188,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;
 
@@ -198,6 +209,7 @@ public DataFileMeta(
         this.creationTime = creationTime;
 
         this.deleteRowCount = deleteRowCount;
+        this.fileSource = fileSource;
     }
 
     public String fileName() {
@@ -291,6 +303,10 @@ public String fileFormat() {
         return split[split.length - 1];
     }
 
+    public Optional<FileSource> fileSource() {
+        return Optional.ofNullable(fileSource);
+    }
+
     public DataFileMeta upgrade(int newLevel) {
         checkArgument(newLevel > this.level);
         return new DataFileMeta(
@@ -308,7 +324,8 @@ public DataFileMeta upgrade(int newLevel) {
                 extraFiles,
                 creationTime,
                 deleteRowCount,
-                embeddedIndex);
+                embeddedIndex,
+                fileSource);
     }
 
     public List<Path> collectFiles(DataFilePathFactory pathFactory) {
@@ -334,7 +351,8 @@ public DataFileMeta copy(List<String> newExtraFiles) {
                 newExtraFiles,
                 creationTime,
                 deleteRowCount,
-                embeddedIndex);
+                embeddedIndex,
+                fileSource);
     }
 
     @Override
@@ -360,7 +378,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
@@ -380,7 +399,8 @@ public int hashCode() {
                 level,
                 extraFiles,
                 creationTime,
-                deleteRowCount);
+                deleteRowCount,
+                fileSource);
     }
 
     @Override
@@ -389,7 +409,8 @@ public String toString() {
                 "{fileName: %s, fileSize: %d, rowCount: %d, embeddedIndex: %s, "
                         + "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,
@@ -404,7 +425,8 @@ public String toString() {
                 level,
                 extraFiles,
                 creationTime,
-                deleteRowCount);
+                deleteRowCount,
+                fileSource);
     }
 
     public static RowType schema() {
@@ -424,6 +446,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);
     }
 

diff --git a/paimon-core/src/main/java/org/apache/paimon/io/DataFileMetaSerializer.java b/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.SimpleStats;
 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)));
     }
 }
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
@@ -28,6 +28,7 @@
 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.stats.SimpleStats;
 import org.apache.paimon.stats.SimpleStatsConverter;
 import org.apache.paimon.types.RowType;
@@ -62,6 +63,7 @@ public class KeyValueDataFileWriter
     private final SimpleStatsConverter keyStatsConverter;
     private final SimpleStatsConverter valueStatsConverter;
     private final InternalRowSerializer keySerializer;
+    private final FileSource fileSource;
 
     private BinaryRow minKey = null;
     private InternalRow maxKey = null;
@@ -80,7 +82,8 @@ public KeyValueDataFileWriter(
             long schemaId,
             int level,
             String compression,
-            CoreOptions options) {
+            CoreOptions options,
+            FileSource fileSource) {
         super(
                 fileIO,
                 factory,
@@ -100,6 +103,7 @@ public KeyValueDataFileWriter(
         this.keyStatsConverter = new SimpleStatsConverter(keyType);
         this.valueStatsConverter = new SimpleStatsConverter(valueType);
         this.keySerializer = new InternalRowSerializer(keyType);
+        this.fileSource = fileSource;
     }
 
     @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,
+                fileSource);
     }
 }
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
@@ -28,6 +28,7 @@
 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.statistics.SimpleColStatsCollector;
 import org.apache.paimon.types.RowType;
 import org.apache.paimon.utils.FileStorePathFactory;
@@ -81,21 +82,27 @@ public DataFilePathFactory pathFactory(int level) {
         return formatContext.pathFactory(level);
     }
 
-    public RollingFileWriter<KeyValue, DataFileMeta> createRollingMergeTreeFileWriter(int level) {
+    public RollingFileWriter<KeyValue, DataFileMeta> createRollingMergeTreeFileWriter(
+            int level, FileSource fileSource) {
         return new RollingFileWriter<>(
-                () -> createDataFileWriter(formatContext.pathFactory(level).newPath(), level),
+                () ->
+                        createDataFileWriter(
+                                formatContext.pathFactory(level).newPath(), level, fileSource),
                 suggestedFileSize);
     }
 
     public RollingFileWriter<KeyValue, DataFileMeta> createRollingChangelogFileWriter(int level) {
         return new RollingFileWriter<>(
                 () ->
                         createDataFileWriter(
-                                formatContext.pathFactory(level).newChangelogPath(), level),
+                                formatContext.pathFactory(level).newChangelogPath(),
+                                level,
+                                FileSource.APPEND),
                 suggestedFileSize);
     }
 
-    private KeyValueDataFileWriter createDataFileWriter(Path path, int level) {
+    private KeyValueDataFileWriter createDataFileWriter(
+            Path path, int level, FileSource fileSource) {
         KeyValueSerializer kvSerializer = new KeyValueSerializer(keyType, valueType);
         return new KeyValueDataFileWriter(
                 fileIO,
@@ -108,7 +115,8 @@ private KeyValueDataFileWriter createDataFileWriter(Path path, int level) {
                 schemaId,
                 level,
                 formatContext.compression(level),
-                options);
+                options,
+                fileSource);
     }
 
     public void deleteFile(String filename, int level) {

diff --git a/paimon-core/src/main/java/org/apache/paimon/io/RowDataFileWriter.java b/paimon-core/src/main/java/org/apache/paimon/io/RowDataFileWriter.java
@@ -24,6 +24,7 @@
 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.statistics.SimpleColStatsCollector;
 import org.apache.paimon.stats.SimpleStats;
 import org.apache.paimon.stats.SimpleStatsConverter;
@@ -48,6 +49,7 @@ public class RowDataFileWriter extends StatsCollectingSingleFileWriter<InternalR
     private final LongCounter seqNumCounter;
     private final SimpleStatsConverter statsArraySerializer;
     @Nullable private final FileIndexWriter fileIndexWriter;
+    private final FileSource fileSource;
 
     public RowDataFileWriter(
             FileIO fileIO,
@@ -59,7 +61,8 @@ public RowDataFileWriter(
             LongCounter seqNumCounter,
             String fileCompression,
             SimpleColStatsCollector.Factory[] statsCollectors,
-            FileIndexOptions fileIndexOptions) {
+            FileIndexOptions fileIndexOptions,
+            FileSource fileSource) {
         super(
                 fileIO,
                 factory,
@@ -75,6 +78,7 @@ public RowDataFileWriter(
         this.fileIndexWriter =
                 FileIndexWriter.create(
                         fileIO, toFileIndexPath(path), writeSchema, fileIndexOptions);
+        this.fileSource = fileSource;
     }
 
     @Override
@@ -111,6 +115,7 @@ public DataFileMeta result() throws IOException {
                 indexResult.independentIndexFile() == null
                         ? Collections.emptyList()
                         : Collections.singletonList(indexResult.independentIndexFile()),
-                indexResult.embeddedIndexBytes());
+                indexResult.embeddedIndexBytes(),
+                fileSource);
     }
 }
diff --git a/paimon-core/src/main/java/org/apache/paimon/io/RowDataRollingFileWriter.java b/paimon-core/src/main/java/org/apache/paimon/io/RowDataRollingFileWriter.java
@@ -23,6 +23,7 @@
 import org.apache.paimon.format.FileFormat;
 import org.apache.paimon.format.avro.AvroFileFormat;
 import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.manifest.FileSource;
 import org.apache.paimon.statistics.SimpleColStatsCollector;
 import org.apache.paimon.types.RowType;
 import org.apache.paimon.utils.LongCounter;
@@ -40,7 +41,8 @@ public RowDataRollingFileWriter(
             LongCounter seqNumCounter,
             String fileCompression,
             SimpleColStatsCollector.Factory[] statsCollectors,
-            FileIndexOptions fileIndexOptions) {
+            FileIndexOptions fileIndexOptions,
+            FileSource fileSource) {
         super(
                 () ->
                         new RowDataFileWriter(
@@ -57,7 +59,8 @@ public RowDataRollingFileWriter(
                                 seqNumCounter,
                                 fileCompression,
                                 statsCollectors,
-                                fileIndexOptions),
+                                fileIndexOptions,
+                                fileSource),
                 targetFileSize);
     }
 }