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

paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java

@@ -23,6 +23,7 @@
 import org.apache.paimon.fs.hadoop.HadoopFileIOLoader;
 import org.apache.paimon.fs.local.LocalFileIO;
 
+import org.apache.commons.io.IOUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -267,6 +268,22 @@ default void overwriteFileUtf8(Path path, String content) throws IOException {
  }
  }
 
+ /**
+ * Copy content of one file into another.
+ *
+ * @return false if targetPath file exists
+ */
+ default boolean copyFile(Path src, Path dst) throws IOException {
+ if (exists(dst)) {
+ return false;
+ }
+ try (SeekableInputStream is = newInputStream(src);
+ PositionOutputStream os = newOutputStream(dst, false)) {
+ IOUtils.copy(is, os);
+ }
+ return true;
+ }
+
  /**
  * Read file to UTF_8 decoding, then write content to one file atomically, initially writes to
  * temp hidden file and only renames to the target file once temp file is closed.
@@ -275,6 +292,7 @@ default void overwriteFileUtf8(Path path, String content) throws IOException {
  */
  default boolean copyFileUtf8(Path sourcePath, Path targetPath) throws IOException {
  String content = readFileUtf8(sourcePath);
+ System.out.println("copyFileUtf8 " + content.length());
  return writeFileUtf8(targetPath, content);
  }
 

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 copyFile(Path src, Path dst) throws IOException {
+ if (exists(dst)) {
+ return false;
+ }
+ Files.copy(toPath(src), toPath(dst), 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-common/src/test/java/org/apache/paimon/fs/FileIOTest.java

@@ -19,19 +19,31 @@
 package org.apache.paimon.fs;
 
 import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.fs.local.LocalFileIO;
 import org.apache.paimon.options.Options;
 
+import org.apache.commons.io.FileUtils;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
 
+import java.io.File;
+import java.io.FileNotFoundException;
 import java.io.IOException;
+import java.nio.file.AccessDeniedException;
+import java.nio.file.DirectoryNotEmptyException;
+import java.nio.file.FileAlreadyExistsException;
+import java.nio.file.Files;
+import java.nio.file.NoSuchFileException;
+import java.nio.file.StandardCopyOption;
 import java.util.Optional;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.ReentrantLock;
 
+import static org.apache.paimon.utils.Preconditions.checkState;
 import static org.assertj.core.api.Assertions.assertThat;
 
-/** Test {@link FileIO}. */
+/** Test static methods and methods with default implementations of {@link FileIO}. */
 public class FileIOTest {
 
  @TempDir java.nio.file.Path tempDir;
@@ -59,6 +71,36 @@ public void testRequireOptions() throws IOException {
  assertThat(fileIO).isInstanceOf(RequireOptionsFileIOLoader.MyFileIO.class);
  }
 
+ @Test
+ public void testCopy() throws Exception {
+ Path srcFile = new Path(tempDir.resolve("src.txt").toUri());
+ Path dstFile = new Path(tempDir.resolve("dst.txt").toUri());
+
+ FileIO fileIO = new DummyFileIO();
+ fileIO.writeFileUtf8(srcFile, "foobar");
+
+ assertThat(fileIO.copyFileUtf8(srcFile, dstFile)).isTrue();
+ assertThat(fileIO.readFileUtf8(dstFile)).isEqualTo("foobar");
+ fileIO.deleteQuietly(dstFile);
+
+ assertThat(fileIO.copyFile(srcFile, dstFile)).isTrue();
+ assertThat(fileIO.readFileUtf8(dstFile)).isEqualTo("foobar");
+ fileIO.deleteQuietly(dstFile);
+
+ fileIO.deleteQuietly(srcFile);
+ srcFile = new Path(this.getClass().getClassLoader().getResource("test-data.orc").toURI());
+
+ fileIO.copyFileUtf8(srcFile, dstFile);
+ assertThat(FileUtils.contentEquals(new File(srcFile.toUri()), new File(dstFile.toUri())))
+ .isFalse();
+ fileIO.deleteQuietly(dstFile);
+
+ fileIO.copyFile(srcFile, dstFile);
+ assertThat(FileUtils.contentEquals(new File(srcFile.toUri()), new File(dstFile.toUri())))
+ .isTrue();
+ fileIO.deleteQuietly(dstFile);
+ }
+
  public static void testOverwriteFileUtf8(Path file, FileIO fileIO) throws InterruptedException {
  AtomicReference<Exception> exception = new AtomicReference<>();
  final int max = 10;
@@ -106,4 +148,148 @@ public static void testOverwriteFileUtf8(Path file, FileIO fileIO) throws Interr
 
  assertThat(exception.get()).isNull();
  }
+
+ /** A {@link FileIO} on local filesystem to test the default copy implementation. */
+ private static class DummyFileIO implements FileIO {
+ private static final ReentrantLock RENAME_LOCK = new ReentrantLock();
+
+ @Override
+ public boolean isObjectStore() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void configure(CatalogContext context) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public SeekableInputStream newInputStream(Path path) throws FileNotFoundException {
+ return new LocalFileIO.LocalSeekableInputStream(toFile(path));
+ }
+
+ @Override
+ public PositionOutputStream newOutputStream(Path path, boolean overwrite)
+ throws IOException {
+ if (exists(path) && !overwrite) {
+ throw new FileAlreadyExistsException("File already exists: " + path);
+ }
+
+ Path parent = path.getParent();
+ if (parent != null && !mkdirs(parent)) {
+ throw new IOException("Mkdirs failed to create " + parent);
+ }
+
+ return new LocalFileIO.LocalPositionOutputStream(toFile(path));
+ }
+
+ @Override
+ public FileStatus getFileStatus(Path path) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public FileStatus[] listStatus(Path path) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean exists(Path path) {
+ return toFile(path).exists();
+ }
+
+ @Override
+ public boolean delete(Path path, boolean recursive) throws IOException {
+ File file = toFile(path);
+ if (file.isFile()) {
+ return file.delete();
+ } else if ((!recursive) && file.isDirectory()) {
+ File[] containedFiles = file.listFiles();
+ if (containedFiles == null) {
+ throw new IOException(
+ "Directory " + file + " does not exist or an I/O error occurred");
+ } else if (containedFiles.length != 0) {
+ throw new IOException("Directory " + file + " is not empty");
+ }
+ }
+
+ return delete(file);
+ }
+
+ private boolean delete(final File f) {
+ if (f.isDirectory()) {
+ final File[] files = f.listFiles();
+ if (files != null) {
+ for (File file : files) {
+ final boolean del = delete(file);
+ if (!del) {
+ return false;
+ }
+ }
+ }
+ } else {
+ return f.delete();
+ }
+
+ // Now directory is empty
+ return f.delete();
+ }
+
+ @Override
+ public boolean mkdirs(Path path) throws IOException {
+ return mkdirsInternal(toFile(path));
+ }
+
+ private boolean mkdirsInternal(File file) throws IOException {
+ if (file.isDirectory()) {
+ return true;
+ } else if (file.exists() && !file.isDirectory()) {
+ // Important: The 'exists()' check above must come before the 'isDirectory()' check
+ // to
+ // be safe when multiple parallel instances try to create the directory
+
+ // exists and is not a directory -> is a regular file
+ throw new FileAlreadyExistsException(file.getAbsolutePath());
+ } else {
+ File parent = file.getParentFile();
+ return (parent == null || mkdirsInternal(parent))
+ && (file.mkdir() || file.isDirectory());
+ }
+ }
+
+ @Override
+ public boolean rename(Path src, Path dst) throws IOException {
+ File srcFile = toFile(src);
+ File dstFile = toFile(dst);
+ File dstParent = dstFile.getParentFile();
+ dstParent.mkdirs();
+ try {
+ RENAME_LOCK.lock();
+ if (dstFile.exists()) {
+ return false;
+ }
+ Files.move(srcFile.toPath(), dstFile.toPath(), StandardCopyOption.ATOMIC_MOVE);
+ return true;
+ } catch (NoSuchFileException
+ | AccessDeniedException
+ | DirectoryNotEmptyException
+ | SecurityException e) {
+ return false;
+ } finally {
+ RENAME_LOCK.unlock();
+ }
+ }
+
+ private File toFile(Path path) {
+ // remove scheme
+ String localPath = path.toUri().getPath();
+ checkState(localPath != null, "Cannot convert a null path to File");
+
+ if (localPath.length() == 0) {
+ return new File(".");
+ }
+
+ return new File(localPath);
+ }
+ }
 }

paimon-common/src/test/java/org/apache/paimon/fs/local/LocalFleIOTest.java

@@ -0,0 +1,66 @@
+/*
+ * 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.fs.local;
+
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+
+import org.apache.commons.io.FileUtils;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link LocalFileIO}. */
+public class LocalFleIOTest {
+
+ @TempDir java.nio.file.Path tempDir;
+
+ @Test
+ public void testCopy() throws Exception {
+ Path srcFile = new Path(tempDir.resolve("src.txt").toUri());
+ Path dstFile = new Path(tempDir.resolve("dst.txt").toUri());
+
+ FileIO fileIO = new LocalFileIO();
+ fileIO.writeFileUtf8(srcFile, "foobar");
+
+ assertThat(fileIO.copyFileUtf8(srcFile, dstFile)).isTrue();
+ assertThat(fileIO.readFileUtf8(dstFile)).isEqualTo("foobar");
+ fileIO.deleteQuietly(dstFile);
+
+ assertThat(fileIO.copyFile(srcFile, dstFile)).isTrue();
+ assertThat(fileIO.readFileUtf8(dstFile)).isEqualTo("foobar");
+ fileIO.deleteQuietly(dstFile);
+
+ fileIO.deleteQuietly(srcFile);
+ srcFile = new Path(this.getClass().getClassLoader().getResource("test-data.orc").toURI());
+
+ fileIO.copyFileUtf8(srcFile, dstFile);
+ assertThat(FileUtils.contentEquals(new File(srcFile.toUri()), new File(dstFile.toUri())))
+ .isFalse();
+ fileIO.deleteQuietly(dstFile);
+
+ fileIO.copyFile(srcFile, dstFile);
+ assertThat(FileUtils.contentEquals(new File(srcFile.toUri()), new File(dstFile.toUri())))
+ .isTrue();
+ fileIO.deleteQuietly(dstFile);
+ }
+}

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);