diff --git a/docs/docs/learn-paimon/understand-files.mdx b/docs/docs/learn-paimon/understand-files.mdx
index 5d0a81296acc..9bf816b2bbff 100644
--- a/docs/docs/learn-paimon/understand-files.mdx
+++ b/docs/docs/learn-paimon/understand-files.mdx
@@ -369,7 +369,9 @@ Let's say all 4 snapshots in the above diagram are about to expire. The expire p
If any directories are left empty after the deletion process, they will be deleted as well,
but only when `snapshot.clean-empty-directories` is enabled (default is `false`).
-By default, empty directories are kept on disk. See [Manage Snapshots](../maintenance/manage-snapshots#expire-snapshots).
+By default, Paimon does not actively delete empty directories or their markers. On an object store,
+an implicit prefix may still cease to be visible after its last object is deleted. See
+[Manage Snapshots](../maintenance/manage-snapshots#expire-snapshots).
Let's say another snapshot, `snapshot-5` is created and snapshot expiration is triggered. `snapshot-1` to `snapshot-4` are
diff --git a/docs/docs/maintenance/manage-partitions.md b/docs/docs/maintenance/manage-partitions.md
index b5a471a6b951..dd9a3aa31eb7 100644
--- a/docs/docs/maintenance/manage-partitions.md
+++ b/docs/docs/maintenance/manage-partitions.md
@@ -49,10 +49,11 @@ __Note:__ After the partition expires, it is logically deleted and the latest sn
files in the file system are not immediately physically deleted, it depends on when the corresponding snapshot expires.
See [Expire Snapshots](./manage-snapshots#expire-snapshots).
-Also, even after the data files are physically deleted by snapshot expiration, the empty partition directories are
-**not** removed by default. To clean up empty directories, set
-`'snapshot.clean-empty-directories' = 'true'` on the table. Please note that on object stores (e.g. OSS, S3)
-this may cause performance issues, which is why the option defaults to `false`.
+Also, even after snapshot expiration physically deletes the data files, Paimon does not actively remove empty
+partition directories or their markers by default. An implicit object-store prefix may nevertheless cease to be
+visible after its last object is deleted. To make Paimon additionally try to remove visible empty directories and
+markers, set `'snapshot.clean-empty-directories' = 'true'` on the table. This may cause performance issues on
+object stores (e.g. OSS, S3), which is why the option defaults to `false`.
:::
diff --git a/docs/docs/maintenance/manage-snapshots.mdx b/docs/docs/maintenance/manage-snapshots.mdx
index 69b972cb2699..237a73ddf287 100644
--- a/docs/docs/maintenance/manage-snapshots.mdx
+++ b/docs/docs/maintenance/manage-snapshots.mdx
@@ -88,14 +88,14 @@ Snapshot expiration is controlled by the following table properties.
No
false
Boolean
-
Whether to try to delete empty directories (e.g. partition and bucket directories) left behind after the data files are deleted during snapshot expiration. Defaults to false: empty directories are kept. Enabling it has caveats: HDFS may print exceptions in NameNode, and object stores (OSS/S3) may suffer performance issues due to the extra prefix operations required to list and delete directory markers.
+
Whether Paimon tries to delete empty directories (e.g. partition and bucket directories) left behind after data files are deleted during snapshot expiration. The default is false, so Paimon does not actively delete those directories or their markers. An object-store prefix that was never explicitly created may still cease to be visible after its last object is deleted. Enabling the option has caveats: HDFS may print exceptions in NameNode, and object stores (OSS/S3) may suffer performance issues due to the extra prefix operations required to list and delete directory markers.
When the number of snapshots is less than `snapshot.num-retained.min`, no snapshots will be expired(even the condition `snapshot.time-retained` meet), after which `snapshot.num-retained.max` and `snapshot.time-retained` will be used to control the snapshot expiration until the remaining snapshot meets the condition.
-Note that snapshot expiration is also what physically deletes data files dropped by [partition expiration](./manage-partitions#expiring-partitions). However, the empty partition and bucket directories left behind after the data files are deleted are **not** removed by default. To clean them up, enable `snapshot.clean-empty-directories` (see the option above). This is off by default because on object stores (OSS/S3) the prefix operations needed to delete directory markers can be expensive.
+Note that snapshot expiration is also what physically deletes data files dropped by [partition expiration](./manage-partitions#expiring-partitions). By default, Paimon does not actively remove empty partition and bucket directories or their markers. On object stores, an implicit prefix may nevertheless cease to be visible when its last object is deleted. To make Paimon additionally try to remove visible empty directories and markers, enable `snapshot.clean-empty-directories` (see the option above). This is off by default because the required prefix operations can be expensive on object stores (OSS/S3).
The following example show more details(`snapshot.num-retained.min` is 2, `snapshot.time-retained` is 1h, `snapshot.num-retained.max` is 5):
diff --git a/docs/docs/program-api/file-io.md b/docs/docs/program-api/file-io.md
new file mode 100644
index 000000000000..3fe653cc06dc
--- /dev/null
+++ b/docs/docs/program-api/file-io.md
@@ -0,0 +1,285 @@
+---
+title: "FileIO API"
+sidebar_position: 4
+---
+
+
+
+# FileIO API
+
+`FileIO` is Paimon's interface for file I/O on local file systems, distributed file systems, and
+object stores. This page is for code that calls `FileIO` directly and for developers implementing a
+new `FileIO` implementation. Most users only need
+[Filesystems](../maintenance/filesystems), which describes the dependencies and options for the
+built-in implementations.
+
+The method sections below describe behavior common to the supported implementations. The final
+section describes how Paimon currently calls the API. Current usage does not narrow the public
+interface: a call remains valid when its method contract allows it, even if Paimon does not make
+that call today.
+
+These contracts describe observable results, not a required sequence of storage requests. An
+implementation can use conditional writes, metadata returned by listings, known file lengths, or
+batch operations. It does not need a preliminary `exists(...)` or `getFileStatus(...)` call when an
+operation can produce the required result directly.
+
+## Read and Write Files
+
+`newInputStream(...)` opens a new `SeekableInputStream`. Each stream has its own position, which
+starts at `0` and is returned by `getPos()`. The stream supports forward and backward seeks from `0`
+through the file length. Reading at the end of the file returns `-1` and does not change the
+position. Opening a missing path or a directory can fail either when the stream is opened or on the
+first read.
+
+`newOutputStream(path, overwrite)` opens a `PositionOutputStream`. Its `getPos()` value is the
+number of bytes written. After the stream closes successfully, the target contains exactly those
+bytes. The `overwrite` argument controls how an existing target is handled:
+
+- `false` preserves the existing file and reports an `IOException`. The error can occur while
+ opening the stream, writing data, or closing it.
+- `true` replaces the existing content.
+
+`flush()` writes buffered data to the underlying stream, but only a successful `close()` guarantees
+that the data is persistent and visible. Writing a file below a missing directory also makes its
+parent paths visible as directories through `exists(...)` and `getFileStatus(...)`. Object store
+implementations do not need to create a physical directory marker for every parent.
+
+A missing directory successfully created by `mkdirs(...)` before its descendants are written is
+explicit: deleting or moving its last child does not delete that directory. A parent that becomes
+visible only because a descendant was written is implicit. It must remain visible while any
+descendant exists, but after the last descendant is deleted or moved away it may either remain as
+an empty directory or become missing. This difference does not require callers to inspect physical
+directory markers.
+
+## File Status and Listing
+
+`getFileStatus(...)` returns a `FileStatus` for an existing path. `getPath()` returns the path,
+`isDir()` distinguishes a directory from a file, `getLen()` returns a file's length, and
+`getModificationTime()` returns the number of milliseconds since the Unix epoch. Modification-time
+precision depends on the storage system. `getAccessTime()` and `getOwner()` may be unavailable; their
+default values are `0` and `null`. A missing path throws `FileNotFoundException`.
+
+The listing methods are defined for existing directories:
+
+- `listStatus(...)` returns the direct files and directories. It returns an empty array for an empty
+ directory.
+- `listFiles(...)` and `listFilesIterative(...)` return files only. With recursive listing enabled,
+ they also return files in nested directories. The iterator may load entries as it is consumed.
+- `listDirectories(...)` returns direct directories only.
+
+Listing order is not guaranteed. `getFileSize(...)` and `isDir(...)` return the corresponding value
+from `getFileStatus(...)`, without requiring a separate `exists(...)` call.
+
+## File and Directory Operations
+
+- `exists(...)` returns `true` for an existing file or directory and `false` for a missing path.
+- `mkdirs(...)` creates the requested directory and any missing parents. It returns `true` when the
+ directory already exists. A file at the target path or in its parent path causes an
+ `IOException`.
+- `delete(path, recursive)` returns `true` after deleting an existing file or empty directory.
+ Deleting a non-empty directory with `recursive=false` throws `IOException` and leaves the
+ directory unchanged. With `recursive=true`, it deletes the complete directory tree. The return
+ value for a missing path is implementation-specific.
+- `rename(...)` moves a file or directory to the exact destination path. Call it with an existing
+ source, a different destination that does not exist, and an existing destination parent in the
+ same underlying file system. On success, it returns `true`, removes the source path, and preserves
+ the file content or complete directory tree. Moving the last descendant out of an explicit source
+ parent leaves that parent as an empty directory; an implicit source parent may become missing.
+- `copyFile(...)` copies the source bytes to the exact destination. An existing destination is
+ replaced only when `overwrite=true`. When a source directory contains files only,
+ `copyFiles(...)` applies the same behavior to each direct file.
+
+`checkOrMkdirs(...)` accepts an existing directory or creates a missing one, but throws
+`IllegalArgumentException` for an existing file. `deleteQuietly(...)`, `deleteFilesQuietly(...)`,
+and `deleteDirectoryQuietly(...)` suppress `IOException`; directory deletion is recursive, while
+file deletion is not. These quiet helpers are best-effort and do not return a success result.
+
+## UTF-8 File Helpers
+
+`writeFile(...)` writes UTF-8 text using the requested overwrite mode. `overwriteFileUtf8(...)` and
+`overwriteHintFile(...)` replace the current content. All three methods close the output stream.
+Use `overwriteHintFile(...)` only for hint files whose temporary absence during an overwrite is
+acceptable.
+
+`readFileUtf8(...)` reads UTF-8 text and closes the input stream. It reads the file line by line and
+does not preserve line separators. `readOverwrittenFileUtf8(...)` returns an empty `Optional` for a
+missing file and retries the remote-file-change errors recognized by the implementation.
+
+`tryToWriteAtomic(...)` returns `true` when it publishes content to a missing target. If the target
+already exists, it returns `false`, keeps the existing content, and removes temporary data. Its
+default implementation writes to a temporary file and then renames it, while storage implementations
+may use native conditional writes. Atomicity between clients therefore depends on the storage
+system.
+
+## Two-Phase Writes
+
+`newTwoPhaseOutputStream(...)` writes data to a staging path. `closeForCommit()` returns a
+serializable committer. The staged data is not visible at the target before `commit(...)` is called;
+a successful `commit(...)` publishes it. If the commit throws an exception, the target state is not
+guaranteed. The `overwrite` argument has the same meaning as it does for `newOutputStream(...)`.
+
+`discard(...)` removes only the data staged by that writer. After a successful commit, `clean(...)`
+can remove resources that the committer no longer needs, but it must not remove another writer's
+data.
+
+The default implementation stages a temporary file and commits it with `rename(...)`. A storage
+system can override the method, for example to use multipart upload.
+
+## Loading and Configuration
+
+`FileIO.get(...)` selects an implementation from a path and `CatalogContext`. When
+`resolving-file-io.enabled` is enabled, it returns a configured `ResolvingFileIO`, which selects an
+underlying `FileIO` for each path. Otherwise, a path without a URI scheme returns `LocalFileIO`
+directly. Its `configure(...)` method is a no-op.
+
+For a path with a scheme, `FileIO.get(...)` first checks the configured preferred loader. If that
+loader is absent or inaccessible, it looks for a discovered loader with the same scheme. Before
+using the preferred or discovered loader, it checks `requiredOptions()`: each returned group lists
+aliases for one required option, and at least one alias from every group must occur in the catalog
+options, matched case-insensitively. A loader with a missing required option is skipped. Selection
+then checks the configured fallback loader and finally Hadoop. The final loader creates a new
+`FileIO`, which `FileIO.get(...)` configures before returning it.
+
+Applications can obtain the storage for an existing table through `Table.fileIO()`. Code that only
+has a path and a `CatalogContext` can call `FileIO.get(...)`. `discoverLoaders()` and
+`checkAccess(...)` support this selection and are not normally called directly.
+
+Call `configure(...)` on factory- or loader-created instances that have not yet received a
+`CatalogContext`. A `FileIO` returned by `Table.fileIO()` is already configured and must not be
+configured again. Some table-bound implementations, including `RESTTokenFileIO`, reject
+reconfiguration.
+
+`setRuntimeContext(...)` supplies optional job-level file system settings. Paimon's Flink
+integration calls it only when `filesystem.job-level-settings.enabled` is enabled. It is not an
+automatic callback after deserialization, and its default implementation does nothing. `close()`
+releases resources owned by an implementation; its default also does nothing.
+
+`FileIO` implementations are serializable and thread-safe. `isObjectStore()` is an implementation
+hint and does not define the behavior of other methods.
+
+## Optional Operations
+
+`archive(...)`, `restoreArchive(...)`, `unarchive(...)`, and `createBlobPresignedUrl(...)` are
+optional. Their default implementations throw `UnsupportedOperationException`.
+
+## Paimon FileIO Usage
+
+This section records the call shapes and results used by Paimon's production code. It helps new
+callers choose the same preconditions and recovery rules. It does not replace the method contracts
+above or remove behavior from methods that have no current caller.
+
+### Status and Listing
+
+Paimon lists paths expected to be directories. They usually come from configuration, a previous
+status or listing result, or an `exists(...)` check. A configured warehouse or object-table
+location can itself be the file system root; callers treat that location as an ordinary directory.
+Core paths do not intentionally pass a regular file to a listing method.
+
+Callers do not depend on listing order or on one snapshot-consistent result while another client is
+changing the directory. Correctness-sensitive traversal handles a directory that disappears after
+its parent was listed by accepting an empty result or catching `FileNotFoundException` and skipping
+that subtree; other `IOException` values fail the operation. Best-effort orphan cleanup may instead
+treat any listing `IOException` as an empty result and skip that subtree. Some callers that want a
+missing directory to mean an empty listing check `exists(...)` first. This listing practice is
+separate from the public `getFileStatus(...)` rule: a missing path from `getFileStatus(...)` must
+throw `FileNotFoundException`.
+
+Paimon consumes modification times for both files and directories, including cleanup cutoffs and a
+branch directory's reported creation time. The value must follow the `FileStatus` contract, while
+its precision and changes during concurrent updates remain file-system-specific. Paimon relies on
+logical parent directories being visible, but does not inspect or require physical directory
+marker objects.
+
+### Output and Copying
+
+Both `newOutputStream(...)` modes are used. Paimon uses `overwrite=false` for UUID- or
+version-derived files expected not to exist; a failed create must preserve an existing target. It
+uses `overwrite=true` for replaceable state and copy destinations. Writers rely on exact
+`getPos()` values and successful `close()` as the publication boundary. Some branch-copy paths also
+create output below parents that have not been created explicitly.
+
+Current `copyFiles(...)` callers copy flat snapshot, schema, and tag metadata directories during
+branch fast-forward. They use the same `FileIO` for source and destination and pass
+`overwrite=true`. Files are copied one at a time; callers do not assume that a failure rolls back
+files copied earlier. This call shape does not remove the public `overwrite=false` behavior from
+`copyFile(...)` or `copyFiles(...)`.
+
+### Directories, Deletion, and Rename
+
+Paimon uses `mkdirs(...)` with directory-semantic paths, including paths that may already exist. It
+relies on parent creation and treats `false` as a creation failure where the result is checked.
+Production code does not intentionally pass an existing file or a child of a file.
+
+Strict `delete(...)` callers normally start with an existing, owned path. They use
+`recursive=false` for files or directories expected to be empty and `recursive=true` for complete
+owned trees. No caller relies on one particular return value for a missing path: it either ignores
+that result or combines a `false` result with `exists(...)`. Quiet deletion is used only where
+best-effort cleanup is acceptable.
+
+Raw `rename(...)` calls use an existing source, a distinct exact destination expected not to exist,
+the same `FileIO` and underlying file system, and an existing or pre-created destination parent. A
+`true` result confirms the move, and some callers check that result. Branch rename currently ignores
+the result and assumes that the selected file system provides atomic rename; it has no coordination
+fallback. Existing-destination recovery is handled by the higher-level conditional and two-phase
+write protocols, not by changing this raw rename shape. Core workflows do not intentionally use
+identical paths or move an item into an existing destination directory; those shapes appear only
+through optional virtual file system passthroughs.
+
+Snapshot publication is the workflow that adds external locking and content checks when atomic
+rename is unavailable. Paimon uses `isObjectStore()` when choosing the default catalog-lock setting,
+but the value alone does not change the `rename(...)` contract. The blob-descriptor source-table path
+also calls `isObjectStore()` before serialization to initialize lazy credentials. That side effect is
+implementation-specific and is not part of the `FileIO` contract.
+
+### Conditional and Two-Phase Publication
+
+Schemas, snapshots, and Iceberg metadata use `tryToWriteAtomic(...)` with an absent target as the
+normal case. `true` means this attempt published its content. `false` means the attempt did not
+publish because a target already exists; that target can be concurrent or stale. Callers inspect the
+existing content, retry, or use an external lock as required by their metadata protocol. Iceberg
+metadata may delete a nonmatching stale target and retry. Callers do not use this method as an
+overwrite operation for arbitrary existing state.
+
+Format-table writers are the current production users of `newTwoPhaseOutputStream(...)`. They pass
+`overwrite=false` and use writer-owned UUID target paths. They rely on staged data remaining hidden,
+a serializable committer from `closeForCommit()`, publication by `commit(...)`, and writer-scoped
+`discard(...)` and `clean(...)` operations. The public API still supports `overwrite=true` even
+though this workflow does not use it.
+
+A remote publication can succeed before reporting an exception, so recovery depends on ownership.
+Paimon preserves an ambiguous mutable target when the caller does not own a unique path. The
+format-table commit path records every attempted committer and can delete an attempted target after
+failure only because each UUID path belongs to that failed batch. This is a format-table recovery
+rule, not permission for arbitrary two-phase callers to delete an uncertain target.
+
+### Lifecycle and Optional Operations
+
+Most table code receives an already configured `FileIO` from `Table.fileIO()`. Factory-created
+instances are either retained by an owner or should be closed by the code that owns their resource
+scope. `CachingFileIO.close()` closes its delegate and then releases its shared cache-manager
+reference.
+
+`archive(...)`, `restoreArchive(...)`, and `unarchive(...)` currently have no production caller or
+implementation, so their default `UnsupportedOperationException` behavior remains in effect.
+`createBlobPresignedUrl(...)` is an active optional operation used by the Blob API and Flink and
+Spark SQL functions. Its callers use the `FileIO` and table root from the same loaded table, pass a
+table-owned blob descriptor, and do not assume that every `FileIO` supports the operation. Spark
+validates a positive whole-second validity before the call. Flink forwards the supplied `Duration`,
+so supporting implementations must reject unsupported validity values.
diff --git a/docs/sidebars.js b/docs/sidebars.js
index be7294db7b03..3ead8cd14948 100644
--- a/docs/sidebars.js
+++ b/docs/sidebars.js
@@ -269,6 +269,7 @@ const sidebars = {
"program-api/rest-api",
"program-api/flink-api",
"program-api/java-api",
+ "program-api/file-io",
"program-api/catalog-api",
"program-api/cpp-api",
"program-api/rust-api",
diff --git a/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java b/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java
index 2b0dcec3f760..e095a4b6c44a 100644
--- a/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java
+++ b/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java
@@ -233,8 +233,8 @@ default FileStatus[] listDirectories(Path path) throws IOException {
* 'mkdir -p'. Existence of the directory hierarchy is not an error.
*
* @param path the directory/directories to be created
- * @return true if at least one new directory has been created, false
- * otherwise
+ * @return true if the directory hierarchy exists after this call, including when
+ * it already existed, false otherwise
* @throws IOException thrown if an I/O error occurs while creating the directory
*/
boolean mkdirs(Path path) throws IOException;
diff --git a/paimon-common/src/main/java/org/apache/paimon/fs/RenamingTwoPhaseOutputStream.java b/paimon-common/src/main/java/org/apache/paimon/fs/RenamingTwoPhaseOutputStream.java
index a8547bd1673b..cad050f99e8e 100644
--- a/paimon-common/src/main/java/org/apache/paimon/fs/RenamingTwoPhaseOutputStream.java
+++ b/paimon-common/src/main/java/org/apache/paimon/fs/RenamingTwoPhaseOutputStream.java
@@ -35,6 +35,7 @@ public class RenamingTwoPhaseOutputStream extends TwoPhaseOutputStream {
private final Path targetPath;
private final Path tempPath;
private final PositionOutputStream tempOutputStream;
+ private final boolean overwrite;
public RenamingTwoPhaseOutputStream(FileIO fileIO, Path targetPath, boolean overwrite)
throws IOException {
@@ -43,6 +44,7 @@ public RenamingTwoPhaseOutputStream(FileIO fileIO, Path targetPath, boolean over
}
this.targetPath = targetPath;
this.tempPath = generateTempPath(targetPath);
+ this.overwrite = overwrite;
// Create temporary file
this.tempOutputStream = fileIO.newOutputStream(tempPath, overwrite);
@@ -81,7 +83,7 @@ public void close() throws IOException {
@Override
public Committer closeForCommit() throws IOException {
close();
- return new TempFileCommitter(tempPath, targetPath);
+ return new TempFileCommitter(tempPath, targetPath, overwrite);
}
/**
@@ -100,10 +102,12 @@ private static class TempFileCommitter implements Committer {
private final Path tempPath;
private final Path targetPath;
+ private final boolean overwrite;
- private TempFileCommitter(Path tempPath, Path targetPath) {
+ private TempFileCommitter(Path tempPath, Path targetPath, boolean overwrite) {
this.tempPath = tempPath;
this.targetPath = targetPath;
+ this.overwrite = overwrite;
}
@Override
@@ -112,7 +116,12 @@ public void commit(FileIO fileIO) throws IOException {
if (parentDir != null && !fileIO.exists(parentDir)) {
fileIO.mkdirs(parentDir);
}
- if (!fileIO.rename(tempPath, targetPath)) {
+ boolean renamed = fileIO.rename(tempPath, targetPath);
+ if (!renamed && overwrite && fileIO.exists(tempPath)) {
+ fileIO.delete(targetPath, false);
+ renamed = fileIO.rename(tempPath, targetPath);
+ }
+ if (!renamed) {
throw new IOException("Failed to rename " + tempPath + " to " + targetPath);
}
if (fileIO.exists(tempPath)) {
@@ -122,12 +131,7 @@ public void commit(FileIO fileIO) throws IOException {
@Override
public void discard(FileIO fileIO) throws IOException {
- if (fileIO.exists(targetPath)) {
- fileIO.deleteQuietly(targetPath);
- }
- if (fileIO.exists(tempPath)) {
- fileIO.deleteQuietly(tempPath);
- }
+ fileIO.deleteQuietly(tempPath);
}
@Override
diff --git a/paimon-common/src/main/java/org/apache/paimon/fs/hadoop/HadoopFileIO.java b/paimon-common/src/main/java/org/apache/paimon/fs/hadoop/HadoopFileIO.java
index 3ff241d6c8f2..ba091e1ba366 100644
--- a/paimon-common/src/main/java/org/apache/paimon/fs/hadoop/HadoopFileIO.java
+++ b/paimon-common/src/main/java/org/apache/paimon/fs/hadoop/HadoopFileIO.java
@@ -176,6 +176,14 @@ public boolean rename(Path src, Path dst) throws IOException {
return getFileSystem(hadoopSrc).rename(hadoopSrc, hadoopDst);
}
+ @Override
+ public boolean tryToWriteAtomic(Path path, String content) throws IOException {
+ if ("file".equalsIgnoreCase(path.toUri().getScheme()) && exists(path)) {
+ return false;
+ }
+ return FileIO.super.tryToWriteAtomic(path, content);
+ }
+
@Override
public void overwriteFileUtf8(Path path, String content) throws IOException {
boolean success = tryAtomicOverwriteViaRename(path, content);
diff --git a/paimon-common/src/main/java/org/apache/paimon/fs/local/LocalFileIO.java b/paimon-common/src/main/java/org/apache/paimon/fs/local/LocalFileIO.java
index 143f9d8c243c..e1e0403b13b5 100644
--- a/paimon-common/src/main/java/org/apache/paimon/fs/local/LocalFileIO.java
+++ b/paimon-common/src/main/java/org/apache/paimon/fs/local/LocalFileIO.java
@@ -243,11 +243,12 @@ public boolean rename(Path src, Path dst) throws IOException {
@Override
public void copyFile(Path sourcePath, Path targetPath, boolean overwrite) throws IOException {
LOG.debug("Invoking copyFile for {} to {}", sourcePath, targetPath);
- if (!overwrite && exists(targetPath)) {
- return;
- }
toPath(targetPath.getParent()).toFile().mkdirs();
- Files.copy(toPath(sourcePath), toPath(targetPath), StandardCopyOption.REPLACE_EXISTING);
+ if (overwrite) {
+ Files.copy(toPath(sourcePath), toPath(targetPath), StandardCopyOption.REPLACE_EXISTING);
+ } else {
+ Files.copy(toPath(sourcePath), toPath(targetPath));
+ }
}
private java.nio.file.Path toPath(Path path) {
diff --git a/paimon-common/src/test/java/org/apache/paimon/fs/FileIOContractCoverageTest.java b/paimon-common/src/test/java/org/apache/paimon/fs/FileIOContractCoverageTest.java
new file mode 100644
index 000000000000..15ec8a98249d
--- /dev/null
+++ b/paimon-common/src/test/java/org/apache/paimon/fs/FileIOContractCoverageTest.java
@@ -0,0 +1,122 @@
+/*
+ * 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;
+
+import org.junit.jupiter.api.Test;
+
+import java.lang.reflect.Method;
+import java.lang.reflect.Modifier;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Guards the ownership of every public method declared by {@link FileIO}. */
+public class FileIOContractCoverageTest {
+
+ private static final Set PROVIDER_CORE =
+ signatures(
+ "isObjectStore()",
+ "newInputStream(Path)",
+ "newOutputStream(Path,boolean)",
+ "getFileStatus(Path)",
+ "listStatus(Path)",
+ "exists(Path)",
+ "delete(Path,boolean)",
+ "mkdirs(Path)",
+ "rename(Path,Path)");
+
+ private static final Set DEFAULT_METHOD =
+ signatures(
+ "newTwoPhaseOutputStream(Path,boolean)",
+ "listFiles(Path,boolean)",
+ "listFilesIterative(Path,boolean)",
+ "listDirectories(Path)",
+ "deleteQuietly(Path)",
+ "deleteFilesQuietly(List)",
+ "deleteDirectoryQuietly(Path)",
+ "getFileSize(Path)",
+ "isDir(Path)",
+ "checkOrMkdirs(Path)",
+ "readFileUtf8(Path)",
+ "tryToWriteAtomic(Path,String)",
+ "writeFile(Path,String,boolean)",
+ "overwriteFileUtf8(Path,String)",
+ "overwriteHintFile(Path,String)",
+ "copyFile(Path,Path,boolean)",
+ "copyFiles(Path,Path,boolean)",
+ "readOverwrittenFileUtf8(Path)");
+
+ private static final Set PROVIDER_LIFECYCLE =
+ signatures("configure(CatalogContext)", "setRuntimeContext(Map)", "close()");
+
+ private static final Set FACTORY =
+ signatures(
+ "get(Path,CatalogContext)",
+ "discoverLoaders()",
+ "checkAccess(FileIOLoader,Path,CatalogContext)");
+
+ private static final Set OPTIONAL_CAPABILITY =
+ signatures(
+ "archive(Path,StorageType)",
+ "restoreArchive(Path,Duration)",
+ "unarchive(Path,StorageType)",
+ "createBlobPresignedUrl(Path,BlobDescriptor,Duration)");
+
+ @Test
+ public void testEveryDeclaredPublicMethodHasExactlyOneOwner() {
+ Set actual =
+ Arrays.stream(FileIO.class.getDeclaredMethods())
+ .filter(method -> Modifier.isPublic(method.getModifiers()))
+ .filter(method -> !method.isSynthetic())
+ .map(FileIOContractCoverageTest::signature)
+ .collect(Collectors.toSet());
+
+ List> categories =
+ Arrays.asList(
+ PROVIDER_CORE,
+ DEFAULT_METHOD,
+ PROVIDER_LIFECYCLE,
+ FACTORY,
+ OPTIONAL_CAPABILITY);
+ Set classified = new HashSet<>();
+ for (Set category : categories) {
+ assertThat(Collections.disjoint(category, classified)).isTrue();
+ classified.addAll(category);
+ }
+
+ assertThat(actual).hasSize(37);
+ assertThat(classified).containsExactlyInAnyOrderElementsOf(actual);
+ }
+
+ private static Set signatures(String... signatures) {
+ return new HashSet<>(Arrays.asList(signatures));
+ }
+
+ private static String signature(Method method) {
+ return method.getName()
+ + Arrays.stream(method.getParameterTypes())
+ .map(Class::getSimpleName)
+ .collect(Collectors.joining(",", "(", ")"));
+ }
+}
diff --git a/paimon-common/src/test/java/org/apache/paimon/fs/FileIOContractTestBase.java b/paimon-common/src/test/java/org/apache/paimon/fs/FileIOContractTestBase.java
new file mode 100644
index 000000000000..857bd8be06ab
--- /dev/null
+++ b/paimon-common/src/test/java/org/apache/paimon/fs/FileIOContractTestBase.java
@@ -0,0 +1,933 @@
+/*
+ * 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;
+
+import org.apache.paimon.utils.InstantiationUtil;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.ByteArrayOutputStream;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Contract tests shared by {@link FileIO} implementations. */
+public abstract class FileIOContractTestBase extends FileIOBehaviorTestBase {
+
+ private static final byte[] DEFAULT_CONTENT = new byte[] {1, 2, 3, 4, 5, 6, 7, 8};
+
+ private FileIO contractFileIO;
+
+ private Path contractBasePath;
+
+ @AfterEach
+ void cleanupContractFixture() throws IOException {
+ if (contractFileIO != null) {
+ contractFileIO.delete(contractBasePath, true);
+ }
+ }
+
+ // ------------------------------------------------------------------------
+ // Input streams
+ // ------------------------------------------------------------------------
+
+ @Test
+ void testInputStreamStartsAtZeroAndReadsCorrectBytes() throws IOException {
+ byte[] content = new byte[] {3, 1, 4, 1, 5, 9};
+ Path file = createRandomFileInDirectory(contractBasePath(), content);
+
+ try (SeekableInputStream in = contractFileIO().newInputStream(file)) {
+ assertThat(in.getPos()).isZero();
+ assertThat(readAll(in)).containsExactly(content);
+ assertThat(in.getPos()).isEqualTo(content.length);
+ }
+ }
+
+ @Test
+ void testInputStreamBulkReadHonorsNonZeroBufferOffset() throws IOException {
+ byte[] content = new byte[] {11, 22, 33};
+ Path file = createRandomFileInDirectory(contractBasePath(), content);
+ byte[] buffer = new byte[] {99, 98, 0, 0, 0, 97, 96};
+
+ try (SeekableInputStream in = contractFileIO().newInputStream(file)) {
+ int totalRead = 0;
+ while (totalRead < content.length) {
+ int read = in.read(buffer, 2 + totalRead, content.length - totalRead);
+ assertThat(read).isPositive();
+ totalRead += read;
+ }
+
+ assertThat(totalRead).isEqualTo(content.length);
+ assertThat(buffer).containsExactly(99, 98, 11, 22, 33, 97, 96);
+ assertThat(in.getPos()).isEqualTo(content.length);
+ }
+ }
+
+ @Test
+ void testInputStreamsHaveIndependentPositions() throws IOException {
+ Path file = createRandomFileInDirectory(contractBasePath(), new byte[] {10, 20, 30});
+
+ try (SeekableInputStream first = contractFileIO().newInputStream(file);
+ SeekableInputStream second = contractFileIO().newInputStream(file)) {
+ assertThat(first.read()).isEqualTo(10);
+ assertThat(first.getPos()).isEqualTo(1);
+ assertThat(second.getPos()).isZero();
+ assertThat(second.read()).isEqualTo(10);
+ assertThat(second.getPos()).isEqualTo(1);
+
+ first.seek(2);
+ assertThat(first.read()).isEqualTo(30);
+ assertThat(second.read()).isEqualTo(20);
+ }
+ }
+
+ @Test
+ void testInputStreamSeeksForwardAndBackward() throws IOException {
+ byte[] content = new byte[] {10, 20, 30, 40, 50, 60};
+ Path file = createRandomFileInDirectory(contractBasePath(), content);
+
+ try (SeekableInputStream in = contractFileIO().newInputStream(file)) {
+ in.seek(4);
+ assertThat(in.getPos()).isEqualTo(4);
+ assertThat(in.read()).isEqualTo(50);
+
+ in.seek(1);
+ assertThat(in.getPos()).isEqualTo(1);
+ assertThat(in.read()).isEqualTo(20);
+ }
+ }
+
+ @Test
+ void testInputStreamReturnsEndOfFileAtFileLength() throws IOException {
+ byte[] content = new byte[] {7, 8, 9};
+ Path file = createRandomFileInDirectory(contractBasePath(), content);
+
+ try (SeekableInputStream in = contractFileIO().newInputStream(file)) {
+ in.seek(content.length);
+ assertThat(in.read()).isEqualTo(-1);
+ assertThat(in.getPos()).isEqualTo(content.length);
+ assertThat(in.read(new byte[2], 0, 2)).isEqualTo(-1);
+ assertThat(in.getPos()).isEqualTo(content.length);
+ }
+ }
+
+ @Test
+ void testInputStreamCanSeekBackToStart() throws IOException {
+ byte[] content = new byte[] {7, 8, 9};
+ Path file = createRandomFileInDirectory(contractBasePath(), content);
+
+ try (SeekableInputStream in = contractFileIO().newInputStream(file)) {
+ assertThat(in.read()).isEqualTo(7);
+ in.seek(0);
+ assertThat(in.getPos()).isZero();
+ assertThat(in.read()).isEqualTo(7);
+ }
+ }
+
+ @Test
+ void testInputStreamSeeksForwardBeyondOneMebibyte() throws IOException {
+ int targetPosition = 1024 * 1024 + 17;
+ byte[] content = new byte[targetPosition + 1];
+ content[targetPosition] = 42;
+ Path file = createRandomFileInDirectory(contractBasePath(), content);
+
+ try (SeekableInputStream in = contractFileIO().newInputStream(file)) {
+ in.seek(targetPosition);
+ assertThat(in.getPos()).isEqualTo(targetPosition);
+ assertThat(in.read()).isEqualTo(42);
+ }
+ }
+
+ @Test
+ void testInputStreamForMissingFileFailsByFirstRead() throws IOException {
+ Path missing = new Path(contractBasePath(), randomName());
+
+ assertOpenOrFirstReadFails(missing);
+ }
+
+ @Test
+ void testInputStreamForDirectoryFailsByFirstRead() throws IOException {
+ Path directory = new Path(contractBasePath(), randomName());
+ contractFileIO().mkdirs(directory);
+
+ assertOpenOrFirstReadFails(directory);
+ }
+
+ // ------------------------------------------------------------------------
+ // Output streams
+ // ------------------------------------------------------------------------
+
+ @Test
+ void testOutputStreamTracksPositionAndPublishesBytesOnClose() throws IOException {
+ Path file = new Path(contractBasePath(), randomName());
+
+ try (PositionOutputStream out = contractFileIO().newOutputStream(file, false)) {
+ assertThat(out.getPos()).isZero();
+ out.write(9);
+ assertThat(out.getPos()).isEqualTo(1);
+ out.write(new byte[] {10, 11, 12, 13}, 1, 2);
+ assertThat(out.getPos()).isEqualTo(3);
+ }
+
+ assertThat(readBytes(file)).containsExactly(9, 11, 12);
+ }
+
+ @Test
+ void testOutputStreamFlushKeepsPositionAndClosePublishesLaterWrites() throws IOException {
+ Path file = new Path(contractBasePath(), randomName());
+
+ try (PositionOutputStream out = contractFileIO().newOutputStream(file, false)) {
+ out.write(new byte[] {1, 2});
+ out.flush();
+ assertThat(out.getPos()).isEqualTo(2);
+ out.write(3);
+ assertThat(out.getPos()).isEqualTo(3);
+ }
+
+ assertThat(readBytes(file)).containsExactly(1, 2, 3);
+ }
+
+ @Test
+ void testOutputStreamCreatesNestedTarget() throws IOException {
+ Path ancestor = new Path(contractBasePath(), randomName());
+ Path parent = new Path(ancestor, randomName());
+ Path file = new Path(parent, randomName());
+ byte[] content = new byte[] {1, 3, 3, 7};
+
+ writeBytes(file, content, false);
+
+ assertThat(readBytes(file)).containsExactly(content);
+ assertThat(contractFileIO().getFileStatus(ancestor).isDir()).isTrue();
+ assertThat(contractFileIO().getFileStatus(parent).isDir()).isTrue();
+ }
+
+ @Test
+ void testOutputStreamOverwriteReplacesOldContent() throws IOException {
+ Path file = createRandomFileInDirectory(contractBasePath(), new byte[] {1, 2, 3, 4, 5});
+
+ writeBytes(file, new byte[] {8, 9}, true);
+
+ assertThat(readBytes(file)).containsExactly(8, 9);
+ }
+
+ @Test
+ void testOutputStreamNoOverwriteFailsAndPreservesOldContent() throws IOException {
+ byte[] oldContent = new byte[] {1, 2, 3};
+ Path file = createRandomFileInDirectory(contractBasePath(), oldContent);
+
+ assertThatThrownBy(() -> writeBytes(file, new byte[] {9, 8, 7}, false))
+ .isInstanceOf(IOException.class);
+ assertThat(readBytes(file)).containsExactly(oldContent);
+ }
+
+ // ------------------------------------------------------------------------
+ // File status
+ // ------------------------------------------------------------------------
+
+ @Test
+ void testGetFileStatusForMissingPathThrowsFileNotFound() throws IOException {
+ Path missing = new Path(contractBasePath(), randomName());
+
+ assertThatThrownBy(() -> contractFileIO().getFileStatus(missing))
+ .isInstanceOf(FileNotFoundException.class);
+ }
+
+ @Test
+ void testGetFileStatusDescribesFile() throws IOException {
+ byte[] content = new byte[] {2, 4, 6, 8, 10};
+ Path file = createRandomFileInDirectory(contractBasePath(), content);
+
+ FileStatus status = contractFileIO().getFileStatus(file);
+
+ assertThat(status.getPath()).isEqualTo(file);
+ assertThat(status.isDir()).isFalse();
+ assertThat(status.getLen()).isEqualTo(content.length);
+ }
+
+ @Test
+ void testGetFileStatusDescribesDirectory() throws IOException {
+ Path directory = new Path(contractBasePath(), randomName());
+ contractFileIO().mkdirs(directory);
+
+ FileStatus status = contractFileIO().getFileStatus(directory);
+
+ assertThat(status.getPath()).isEqualTo(directory);
+ assertThat(status.isDir()).isTrue();
+ }
+
+ @Test
+ void testFileStatusProvidesConsistentModificationTime() throws IOException {
+ Path file = createRandomFileInDirectory(contractBasePath());
+
+ FileStatus directStatus = contractFileIO().getFileStatus(file);
+ FileStatus listedStatus = statusFor(contractFileIO().listStatus(contractBasePath()), file);
+
+ long directModificationTime = directStatus.getModificationTime();
+ long listedModificationTime = listedStatus.getModificationTime();
+ assertThat(directModificationTime).isGreaterThan(1_000_000_000_000L);
+ assertThat(listedModificationTime).isGreaterThan(1_000_000_000_000L);
+ assertThat(Math.abs(listedModificationTime - directModificationTime))
+ .isLessThanOrEqualTo(1_000L);
+ }
+
+ @Test
+ void testExistsRecognizesDirectory() throws IOException {
+ Path directory = new Path(contractBasePath(), randomName());
+ contractFileIO().mkdirs(directory);
+
+ assertThat(contractFileIO().exists(directory)).isTrue();
+ }
+
+ @Test
+ void testStatusHelpersDescribeFilesAndDirectories() throws IOException {
+ byte[] content = new byte[] {1, 4, 9, 16};
+ Path file = createRandomFileInDirectory(contractBasePath(), content);
+
+ assertThat(contractFileIO().getFileSize(file)).isEqualTo(content.length);
+ assertThat(contractFileIO().isDir(file)).isFalse();
+ assertThat(contractFileIO().isDir(contractBasePath())).isTrue();
+ }
+
+ // ------------------------------------------------------------------------
+ // Listings
+ // ------------------------------------------------------------------------
+
+ @Test
+ void testListStatusOfEmptyDirectoryReturnsNonNullEmptyArray() throws IOException {
+ FileStatus[] statuses = contractFileIO().listStatus(contractBasePath());
+
+ assertThat(statuses).isEmpty();
+ }
+
+ @Test
+ void testListStatusReturnsOnlyCorrectDirectChildren() throws IOException {
+ byte[] firstContent = new byte[] {1, 2, 3, 4};
+ byte[] secondContent = new byte[] {5, 6};
+ Path firstFile = createRandomFileInDirectory(contractBasePath(), firstContent);
+ Path secondFile = createRandomFileInDirectory(contractBasePath(), secondContent);
+ Path firstDirectory = new Path(contractBasePath(), randomName());
+ Path secondDirectory = new Path(contractBasePath(), randomName());
+ Path nestedDirectory = new Path(firstDirectory, randomName());
+ createRandomFileInDirectory(nestedDirectory, new byte[] {9});
+ contractFileIO().mkdirs(secondDirectory);
+
+ FileStatus[] statuses = contractFileIO().listStatus(contractBasePath());
+
+ assertThat(statuses)
+ .extracting(FileStatus::getPath)
+ .containsExactlyInAnyOrder(firstFile, secondFile, firstDirectory, secondDirectory);
+ FileStatus firstFileStatus = statusFor(statuses, firstFile);
+ assertThat(firstFileStatus.isDir()).isFalse();
+ assertThat(firstFileStatus.getLen()).isEqualTo(firstContent.length);
+ FileStatus secondFileStatus = statusFor(statuses, secondFile);
+ assertThat(secondFileStatus.isDir()).isFalse();
+ assertThat(secondFileStatus.getLen()).isEqualTo(secondContent.length);
+ assertThat(statusFor(statuses, firstDirectory).isDir()).isTrue();
+ assertThat(statusFor(statuses, secondDirectory).isDir()).isTrue();
+ }
+
+ @Test
+ void testListFilesNonRecursiveReturnsOnlyDirectFilesAndMatchesIterator() throws IOException {
+ Path firstDirectFile = createRandomFileInDirectory(contractBasePath());
+ Path secondDirectFile = createRandomFileInDirectory(contractBasePath());
+ Path directory = new Path(contractBasePath(), randomName());
+ createRandomFileInDirectory(directory);
+
+ FileStatus[] arrayResult = contractFileIO().listFiles(contractBasePath(), false);
+ List iteratorResult =
+ collect(contractFileIO().listFilesIterative(contractBasePath(), false));
+
+ assertThat(arrayResult)
+ .extracting(FileStatus::getPath)
+ .containsExactlyInAnyOrder(firstDirectFile, secondDirectFile);
+ assertThat(iteratorResult)
+ .extracting(FileStatus::getPath)
+ .containsExactlyInAnyOrder(firstDirectFile, secondDirectFile);
+ assertThat(Arrays.stream(arrayResult).allMatch(status -> !status.isDir())).isTrue();
+ assertThat(iteratorResult.stream().allMatch(status -> !status.isDir())).isTrue();
+ }
+
+ @Test
+ void testListFilesRecursiveReturnsAllFilesAndMatchesIterator() throws IOException {
+ Path firstDirectFile = createRandomFileInDirectory(contractBasePath());
+ Path secondDirectFile = createRandomFileInDirectory(contractBasePath());
+ Path firstLevelDirectory = new Path(contractBasePath(), randomName());
+ Path firstLevelFile = createRandomFileInDirectory(firstLevelDirectory);
+ Path secondLevelDirectory = new Path(firstLevelDirectory, randomName());
+ Path secondLevelFile = createRandomFileInDirectory(secondLevelDirectory);
+
+ FileStatus[] arrayResult = contractFileIO().listFiles(contractBasePath(), true);
+ List iteratorResult =
+ collect(contractFileIO().listFilesIterative(contractBasePath(), true));
+
+ assertThat(arrayResult)
+ .extracting(FileStatus::getPath)
+ .containsExactlyInAnyOrder(
+ firstDirectFile, secondDirectFile, firstLevelFile, secondLevelFile);
+ assertThat(iteratorResult)
+ .extracting(FileStatus::getPath)
+ .containsExactlyInAnyOrder(
+ firstDirectFile, secondDirectFile, firstLevelFile, secondLevelFile);
+ assertThat(Arrays.stream(arrayResult).allMatch(status -> !status.isDir())).isTrue();
+ assertThat(iteratorResult.stream().allMatch(status -> !status.isDir())).isTrue();
+ }
+
+ @Test
+ void testListDirectoriesReturnsOnlyDirectDirectories() throws IOException {
+ createRandomFileInDirectory(contractBasePath());
+ Path firstDirectDirectory = new Path(contractBasePath(), randomName());
+ Path secondDirectDirectory = new Path(contractBasePath(), randomName());
+ Path nestedDirectory = new Path(firstDirectDirectory, randomName());
+ contractFileIO().mkdirs(nestedDirectory);
+ contractFileIO().mkdirs(secondDirectDirectory);
+
+ FileStatus[] statuses = contractFileIO().listDirectories(contractBasePath());
+
+ assertThat(statuses)
+ .extracting(FileStatus::getPath)
+ .containsExactlyInAnyOrder(firstDirectDirectory, secondDirectDirectory);
+ assertThat(Arrays.stream(statuses).allMatch(FileStatus::isDir)).isTrue();
+ }
+
+ // ------------------------------------------------------------------------
+ // Delete
+ // ------------------------------------------------------------------------
+
+ @Test
+ void testDeleteReturnsTrueForExistingTargets() throws IOException {
+ Path file = createRandomFileInDirectory(contractBasePath());
+ Path directory = new Path(contractBasePath(), randomName());
+ contractFileIO().mkdirs(directory);
+
+ assertThat(contractFileIO().delete(file, false)).isTrue();
+ assertThat(contractFileIO().delete(directory, false)).isTrue();
+ }
+
+ @Test
+ void testDeleteLastChildKeepsExplicitParentDirectory() throws IOException {
+ Path parent = new Path(contractBasePath(), randomName());
+ contractFileIO().mkdirs(parent);
+ Path child = new Path(parent, randomName());
+ writeBytes(child, DEFAULT_CONTENT, false);
+
+ assertThat(contractFileIO().delete(child, false)).isTrue();
+
+ assertThat(contractFileIO().getFileStatus(parent).isDir()).isTrue();
+ assertThat(contractFileIO().listStatus(parent)).isEmpty();
+ }
+
+ @Test
+ void testDeleteLastChildAllowsImplicitParentToDisappear() throws IOException {
+ Path parent = new Path(contractBasePath(), randomName());
+ Path child = new Path(parent, randomName());
+ writeBytes(child, DEFAULT_CONTENT, false);
+ assertThat(contractFileIO().getFileStatus(parent).isDir()).isTrue();
+
+ assertThat(contractFileIO().delete(child, false)).isTrue();
+
+ assertMissingOrEmptyDirectory(parent);
+ }
+
+ @Test
+ void testDeleteChildKeepsImplicitParentVisibleWhileSiblingExists() throws IOException {
+ Path parent = new Path(contractBasePath(), randomName());
+ Path deleted = new Path(parent, randomName());
+ Path sibling = new Path(parent, randomName());
+ writeBytes(deleted, new byte[] {1}, false);
+ writeBytes(sibling, new byte[] {2}, false);
+
+ assertThat(contractFileIO().delete(deleted, false)).isTrue();
+
+ assertThat(contractFileIO().getFileStatus(parent).isDir()).isTrue();
+ assertThat(readBytes(sibling)).containsExactly(2);
+ assertThat(contractFileIO().listStatus(parent))
+ .extracting(FileStatus::getPath)
+ .containsExactly(sibling);
+ }
+
+ // ------------------------------------------------------------------------
+ // Rename
+ // ------------------------------------------------------------------------
+
+ @Test
+ void testRenameFileMovesExactBytesToMissingDestination() throws IOException {
+ byte[] content = new byte[] {4, 2, 4, 2};
+ Path source = createRandomFileInDirectory(contractBasePath(), content);
+ Path destination = new Path(contractBasePath(), randomName());
+
+ assertThat(contractFileIO().rename(source, destination)).isTrue();
+
+ assertThat(contractFileIO().exists(source)).isFalse();
+ assertThat(readBytes(destination)).containsExactly(content);
+ }
+
+ @Test
+ void testRenameLastChildKeepsExplicitSourceParentDirectory() throws IOException {
+ Path sourceParent = new Path(contractBasePath(), randomName());
+ contractFileIO().mkdirs(sourceParent);
+ Path source = new Path(sourceParent, randomName());
+ Path destination = new Path(contractBasePath(), randomName());
+ writeBytes(source, DEFAULT_CONTENT, false);
+
+ assertThat(contractFileIO().rename(source, destination)).isTrue();
+
+ assertThat(contractFileIO().getFileStatus(sourceParent).isDir()).isTrue();
+ assertThat(contractFileIO().listStatus(sourceParent)).isEmpty();
+ assertThat(readBytes(destination)).containsExactly(DEFAULT_CONTENT);
+ }
+
+ @Test
+ void testRenameLastChildAllowsImplicitSourceParentToDisappear() throws IOException {
+ Path sourceParent = new Path(contractBasePath(), randomName());
+ Path source = new Path(sourceParent, randomName());
+ Path destination = new Path(contractBasePath(), randomName());
+ writeBytes(source, DEFAULT_CONTENT, false);
+ assertThat(contractFileIO().getFileStatus(sourceParent).isDir()).isTrue();
+
+ assertThat(contractFileIO().rename(source, destination)).isTrue();
+
+ assertMissingOrEmptyDirectory(sourceParent);
+ assertThat(readBytes(destination)).containsExactly(DEFAULT_CONTENT);
+ }
+
+ @Test
+ void testRenameChildKeepsImplicitSourceParentVisibleWhileSiblingExists() throws IOException {
+ Path sourceParent = new Path(contractBasePath(), randomName());
+ Path source = new Path(sourceParent, randomName());
+ Path sibling = new Path(sourceParent, randomName());
+ Path destination = new Path(contractBasePath(), randomName());
+ writeBytes(source, new byte[] {1}, false);
+ writeBytes(sibling, new byte[] {2}, false);
+
+ assertThat(contractFileIO().rename(source, destination)).isTrue();
+
+ assertThat(contractFileIO().getFileStatus(sourceParent).isDir()).isTrue();
+ assertThat(readBytes(sibling)).containsExactly(2);
+ assertThat(contractFileIO().listStatus(sourceParent))
+ .extracting(FileStatus::getPath)
+ .containsExactly(sibling);
+ assertThat(readBytes(destination)).containsExactly(1);
+ }
+
+ @Test
+ void testRenameDirectoryMovesExactTreeToMissingDestination() throws IOException {
+ Path source = new Path(contractBasePath(), randomName());
+ Path child = createRandomFileInDirectory(source, new byte[] {1, 2});
+ Path nestedDirectory = new Path(source, randomName());
+ Path nestedChild = createRandomFileInDirectory(nestedDirectory, new byte[] {3, 4, 5});
+ Path destination = new Path(contractBasePath(), randomName());
+
+ assertThat(contractFileIO().rename(source, destination)).isTrue();
+
+ assertThat(contractFileIO().exists(source)).isFalse();
+ assertThat(contractFileIO().exists(child)).isFalse();
+ assertThat(contractFileIO().exists(nestedDirectory)).isFalse();
+ assertThat(contractFileIO().exists(nestedChild)).isFalse();
+ assertThat(readBytes(new Path(destination, child.getName()))).containsExactly(1, 2);
+ assertThat(
+ readBytes(
+ new Path(
+ new Path(destination, nestedDirectory.getName()),
+ nestedChild.getName())))
+ .containsExactly(3, 4, 5);
+ }
+
+ // ------------------------------------------------------------------------
+ // Copy
+ // ------------------------------------------------------------------------
+
+ @Test
+ void testCopyFileCreatesDestinationWithSourceBytes() throws IOException {
+ byte[] content = new byte[] {6, 2, 6, 4, 3};
+ Path source = createRandomFileInDirectory(contractBasePath(), content);
+ Path destination = new Path(contractBasePath(), randomName());
+
+ contractFileIO().copyFile(source, destination, false);
+
+ assertThat(readBytes(destination)).containsExactly(content);
+ assertThat(readBytes(source)).containsExactly(content);
+ }
+
+ @Test
+ void testCopyFileOverwriteReplacesDestination() throws IOException {
+ byte[] content = new byte[] {7, 7};
+ Path source = createRandomFileInDirectory(contractBasePath(), content);
+ Path destination = createRandomFileInDirectory(contractBasePath(), new byte[] {1, 2, 3, 4});
+
+ contractFileIO().copyFile(source, destination, true);
+
+ assertThat(readBytes(destination)).containsExactly(content);
+ assertThat(readBytes(source)).containsExactly(content);
+ }
+
+ @Test
+ void testCopyFileNoOverwriteFailsAndPreservesDestination() throws IOException {
+ byte[] sourceContent = new byte[] {9, 9};
+ Path source = createRandomFileInDirectory(contractBasePath(), sourceContent);
+ byte[] destinationContent = new byte[] {1, 2, 3};
+ Path destination = createRandomFileInDirectory(contractBasePath(), destinationContent);
+
+ assertThatThrownBy(() -> contractFileIO().copyFile(source, destination, false))
+ .isInstanceOf(IOException.class);
+ assertThat(readBytes(destination)).containsExactly(destinationContent);
+ assertThat(readBytes(source)).containsExactly(sourceContent);
+ }
+
+ @Test
+ void testCopyFilesCopiesEveryDirectFile() throws IOException {
+ Path sourceDirectory = new Path(contractBasePath(), randomName());
+ Path first = createRandomFileInDirectory(sourceDirectory, new byte[] {1, 3});
+ Path second = createRandomFileInDirectory(sourceDirectory, new byte[] {2, 4, 6});
+ Path targetDirectory = new Path(contractBasePath(), randomName());
+ contractFileIO().mkdirs(targetDirectory);
+
+ contractFileIO().copyFiles(sourceDirectory, targetDirectory, false);
+
+ assertThat(readBytes(new Path(targetDirectory, first.getName()))).containsExactly(1, 3);
+ assertThat(readBytes(new Path(targetDirectory, second.getName()))).containsExactly(2, 4, 6);
+ assertThat(readBytes(first)).containsExactly(1, 3);
+ assertThat(readBytes(second)).containsExactly(2, 4, 6);
+ }
+
+ // ------------------------------------------------------------------------
+ // Text and atomic helpers
+ // ------------------------------------------------------------------------
+
+ @Test
+ void testUtf8ReadWriteHelpersPreserveContent() throws IOException {
+ Path file = new Path(contractBasePath(), randomName());
+ String content = "Paimon-文件-IO";
+
+ contractFileIO().writeFile(file, content, false);
+
+ assertThat(contractFileIO().readFileUtf8(file)).isEqualTo(content);
+ assertThat(readBytes(file)).containsExactly(content.getBytes(StandardCharsets.UTF_8));
+ }
+
+ @Test
+ void testOverwriteHelpersReplaceVisibleContent() throws IOException {
+ Path file = new Path(contractBasePath(), randomName());
+ contractFileIO().writeFile(file, "old", false);
+
+ contractFileIO().overwriteFileUtf8(file, "new");
+ assertThat(contractFileIO().readFileUtf8(file)).isEqualTo("new");
+
+ contractFileIO().overwriteHintFile(file, "hint");
+ assertThat(contractFileIO().readFileUtf8(file)).isEqualTo("hint");
+ }
+
+ @Test
+ void testTryToWriteAtomicPublishesMissingTarget() throws IOException {
+ Path target = new Path(contractBasePath(), randomName());
+
+ assertThat(contractFileIO().tryToWriteAtomic(target, "atomic")).isTrue();
+ assertThat(contractFileIO().readFileUtf8(target)).isEqualTo("atomic");
+ }
+
+ @Test
+ void testTryToWriteAtomicPreservesExistingTarget() throws IOException {
+ Path target = new Path(contractBasePath(), randomName());
+ contractFileIO().writeFile(target, "existing", false);
+
+ assertThat(contractFileIO().tryToWriteAtomic(target, "replacement")).isFalse();
+ assertThat(contractFileIO().readFileUtf8(target)).isEqualTo("existing");
+ }
+
+ // ------------------------------------------------------------------------
+ // Two-phase output
+ // ------------------------------------------------------------------------
+
+ @Test
+ void testTwoPhaseOutputPublishesOnlyAfterCommit() throws IOException {
+ Path target = new Path(contractBasePath(), randomName());
+ byte[] content = new byte[] {5, 4, 3, 2, 1};
+ TwoPhaseOutputStream out = contractFileIO().newTwoPhaseOutputStream(target, false);
+ assertThat(out.getPos()).isZero();
+ out.write(content);
+ assertThat(out.getPos()).isEqualTo(content.length);
+ assertThat(contractFileIO().exists(target)).isFalse();
+ TwoPhaseOutputStream.Committer committer = out.closeForCommit();
+
+ assertThat(committer.targetPath()).isEqualTo(target);
+ assertThat(contractFileIO().exists(target)).isFalse();
+
+ committer.commit(contractFileIO());
+ assertThat(readBytes(target)).containsExactly(content);
+ }
+
+ @Test
+ void testTwoPhaseNoOverwritePreservesExistingTarget() throws IOException {
+ Path target = new Path(contractBasePath(), randomName());
+ byte[] existing = new byte[] {9, 8, 7};
+ writeBytes(target, existing, false);
+
+ AtomicReference staged = new AtomicReference<>();
+ try {
+ assertThatThrownBy(
+ () -> {
+ TwoPhaseOutputStream out =
+ contractFileIO().newTwoPhaseOutputStream(target, false);
+ out.write(new byte[] {1, 2, 3});
+ staged.set(out.closeForCommit());
+ staged.get().commit(contractFileIO());
+ })
+ .isInstanceOf(IOException.class);
+ } finally {
+ if (staged.get() != null) {
+ staged.get().discard(contractFileIO());
+ }
+ }
+ assertThat(readBytes(target)).containsExactly(existing);
+ }
+
+ @Test
+ void testTwoPhaseOverwriteReplacesExistingTargetOnCommit() throws IOException {
+ Path target = new Path(contractBasePath(), randomName());
+ writeBytes(target, new byte[] {9, 8, 7}, false);
+ byte[] replacement = new byte[] {1, 2, 3};
+
+ TwoPhaseOutputStream out = contractFileIO().newTwoPhaseOutputStream(target, true);
+ out.write(replacement);
+ out.closeForCommit().commit(contractFileIO());
+
+ assertThat(readBytes(target)).containsExactly(replacement);
+ }
+
+ @Test
+ void testTwoPhaseDiscardDoesNotPublishAbandonedData() throws IOException {
+ Path target = new Path(contractBasePath(), randomName());
+ TwoPhaseOutputStream.Committer committer =
+ stageTwoPhaseOutput(target, new byte[] {1, 2, 3});
+
+ committer.discard(contractFileIO());
+
+ assertThat(contractFileIO().exists(target)).isFalse();
+ }
+
+ @Test
+ void testTwoPhaseDiscardPreservesPreExistingTarget() throws IOException {
+ byte[] oldContent = new byte[] {8, 6, 7, 5};
+ Path target = new Path(contractBasePath(), randomName());
+ TwoPhaseOutputStream.Committer committer =
+ stageTwoPhaseOutput(target, new byte[] {3, 0, 9});
+ assertThat(contractFileIO().exists(target)).isFalse();
+
+ writeBytes(target, oldContent, false);
+ assertThat(readBytes(target)).containsExactly(oldContent);
+
+ committer.discard(contractFileIO());
+
+ assertThat(readBytes(target)).containsExactly(oldContent);
+ }
+
+ @Test
+ void testTwoPhaseDiscardDoesNotAffectAnotherWriter() throws IOException {
+ Path target = new Path(contractBasePath(), randomName());
+ TwoPhaseOutputStream.Committer abandoned =
+ stageTwoPhaseOutput(target, new byte[] {1, 1, 1});
+
+ byte[] committedContent = new byte[] {2, 2, 2};
+ TwoPhaseOutputStream.Committer successful = stageTwoPhaseOutput(target, committedContent);
+
+ abandoned.discard(contractFileIO());
+ successful.commit(contractFileIO());
+
+ assertThat(readBytes(target)).containsExactly(committedContent);
+ }
+
+ @Test
+ void testTwoPhaseCleanPreservesCommittedTarget() throws IOException {
+ byte[] content = new byte[] {2, 7, 1, 8};
+ Path target = new Path(contractBasePath(), randomName());
+ TwoPhaseOutputStream.Committer committer = stageTwoPhaseOutput(target, content);
+ committer.commit(contractFileIO());
+
+ committer.clean(contractFileIO());
+
+ assertThat(readBytes(target)).containsExactly(content);
+ }
+
+ @Test
+ void testTwoPhaseCleanDoesNotAffectAnotherWriter() throws IOException {
+ Path target = new Path(contractBasePath(), randomName());
+ TwoPhaseOutputStream.Committer first = stageTwoPhaseOutput(target, new byte[] {1, 2, 3});
+ byte[] secondContent = new byte[] {4, 5, 6};
+ TwoPhaseOutputStream.Committer second = stageTwoPhaseOutput(target, secondContent);
+
+ first.commit(contractFileIO());
+ assertThat(contractFileIO().delete(target, false)).isTrue();
+ first.clean(contractFileIO());
+ second.commit(contractFileIO());
+
+ assertThat(readBytes(target)).containsExactly(secondContent);
+ }
+
+ @Test
+ void testTwoPhaseCommitterSurvivesSerialization() throws Exception {
+ Path target = new Path(contractBasePath(), randomName());
+ byte[] content = new byte[] {8, 5, 3, 0, 9};
+ TwoPhaseOutputStream.Committer committer = stageTwoPhaseOutput(target, content);
+
+ TwoPhaseOutputStream.Committer restored = InstantiationUtil.clone(committer);
+
+ assertThat(restored.targetPath()).isEqualTo(target);
+ restored.commit(contractFileIO());
+ assertThat(readBytes(target)).containsExactly(content);
+
+ Path discardedTarget = new Path(contractBasePath(), randomName());
+ TwoPhaseOutputStream.Committer discarded =
+ InstantiationUtil.clone(
+ stageTwoPhaseOutput(discardedTarget, new byte[] {1, 4, 1, 4}));
+ discarded.discard(contractFileIO());
+ assertThat(contractFileIO().exists(discardedTarget)).isFalse();
+
+ Path overwrittenTarget = new Path(contractBasePath(), randomName());
+ writeBytes(overwrittenTarget, new byte[] {9, 9, 9}, false);
+ byte[] replacement = new byte[] {2, 6, 5, 3};
+ TwoPhaseOutputStream.Committer overwriting =
+ InstantiationUtil.clone(stageTwoPhaseOutput(overwrittenTarget, replacement, true));
+ overwriting.commit(contractFileIO());
+ assertThat(readBytes(overwrittenTarget)).containsExactly(replacement);
+ }
+
+ private FileIO contractFileIO() throws IOException {
+ initializeContractFixture();
+ return contractFileIO;
+ }
+
+ private Path contractBasePath() throws IOException {
+ initializeContractFixture();
+ return contractBasePath;
+ }
+
+ private void initializeContractFixture() throws IOException {
+ if (contractFileIO != null) {
+ return;
+ }
+
+ try {
+ FileIO fileIO = getFileSystem();
+ Path basePath = new Path(getBasePath(), randomName());
+ fileIO.mkdirs(basePath);
+ contractFileIO = fileIO;
+ contractBasePath = basePath;
+ } catch (IOException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new IOException(e);
+ }
+ }
+
+ private void writeBytes(Path file, byte[] content, boolean overwrite) throws IOException {
+ try (PositionOutputStream out = contractFileIO().newOutputStream(file, overwrite)) {
+ out.write(content);
+ }
+ }
+
+ private TwoPhaseOutputStream.Committer stageTwoPhaseOutput(Path target, byte[] content)
+ throws IOException {
+ return stageTwoPhaseOutput(target, content, false);
+ }
+
+ private TwoPhaseOutputStream.Committer stageTwoPhaseOutput(
+ Path target, byte[] content, boolean overwrite) throws IOException {
+ TwoPhaseOutputStream out = contractFileIO().newTwoPhaseOutputStream(target, overwrite);
+ out.write(content);
+ return out.closeForCommit();
+ }
+
+ private byte[] readBytes(Path file) throws IOException {
+ try (SeekableInputStream in = contractFileIO().newInputStream(file)) {
+ return readAll(in);
+ }
+ }
+
+ private void assertOpenOrFirstReadFails(Path path) throws IOException {
+ final SeekableInputStream in;
+ try {
+ in = contractFileIO().newInputStream(path);
+ } catch (IOException expectedAtOpen) {
+ return;
+ }
+
+ try {
+ assertThatThrownBy(in::read).isInstanceOf(IOException.class);
+ } finally {
+ try {
+ in.close();
+ } catch (IOException ignoredAtClose) {
+ }
+ }
+ }
+
+ private void assertMissingOrEmptyDirectory(Path path) throws IOException {
+ if (!contractFileIO().exists(path)) {
+ return;
+ }
+ assertThat(contractFileIO().getFileStatus(path).isDir()).isTrue();
+ assertThat(contractFileIO().listStatus(path)).isEmpty();
+ }
+
+ private static byte[] readAll(SeekableInputStream in) throws IOException {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ byte[] buffer = new byte[4];
+ int read;
+ while ((read = in.read(buffer, 0, buffer.length)) != -1) {
+ out.write(buffer, 0, read);
+ }
+ return out.toByteArray();
+ }
+
+ private static List collect(RemoteIterator iterator)
+ throws IOException {
+ List statuses = new ArrayList<>();
+ while (iterator.hasNext()) {
+ statuses.add(iterator.next());
+ }
+ return statuses;
+ }
+
+ private static FileStatus statusFor(FileStatus[] statuses, Path path) {
+ for (FileStatus status : statuses) {
+ if (status.getPath().equals(path)) {
+ return status;
+ }
+ }
+ throw new AssertionError("No status for " + path);
+ }
+
+ private Path createRandomFileInDirectory(Path directory) throws IOException {
+ return createRandomFileInDirectory(directory, DEFAULT_CONTENT);
+ }
+
+ private Path createRandomFileInDirectory(Path directory, byte[] content) throws IOException {
+ contractFileIO().mkdirs(directory);
+ Path file = new Path(directory, randomName());
+ writeBytes(file, content, false);
+ return file;
+ }
+}
diff --git a/paimon-common/src/test/java/org/apache/paimon/fs/FileIODefaultMethodTest.java b/paimon-common/src/test/java/org/apache/paimon/fs/FileIODefaultMethodTest.java
new file mode 100644
index 000000000000..014463834b63
--- /dev/null
+++ b/paimon-common/src/test/java/org/apache/paimon/fs/FileIODefaultMethodTest.java
@@ -0,0 +1,487 @@
+/*
+ * 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;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import static org.apache.paimon.fs.RecordingFileIO.Method.DELETE;
+import static org.apache.paimon.fs.RecordingFileIO.Method.EXISTS;
+import static org.apache.paimon.fs.RecordingFileIO.Method.GET_FILE_STATUS;
+import static org.apache.paimon.fs.RecordingFileIO.Method.INPUT_READ;
+import static org.apache.paimon.fs.RecordingFileIO.Method.LIST_STATUS;
+import static org.apache.paimon.fs.RecordingFileIO.Method.MKDIRS;
+import static org.apache.paimon.fs.RecordingFileIO.Method.NEW_INPUT_STREAM;
+import static org.apache.paimon.fs.RecordingFileIO.Method.NEW_OUTPUT_STREAM;
+import static org.apache.paimon.fs.RecordingFileIO.Method.OUTPUT_WRITE;
+import static org.apache.paimon.fs.RecordingFileIO.Method.RENAME;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Contract tests for methods implemented directly on {@link FileIO}. */
+class FileIODefaultMethodTest {
+
+ private RecordingFileIO fileIO;
+ private Path root;
+
+ @BeforeEach
+ void beforeEach() {
+ fileIO = new RecordingFileIO();
+ root = new Path("test:///root");
+ fileIO.putDirectory(root);
+ fileIO.reset();
+ }
+
+ @Test
+ void lifecycleDefaultsAreNoOps() throws Exception {
+ fileIO.setRuntimeContext(Collections.singletonMap("key", "value"));
+ fileIO.close();
+
+ assertThat(fileIO.calls()).isEmpty();
+ }
+
+ @Test
+ void statusHelpersUseOnlyFileStatus() throws Exception {
+ Path file = new Path(root, "file");
+ Path directory = new Path(root, "directory");
+ fileIO.putFile(file, "你好");
+ fileIO.putDirectory(directory);
+ fileIO.reset();
+
+ assertThat(fileIO.getFileSize(file))
+ .isEqualTo("你好".getBytes(StandardCharsets.UTF_8).length);
+ assertThat(fileIO.calls()).containsExactly(RecordingFileIO.call(GET_FILE_STATUS, file));
+
+ fileIO.reset();
+ assertThat(fileIO.isDir(file)).isFalse();
+ assertThat(fileIO.calls()).containsExactly(RecordingFileIO.call(GET_FILE_STATUS, file));
+
+ fileIO.reset();
+ assertThat(fileIO.isDir(directory)).isTrue();
+ assertThat(fileIO.calls())
+ .containsExactly(RecordingFileIO.call(GET_FILE_STATUS, directory));
+
+ fileIO.reset();
+ fileIO.failNext(GET_FILE_STATUS, new IOException("status failed"));
+ assertThatThrownBy(() -> fileIO.getFileSize(file))
+ .isInstanceOf(IOException.class)
+ .hasMessage("status failed");
+
+ fileIO.reset();
+ fileIO.failNext(GET_FILE_STATUS, new IOException("type failed"));
+ assertThatThrownBy(() -> fileIO.isDir(file))
+ .isInstanceOf(IOException.class)
+ .hasMessage("type failed");
+ }
+
+ @Test
+ void checkOrMkdirsAvoidsUnneededMutations() throws Exception {
+ Path existingDirectory = new Path(root, "directory");
+ fileIO.putDirectory(existingDirectory);
+ fileIO.reset();
+
+ fileIO.checkOrMkdirs(existingDirectory);
+
+ assertThat(fileIO.callCount(EXISTS) + fileIO.callCount(GET_FILE_STATUS)).isBetween(1L, 2L);
+ assertThat(fileIO.callCount(MKDIRS)).isZero();
+ assertOnlyCalls(EXISTS, GET_FILE_STATUS);
+
+ Path missing = new Path(root, "created");
+ fileIO.reset();
+ fileIO.checkOrMkdirs(missing);
+ assertThat(fileIO.isDirectoryInMemory(missing)).isTrue();
+ assertThat(fileIO.callCount(EXISTS) + fileIO.callCount(GET_FILE_STATUS))
+ .isLessThanOrEqualTo(1L);
+ assertThat(fileIO.callCount(MKDIRS)).isEqualTo(1);
+ assertOnlyCalls(EXISTS, GET_FILE_STATUS, MKDIRS);
+
+ Path file = new Path(root, "file");
+ fileIO.putFile(file, "content");
+ fileIO.reset();
+ assertThatThrownBy(() -> fileIO.checkOrMkdirs(file))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("should be a directory");
+ assertThat(fileIO.callCount(MKDIRS)).isZero();
+ assertOnlyCalls(EXISTS, GET_FILE_STATUS);
+ }
+
+ @Test
+ void quietDeletesUseTheRequestedRecursionAndSuppressIoFailures() {
+ Path file = new Path(root, "file");
+ fileIO.putFile(file, "content");
+ fileIO.deleteQuietly(file);
+ assertThat(fileIO.existsInMemory(file)).isFalse();
+ assertThat(fileIO.calls(DELETE)).containsExactly(RecordingFileIO.call(DELETE, file, false));
+ assertThat(fileIO.callCount(EXISTS)).isZero();
+
+ Path first = new Path(root, "first");
+ Path second = new Path(root, "second");
+ fileIO.putFile(first, "1");
+ fileIO.putFile(second, "2");
+ fileIO.reset();
+ fileIO.deleteFilesQuietly(Arrays.asList(first, second));
+ assertThat(fileIO.existsInMemory(first)).isFalse();
+ assertThat(fileIO.existsInMemory(second)).isFalse();
+ assertThat(fileIO.calls())
+ .containsExactly(
+ RecordingFileIO.call(DELETE, first, false),
+ RecordingFileIO.call(DELETE, second, false));
+
+ fileIO.putFile(first, "1");
+ fileIO.putFile(second, "2");
+ fileIO.reset();
+ fileIO.failNext(DELETE, new IOException("first failed"));
+ fileIO.deleteFilesQuietly(Arrays.asList(first, second));
+ assertThat(fileIO.existsInMemory(first)).isTrue();
+ assertThat(fileIO.existsInMemory(second)).isFalse();
+ assertThat(fileIO.calls(DELETE))
+ .containsExactly(
+ RecordingFileIO.call(DELETE, first, false),
+ RecordingFileIO.call(DELETE, second, false));
+
+ Path directory = new Path(root, "directory");
+ fileIO.putDirectory(directory);
+ fileIO.reset();
+ fileIO.deleteDirectoryQuietly(directory);
+ assertThat(fileIO.calls(DELETE))
+ .containsExactly(RecordingFileIO.call(DELETE, directory, true));
+
+ Path missing = new Path(root, "missing");
+ fileIO.reset();
+ fileIO.deleteQuietly(missing);
+ assertThat(fileIO.callCount(DELETE)).isEqualTo(1);
+ assertThat(fileIO.callCount(EXISTS)).isLessThanOrEqualTo(1);
+
+ fileIO.putFile(file, "content");
+ fileIO.reset();
+ fileIO.failNext(DELETE, new IOException("planned"));
+ fileIO.deleteQuietly(file);
+ assertThat(fileIO.existsInMemory(file)).isTrue();
+ assertThat(fileIO.callCount(DELETE)).isEqualTo(1);
+ assertThat(fileIO.callCount(EXISTS)).isZero();
+ }
+
+ @Test
+ void utf8HelpersPreserveContentAndForwardOverwriteMode() throws Exception {
+ Path file = new Path(root, "unicode");
+ String content = "Paimon-文件-🙂";
+
+ fileIO.writeFile(file, content, false);
+ assertThat(fileIO.fileContent(file)).isEqualTo(content);
+ assertThat(fileIO.openOutputStreams()).isZero();
+ assertThat(fileIO.calls(NEW_OUTPUT_STREAM))
+ .containsExactly(RecordingFileIO.call(NEW_OUTPUT_STREAM, file, false));
+
+ fileIO.reset();
+ assertThat(fileIO.readFileUtf8(file)).isEqualTo(content);
+ assertThat(fileIO.openInputStreams()).isZero();
+ assertThat(fileIO.calls(NEW_INPUT_STREAM))
+ .containsExactly(RecordingFileIO.call(NEW_INPUT_STREAM, file));
+
+ fileIO.reset();
+ fileIO.overwriteFileUtf8(file, "replacement");
+ assertThat(fileIO.fileContent(file)).isEqualTo("replacement");
+ assertThat(fileIO.calls(NEW_OUTPUT_STREAM))
+ .containsExactly(RecordingFileIO.call(NEW_OUTPUT_STREAM, file, true));
+
+ fileIO.reset();
+ fileIO.overwriteHintFile(file, "hint");
+ assertThat(fileIO.fileContent(file)).isEqualTo("hint");
+ assertThat(fileIO.calls(NEW_OUTPUT_STREAM))
+ .containsExactly(RecordingFileIO.call(NEW_OUTPUT_STREAM, file, true));
+ }
+
+ @Test
+ void listingDefaultsReturnFilesOrDirectoriesWithoutEagerRelisting() throws Exception {
+ Path topFile = new Path(root, "top");
+ Path directory = new Path(root, "directory");
+ Path nestedFile = new Path(directory, "nested");
+ fileIO.putFile(topFile, "top");
+ fileIO.putDirectory(directory);
+ fileIO.putFile(nestedFile, "nested");
+ fileIO.reset();
+
+ assertThat(fileIO.listFiles(root, false))
+ .extracting(FileStatus::getPath)
+ .containsExactlyInAnyOrder(topFile);
+ assertThat(fileIO.callCount(LIST_STATUS)).isEqualTo(1);
+ assertNoMetadataPreflight();
+
+ fileIO.reset();
+ assertThat(fileIO.listFiles(root, true))
+ .extracting(FileStatus::getPath)
+ .containsExactlyInAnyOrder(topFile, nestedFile);
+ assertThat(fileIO.callCount(LIST_STATUS)).isLessThanOrEqualTo(2);
+ assertNoMetadataPreflight();
+
+ fileIO.reset();
+ RemoteIterator iterator = fileIO.listFilesIterative(root, true);
+ assertThat(iterator.hasNext()).isTrue();
+ assertThat(iterator.hasNext()).isTrue();
+ assertThat(collect(iterator))
+ .extracting(FileStatus::getPath)
+ .containsExactlyInAnyOrder(topFile, nestedFile);
+ assertThat(fileIO.callCount(LIST_STATUS)).isLessThanOrEqualTo(2);
+ assertNoMetadataPreflight();
+
+ fileIO.reset();
+ assertThat(fileIO.listDirectories(root))
+ .extracting(FileStatus::getPath)
+ .containsExactlyInAnyOrder(directory);
+ assertThat(fileIO.callCount(LIST_STATUS)).isEqualTo(1);
+ assertNoMetadataPreflight();
+
+ Path empty = new Path(root, "empty");
+ fileIO.putDirectory(empty);
+ fileIO.reset();
+ assertThat(fileIO.listFiles(empty, true)).isEmpty();
+ assertThat(fileIO.listDirectories(empty)).isEmpty();
+
+ fileIO.reset();
+ fileIO.failNext(LIST_STATUS, new IOException("lazy listing failed"));
+ RemoteIterator failing = fileIO.listFilesIterative(root, true);
+ assertThatThrownBy(failing::hasNext)
+ .isInstanceOf(IOException.class)
+ .hasMessage("lazy listing failed");
+ }
+
+ @Test
+ void copyDefaultsTransferBytesAndForwardOverwriteMode() throws Exception {
+ Path source = new Path(root, "source");
+ Path target = new Path(root, "target");
+ fileIO.putFile(source, "source-文件");
+ fileIO.reset();
+
+ fileIO.copyFile(source, target, false);
+
+ assertThat(fileIO.fileContent(target)).isEqualTo("source-文件");
+ assertThat(fileIO.fileContent(source)).isEqualTo("source-文件");
+ assertThat(fileIO.openInputStreams()).isZero();
+ assertThat(fileIO.openOutputStreams()).isZero();
+ assertThat(fileIO.calls(NEW_INPUT_STREAM))
+ .containsExactly(RecordingFileIO.call(NEW_INPUT_STREAM, source));
+ assertThat(fileIO.calls(NEW_OUTPUT_STREAM))
+ .containsExactly(RecordingFileIO.call(NEW_OUTPUT_STREAM, target, false));
+ assertNoCopyPreflights(0);
+
+ Path sourceDirectory = new Path(root, "sources");
+ Path targetDirectory = new Path(root, "targets");
+ Path first = new Path(sourceDirectory, "first");
+ Path second = new Path(sourceDirectory, "second");
+ fileIO.putDirectory(sourceDirectory);
+ fileIO.putDirectory(targetDirectory);
+ fileIO.putFile(first, "1");
+ fileIO.putFile(second, "2");
+ fileIO.reset();
+
+ fileIO.copyFiles(sourceDirectory, targetDirectory, true);
+
+ assertThat(fileIO.fileContent(new Path(targetDirectory, "first"))).isEqualTo("1");
+ assertThat(fileIO.fileContent(new Path(targetDirectory, "second"))).isEqualTo("2");
+ assertThat(fileIO.fileContent(first)).isEqualTo("1");
+ assertThat(fileIO.fileContent(second)).isEqualTo("2");
+ assertThat(fileIO.callCount(LIST_STATUS)).isEqualTo(1);
+ assertThat(fileIO.callCount(NEW_INPUT_STREAM)).isEqualTo(2);
+ assertThat(fileIO.calls(NEW_OUTPUT_STREAM))
+ .containsExactlyInAnyOrder(
+ RecordingFileIO.call(
+ NEW_OUTPUT_STREAM, new Path(targetDirectory, "first"), true),
+ RecordingFileIO.call(
+ NEW_OUTPUT_STREAM, new Path(targetDirectory, "second"), true));
+ assertNoCopyPreflights(1);
+
+ Path failedTarget = new Path(root, "failed-target");
+ fileIO.reset();
+ fileIO.failNext(NEW_OUTPUT_STREAM, new IOException("target open failed"));
+ assertThatThrownBy(() -> fileIO.copyFile(source, failedTarget, false))
+ .isInstanceOf(IOException.class)
+ .hasMessage("target open failed");
+ assertThat(fileIO.openInputStreams()).isZero();
+ assertThat(fileIO.openOutputStreams()).isZero();
+ assertThat(fileIO.existsInMemory(failedTarget)).isFalse();
+ assertThat(fileIO.existsInMemory(source)).isTrue();
+ assertNoCopyPreflights(0);
+
+ fileIO.reset();
+ fileIO.failNext(INPUT_READ, new IOException("source read failed"));
+ assertThatThrownBy(() -> fileIO.copyFile(source, new Path(root, "read-failed"), false))
+ .isInstanceOf(IOException.class)
+ .hasMessage("source read failed");
+ assertThat(fileIO.openInputStreams()).isZero();
+ assertThat(fileIO.openOutputStreams()).isZero();
+ assertNoCopyPreflights(0);
+
+ fileIO.reset();
+ fileIO.failNext(OUTPUT_WRITE, new IOException("target write failed"));
+ assertThatThrownBy(() -> fileIO.copyFile(source, new Path(root, "write-failed"), false))
+ .isInstanceOf(IOException.class)
+ .hasMessage("target write failed");
+ assertThat(fileIO.openInputStreams()).isZero();
+ assertThat(fileIO.openOutputStreams()).isZero();
+ assertNoCopyPreflights(0);
+ }
+
+ @Test
+ void tryToWriteAtomicPublishesOrCleansUpTheTemporaryFile() throws Exception {
+ Path target = new Path(root, "atomic");
+
+ assertThat(fileIO.tryToWriteAtomic(target, "new")).isTrue();
+ assertThat(fileIO.fileContent(target)).isEqualTo("new");
+ assertThat(fileIO.callCount(NEW_OUTPUT_STREAM)).isEqualTo(1);
+ assertThat(fileIO.callCount(RENAME)).isEqualTo(1);
+ assertThat(fileIO.callCount(DELETE)).isZero();
+ assertThat(fileIO.callCount(EXISTS)).isZero();
+ assertThat(fileIO.callCount(GET_FILE_STATUS)).isZero();
+ assertThat(fileIO.callCount(LIST_STATUS)).isZero();
+ assertAtomicTemporaryPath(target, fileIO.calls(RENAME).get(0));
+
+ Path failedTarget = new Path(root, "failed-atomic");
+ fileIO.reset();
+ fileIO.failNext(RENAME, new IOException("rename failed"));
+ assertThatThrownBy(() -> fileIO.tryToWriteAtomic(failedTarget, "content"))
+ .isInstanceOf(IOException.class)
+ .hasMessage("rename failed");
+ RecordingFileIO.Call failedRename = fileIO.calls(RENAME).get(0);
+ Path failedTemporary = failedRename.argument(0, Path.class);
+ assertThat(fileIO.existsInMemory(failedTemporary)).isFalse();
+ assertThat(fileIO.existsInMemory(failedTarget)).isFalse();
+ assertThat(fileIO.callCount(DELETE)).isEqualTo(1);
+ assertThat(fileIO.callCount(LIST_STATUS)).isZero();
+
+ fileIO.putFile(target, "existing");
+ fileIO.reset();
+ assertThat(fileIO.tryToWriteAtomic(target, "replacement")).isFalse();
+ assertThat(fileIO.fileContent(target)).isEqualTo("existing");
+ assertThat(fileIO.callCount(NEW_OUTPUT_STREAM)).isEqualTo(1);
+ assertThat(fileIO.callCount(RENAME)).isEqualTo(1);
+ assertThat(fileIO.callCount(DELETE)).isEqualTo(1);
+ assertThat(fileIO.callCount(EXISTS)).isZero();
+ assertThat(fileIO.callCount(LIST_STATUS)).isZero();
+ assertAtomicTemporaryPath(target, fileIO.calls(RENAME).get(0));
+ }
+
+ @Test
+ void defaultTwoPhaseOutputStagesThenPublishesOnCommit() throws Exception {
+ Path target = new Path(root, "two-phase");
+ String content = "staged-文件";
+
+ TwoPhaseOutputStream stream = fileIO.newTwoPhaseOutputStream(target, false);
+ stream.write(content.getBytes(StandardCharsets.UTF_8));
+ TwoPhaseOutputStream.Committer committer = stream.closeForCommit();
+
+ assertThat(fileIO.existsInMemory(target)).isFalse();
+ assertThat(committer.targetPath()).isEqualTo(target);
+ committer.commit(fileIO);
+ assertThat(fileIO.fileContent(target)).isEqualTo(content);
+ assertThat(fileIO.openOutputStreams()).isZero();
+ }
+
+ @Test
+ void overwrittenReadReturnsEmptyForMissingFilesWithoutExistenceProbe() throws Exception {
+ Path missing = new Path(root, "missing");
+
+ assertThat(fileIO.readOverwrittenFileUtf8(missing)).isEmpty();
+ assertThat(fileIO.callCount(NEW_INPUT_STREAM)).isEqualTo(1);
+ assertThat(fileIO.callCount(EXISTS)).isZero();
+
+ Path disappeared = new Path(root, "disappeared");
+ fileIO.reset();
+ fileIO.failNext(NEW_INPUT_STREAM, new IOException("transient"));
+ assertThat(fileIO.readOverwrittenFileUtf8(disappeared)).isEmpty();
+ assertThat(fileIO.callCount(EXISTS)).isLessThanOrEqualTo(1);
+ }
+
+ @Test
+ void overwrittenReadRetriesOnlyKnownConcurrentChangeFailures() throws Exception {
+ Path file = new Path(root, "overwritten");
+ fileIO.putFile(file, "stable");
+ fileIO.failNext(NEW_INPUT_STREAM, blocklistChanged());
+ fileIO.failNext(NEW_INPUT_STREAM, blocklistChanged());
+
+ assertThat(fileIO.readOverwrittenFileUtf8(file)).contains("stable");
+ assertThat(fileIO.callCount(NEW_INPUT_STREAM)).isEqualTo(3);
+ assertThat(fileIO.callCount(EXISTS)).isLessThanOrEqualTo(2);
+
+ fileIO.reset();
+ fileIO.failNext(NEW_INPUT_STREAM, new IOException("unrelated"));
+ assertThatThrownBy(() -> fileIO.readOverwrittenFileUtf8(file))
+ .isInstanceOf(IOException.class)
+ .hasMessage("unrelated");
+ assertThat(fileIO.callCount(NEW_INPUT_STREAM)).isEqualTo(1);
+ assertThat(fileIO.callCount(EXISTS)).isLessThanOrEqualTo(1);
+ }
+
+ @Test
+ void overwrittenReadStopsAfterFiveKnownFailures() {
+ Path file = new Path(root, "overwritten");
+ fileIO.putFile(file, "stable");
+ for (int i = 0; i < 5; i++) {
+ fileIO.failNext(NEW_INPUT_STREAM, blocklistChanged());
+ }
+
+ assertThatThrownBy(() -> fileIO.readOverwrittenFileUtf8(file))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("Blocklist for");
+ assertThat(fileIO.callCount(NEW_INPUT_STREAM)).isEqualTo(5);
+ assertThat(fileIO.callCount(EXISTS)).isLessThanOrEqualTo(5);
+ }
+
+ private static IOException blocklistChanged() {
+ return new IOException("Blocklist for test has changed");
+ }
+
+ private void assertNoMetadataPreflight() {
+ assertThat(fileIO.callCount(GET_FILE_STATUS)).isZero();
+ assertThat(fileIO.callCount(EXISTS)).isZero();
+ }
+
+ private void assertNoCopyPreflights(long expectedListCalls) {
+ assertNoMetadataPreflight();
+ assertThat(fileIO.callCount(DELETE)).isZero();
+ assertThat(fileIO.callCount(LIST_STATUS)).isEqualTo(expectedListCalls);
+ }
+
+ private void assertOnlyCalls(RecordingFileIO.Method... allowedMethods) {
+ List allowed = Arrays.asList(allowedMethods);
+ assertThat(fileIO.calls()).allMatch(call -> allowed.contains(call.method()));
+ }
+
+ private static List collect(RemoteIterator iterator)
+ throws IOException {
+ List result = new ArrayList<>();
+ while (iterator.hasNext()) {
+ result.add(iterator.next());
+ }
+ return result;
+ }
+
+ private static void assertAtomicTemporaryPath(Path target, RecordingFileIO.Call renameCall) {
+ Path temporary = renameCall.argument(0, Path.class);
+ assertThat(renameCall.argument(1, Path.class)).isEqualTo(target);
+ assertThat(temporary.getParent()).isEqualTo(target.getParent());
+ assertThat(temporary.getName()).startsWith("." + target.getName() + ".").endsWith(".tmp");
+ }
+}
diff --git a/paimon-common/src/test/java/org/apache/paimon/fs/FileIOReturnTypeTest.java b/paimon-common/src/test/java/org/apache/paimon/fs/FileIOReturnTypeTest.java
new file mode 100644
index 000000000000..0ffc4015d9ec
--- /dev/null
+++ b/paimon-common/src/test/java/org/apache/paimon/fs/FileIOReturnTypeTest.java
@@ -0,0 +1,99 @@
+/*
+ * 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;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.same;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/** Tests default return values and adapters in the public file I/O API. */
+public class FileIOReturnTypeTest {
+
+ @Test
+ public void testFileStatusDefaultValues() {
+ FileStatus status =
+ new FileStatus() {
+ @Override
+ public long getLen() {
+ return 0;
+ }
+
+ @Override
+ public boolean isDir() {
+ return false;
+ }
+
+ @Override
+ public Path getPath() {
+ return new Path("file:/status");
+ }
+
+ @Override
+ public long getModificationTime() {
+ return 0;
+ }
+ };
+
+ assertThat(status.getAccessTime()).isZero();
+ assertThat(status.getOwner()).isNull();
+ }
+
+ @Test
+ public void testSeekableInputStreamWrapForwardsReadAndClose() throws IOException {
+ InputStream input = mock(InputStream.class);
+ byte[] buffer = new byte[4];
+ when(input.read()).thenReturn(17);
+ when(input.read(buffer, 1, 2)).thenReturn(2);
+
+ SeekableInputStream wrapped = SeekableInputStream.wrap(input);
+ assertThat(wrapped.read()).isEqualTo(17);
+ assertThat(wrapped.read(buffer, 1, 2)).isEqualTo(2);
+ wrapped.close();
+
+ verify(input).read();
+ verify(input).read(same(buffer), eq(1), eq(2));
+ verify(input).close();
+ }
+
+ @Test
+ public void testSeekableInputStreamWrapRejectsSeek() {
+ SeekableInputStream wrapped =
+ SeekableInputStream.wrap(new ByteArrayInputStream(new byte[0]));
+
+ assertThatThrownBy(() -> wrapped.seek(0)).isInstanceOf(UnsupportedOperationException.class);
+ }
+
+ @Test
+ public void testSeekableInputStreamWrapRejectsGetPos() {
+ SeekableInputStream wrapped =
+ SeekableInputStream.wrap(new ByteArrayInputStream(new byte[0]));
+
+ assertThatThrownBy(wrapped::getPos).isInstanceOf(UnsupportedOperationException.class);
+ }
+}
diff --git a/paimon-common/src/test/java/org/apache/paimon/fs/FileIOTest.java b/paimon-common/src/test/java/org/apache/paimon/fs/FileIOTest.java
index 96e023d1eb46..ada55a09c153 100644
--- a/paimon-common/src/test/java/org/apache/paimon/fs/FileIOTest.java
+++ b/paimon-common/src/test/java/org/apache/paimon/fs/FileIOTest.java
@@ -19,12 +19,15 @@
package org.apache.paimon.fs;
import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.data.BlobDescriptor;
import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.options.CatalogOptions;
import org.apache.paimon.options.Options;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import org.mockito.MockedStatic;
import java.io.File;
import java.io.FileNotFoundException;
@@ -37,14 +40,20 @@
import java.nio.file.StandardCopyOption;
import java.time.Duration;
import java.util.Arrays;
+import java.util.Collections;
import java.util.Comparator;
+import java.util.List;
import java.util.Optional;
+import java.util.ServiceLoader;
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;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.when;
/** Test static methods and methods with default implementations of {@link FileIO}. */
public class FileIOTest {
@@ -74,6 +83,96 @@ public void testRequireOptions() throws IOException {
assertThat(fileIO).isInstanceOf(RequireOptionsFileIOLoader.MyFileIO.class);
}
+ @Test
+ public void testGetSchemelessPathUsesLocalFileIO() throws IOException {
+ Path path = new Path(tempDir.resolve("local").toString());
+
+ FileIO fileIO = FileIO.get(path, CatalogContext.create(new Options()));
+
+ assertThat(fileIO).isInstanceOf(LocalFileIO.class);
+ }
+
+ @Test
+ public void testGetUsesResolvingFileIOWhenEnabled() throws IOException {
+ Options options = new Options();
+ options.set(CatalogOptions.RESOLVING_FILE_IO_ENABLED, true);
+ Path path = new Path(tempDir.resolve("resolving").toUri());
+
+ FileIO fileIO = FileIO.get(path, CatalogContext.create(options));
+
+ assertThat(fileIO).isInstanceOf(ResolvingFileIO.class);
+ fileIO.writeFile(path, "configured", false);
+ assertThat(fileIO.readFileUtf8(path)).isEqualTo("configured");
+ }
+
+ @Test
+ public void testGetRejectsLocalPathWithAuthority() {
+ Path malformed = new Path("file://host/tmp/table");
+
+ assertThatThrownBy(() -> FileIO.get(malformed, CatalogContext.create(new Options())))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("authority 'host'")
+ .hasMessageContaining("file:///host/tmp/table");
+ }
+
+ @Test
+ public void testDiscoverLoadersIncludesTestService() {
+ assertThat(FileIO.discoverLoaders().get("require-options"))
+ .isInstanceOf(RequireOptionsFileIOLoader.class);
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testDiscoverLoadersRejectsDuplicateSchemes() {
+ FileIOLoader first = new TrackingLoader("duplicate");
+ FileIOLoader second = new TrackingLoader("duplicate");
+ ServiceLoader services = mock(ServiceLoader.class);
+ when(services.iterator()).thenReturn(Arrays.asList(first, second).iterator());
+
+ try (MockedStatic serviceLoader = mockStatic(ServiceLoader.class)) {
+ serviceLoader
+ .when(
+ () ->
+ ServiceLoader.load(
+ FileIOLoader.class,
+ FileIOLoader.class.getClassLoader()))
+ .thenReturn(services);
+
+ assertThatThrownBy(FileIO::discoverLoaders)
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("Multiple FileIO for scheme 'duplicate'");
+ }
+ }
+
+ @Test
+ public void testPreferredLoaderWithMissingOptionsFallsBackToAccessibleLoader()
+ throws IOException {
+ TrackingLoader preferred = new TrackingLoader("preferred", "required-by-preferred");
+ TrackingLoader fallback = new TrackingLoader("fallback");
+ Path path = new Path("unregistered:///warehouse");
+
+ TrackingLocalFileIO selected =
+ (TrackingLocalFileIO)
+ FileIO.get(path, CatalogContext.create(new Options(), preferred, fallback));
+
+ assertThat(selected.owner).isEqualTo("fallback");
+ assertThat(selected.configured).isTrue();
+ }
+
+ @Test
+ public void testAccessiblePreferredLoaderIsSelectedBeforeFallback() throws IOException {
+ TrackingLoader preferred = new TrackingLoader("preferred");
+ TrackingLoader fallback = new TrackingLoader("fallback");
+ Path path = new Path("unregistered:///warehouse");
+
+ TrackingLocalFileIO selected =
+ (TrackingLocalFileIO)
+ FileIO.get(path, CatalogContext.create(new Options(), preferred, fallback));
+
+ assertThat(selected.owner).isEqualTo("preferred");
+ assertThat(selected.configured).isTrue();
+ }
+
@Test
public void testCopy() throws Exception {
Path srcFile = new Path(tempDir.resolve("src.txt").toUri());
@@ -196,6 +295,73 @@ public void testDefaultArchiveUnsupported() {
.isInstanceOf(UnsupportedOperationException.class)
.hasMessageContaining(DummyFileIO.class.getName())
.hasMessageContaining("unarchive");
+
+ BlobDescriptor descriptor = new BlobDescriptor(path.toString(), 0, 1);
+ assertThatThrownBy(
+ () ->
+ fileIO.createBlobPresignedUrl(
+ new Path(tempDir.toUri()),
+ descriptor,
+ Duration.ofMinutes(5)))
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessageContaining(DummyFileIO.class.getName())
+ .hasMessageContaining("presigned");
+ }
+
+ private static class TrackingLoader implements FileIOLoader {
+
+ private static final long serialVersionUID = 1L;
+
+ private final String scheme;
+ private final String requiredOption;
+
+ private TrackingLoader(String scheme) {
+ this(scheme, null);
+ }
+
+ private TrackingLoader(String scheme, String requiredOption) {
+ this.scheme = scheme;
+ this.requiredOption = requiredOption;
+ }
+
+ @Override
+ public String getScheme() {
+ return scheme;
+ }
+
+ @Override
+ public List requiredOptions() {
+ return requiredOption == null
+ ? Collections.emptyList()
+ : Collections.singletonList(new String[] {requiredOption});
+ }
+
+ @Override
+ public FileIO load(Path path) {
+ return new TrackingLocalFileIO(scheme);
+ }
+ }
+
+ private static class TrackingLocalFileIO extends LocalFileIO {
+
+ private static final long serialVersionUID = 1L;
+
+ private final String owner;
+ private boolean configured;
+
+ private TrackingLocalFileIO(String owner) {
+ this.owner = owner;
+ }
+
+ @Override
+ public void configure(CatalogContext context) {
+ configured = true;
+ }
+
+ @Override
+ public boolean exists(Path path) {
+ return true;
+ }
}
/** A {@link FileIO} on local filesystem to test various default implementations. */
diff --git a/paimon-common/src/test/java/org/apache/paimon/fs/HadoopLocalFileIOBehaviorTest.java b/paimon-common/src/test/java/org/apache/paimon/fs/HadoopLocalFileIOBehaviorTest.java
index 6f6ed70b1ad7..14f4887ee279 100644
--- a/paimon-common/src/test/java/org/apache/paimon/fs/HadoopLocalFileIOBehaviorTest.java
+++ b/paimon-common/src/test/java/org/apache/paimon/fs/HadoopLocalFileIOBehaviorTest.java
@@ -23,14 +23,16 @@
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.RawLocalFileSystem;
import org.apache.hadoop.util.VersionInfo;
+import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import java.net.URI;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assumptions.assumeThat;
/** Behavior tests for Hadoop Local. */
-class HadoopLocalFileIOBehaviorTest extends FileIOBehaviorTestBase {
+class HadoopLocalFileIOBehaviorTest extends FileIOContractTestBase {
@TempDir private java.nio.file.Path tmp;
@@ -48,6 +50,11 @@ protected Path getBasePath() {
return new Path(tmp.toUri());
}
+ @Test
+ void testIsNotObjectStore() throws Exception {
+ assertThat(getFileSystem().isObjectStore()).isFalse();
+ }
+
// ------------------------------------------------------------------------
/** This test needs to be skipped for earlier Hadoop versions because those have a bug. */
diff --git a/paimon-common/src/test/java/org/apache/paimon/fs/HdfsBehaviorTest.java b/paimon-common/src/test/java/org/apache/paimon/fs/HdfsBehaviorTest.java
index b719038caa07..0a5479a8de89 100644
--- a/paimon-common/src/test/java/org/apache/paimon/fs/HdfsBehaviorTest.java
+++ b/paimon-common/src/test/java/org/apache/paimon/fs/HdfsBehaviorTest.java
@@ -38,7 +38,7 @@
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
/** Behavior tests for HDFS. */
-class HdfsBehaviorTest extends FileIOBehaviorTestBase {
+class HdfsBehaviorTest extends FileIOContractTestBase {
private static MiniDFSCluster hdfsCluster;
diff --git a/paimon-common/src/test/java/org/apache/paimon/fs/LocalFileIOBehaviorTest.java b/paimon-common/src/test/java/org/apache/paimon/fs/LocalFileIOBehaviorTest.java
index 2477bdcdbad1..209921fe3f09 100644
--- a/paimon-common/src/test/java/org/apache/paimon/fs/LocalFileIOBehaviorTest.java
+++ b/paimon-common/src/test/java/org/apache/paimon/fs/LocalFileIOBehaviorTest.java
@@ -20,10 +20,13 @@
import org.apache.paimon.fs.local.LocalFileIO;
+import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import static org.assertj.core.api.Assertions.assertThat;
+
/** Test for {@link LocalFileIO}. */
-public class LocalFileIOBehaviorTest extends FileIOBehaviorTestBase {
+public class LocalFileIOBehaviorTest extends FileIOContractTestBase {
@TempDir private java.nio.file.Path tmp;
@@ -36,4 +39,9 @@ protected FileIO getFileSystem() {
protected Path getBasePath() {
return new Path(tmp.toUri());
}
+
+ @Test
+ void testIsNotObjectStore() {
+ assertThat(new LocalFileIO().isObjectStore()).isFalse();
+ }
}
diff --git a/paimon-common/src/test/java/org/apache/paimon/fs/RecordingFileIO.java b/paimon-common/src/test/java/org/apache/paimon/fs/RecordingFileIO.java
new file mode 100644
index 000000000000..0ebeeb1b7fdf
--- /dev/null
+++ b/paimon-common/src/test/java/org/apache/paimon/fs/RecordingFileIO.java
@@ -0,0 +1,427 @@
+/*
+ * 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;
+
+import org.apache.paimon.catalog.CatalogContext;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.FileAlreadyExistsException;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.Deque;
+import java.util.EnumMap;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/** Deterministic in-memory implementation of the primitive {@link FileIO} operations. */
+final class RecordingFileIO implements FileIO {
+
+ enum Method {
+ GET_FILE_STATUS,
+ LIST_STATUS,
+ EXISTS,
+ DELETE,
+ MKDIRS,
+ RENAME,
+ NEW_INPUT_STREAM,
+ NEW_OUTPUT_STREAM,
+ INPUT_READ,
+ OUTPUT_WRITE
+ }
+
+ static final class Call {
+ private final Method method;
+ private final List