[core] Generate changelog by copying data when records are insert-only

paimon-common/src/main/java/org/apache/paimon/fs/local/LocalFileIO.java

@@ -217,6 +217,19 @@ public boolean rename(Path src, Path dst) throws IOException {
  }
  }
 
+ @Override
+ public boolean copyFileUtf8(Path sourcePath, Path targetPath) throws IOException {
+ if (Files.exists(toPath(targetPath))) {
+ return false;
+ }
+ Files.copy(toPath(sourcePath), toPath(targetPath), StandardCopyOption.COPY_ATTRIBUTES);
+ return true;
+ }
+
+ private java.nio.file.Path toPath(Path path) {
+ return toFile(path).toPath();
+ }
+
  /**
  * Converts the given Path to a File for this file system. If the path is empty, we will return
  * <tt>new File(".")</tt> instead of <tt>new File("")</tt>, since the latter returns

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

@@ -32,6 +32,9 @@
 import org.apache.paimon.table.BucketMode;
 import org.apache.paimon.table.CatalogEnvironment;
 import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.RecordAttributeManager;
+
+import javax.annotation.Nullable;
 
 import java.util.Comparator;
 import java.util.List;
@@ -87,12 +90,14 @@ public RawFileSplitRead newRead() {
 
  @Override
  public AppendOnlyFileStoreWrite newWrite(String commitUser) {
- return newWrite(commitUser, null);
+ return newWrite(commitUser, null, null);
  }
 
  @Override
  public AppendOnlyFileStoreWrite newWrite(
- String commitUser, ManifestCacheFilter manifestFilter) {
+ String commitUser,
+ ManifestCacheFilter manifestFilter,
+ @Nullable RecordAttributeManager recordAttributeManager) {
  return new AppendOnlyFileStoreWrite(
  fileIO,
  newRead(),

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

@@ -37,6 +37,7 @@
 import org.apache.paimon.tag.TagAutoManager;
 import org.apache.paimon.types.RowType;
 import org.apache.paimon.utils.FileStorePathFactory;
+import org.apache.paimon.utils.RecordAttributeManager;
 import org.apache.paimon.utils.SnapshotManager;
 import org.apache.paimon.utils.TagManager;
 
@@ -76,7 +77,10 @@ public interface FileStore<T> extends Serializable {
 
  FileStoreWrite<T> newWrite(String commitUser);
 
- FileStoreWrite<T> newWrite(String commitUser, ManifestCacheFilter manifestFilter);
+ FileStoreWrite<T> newWrite(
+ String commitUser,
+ ManifestCacheFilter manifestFilter,
+ @Nullable RecordAttributeManager recordAttributeManager);
 
  FileStoreCommit newCommit(String commitUser);
 

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

@@ -42,9 +42,12 @@
 import org.apache.paimon.types.RowType;
 import org.apache.paimon.utils.FileStorePathFactory;
 import org.apache.paimon.utils.KeyComparatorSupplier;
+import org.apache.paimon.utils.RecordAttributeManager;
 import org.apache.paimon.utils.UserDefinedSeqComparator;
 import org.apache.paimon.utils.ValueEqualiserSupplier;
 
+import javax.annotation.Nullable;
+
 import java.util.Comparator;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -152,11 +155,14 @@ public KeyValueFileReaderFactory.Builder newReaderFactoryBuilder() {
 
  @Override
  public KeyValueFileStoreWrite newWrite(String commitUser) {
- return newWrite(commitUser, null);
+ return newWrite(commitUser, null, null);
  }
 
  @Override
- public KeyValueFileStoreWrite newWrite(String commitUser, ManifestCacheFilter manifestFilter) {
+ public KeyValueFileStoreWrite newWrite(
+ String commitUser,
+ ManifestCacheFilter manifestFilter,
+ @Nullable RecordAttributeManager recordAttributeManager) {
  IndexMaintainer.Factory<KeyValue> indexFactory = null;
  if (bucketMode() == BucketMode.HASH_DYNAMIC) {
  indexFactory = new HashIndexMaintainer.Factory(newIndexFileHandler());
@@ -185,7 +191,8 @@ public KeyValueFileStoreWrite newWrite(String commitUser, ManifestCacheFilter ma
  deletionVectorsMaintainerFactory,
  options,
  keyValueFieldsExtractor,
- tableName);
+ tableName,
+ recordAttributeManager);
  }
 
  private Map<String, FileStorePathFactory> format2PathFactory() {

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

@@ -349,6 +349,26 @@ public DataFileMeta upgrade(int newLevel) {
  fileSource);
  }
 
+ public DataFileMeta rename(String newFileName) {
+ return new DataFileMeta(
+ newFileName,
+ fileSize,
+ rowCount,
+ minKey,
+ maxKey,
+ keyStats,
+ valueStats,
+ minSequenceNumber,
+ maxSequenceNumber,
+ schemaId,
+ level,
+ extraFiles,
+ creationTime,
+ deleteRowCount,
+ embeddedIndex,
+ fileSource);
+ }
+
  public List<Path> collectFiles(DataFilePathFactory pathFactory) {
  List<Path> paths = new ArrayList<>();
  paths.add(pathFactory.toPath(fileName));

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

@@ -36,6 +36,7 @@
 
 import javax.annotation.Nullable;
 
+import java.io.IOException;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Optional;
@@ -123,6 +124,21 @@ public void deleteFile(String filename, int level) {
  fileIO.deleteQuietly(formatContext.pathFactory(level).toPath(filename));
  }
 
+ public void copyFile(String sourceFileName, String targetFileName, int level)
+ throws IOException {
+ Path sourcePath = formatContext.pathFactory(level).toPath(sourceFileName);
+ Path targetPath = formatContext.pathFactory(level).toPath(targetFileName);
+ fileIO.copyFileUtf8(sourcePath, targetPath);
+ }
+
+ public FileIO getFileIO() {
+ return fileIO;
+ }
+
+ public Path newChangelogPath(int level) {
+ return formatContext.pathFactory(level).newChangelogPath();
+ }
+
  public static Builder builder(
  FileIO fileIO,
  long schemaId,

paimon-core/src/main/java/org/apache/paimon/mergetree/MergeTreeWriter.java

@@ -26,6 +26,8 @@
 import org.apache.paimon.compact.CompactResult;
 import org.apache.paimon.data.InternalRow;
 import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
 import org.apache.paimon.io.CompactIncrement;
 import org.apache.paimon.io.DataFileMeta;
 import org.apache.paimon.io.DataIncrement;
@@ -37,6 +39,7 @@
 import org.apache.paimon.mergetree.compact.MergeFunction;
 import org.apache.paimon.options.MemorySize;
 import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.BucketRecordAttributeManager;
 import org.apache.paimon.utils.CommitIncrement;
 import org.apache.paimon.utils.FieldsComparator;
 import org.apache.paimon.utils.RecordWriter;
@@ -78,6 +81,9 @@ public class MergeTreeWriter implements RecordWriter<KeyValue>, MemoryOwner {
  private final LinkedHashMap<String, DataFileMeta> compactBefore;
  private final LinkedHashSet<DataFileMeta> compactAfter;
  private final LinkedHashSet<DataFileMeta> compactChangelog;
+ @Nullable private final BucketRecordAttributeManager recordAttributeManager;
+
+ private final boolean canChangelogOptimizedToCopyPreconditions;
 
  @Nullable private CompactDeletionFile compactDeletionFile;
 
@@ -98,7 +104,8 @@ public MergeTreeWriter(
  boolean commitForceCompact,
  ChangelogProducer changelogProducer,
  @Nullable CommitIncrement increment,
- @Nullable FieldsComparator userDefinedSeqComparator) {
+ @Nullable FieldsComparator userDefinedSeqComparator,
+ @Nullable BucketRecordAttributeManager recordAttributeManager) {
  this.writeBufferSpillable = writeBufferSpillable;
  this.maxDiskSize = maxDiskSize;
  this.sortMaxFan = sortMaxFan;
@@ -114,6 +121,7 @@ public MergeTreeWriter(
  this.commitForceCompact = commitForceCompact;
  this.changelogProducer = changelogProducer;
  this.userDefinedSeqComparator = userDefinedSeqComparator;
+ this.recordAttributeManager = recordAttributeManager;
 
  this.newFiles = new LinkedHashSet<>();
  this.deletedFiles = new LinkedHashSet<>();
@@ -133,6 +141,25 @@ public MergeTreeWriter(
  compactChangelog.addAll(increment.compactIncrement().changelogFiles());
  updateCompactDeletionFile(increment.compactDeletionFile());
  }
+
+ // TODO: Verify the performance of the default implementation of copy,
+ // and remove the following code block if proved to be of no regression.
+ boolean isCopyMethodOverridden;
+ try {
+ isCopyMethodOverridden =
+ !writerFactory
+ .getFileIO()
+ .getClass()
+ .getMethod("copyFileUtf8", Path.class, Path.class)
+ .getDeclaringClass()
+ .equals(FileIO.class);
+ } catch (NoSuchMethodException e) {
+ throw new RuntimeException(e);
+ }
+ this.canChangelogOptimizedToCopyPreconditions =
+ isCopyMethodOverridden
+ && (changelogProducer == ChangelogProducer.INPUT)
+ && recordAttributeManager != null;
  }
 
  private long newSequenceNumber() {
@@ -213,7 +240,7 @@ private void flushWriteBuffer(boolean waitForLatestCompaction, boolean forcedFul
  }
 
  final RollingFileWriter<KeyValue, DataFileMeta> changelogWriter =
- changelogProducer == ChangelogProducer.INPUT
+ changelogProducer == ChangelogProducer.INPUT & !canChangelogOptimizedToCopy()
  ? writerFactory.createRollingChangelogFileWriter(0)
  : null;
  final RollingFileWriter<KeyValue, DataFileMeta> dataWriter =
@@ -232,22 +259,49 @@ private void flushWriteBuffer(boolean waitForLatestCompaction, boolean forcedFul
  dataWriter.close();
  }
 
+ List<DataFileMeta> fileMetas = dataWriter.result();
  if (changelogWriter != null) {
  newFilesChangelog.addAll(changelogWriter.result());
+ } else if (canChangelogOptimizedToCopy()) {
+ List<DataFileMeta> changelogMetas = getChangelogFileMetaFromDataFile(fileMetas);
+ for (int i = 0; i < changelogMetas.size(); i++) {
+ writerFactory.copyFile(
+ fileMetas.get(i).fileName(), changelogMetas.get(i).fileName(), 0);
+ }
+ newFilesChangelog.addAll(changelogMetas);
  }
 
- for (DataFileMeta fileMeta : dataWriter.result()) {
+ for (DataFileMeta fileMeta : fileMetas) {
  newFiles.add(fileMeta);
  compactManager.addNewFile(fileMeta);
  }
 
  writeBuffer.clear();
+ if (recordAttributeManager != null) {
+ recordAttributeManager.onFlush();
+ }
  }
 
  trySyncLatestCompaction(waitForLatestCompaction);
  compactManager.triggerCompaction(forcedFullCompaction);
  }
 
+ private List<DataFileMeta> getChangelogFileMetaFromDataFile(
+ List<DataFileMeta> dataFileMetaList) {
+ List<DataFileMeta> changelogFileMetaList = new ArrayList<>();
+ for (DataFileMeta dataFileMeta : dataFileMetaList) {
+ DataFileMeta changelogFileMeta =
+ dataFileMeta.rename(writerFactory.newChangelogPath(0).getName());
+ changelogFileMetaList.add(changelogFileMeta);
+ }
+ return changelogFileMetaList;
+ }
+
+ private boolean canChangelogOptimizedToCopy() {
+ return canChangelogOptimizedToCopyPreconditions
+ && recordAttributeManager.areAllRecordsInsertOnlySinceLastFlush();
+ }
+
  @Override
  public CommitIncrement prepareCommit(boolean waitCompaction) throws Exception {
  flushWriteBuffer(waitCompaction, false);

paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java

@@ -67,6 +67,7 @@
 import org.apache.paimon.utils.CommitIncrement;
 import org.apache.paimon.utils.FieldsComparator;
 import org.apache.paimon.utils.FileStorePathFactory;
+import org.apache.paimon.utils.RecordAttributeManager;
 import org.apache.paimon.utils.SnapshotManager;
 import org.apache.paimon.utils.UserDefinedSeqComparator;
 
@@ -101,6 +102,7 @@ public class KeyValueFileStoreWrite extends MemoryFileStoreWrite<KeyValue> {
  private final RowType keyType;
  private final RowType valueType;
  @Nullable private final RecordLevelExpire recordLevelExpire;
+ @Nullable private final RecordAttributeManager recordAttributeManager;
 
  public KeyValueFileStoreWrite(
  FileIO fileIO,
@@ -121,7 +123,8 @@ public KeyValueFileStoreWrite(
  @Nullable DeletionVectorsMaintainer.Factory deletionVectorsMaintainerFactory,
  CoreOptions options,
  KeyValueFieldsExtractor extractor,
- String tableName) {
+ String tableName,
+ @Nullable RecordAttributeManager recordAttributeManager) {
  super(
  commitUser,
  snapshotManager,
@@ -134,6 +137,7 @@ public KeyValueFileStoreWrite(
  this.keyType = keyType;
  this.valueType = valueType;
  this.udsComparatorSupplier = udsComparatorSupplier;
+ this.recordAttributeManager = recordAttributeManager;
  this.readerFactoryBuilder =
  KeyValueFileReaderFactory.builder(
  fileIO,
@@ -210,7 +214,11 @@ protected MergeTreeWriter createWriter(
  options.commitForceCompact(),
  options.changelogProducer(),
  restoreIncrement,
- UserDefinedSeqComparator.create(valueType, options));
+ UserDefinedSeqComparator.create(valueType, options),
+ recordAttributeManager == null
+ ? null
+ : recordAttributeManager.getBucketRecordAttributeManager(
+ partition, bucket));
  }
 
  @VisibleForTesting

paimon-core/src/main/java/org/apache/paimon/privilege/PrivilegedFileStore.java

@@ -40,6 +40,7 @@
 import org.apache.paimon.tag.TagAutoManager;
 import org.apache.paimon.types.RowType;
 import org.apache.paimon.utils.FileStorePathFactory;
+import org.apache.paimon.utils.RecordAttributeManager;
 import org.apache.paimon.utils.SnapshotManager;
 import org.apache.paimon.utils.TagManager;
 
@@ -128,9 +129,12 @@ public FileStoreWrite<T> newWrite(String commitUser) {
  }
 
  @Override
- public FileStoreWrite<T> newWrite(String commitUser, ManifestCacheFilter manifestFilter) {
+ public FileStoreWrite<T> newWrite(
+ String commitUser,
+ ManifestCacheFilter manifestFilter,
+ @Nullable RecordAttributeManager recordAttributeManager) {
  privilegeChecker.assertCanInsert(identifier);
- return wrapped.newWrite(commitUser, manifestFilter);
+ return wrapped.newWrite(commitUser, manifestFilter, recordAttributeManager);
  }
 
  @Override

paimon-core/src/main/java/org/apache/paimon/privilege/PrivilegedFileStoreTable.java

@@ -39,9 +39,12 @@
 import org.apache.paimon.table.source.StreamDataTableScan;
 import org.apache.paimon.table.source.snapshot.SnapshotReader;
 import org.apache.paimon.utils.BranchManager;
+import org.apache.paimon.utils.RecordAttributeManager;
 import org.apache.paimon.utils.SnapshotManager;
 import org.apache.paimon.utils.TagManager;
 
+import javax.annotation.Nullable;
+
 import java.time.Duration;
 import java.util.Map;
 import java.util.Objects;
@@ -265,9 +268,12 @@ public TableWriteImpl<?> newWrite(String commitUser) {
  }
 
  @Override
- public TableWriteImpl<?> newWrite(String commitUser, ManifestCacheFilter manifestFilter) {
+ public TableWriteImpl<?> newWrite(
+ String commitUser,
+ ManifestCacheFilter manifestFilter,
+ @Nullable RecordAttributeManager recordAttributeManager) {
  privilegeChecker.assertCanInsert(identifier);
- return wrapped.newWrite(commitUser, manifestFilter);
+ return wrapped.newWrite(commitUser, manifestFilter, recordAttributeManager);
  }
 
  @Override