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 f5767d61c5 [filesystem][s3] Cache FileSystem by scheme and authority
in HadoopCompliantFileIO (#10182)
f5767d61c5 is described below
commit f5767d61c5d781ca054ab891ea1e6fb82198e195
Author: Eunbin Son <[email protected]>
AuthorDate: Sat Oct 3 18:11:44 2026 +0900
[filesystem][s3] Cache FileSystem by scheme and authority in
HadoopCompliantFileIO (#10182)
---
.../apache/paimon/s3/HadoopCompliantFileIO.java | 18 +++--
.../apache/paimon/s3/S3FileIOSchemeCacheTest.java | 92 ++++++++++++++++++++++
2 files changed, 102 insertions(+), 8 deletions(-)
diff --git
a/paimon-filesystems/paimon-s3-impl/src/main/java/org/apache/paimon/s3/HadoopCompliantFileIO.java
b/paimon-filesystems/paimon-s3-impl/src/main/java/org/apache/paimon/s3/HadoopCompliantFileIO.java
index 5965d7e670..298a088c0e 100644
---
a/paimon-filesystems/paimon-s3-impl/src/main/java/org/apache/paimon/s3/HadoopCompliantFileIO.java
+++
b/paimon-filesystems/paimon-s3-impl/src/main/java/org/apache/paimon/s3/HadoopCompliantFileIO.java
@@ -24,6 +24,7 @@ import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.PositionOutputStream;
import org.apache.paimon.fs.RemoteIterator;
import org.apache.paimon.fs.SeekableInputStream;
+import org.apache.paimon.utils.Pair;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
@@ -31,6 +32,7 @@ import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.io.IOUtils;
import java.io.IOException;
+import java.net.URI;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
@@ -43,7 +45,7 @@ public abstract class HadoopCompliantFileIO implements FileIO
{
private static final long serialVersionUID = 1L;
- protected transient volatile Map<String, FileSystem> fsMap;
+ protected transient volatile Map<Pair<String, String>, FileSystem> fsMap;
@Override
public SeekableInputStream newInputStream(Path path) throws IOException {
@@ -137,16 +139,16 @@ public abstract class HadoopCompliantFileIO implements
FileIO {
}
}
- Map<String, FileSystem> map = fsMap;
+ Map<Pair<String, String>, FileSystem> map = fsMap;
- String authority = path.toUri().getAuthority();
- if (authority == null) {
- authority = "DEFAULT";
- }
- FileSystem fs = map.get(authority);
+ URI uri = path.toUri();
+ String scheme = uri.getScheme();
+ String authority = uri.getAuthority();
+ Pair<String, String> key = Pair.of(scheme, authority);
+ FileSystem fs = map.get(key);
if (fs == null) {
fs = createFileSystem(path);
- map.put(authority, fs);
+ map.put(key, fs);
}
return fs;
}
diff --git
a/paimon-filesystems/paimon-s3-impl/src/test/java/org/apache/paimon/s3/S3FileIOSchemeCacheTest.java
b/paimon-filesystems/paimon-s3-impl/src/test/java/org/apache/paimon/s3/S3FileIOSchemeCacheTest.java
new file mode 100644
index 0000000000..1aebd61a79
--- /dev/null
+++
b/paimon-filesystems/paimon-s3-impl/src/test/java/org/apache/paimon/s3/S3FileIOSchemeCacheTest.java
@@ -0,0 +1,92 @@
+/*
+ * 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.options.Options;
+
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests that one {@link S3FileIO} instance serves both {@code s3://} and
{@code s3a://} paths of
+ * the same bucket. No S3 backend is contacted: the bucket probe is disabled
and the endpoint is
+ * unroutable, and only {@link FileSystem#makeQualified} is called on the
returned file systems.
+ */
+class S3FileIOSchemeCacheTest {
+
+ private static final String BUCKET = "paimon-scheme-cache-bucket";
+
+ @Test
+ void testS3ThenS3aOnSameBucket() throws Exception {
+ assertSchemesAreServedSeparately("s3", "s3a");
+ }
+
+ @Test
+ void testS3aThenS3OnSameBucket() throws Exception {
+ assertSchemesAreServedSeparately("s3a", "s3");
+ }
+
+ @Test
+ void testSameSchemeAndBucketReusesFileSystem() throws Exception {
+ S3FileIO fileIO = newFileIO();
+ for (String scheme : new String[] {"s3", "s3a"}) {
+ FileSystem first = fileIO.getFileSystem(path(scheme,
"warehouse/a"));
+ FileSystem second = fileIO.getFileSystem(path(scheme,
"warehouse/b"));
+ assertThat(second).isSameAs(first);
+ }
+ }
+
+ private static void assertSchemesAreServedSeparately(String firstScheme,
String secondScheme)
+ throws Exception {
+ S3FileIO fileIO = newFileIO();
+ Path firstPath = path(firstScheme, "warehouse/db/t");
+ Path secondPath = path(secondScheme, "hive/t/part-0.parquet");
+
+ FileSystem firstFs = fileIO.getFileSystem(firstPath);
+ FileSystem secondFs = fileIO.getFileSystem(secondPath);
+
+ // makeQualified runs FileSystem#checkPath, which rejects a path of
another scheme.
+ assertThat(firstFs.makeQualified(firstPath)).isEqualTo(firstPath);
+ assertThat(secondFs.makeQualified(secondPath)).isEqualTo(secondPath);
+ assertThat(firstFs.getUri().getScheme()).isEqualTo(firstScheme);
+ assertThat(secondFs.getUri().getScheme()).isEqualTo(secondScheme);
+ assertThat(secondFs).isNotSameAs(firstFs);
+ }
+
+ private static S3FileIO newFileIO() {
+ Options options = new Options();
+ // Never reach the network: no bucket probe, dummy credentials and an
unroutable endpoint.
+ options.set("s3.bucket.probe", "0");
+ options.set("s3.endpoint", "http://localhost:1";);
+ options.set("s3.endpoint.region", "us-east-1");
+ options.set("s3.access-key", "dummy-access-key");
+ options.set("s3.secret-key", "dummy-secret-key");
+ S3FileIO fileIO = new S3FileIO();
+ fileIO.configure(CatalogContext.create(options));
+ return fileIO;
+ }
+
+ private static Path path(String scheme, String key) {
+ return new Path(scheme + "://" + BUCKET + "/" + key);
+ }
+}
