[core] Make HadoopFileIO determine isObjectStore based on the path (#6393)

Zouxxyy · web-flow · commit ccd33b1701b6 · 2025-10-14T10:28:11.000+02:00
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
@@ -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.utils.FileIOUtils;
 import org.apache.paimon.utils.FunctionWithException;
 import org.apache.paimon.utils.Pair;
 import org.apache.paimon.utils.ReflectionUtils;
@@ -43,6 +44,7 @@
 import java.lang.reflect.Method;
 import java.net.URI;
 import java.nio.charset.StandardCharsets;
+import java.util.Locale;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicReference;
@@ -58,15 +60,21 @@ public class HadoopFileIO implements FileIO {
 
     protected transient volatile Map<Pair<String, String>, FileSystem> fsMap;
 
+    private final Path path;
+
+    public HadoopFileIO(Path path) {
+        this.path = path;
+    }
+
     @VisibleForTesting
-    public void setFileSystem(Path path, FileSystem fs) throws IOException {
-        org.apache.hadoop.fs.Path hadoopPath = path(path);
-        getFileSystem(hadoopPath, p -> fs);
+    public void setFileSystem(FileSystem fs) throws IOException {
+        getFileSystem(path(path), p -> fs);
     }
 
     @Override
     public boolean isObjectStore() {
-        return false;
+        String scheme = path.toUri().getScheme().toLowerCase(Locale.US);
+        return FileIOUtils.isObjectStore(scheme);
     }
 
     @Override
diff --git a/paimon-common/src/main/java/org/apache/paimon/fs/hadoop/HadoopFileIOLoader.java b/paimon-common/src/main/java/org/apache/paimon/fs/hadoop/HadoopFileIOLoader.java
@@ -33,6 +33,6 @@ public String getScheme() {
 
     @Override
     public HadoopFileIO load(Path path) {
-        return new HadoopFileIO();
+        return new HadoopFileIO(path);
     }
 }
diff --git a/paimon-common/src/main/java/org/apache/paimon/fs/hadoop/HadoopViewFsFileIOLoader.java b/paimon-common/src/main/java/org/apache/paimon/fs/hadoop/HadoopViewFsFileIOLoader.java
@@ -33,6 +33,6 @@ public String getScheme() {
 
     @Override
     public HadoopFileIO load(Path path) {
-        return new HadoopFileIO();
+        return new HadoopFileIO(path);
     }
 }
diff --git a/paimon-common/src/main/java/org/apache/paimon/utils/FileIOUtils.java b/paimon-common/src/main/java/org/apache/paimon/utils/FileIOUtils.java
@@ -382,4 +382,27 @@ public static void deleteDirectoryQuietly(File directory) {
         } catch (Exception ignored) {
         }
     }
+
+    public static boolean isObjectStore(String scheme) {
+        if (scheme.startsWith("s3")
+                || scheme.startsWith("emr")
+                || scheme.startsWith("oss")
+                || scheme.startsWith("wasb")
+                || scheme.startsWith("abfs")
+                || scheme.startsWith("gs")
+                || scheme.startsWith("cosn")) {
+            // the Amazon S3 storage or Aliyun OSS storage or Azure Blob Storage
+            // or Google Cloud Storage
+            return true;
+        } else if (scheme.startsWith("http") || scheme.startsWith("ftp")) {
+            // file servers instead of file systems
+            // they might actually be consistent, but we have no hard guarantees
+            // currently to rely on that
+            return true;
+        } else {
+            // the remainder should include hdfs, kosmos, ceph, ...
+            // this also includes federated HDFS (viewfs).
+            return false;
+        }
+    }
 }
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
@@ -38,8 +38,8 @@ class HadoopLocalFileIOBehaviorTest extends FileIOBehaviorTestBase {
     protected FileIO getFileSystem() throws Exception {
         org.apache.hadoop.fs.FileSystem fs = new RawLocalFileSystem();
         fs.initialize(URI.create("file:///"), new Configuration());
-        HadoopFileIO fileIO = new HadoopFileIO();
-        fileIO.setFileSystem(getBasePath(), fs);
+        HadoopFileIO fileIO = new HadoopFileIO(getBasePath());
+        fileIO.setFileSystem(fs);
         return fileIO;
     }
 
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
@@ -28,11 +28,14 @@
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
 
+import java.io.ByteArrayOutputStream;
 import java.io.File;
 import java.io.IOException;
+import java.io.ObjectOutputStream;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assumptions.assumeThat;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 
 /** Behavior tests for HDFS. */
 class HdfsBehaviorTest extends FileIOBehaviorTestBase {
@@ -62,8 +65,8 @@ static void createHDFS(@TempDir File tmp) throws Exception {
         org.apache.hadoop.fs.FileSystem hdfs = hdfsCluster.getFileSystem();
 
         basePath = new Path(hdfs.getUri().toString() + "/tests");
-        fs = new HadoopFileIO();
-        fs.setFileSystem(basePath, hdfs);
+        fs = new HadoopFileIO(basePath);
+        fs.setFileSystem(hdfs);
     }
 
     @AfterAll
@@ -102,4 +105,20 @@ public void testAtomicWrite() throws IOException {
     public void testAtomicWriteMultipleThreads() throws InterruptedException {
         FileIOTest.testOverwriteFileUtf8(new Path(getBasePath(), randomName()), fs);
     }
+
+    @Test
+    public void testIsObjectStore() {
+        assertThat(fs.isObjectStore()).isEqualTo(false);
+    }
+
+    @Test
+    public void testSerializable() {
+        assertDoesNotThrow(
+                () -> {
+                    try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
+                            ObjectOutputStream oos = new ObjectOutputStream(baos)) {
+                        oos.writeObject(fs);
+                    }
+                });
+    }
 }
diff --git a/paimon-common/src/test/java/org/apache/paimon/fs/hadoop/HadoopSecuredFileSystemTest.java b/paimon-common/src/test/java/org/apache/paimon/fs/hadoop/HadoopSecuredFileSystemTest.java
@@ -19,6 +19,7 @@
 package org.apache.paimon.fs.hadoop;
 
 import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.fs.Path;
 import org.apache.paimon.options.Options;
 
 import org.junit.jupiter.api.Test;
@@ -41,7 +42,7 @@ public void test() throws Exception {
         options.set("security.kerberos.login.principal", "test-user");
         options.set("security.kerberos.login.keytab", keytabFile.getAbsolutePath());
 
-        HadoopFileIO fileIO = new HadoopFileIO();
+        HadoopFileIO fileIO = new HadoopFileIO(new Path("file:///tmp/test"));
         fileIO.configure(CatalogContext.create(options));
         assertThat(fileIO.getFileSystem(new org.apache.hadoop.fs.Path("file:///tmp/test")))
                 .isInstanceOf(HadoopSecuredFileSystem.class);
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkFileIO.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkFileIO.java
@@ -24,6 +24,7 @@
 import org.apache.paimon.fs.Path;
 import org.apache.paimon.fs.PositionOutputStream;
 import org.apache.paimon.fs.SeekableInputStream;
+import org.apache.paimon.utils.FileIOUtils;
 
 import org.apache.flink.core.fs.FSDataInputStream;
 import org.apache.flink.core.fs.FSDataOutputStream;
@@ -50,27 +51,7 @@ public boolean isObjectStore() {
         try {
             FileSystem fs = path.getFileSystem();
             String scheme = fs.getUri().getScheme().toLowerCase(Locale.US);
-
-            if (scheme.startsWith("s3")
-                    || scheme.startsWith("emr")
-                    || scheme.startsWith("oss")
-                    || scheme.startsWith("wasb")
-                    || scheme.startsWith("abfs")
-                    || scheme.startsWith("gs")
-                    || scheme.startsWith("cosn")) {
-                // the Amazon S3 storage or Aliyun OSS storage or Azure Blob Storage
-                // or Google Cloud Storage
-                return true;
-            } else if (scheme.startsWith("http") || scheme.startsWith("ftp")) {
-                // file servers instead of file systems
-                // they might actually be consistent, but we have no hard guarantees
-                // currently to rely on that
-                return true;
-            } else {
-                // the remainder should include hdfs, kosmos, ceph, ...
-                // this also includes federated HDFS (viewfs).
-                return false;
-            }
+            return FileIOUtils.isObjectStore(scheme);
         } catch (IOException e) {
             throw new UncheckedIOException(e);
         }

Original file line number	Diff line number	Diff line change
`@@ -33,6 +33,6 @@ public String getScheme() {`
`33`	`33`
`34`	`34`	`@Override`
`35`	`35`	`public HadoopFileIO load(Path path) {`
`36`		`- return new HadoopFileIO();`
	`36`	`+ return new HadoopFileIO(path);`
`37`	`37`	`}`
`38`	`38`	`}`