This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 35a8e3dd79e7250ec1a4b25444deecf1e269ead1 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 8055959ffb..57711bf714 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)); + } +}
