This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 2da3900f6c [filesystem][s3] Serve s3a:// paths with the S3 plugin
(#9814)
2da3900f6c is described below
commit 2da3900f6c0c2c15e072c75b6b6adef65fabbd9e
Author: κΉλν <[email protected]>
AuthorDate: Tue Sep 15 12:18:20 2026 +0900
[filesystem][s3] Serve s3a:// paths with the S3 plugin (#9814)
---
docs/docs/maintenance/filesystems.mdx | 2 +-
paimon-filesystems/paimon-s3/pom.xml | 14 ++
.../main/java/org/apache/paimon/s3/S3ALoader.java | 35 +++++
.../services/org.apache.paimon.fs.FileIOLoader | 1 +
.../java/org/apache/paimon/s3/S3ASchemeTest.java | 151 +++++++++++++++++++++
5 files changed, 202 insertions(+), 1 deletion(-)
diff --git a/docs/docs/maintenance/filesystems.mdx
b/docs/docs/maintenance/filesystems.mdx
index d80c942847..54f0817f99 100644
--- a/docs/docs/maintenance/filesystems.mdx
+++ b/docs/docs/maintenance/filesystems.mdx
@@ -43,7 +43,7 @@ FileSystem pluggable jars for user to query tables from
Spark/Hive side.
| Local File System | file:// | N | Built-in Support
|
| HDFS | hdfs:// | N | Built-in Support,
ensure that the cluster is in the hadoop environment |
| Aliyun OSS | oss:// | Y |
|
-| S3 | s3:// | Y |
|
+| S3 | s3://, s3a:// | Y | Both schemes are
served by `S3AFileSystem` |
| Tencent Cloud Object Storage | cosn:// | Y |
|
| Microsoft Azure Storage | abfs:// | Y |
|
| Huawei OBS | obs:// | Y |
|
diff --git a/paimon-filesystems/paimon-s3/pom.xml
b/paimon-filesystems/paimon-s3/pom.xml
index 3137cb143a..39f45c163c 100644
--- a/paimon-filesystems/paimon-s3/pom.xml
+++ b/paimon-filesystems/paimon-s3/pom.xml
@@ -52,6 +52,20 @@
</exclusion>
</exclusions>
</dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-hdfs-client</artifactId>
+ <version>${hadoop.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.paimon</groupId>
+ <artifactId>paimon-hadoop-shaded-3.4</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
<dependency>
<groupId>org.apache.paimon</groupId>
<artifactId>paimon-test-utils</artifactId>
diff --git
a/paimon-filesystems/paimon-s3/src/main/java/org/apache/paimon/s3/S3ALoader.java
b/paimon-filesystems/paimon-s3/src/main/java/org/apache/paimon/s3/S3ALoader.java
new file mode 100644
index 0000000000..5d87f30c09
--- /dev/null
+++
b/paimon-filesystems/paimon-s3/src/main/java/org/apache/paimon/s3/S3ALoader.java
@@ -0,0 +1,35 @@
+/*
+ * 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.s3;
+
+/**
+ * A {@link S3Loader} for the {@code s3a} scheme.
+ *
+ * <p>The underlying {@code S3FileIO} creates an {@code S3AFileSystem} and
initializes it with the
+ * given URI, so {@code s3a://} paths are served by exactly the same code path
as {@code s3://}.
+ */
+public class S3ALoader extends S3Loader {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public String getScheme() {
+ return "s3a";
+ }
+}
diff --git
a/paimon-filesystems/paimon-s3/src/main/resources/META-INF/services/org.apache.paimon.fs.FileIOLoader
b/paimon-filesystems/paimon-s3/src/main/resources/META-INF/services/org.apache.paimon.fs.FileIOLoader
index bf91f2b836..2c3f901044 100644
---
a/paimon-filesystems/paimon-s3/src/main/resources/META-INF/services/org.apache.paimon.fs.FileIOLoader
+++
b/paimon-filesystems/paimon-s3/src/main/resources/META-INF/services/org.apache.paimon.fs.FileIOLoader
@@ -14,3 +14,4 @@
# limitations under the License.
org.apache.paimon.s3.S3Loader
+org.apache.paimon.s3.S3ALoader
diff --git
a/paimon-filesystems/paimon-s3/src/test/java/org/apache/paimon/s3/S3ASchemeTest.java
b/paimon-filesystems/paimon-s3/src/test/java/org/apache/paimon/s3/S3ASchemeTest.java
new file mode 100644
index 0000000000..23394cb0fb
--- /dev/null
+++
b/paimon-filesystems/paimon-s3/src/test/java/org/apache/paimon/s3/S3ASchemeTest.java
@@ -0,0 +1,151 @@
+/*
+ * 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.s3;
+
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.FileIOLoader;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.options.Options;
+
+import org.apache.hadoop.conf.Configuration;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.net.URI;
+import java.util.Map;
+import java.util.UUID;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/** Tests that the {@code s3a} scheme is served by the S3 plugin. */
+public class S3ASchemeTest {
+
+ @RegisterExtension public static final MinioTestContainer MINIO = new
MinioTestContainer();
+
+ @Test
+ public void testS3aSchemeIsRegistered() {
+ Map<String, FileIOLoader> loaders = FileIO.discoverLoaders();
+ assertTrue(loaders.containsKey("s3"), "s3 loader missing");
+ assertTrue(loaders.containsKey("s3a"), "s3a loader missing");
+ assertEquals(S3Loader.class, loaders.get("s3").getClass());
+ assertEquals(S3ALoader.class, loaders.get("s3a").getClass());
+ }
+
+ @Test
+ public void testReadWriteOverS3aScheme() throws Exception {
+ String bucket =
MINIO.getS3UriForDefaultBucket().substring("s3://".length());
+ Path path = new Path("s3a://" + bucket + "/s3a-test/" +
UUID.randomUUID() + ".txt");
+
+ Options options = new Options();
+ MINIO.getS3ConfigOptions().forEach(options::set);
+ FileIO fileIO = FileIO.get(path, CatalogContext.create(options, new
Configuration()));
+
+ System.out.println("[s3a] FileIO class = " +
fileIO.getClass().getName());
+ assertTrue(
+
fileIO.getClass().getName().startsWith("org.apache.paimon.s3.S3Loader"),
+ "expected the S3 plugin FileIO, got " +
fileIO.getClass().getName());
+
+ fileIO.writeFile(path, "hello-s3a", true);
+ assertTrue(fileIO.exists(path));
+ assertEquals("hello-s3a", fileIO.readFileUtf8(path));
+ assertTrue(fileIO.listStatus(path.getParent()).length >= 1);
+ assertTrue(fileIO.delete(path, false));
+ }
+
+ @Test
+ public void testS3AndS3aAreBothUsable() throws Exception {
+ String bucket =
MINIO.getS3UriForDefaultBucket().substring("s3://".length());
+ String name = "both-" + UUID.randomUUID() + ".txt";
+ Options options = new Options();
+ MINIO.getS3ConfigOptions().forEach(options::set);
+ CatalogContext context = CatalogContext.create(options, new
Configuration());
+
+ Path s3Path = new Path("s3://" + bucket + "/s3a-test/" + name);
+ Path s3aPath = new Path("s3a://" + bucket + "/s3a-test/" + name);
+
+ FileIO.get(s3Path, context).writeFile(s3Path, "written-via-s3", true);
+ // the very same object must be visible through the s3a scheme
+ assertEquals("written-via-s3", FileIO.get(s3aPath,
context).readFileUtf8(s3aPath));
+ }
+
+ @Test
+ public void testUnderlyingFileSystemIsS3A() throws Exception {
+ assertUnderlyingFileSystemIsS3A("s3a");
+ }
+
+ @Test
+ public void testUnderlyingFileSystemIsS3AForPlainS3Scheme() throws
Exception {
+ assertUnderlyingFileSystemIsS3A("s3");
+ }
+
+ private void assertUnderlyingFileSystemIsS3A(String scheme) throws
Exception {
+ String bucket =
MINIO.getS3UriForDefaultBucket().substring("s3://".length());
+ Path path =
+ new Path(scheme + "://" + bucket + "/s3a-test/fs-" +
UUID.randomUUID() + ".txt");
+
+ Options options = new Options();
+ MINIO.getS3ConfigOptions().forEach(options::set);
+ FileIO fileIO = FileIO.get(path, CatalogContext.create(options, new
Configuration()));
+ fileIO.writeFile(path, "probe", true);
+
+ // unwrap PluginFileIO -> S3FileIO (loaded in the plugin classloader)
+ Field lazy =
+
Class.forName("org.apache.paimon.fs.PluginFileIO").getDeclaredField("lazyFileIO");
+ lazy.setAccessible(true);
+ Object delegate = lazy.get(fileIO);
+ System.out.println("[" + scheme + "] plugin FileIO = " +
delegate.getClass().getName());
+
+ // unwrap S3FileIO -> the Hadoop FileSystem it actually created
+ Field fsMapField = null;
+ for (Class<?> c = delegate.getClass(); c != null; c =
c.getSuperclass()) {
+ try {
+ fsMapField = c.getDeclaredField("fsMap");
+ break;
+ } catch (NoSuchFieldException ignored) {
+ // keep walking
+ }
+ }
+ fsMapField.setAccessible(true);
+ Map<?, ?> fsMap = (Map<?, ?>) fsMapField.get(delegate);
+ Object fs = fsMap.values().iterator().next();
+
+ Method getScheme = fs.getClass().getMethod("getScheme");
+ Method getUri = fs.getClass().getMethod("getUri");
+ URI fsUri = (URI) getUri.invoke(fs);
+ System.out.println(
+ "["
+ + scheme
+ + "] hadoop FileSystem = "
+ + fs.getClass().getName()
+ + ", getScheme() = "
+ + getScheme.invoke(fs)
+ + ", getUri() = "
+ + fsUri);
+
+ assertEquals("org.apache.hadoop.fs.s3a.S3AFileSystem",
fs.getClass().getName());
+ assertEquals(scheme, fsUri.getScheme());
+ assertEquals(bucket, fsUri.getHost());
+ assertEquals("probe", fileIO.readFileUtf8(path));
+ assertTrue(fileIO.delete(path, false));
+ }
+}
