[core] Introduce 'file-io.atomic-rename.enabled' for atomic rename control

docs/layouts/shortcodes/generated/catalog_configuration.html

@@ -98,6 +98,12 @@
             <td>Boolean</td>
             <td>Whether to allow static cache in file io implementation. If not allowed, this means that there may be a large number of FileIO instances generated, enabling caching can lead to resource leakage.</td>
         </tr>
+        <tr>
+            <td><h5>file-io.atomic-rename.enabled</h5></td>
+            <td style="word-wrap: break-word;">true</td>
+            <td>Boolean</td>
+            <td>Whether to enable atomic rename for file overwrite operations. When enabled, Paimon attempts atomic rename via temp file, supported on HDFS (DistributedFileSystem, ViewFileSystem). Falls back to direct overwrite on object stores like S3/OSS. Set to false to skip atomic rename and avoid reflection/temp file overhead.</td>
+        </tr>
         <tr>
             <td><h5>format-table.enabled</h5></td>
             <td style="word-wrap: break-word;">true</td>

paimon-api/src/main/java/org/apache/paimon/options/CatalogOptions.java

@@ -184,4 +184,14 @@ public class CatalogOptions {
                             "Whether to allow static cache in file io implementation. If not allowed, this means that "
                                     + "there may be a large number of FileIO instances generated, enabling caching can "
                                     + "lead to resource leakage.");
+
+    public static final ConfigOption<Boolean> FILE_IO_ATOMIC_RENAME_ENABLED =
+            ConfigOptions.key("file-io.atomic-rename.enabled")
+                    .booleanType()
+                    .defaultValue(true)
+                    .withDescription(
+                            "Whether to enable atomic rename for file overwrite operations. "
+                                    + "When enabled, Paimon attempts atomic rename via temp file, supported on HDFS (DistributedFileSystem, ViewFileSystem). "
+                                    + "Falls back to direct overwrite on object stores like S3/OSS. "
+                                    + "Set to false to skip atomic rename and avoid reflection/temp file overhead.");
 }

paimon-common/src/main/java/org/apache/paimon/fs/hadoop/HadoopFileIO.java

@@ -27,6 +27,7 @@
 import org.apache.paimon.fs.RemoteIterator;
 import org.apache.paimon.fs.SeekableInputStream;
 import org.apache.paimon.hadoop.SerializableConfiguration;
+import org.apache.paimon.options.CatalogOptions;
 import org.apache.paimon.utils.FileIOUtils;
 import org.apache.paimon.utils.FunctionWithException;
 import org.apache.paimon.utils.Pair;
@@ -58,6 +59,8 @@ public class HadoopFileIO implements FileIO {
 
     private org.apache.paimon.options.Options options;
 
+    private boolean atomicRenameEnabled = true;
+
     protected transient volatile Map<Pair<String, String>, FileSystem> fsMap;
 
     private final Path path;
@@ -81,6 +84,7 @@ public boolean isObjectStore() {
     public void configure(CatalogContext context) {
         this.hadoopConf = new SerializableConfiguration(context.hadoopConf());
         this.options = context.options();
+        this.atomicRenameEnabled = options.get(CatalogOptions.FILE_IO_ATOMIC_RENAME_ENABLED);
     }
 
     public Configuration hadoopConf() {
@@ -168,8 +172,12 @@ public boolean rename(Path src, Path dst) throws IOException {
 
     @Override
     public void overwriteFileUtf8(Path path, String content) throws IOException {
-        boolean success = tryAtomicOverwriteViaRename(path, content);
-        if (!success) {
+        if (atomicRenameEnabled) {
+            boolean success = tryAtomicOverwriteViaRename(path, content);
+            if (!success) {
+                FileIO.super.overwriteFileUtf8(path, content);
+            }
+        } else {
             FileIO.super.overwriteFileUtf8(path, content);
         }
     }