Changes for supporting Atomic writes with Hadoop 3.4+

Author: junmuz

paimon-filesystems/paimon-azure-impl/src/main/java/org/apache/paimon/azure/AzureFileIO.java

@@ -20,9 +20,13 @@
 
 import org.apache.paimon.catalog.CatalogContext;
 import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
 import org.apache.paimon.options.Options;
 
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FSDataOutputStreamBuilder;
+import org.apache.hadoop.fs.FileAlreadyExistsException;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.azure.NativeAzureFileSystem;
 import org.apache.hadoop.fs.azurebfs.AzureBlobFileSystem;
@@ -32,6 +36,7 @@
 import java.io.IOException;
 import java.io.UncheckedIOException;
 import java.net.URI;
+import java.nio.charset.StandardCharsets;
 import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.ConcurrentHashMap;
@@ -66,6 +71,33 @@ public void configure(CatalogContext context) {
         this.hadoopOptions = mirrorCertainHadoopConfig(loadHadoopConfigFromContext(context));
     }
 
+    /**
+     * Write content atomically using Azure conditional writes.
+     *
+     * @param path the target file path
+     * @param content the content to write
+     * @return true if write succeeded, false if file already exists
+     * @throws IOException on I/O errors
+     */
+    @Override
+    public boolean tryToWriteAtomic(Path path, String content) throws IOException {
+        org.apache.hadoop.fs.Path hadoopPath = path(path);
+        FileSystem fs = getFileSystem(hadoopPath);
+
+        byte[] contentBytes = content.getBytes(StandardCharsets.UTF_8);
+
+        FSDataOutputStreamBuilder builder = fs.createFile(hadoopPath);
+        builder.must("fs.option.create.conditional.overwrite", true);
+
+        try (FSDataOutputStream out = builder.create().overwrite(false).build()) {
+            out.write(contentBytes);
+            return true;
+        } catch (FileAlreadyExistsException e) {
+            LOG.debug("Conditional write failed, file already exists: {}", path);
+            return false;
+        }
+    }
+
     // add additional config entries from the IO config to the Hadoop config
     private Options loadHadoopConfigFromContext(CatalogContext context) {
         Options hadoopConfig = new Options();

paimon-filesystems/paimon-azure-impl/src/main/java/org/apache/paimon/azure/HadoopCompliantFileIO.java

@@ -102,11 +102,11 @@ public boolean rename(Path src, Path dst) throws IOException {
         return getFileSystem(hadoopSrc).rename(hadoopSrc, hadoopDst);
     }
 
-    private org.apache.hadoop.fs.Path path(Path path) {
+    protected org.apache.hadoop.fs.Path path(Path path) {
         return new org.apache.hadoop.fs.Path(path.toUri());
     }
 
-    private FileSystem getFileSystem(org.apache.hadoop.fs.Path path) throws IOException {
+    protected FileSystem getFileSystem(org.apache.hadoop.fs.Path path) throws IOException {
         if (fs == null) {
             synchronized (this) {
                 if (fs == null) {

paimon-filesystems/paimon-s3-impl/src/main/java/org/apache/paimon/s3/S3FileIO.java

@@ -25,6 +25,9 @@
 import org.apache.paimon.options.Options;
 
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FSDataOutputStreamBuilder;
+import org.apache.hadoop.fs.FileAlreadyExistsException;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.s3a.S3AFileSystem;
 import org.slf4j.Logger;
@@ -33,6 +36,7 @@
 import java.io.IOException;
 import java.io.UncheckedIOException;
 import java.net.URI;
+import java.nio.charset.StandardCharsets;
 import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.ConcurrentHashMap;
@@ -85,6 +89,33 @@ public TwoPhaseOutputStream newTwoPhaseOutputStream(Path path, boolean overwrite
                 new S3MultiPartUpload(fs, fs.getConf()), hadoopPath, path);
     }
 
+    /**
+     * Write content atomically using S3 conditional writes via Hadoop 3.4+ native API.
+     *
+     * @param path the target file path
+     * @param content the content to write
+     * @return true if write succeeded, false if file already exists
+     * @throws IOException on I/O errors
+     */
+    @Override
+    public boolean tryToWriteAtomic(Path path, String content) throws IOException {
+        org.apache.hadoop.fs.Path hadoopPath = path(path);
+        S3AFileSystem fs = (S3AFileSystem) getFileSystem(hadoopPath);
+
+        byte[] contentBytes = content.getBytes(StandardCharsets.UTF_8);
+
+        FSDataOutputStreamBuilder builder = fs.createFile(hadoopPath);
+        builder.must("fs.option.create.conditional.overwrite", true);
+
+        try (FSDataOutputStream out = builder.create().overwrite(false).build()) {
+            out.write(contentBytes);
+            return true;
+        } catch (FileAlreadyExistsException e) {
+            LOG.debug("Conditional write failed, file already exists: {}", path);
+            return false;
+        }
+    }
+
     // add additional config entries from the IO config to the Hadoop config
     private Options loadHadoopConfigFromContext(CatalogContext context) {
         Options hadoopConfig = new Options();