You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by mo...@apache.org on 2026/03/31 05:24:28 UTC
(doris) branch branch-fs-spi updated: [refactor](fe) Phase3: split filesystem implementations into independent SPI modules (#61919)
This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch branch-fs-spi
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-fs-spi by this push:
new 23ed5e323e1 [refactor](fe) Phase3: split filesystem implementations into independent SPI modules (#61919)
23ed5e323e1 is described below
commit 23ed5e323e1b6a2ad14f6db4b6d402e6129fa06d
Author: Mingyu Chen (Rayner) <yu...@selectdb.com>
AuthorDate: Mon Mar 30 22:24:20 2026 -0700
[refactor](fe) Phase3: split filesystem implementations into independent SPI modules (#61919)
### What problem does this PR solve?
Issue Number: N/A
Problem Summary:
Refactor the Apache Doris FE filesystem layer so that each storage
backend
(S3, OSS, COS, OBS, Azure, HDFS, Local) lives in its own independent
Maven
module and is discovered at runtime via Java ServiceLoader (SPI
pattern).
Adding a new storage backend now requires zero changes to fe-core.
Changes by step:
**P3.0b** – Rename `FileSystemProvider` →
`LegacyFileSystemProviderFactory`
to free up the name `FileSystemProvider` for the new SPI interface.
**Step 1** – Create `fe/fe-filesystem/` aggregator POM and
`fe-filesystem-spi`
module. Defines the stable SPI contracts:
- `FileSystem`, `ObjStorage`, `FileSystemProvider` interfaces
- `HadoopAuthenticator`, `IOCallable` for HDFS/Kerberos
- Value objects: `FileEntry`, `FileIterator`, `InputFile`, `OutputFile`,
`RequestBody`, `ObjectMetadata`, `MultipartPart`, `UploadedPart`
**Step 2** – `fe-filesystem-s3` module:
- `S3Uri`, `S3ObjStorage`, `S3FileSystem`, `S3OutputStream`
- `S3FileSystemProvider` registered via META-INF/services
**Step 3** – `fe-filesystem-oss`, `fe-filesystem-cos`,
`fe-filesystem-obs`:
- Thin providers that reuse S3FileSystem via S3-compatible APIs
- Each translates vendor-specific property keys to AWS_* keys
**Step 3b** – `fe-filesystem-azure` module:
- `AzureUri` (wasb/wasbs/abfs/abfss/https), `AzureObjStorage`
- Supports shared-key, service-principal, and default-credential auth
- Multipart upload via Azure block-blob stageBlock/commitBlockList
**Step 4** – `fe-filesystem-hdfs` module:
- `HdfsConfigBuilder`, `SimpleHadoopAuthenticator`,
`KerberosHadoopAuthenticator`
- `DFSFileSystem` implements `spi.FileSystem` directly
- Supports hdfs/viewfs/ofs/jfs/oss schemes
**Step 6** – `fe-filesystem-local` (test-only):
- `LocalFileSystem` using `java.nio.file` for unit-test isolation
**Step 7** – `fe-core` integration:
- `fe-filesystem-spi` added as compile dep; all impl modules as runtime
deps
- New `StoragePropertiesConverter` translates `StorageProperties` → Map
- `FileSystemFactory` extended with `getFileSystem(Map)` /
`getFileSystem(StorageProperties)` using ServiceLoader (double-checked
locking); old `get()` methods kept `@Deprecated` for Phase-4 migration
- Fix checkstyle violations (import order, line length) in HMS classes
introduced by P3.0b rename
### Release note
None (internal refactoring; no user-visible behavior change)
### Check List (For Author)
- Test: `mvn install -pl fe-core -am -DskipTests` passes (BUILD SUCCESS)
- Behavior changed: No (legacy API fully preserved; new SPI API added
alongside)
- Does this need documentation: No
Co-authored-by: Copilot <22...@users.noreply.github.com>
---
fe/fe-core/pom.xml | 43 +++
.../doris/datasource/hive/HMSExternalCatalog.java | 7 +-
.../doris/datasource/hive/HMSTransaction.java | 5 +-
.../org/apache/doris/fs/FileSystemFactory.java | 124 +++++++-
.../apache/doris/fs/FileSystemProviderImpl.java | 2 +-
...r.java => LegacyFileSystemProviderFactory.java} | 7 +-
.../doris/fs/StoragePropertiesConverter.java | 71 +++++
.../doris/transaction/HiveTransactionManager.java | 6 +-
.../transaction/TransactionManagerFactory.java | 4 +-
.../datasource/hive/HMSTransactionPathTest.java | 4 +-
.../doris/datasource/hive/HmsCommitTest.java | 4 +-
fe/fe-filesystem/fe-filesystem-azure/pom.xml | 65 ++++
.../doris/filesystem/azure/AzureFileSystem.java | 286 +++++++++++++++++
.../filesystem/azure/AzureFileSystemProvider.java | 60 ++++
.../doris/filesystem/azure/AzureObjStorage.java | 342 +++++++++++++++++++++
.../apache/doris/filesystem/azure/AzureUri.java | 146 +++++++++
....apache.doris.filesystem.spi.FileSystemProvider | 1 +
fe/fe-filesystem/fe-filesystem-cos/pom.xml | 44 +++
.../filesystem/cos/CosFileSystemProvider.java | 68 ++++
....apache.doris.filesystem.spi.FileSystemProvider | 1 +
fe/fe-filesystem/fe-filesystem-hdfs/pom.xml | 75 +++++
.../doris/filesystem/hdfs/DFSFileSystem.java | 149 +++++++++
.../doris/filesystem/hdfs/HdfsConfigBuilder.java | 58 ++++
.../doris/filesystem/hdfs/HdfsFileIterator.java | 56 ++++
.../filesystem/hdfs/HdfsFileSystemProvider.java | 61 ++++
.../doris/filesystem/hdfs/HdfsInputFile.java | 60 ++++
.../doris/filesystem/hdfs/HdfsOutputFile.java | 60 ++++
.../hdfs/KerberosHadoopAuthenticator.java | 65 ++++
.../hdfs/SimpleHadoopAuthenticator.java} | 19 +-
....apache.doris.filesystem.spi.FileSystemProvider | 1 +
fe/fe-filesystem/fe-filesystem-local/pom.xml | 44 +++
.../doris/filesystem/local/LocalFileSystem.java | 180 +++++++++++
.../filesystem/local/LocalFileSystemProvider.java | 47 +++
....apache.doris.filesystem.spi.FileSystemProvider | 1 +
fe/fe-filesystem/fe-filesystem-obs/pom.xml | 44 +++
.../filesystem/obs/ObsFileSystemProvider.java | 68 ++++
....apache.doris.filesystem.spi.FileSystemProvider | 1 +
fe/fe-filesystem/fe-filesystem-oss/pom.xml | 44 +++
.../filesystem/oss/OssFileSystemProvider.java | 68 ++++
....apache.doris.filesystem.spi.FileSystemProvider | 1 +
fe/fe-filesystem/fe-filesystem-s3/pom.xml | 73 +++++
.../apache/doris/filesystem/s3/S3FileSystem.java | 232 ++++++++++++++
.../doris/filesystem/s3/S3FileSystemProvider.java | 57 ++++
.../apache/doris/filesystem/s3/S3ObjStorage.java | 331 ++++++++++++++++++++
.../apache/doris/filesystem/s3/S3OutputStream.java | 71 +++++
.../java/org/apache/doris/filesystem/s3/S3Uri.java | 63 ++++
....apache.doris.filesystem.spi.FileSystemProvider | 1 +
fe/fe-filesystem/fe-filesystem-spi/pom.xml | 46 +++
.../apache/doris/filesystem/spi/BlockInfo.java} | 30 +-
.../doris/filesystem/spi/DorisInputFile.java} | 17 +-
.../doris/filesystem/spi/DorisOutputFile.java} | 17 +-
.../org/apache/doris/filesystem/spi/FileEntry.java | 54 ++++
.../apache/doris/filesystem/spi/FileIterator.java} | 15 +-
.../apache/doris/filesystem/spi/FileSystem.java} | 28 +-
.../doris/filesystem/spi/FileSystemProvider.java | 58 ++++
.../doris/filesystem/spi/HadoopAuthenticator.java | 45 +++
.../apache/doris/filesystem/spi/IOCallable.java} | 16 +-
.../org/apache/doris/filesystem/spi/Location.java | 71 +++++
.../apache/doris/filesystem/spi/ObjFileSystem.java | 64 ++++
.../apache/doris/filesystem/spi/ObjStorage.java | 58 ++++
.../apache/doris/filesystem/spi/RemoteObject.java} | 36 ++-
.../apache/doris/filesystem/spi/RemoteObjects.java | 48 +++
.../apache/doris/filesystem/spi/RequestBody.java} | 32 +-
.../doris/filesystem/spi/UploadPartResult.java} | 24 +-
fe/fe-filesystem/pom.xml | 50 +++
fe/pom.xml | 1 +
66 files changed, 3871 insertions(+), 59 deletions(-)
diff --git a/fe/fe-core/pom.xml b/fe/fe-core/pom.xml
index 7a6feee5e79..1596f0976bb 100644
--- a/fe/fe-core/pom.xml
+++ b/fe/fe-core/pom.xml
@@ -69,6 +69,49 @@ under the License.
<artifactId>fe-common</artifactId>
<version>${project.version}</version>
</dependency>
+ <!-- fe-filesystem-spi: compile-time SPI interfaces for FileSystemFactory -->
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>fe-filesystem-spi</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <!-- fe-filesystem impl modules: runtime-only, discovered via ServiceLoader -->
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>fe-filesystem-s3</artifactId>
+ <version>${project.version}</version>
+ <scope>runtime</scope>
+ </dependency>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>fe-filesystem-azure</artifactId>
+ <version>${project.version}</version>
+ <scope>runtime</scope>
+ </dependency>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>fe-filesystem-oss</artifactId>
+ <version>${project.version}</version>
+ <scope>runtime</scope>
+ </dependency>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>fe-filesystem-cos</artifactId>
+ <version>${project.version}</version>
+ <scope>runtime</scope>
+ </dependency>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>fe-filesystem-obs</artifactId>
+ <version>${project.version}</version>
+ <scope>runtime</scope>
+ </dependency>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>fe-filesystem-hdfs</artifactId>
+ <version>${project.version}</version>
+ <scope>runtime</scope>
+ </dependency>
<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>fe-thrift</artifactId>
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HMSExternalCatalog.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HMSExternalCatalog.java
index abb6369a9e4..e4b1c3c48cb 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HMSExternalCatalog.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HMSExternalCatalog.java
@@ -33,8 +33,8 @@ import org.apache.doris.datasource.iceberg.IcebergUtils;
import org.apache.doris.datasource.metacache.CacheSpec;
import org.apache.doris.datasource.operations.ExternalMetadataOperations;
import org.apache.doris.datasource.property.metastore.AbstractHiveProperties;
-import org.apache.doris.fs.FileSystemProvider;
import org.apache.doris.fs.FileSystemProviderImpl;
+import org.apache.doris.fs.LegacyFileSystemProviderFactory;
import org.apache.doris.fs.remote.dfs.DFSFileSystem;
import org.apache.doris.transaction.TransactionManagerFactory;
@@ -146,8 +146,9 @@ public class HMSExternalCatalog extends ExternalCatalog {
String.format("hms_iceberg_catalog_%s_executor_pool", name),
true,
executionAuthenticator);
- FileSystemProvider fileSystemProvider = new FileSystemProviderImpl(Env.getCurrentEnv().getExtMetaCacheMgr(),
- this.catalogProperty.getStoragePropertiesMap());
+ LegacyFileSystemProviderFactory fileSystemProvider =
+ new FileSystemProviderImpl(Env.getCurrentEnv().getExtMetaCacheMgr(),
+ this.catalogProperty.getStoragePropertiesMap());
this.fileSystemExecutor = ThreadPoolManager.newDaemonFixedThreadPool(FILE_SYSTEM_EXECUTOR_THREAD_NUM,
Integer.MAX_VALUE, String.format("hms_committer_%s_file_system_executor_pool", name), true);
transactionManager = TransactionManagerFactory.createHiveTransactionManager(hiveOps, fileSystemProvider,
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HMSTransaction.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HMSTransaction.java
index 362658b2007..1e9e36b21a5 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HMSTransaction.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HMSTransaction.java
@@ -28,9 +28,9 @@ import org.apache.doris.common.profile.SummaryProfile;
import org.apache.doris.datasource.NameMapping;
import org.apache.doris.datasource.statistics.CommonStatistics;
import org.apache.doris.foundation.util.PathUtils;
-import org.apache.doris.fs.FileSystemProvider;
import org.apache.doris.fs.FileSystemUtil;
import org.apache.doris.fs.LegacyFileSystemApi;
+import org.apache.doris.fs.LegacyFileSystemProviderFactory;
import org.apache.doris.fs.remote.MultipartUploadCapable;
import org.apache.doris.fs.remote.RemoteFile;
import org.apache.doris.fs.remote.S3FileSystem;
@@ -138,7 +138,8 @@ public class HMSTransaction implements Transaction {
private final Set<UncompletedMpuPendingUpload> uncompletedMpuPendingUploads = new HashSet<>();
- public HMSTransaction(HiveMetadataOps hiveOps, FileSystemProvider fileSystemProvider, Executor fileSystemExecutor) {
+ public HMSTransaction(HiveMetadataOps hiveOps, LegacyFileSystemProviderFactory fileSystemProvider,
+ Executor fileSystemExecutor) {
this.hiveOps = hiveOps;
this.fs = fileSystemProvider.get(null);
if (!(fs instanceof SwitchingFileSystem)) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemFactory.java b/fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemFactory.java
index 11536418fd9..fd65cab5ec7 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemFactory.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemFactory.java
@@ -19,22 +19,69 @@ package org.apache.doris.fs;
import org.apache.doris.common.UserException;
import org.apache.doris.datasource.property.storage.StorageProperties;
+import org.apache.doris.filesystem.spi.FileSystemProvider;
import org.apache.doris.fs.remote.RemoteFileSystem;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.io.IOException;
+import java.util.ArrayList;
import java.util.List;
import java.util.Map;
+import java.util.ServiceLoader;
+
+/**
+ * Factory for filesystem instances.
+ *
+ * <h2>Two APIs</h2>
+ * <ul>
+ * <li><b>Legacy API</b> ({@code get(StorageProperties)} / {@code get(FileSystemType, Map)}) —
+ * delegates to {@link StorageTypeMapper}; returns {@link RemoteFileSystem}. All existing
+ * fe-core callers continue to use this path unchanged.</li>
+ * <li><b>SPI API</b> ({@code getFileSystem(Map)} / {@code getFileSystem(StorageProperties)}) —
+ * uses Java ServiceLoader to discover {@link FileSystemProvider} implementations at runtime;
+ * returns {@code org.apache.doris.filesystem.spi.FileSystem}. New code should use this path.
+ * Adding a new storage backend requires only a new Maven module; zero changes here.</li>
+ * </ul>
+ *
+ * <p>Compile-time dependency on {@code fe-filesystem-spi} only. All implementation modules
+ * ({@code fe-filesystem-s3}, {@code fe-filesystem-hdfs}, …) are runtime-only dependencies
+ * discovered via {@code META-INF/services}.
+ */
+public final class FileSystemFactory {
+
+ private static final Logger LOG = LogManager.getLogger(FileSystemFactory.class);
+
+ // SPI provider cache (loaded once, reloadable in tests)
+ private static volatile List<FileSystemProvider> cachedProviders = null;
+
+ private FileSystemFactory() {}
-public class FileSystemFactory {
+ // =========================================================
+ // Legacy API — backward compatible, returns RemoteFileSystem
+ // =========================================================
+ /**
+ * Legacy entry point. Returns a {@link RemoteFileSystem} via {@link StorageTypeMapper}.
+ *
+ * @deprecated New code should use {@link #getFileSystem(Map)} instead.
+ */
+ @Deprecated
public static RemoteFileSystem get(StorageProperties storageProperties) {
return StorageTypeMapper.create(storageProperties);
}
- //todo remove when catalog use storage properties
+ /**
+ * Legacy entry point by enum type. Returns a {@link RemoteFileSystem} via
+ * {@link StorageTypeMapper}.
+ *
+ * @deprecated New code should use {@link #getFileSystem(Map)} instead.
+ */
+ @Deprecated
public static RemoteFileSystem get(FileSystemType fileSystemType, Map<String, String> properties)
throws UserException {
List<StorageProperties> storagePropertiesList = StorageProperties.createAll(properties);
-
for (StorageProperties storageProperties : storagePropertiesList) {
if (storageProperties.getStorageName().equalsIgnoreCase(fileSystemType.name())) {
return StorageTypeMapper.create(storageProperties);
@@ -42,4 +89,75 @@ public class FileSystemFactory {
}
throw new RuntimeException("Unsupported file system type: " + fileSystemType);
}
+
+ // =========================================================
+ // SPI API — ServiceLoader-based, returns spi.FileSystem
+ // =========================================================
+
+ /**
+ * SPI entry point: discovers and selects a {@link FileSystemProvider} via ServiceLoader,
+ * then creates the filesystem.
+ *
+ * @param properties key-value storage configuration
+ * @return initialized {@code org.apache.doris.filesystem.spi.FileSystem}
+ * @throws IOException if no provider matches or creation fails
+ */
+ public static org.apache.doris.filesystem.spi.FileSystem getFileSystem(Map<String, String> properties)
+ throws IOException {
+ List<FileSystemProvider> providers = getProviders();
+ List<String> tried = new ArrayList<>();
+ for (FileSystemProvider provider : providers) {
+ if (provider.supports(properties)) {
+ LOG.debug("FileSystemFactory: selected SPI provider '{}' for keys={}",
+ provider.name(), properties.keySet());
+ return provider.create(properties);
+ }
+ tried.add(provider.name());
+ }
+ throw new IOException(String.format(
+ "No FileSystemProvider found for properties %s. Tried: %s. "
+ + "Ensure the corresponding fe-filesystem-xxx jar is on the classpath.",
+ properties.keySet(), tried));
+ }
+
+ /**
+ * SPI entry point accepting legacy {@link StorageProperties}.
+ * Converts via {@link StoragePropertiesConverter} then delegates to
+ * {@link #getFileSystem(Map)}.
+ */
+ public static org.apache.doris.filesystem.spi.FileSystem getFileSystem(StorageProperties storageProperties)
+ throws IOException {
+ return getFileSystem(StoragePropertiesConverter.toMap(storageProperties));
+ }
+
+ /**
+ * Returns all discovered SPI providers. Uses cached list after first load.
+ * Package-private for testing.
+ */
+ static List<FileSystemProvider> getProviders() {
+ if (cachedProviders == null) {
+ synchronized (FileSystemFactory.class) {
+ if (cachedProviders == null) {
+ List<FileSystemProvider> providers = new ArrayList<>();
+ ServiceLoader<FileSystemProvider> loader = ServiceLoader.load(
+ FileSystemProvider.class,
+ Thread.currentThread().getContextClassLoader());
+ loader.forEach(providers::add);
+ LOG.info("FileSystemFactory: loaded {} SPI provider(s): {}",
+ providers.size(),
+ providers.stream().map(FileSystemProvider::name)
+ .collect(java.util.stream.Collectors.joining(", ")));
+ cachedProviders = providers;
+ }
+ }
+ }
+ return cachedProviders;
+ }
+
+ /**
+ * Clears the SPI provider cache. For testing only.
+ */
+ static void clearProviderCache() {
+ cachedProviders = null;
+ }
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemProviderImpl.java b/fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemProviderImpl.java
index bd375add070..69ab5fb783e 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemProviderImpl.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemProviderImpl.java
@@ -24,7 +24,7 @@ import org.apache.doris.fs.remote.SwitchingFileSystem;
import java.util.Map;
-public class FileSystemProviderImpl implements FileSystemProvider {
+public class FileSystemProviderImpl implements LegacyFileSystemProviderFactory {
private ExternalMetaCacheMgr extMetaCacheMgr;
private Map<StorageProperties.Type, StorageProperties> storagePropertiesMap;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemProvider.java b/fe/fe-core/src/main/java/org/apache/doris/fs/LegacyFileSystemProviderFactory.java
similarity index 76%
copy from fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemProvider.java
copy to fe/fe-core/src/main/java/org/apache/doris/fs/LegacyFileSystemProviderFactory.java
index 1dd276fc12c..84f04643210 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemProvider.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/fs/LegacyFileSystemProviderFactory.java
@@ -19,6 +19,11 @@ package org.apache.doris.fs;
import org.apache.doris.datasource.SessionContext;
-public interface FileSystemProvider {
+/**
+ * Factory interface for the legacy (pre-SPI) filesystem provider system.
+ * Renamed from FileSystemProvider to avoid naming conflict with the new
+ * org.apache.doris.filesystem.spi.FileSystemProvider SPI interface (Phase 3).
+ */
+public interface LegacyFileSystemProviderFactory {
LegacyFileSystemApi get(SessionContext ctx);
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/fs/StoragePropertiesConverter.java b/fe/fe-core/src/main/java/org/apache/doris/fs/StoragePropertiesConverter.java
new file mode 100644
index 00000000000..dcfa91ed2c1
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/fs/StoragePropertiesConverter.java
@@ -0,0 +1,71 @@
+// 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.doris.fs;
+
+import org.apache.doris.datasource.property.storage.AbstractS3CompatibleProperties;
+import org.apache.doris.datasource.property.storage.AzureProperties;
+import org.apache.doris.datasource.property.storage.BrokerProperties;
+import org.apache.doris.datasource.property.storage.HdfsCompatibleProperties;
+import org.apache.doris.datasource.property.storage.StorageProperties;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Converts legacy {@link StorageProperties} objects to the {@code Map<String, String>} format
+ * expected by {@link org.apache.doris.filesystem.spi.FileSystemProvider#supports} and
+ * {@link org.apache.doris.filesystem.spi.FileSystemProvider#create}.
+ *
+ * <p>This class is the ONLY place in fe-core that knows about the mapping between
+ * StorageProperties subtypes and filesystem property keys. All other code in fe-core
+ * should use {@link FileSystemFactory#get(Map)} or {@link FileSystemFactory#get(StorageProperties)}.
+ */
+public final class StoragePropertiesConverter {
+
+ private StoragePropertiesConverter() {}
+
+ /**
+ * Converts a {@link StorageProperties} instance to a flat {@code Map<String, String>}
+ * suitable for passing to {@code FileSystemProvider.supports()} and
+ * {@code FileSystemProvider.create()}.
+ *
+ * <p>Uses {@code getBackendConfigProperties()} as the base map and injects
+ * additional type-marker keys so that providers can unambiguously identify themselves.
+ */
+ public static Map<String, String> toMap(StorageProperties props) {
+ Map<String, String> map = new HashMap<>(props.getBackendConfigProperties());
+
+ if (props instanceof AbstractS3CompatibleProperties) {
+ AbstractS3CompatibleProperties s3Props = (AbstractS3CompatibleProperties) props;
+ map.put("AWS_ENDPOINT", s3Props.getEndpoint());
+ map.put("AWS_ACCESS_KEY", s3Props.getAccessKey());
+ map.put("AWS_REGION", s3Props.getRegion());
+ map.put("_STORAGE_TYPE_", s3Props.getStorageName());
+ } else if (props instanceof AzureProperties) {
+ AzureProperties azureProps = (AzureProperties) props;
+ map.put("AZURE_ACCOUNT_NAME", azureProps.getAccountName());
+ map.put("_STORAGE_TYPE_", "AZURE");
+ } else if (props instanceof HdfsCompatibleProperties) {
+ map.put("_STORAGE_TYPE_", "HDFS");
+ } else if (props instanceof BrokerProperties) {
+ map.put("_STORAGE_TYPE_", "BROKER");
+ }
+
+ return map;
+ }
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/transaction/HiveTransactionManager.java b/fe/fe-core/src/main/java/org/apache/doris/transaction/HiveTransactionManager.java
index 65f0c2bd5e3..c78c275da14 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/transaction/HiveTransactionManager.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/transaction/HiveTransactionManager.java
@@ -19,16 +19,16 @@ package org.apache.doris.transaction;
import org.apache.doris.datasource.hive.HMSTransaction;
import org.apache.doris.datasource.hive.HiveMetadataOps;
-import org.apache.doris.fs.FileSystemProvider;
+import org.apache.doris.fs.LegacyFileSystemProviderFactory;
import java.util.concurrent.Executor;
public class HiveTransactionManager extends AbstractExternalTransactionManager<HMSTransaction> {
- private final FileSystemProvider fileSystemProvider;
+ private final LegacyFileSystemProviderFactory fileSystemProvider;
private final Executor fileSystemExecutor;
- public HiveTransactionManager(HiveMetadataOps ops, FileSystemProvider fileSystemProvider,
+ public HiveTransactionManager(HiveMetadataOps ops, LegacyFileSystemProviderFactory fileSystemProvider,
Executor fileSystemExecutor) {
super(ops);
this.fileSystemProvider = fileSystemProvider;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/transaction/TransactionManagerFactory.java b/fe/fe-core/src/main/java/org/apache/doris/transaction/TransactionManagerFactory.java
index e08f13ad0a8..cab7a1dd6fb 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/transaction/TransactionManagerFactory.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/transaction/TransactionManagerFactory.java
@@ -20,14 +20,14 @@ package org.apache.doris.transaction;
import org.apache.doris.datasource.hive.HiveMetadataOps;
import org.apache.doris.datasource.iceberg.IcebergMetadataOps;
import org.apache.doris.datasource.maxcompute.MaxComputeExternalCatalog;
-import org.apache.doris.fs.FileSystemProvider;
+import org.apache.doris.fs.LegacyFileSystemProviderFactory;
import java.util.concurrent.Executor;
public class TransactionManagerFactory {
public static TransactionManager createHiveTransactionManager(HiveMetadataOps ops,
- FileSystemProvider fileSystemProvider, Executor fileSystemExecutor) {
+ LegacyFileSystemProviderFactory fileSystemProvider, Executor fileSystemExecutor) {
return new HiveTransactionManager(ops, fileSystemProvider, fileSystemExecutor);
}
diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/hive/HMSTransactionPathTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/hive/HMSTransactionPathTest.java
index b8284c13200..52f4ab0e99a 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/datasource/hive/HMSTransactionPathTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/hive/HMSTransactionPathTest.java
@@ -18,8 +18,8 @@
package org.apache.doris.datasource.hive;
import org.apache.doris.backup.Status;
-import org.apache.doris.fs.FileSystemProvider;
import org.apache.doris.fs.LegacyFileSystemApi;
+import org.apache.doris.fs.LegacyFileSystemProviderFactory;
import org.apache.doris.fs.LocalDfsFileSystem;
import org.apache.doris.fs.remote.RemoteFile;
import org.apache.doris.fs.remote.SwitchingFileSystem;
@@ -163,7 +163,7 @@ public class HMSTransactionPathTest {
private static HMSTransaction createTransaction(LegacyFileSystemApi delegate) {
SwitchingFileSystem switchingFs = new TestSwitchingFileSystem(delegate);
- FileSystemProvider provider = ctx -> switchingFs;
+ LegacyFileSystemProviderFactory provider = ctx -> switchingFs;
return new HMSTransaction(null, provider, Runnable::run);
}
diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/hive/HmsCommitTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/hive/HmsCommitTest.java
index 7e946b5ae47..1fcb0a85395 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/datasource/hive/HmsCommitTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/hive/HmsCommitTest.java
@@ -25,8 +25,8 @@ import org.apache.doris.common.security.authentication.ExecutionAuthenticator;
import org.apache.doris.common.util.DebugUtil;
import org.apache.doris.datasource.NameMapping;
import org.apache.doris.datasource.TestHMSCachedClient;
-import org.apache.doris.fs.FileSystemProvider;
import org.apache.doris.fs.LegacyFileSystemApi;
+import org.apache.doris.fs.LegacyFileSystemProviderFactory;
import org.apache.doris.fs.LocalDfsFileSystem;
import org.apache.doris.fs.remote.SwitchingFileSystem;
import org.apache.doris.nereids.trees.plans.commands.insert.HiveInsertCommandContext;
@@ -71,7 +71,7 @@ public class HmsCommitTest {
private static HiveMetadataOps hmsOps;
private static HMSCachedClient hmsClient;
- private static FileSystemProvider fileSystemProvider;
+ private static LegacyFileSystemProviderFactory fileSystemProvider;
private static final String dbName = "test_db";
private static final String tbWithPartition = "test_tb_with_partition";
private static final String tbWithoutPartition = "test_tb_without_partition";
diff --git a/fe/fe-filesystem/fe-filesystem-azure/pom.xml b/fe/fe-filesystem/fe-filesystem-azure/pom.xml
new file mode 100644
index 00000000000..26d2bd833c0
--- /dev/null
+++ b/fe/fe-filesystem/fe-filesystem-azure/pom.xml
@@ -0,0 +1,65 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.doris</groupId>
+ <artifactId>fe-filesystem</artifactId>
+ <version>${revision}</version>
+ <relativePath>../pom.xml</relativePath>
+ </parent>
+
+ <artifactId>fe-filesystem-azure</artifactId>
+ <packaging>jar</packaging>
+ <name>Doris FE Filesystem - Azure</name>
+ <description>Azure Blob Storage filesystem implementation for Doris FE.</description>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.doris</groupId>
+ <artifactId>fe-filesystem-spi</artifactId>
+ <version>${revision}</version>
+ </dependency>
+
+ <!-- Azure SDK (versions managed by azure-sdk-bom in fe/pom.xml dependencyManagement) -->
+ <dependency>
+ <groupId>com.azure</groupId>
+ <artifactId>azure-storage-blob</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.azure</groupId>
+ <artifactId>azure-identity</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-api</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+</project>
diff --git a/fe/fe-filesystem/fe-filesystem-azure/src/main/java/org/apache/doris/filesystem/azure/AzureFileSystem.java b/fe/fe-filesystem/fe-filesystem-azure/src/main/java/org/apache/doris/filesystem/azure/AzureFileSystem.java
new file mode 100644
index 00000000000..2387620dccd
--- /dev/null
+++ b/fe/fe-filesystem/fe-filesystem-azure/src/main/java/org/apache/doris/filesystem/azure/AzureFileSystem.java
@@ -0,0 +1,286 @@
+// 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.doris.filesystem.azure;
+
+import org.apache.doris.filesystem.spi.DorisInputFile;
+import org.apache.doris.filesystem.spi.DorisOutputFile;
+import org.apache.doris.filesystem.spi.FileEntry;
+import org.apache.doris.filesystem.spi.FileIterator;
+import org.apache.doris.filesystem.spi.Location;
+import org.apache.doris.filesystem.spi.ObjFileSystem;
+import org.apache.doris.filesystem.spi.RemoteObject;
+import org.apache.doris.filesystem.spi.RemoteObjects;
+import org.apache.doris.filesystem.spi.RequestBody;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Azure Blob Storage-backed {@link org.apache.doris.filesystem.spi.FileSystem} implementation.
+ *
+ * <p>Does not depend on fe-core, fe-common, or fe-catalog.
+ * Azure does not support atomic directory renames; {@link #rename} is limited to single blobs.
+ */
+public class AzureFileSystem extends ObjFileSystem {
+
+ private static final Logger LOG = LogManager.getLogger(AzureFileSystem.class);
+
+ private static final String DIR_MARKER_SUFFIX = "/";
+
+ public AzureFileSystem(AzureObjStorage objStorage) {
+ super("AZURE", objStorage);
+ }
+
+ @Override
+ protected boolean isNotFoundError(IOException e) {
+ return e instanceof java.io.FileNotFoundException
+ || (e.getMessage() != null && e.getMessage().contains("404"));
+ }
+
+ @Override
+ public void mkdirs(Location location) throws IOException {
+ // Azure is a flat namespace; create a zero-byte directory marker for compatibility.
+ String path = location.uri().endsWith(DIR_MARKER_SUFFIX)
+ ? location.uri()
+ : location.uri() + DIR_MARKER_SUFFIX;
+ objStorage.putObject(path, RequestBody.of(InputStream.nullInputStream(), 0));
+ }
+
+ @Override
+ public void delete(Location location, boolean recursive) throws IOException {
+ if (recursive) {
+ String prefix = location.uri().endsWith(DIR_MARKER_SUFFIX)
+ ? location.uri() : location.uri() + DIR_MARKER_SUFFIX;
+ deleteRecursive(prefix);
+ }
+ try {
+ objStorage.deleteObject(location.uri());
+ } catch (IOException e) {
+ if (!isNotFoundError(e)) {
+ throw e;
+ }
+ }
+ }
+
+ private void deleteRecursive(String prefix) throws IOException {
+ String continuationToken = null;
+ do {
+ RemoteObjects batch = objStorage.listObjects(prefix, continuationToken);
+ for (RemoteObject obj : batch.getObjectList()) {
+ objStorage.deleteObject(rebuildUri(prefix, obj.getKey()));
+ }
+ continuationToken = batch.isTruncated() ? batch.getContinuationToken() : null;
+ } while (continuationToken != null);
+ }
+
+ /**
+ * Azure Blob Storage does not support atomic directory renames.
+ * Single-blob renames are supported via copy-then-delete.
+ *
+ * @throws UnsupportedOperationException if the source appears to be a directory prefix
+ */
+ @Override
+ public void rename(Location src, Location dst) throws IOException {
+ if (src.uri().endsWith(DIR_MARKER_SUFFIX)) {
+ throw new UnsupportedOperationException(
+ "Renaming directories is not supported in Azure Blob Storage.");
+ }
+ objStorage.copyObject(src.uri(), dst.uri());
+ objStorage.deleteObject(src.uri());
+ }
+
+ @Override
+ public FileIterator list(Location location) throws IOException {
+ return new AzureFileIterator(location.uri());
+ }
+
+ @Override
+ public DorisInputFile newInputFile(Location location) throws IOException {
+ return new AzureInputFile(location);
+ }
+
+ @Override
+ public DorisOutputFile newOutputFile(Location location) throws IOException {
+ return new AzureOutputFile(location);
+ }
+
+ private static String rebuildUri(String prefix, String key) {
+ // prefix is like "wasbs://container@account.blob.core.windows.net/path/",
+ // key is the full blob name; reconstruct the full URI for the key.
+ int schemeEnd = prefix.indexOf("://");
+ if (schemeEnd < 0) {
+ return prefix + key;
+ }
+ String scheme = prefix.substring(0, schemeEnd);
+ String withoutScheme = prefix.substring(schemeEnd + 3);
+ // For wasb/abfs: "container@account.host/path/"
+ int firstSlash = withoutScheme.indexOf('/');
+ if (firstSlash < 0) {
+ return prefix + key;
+ }
+ String authority = withoutScheme.substring(0, firstSlash);
+ return scheme + "://" + authority + "/" + key;
+ }
+
+ /** Lazy-paginating FileIterator over Azure list results. */
+ private class AzureFileIterator implements FileIterator {
+ private final String prefix;
+ private String continuationToken;
+ private List<FileEntry> buffer = new ArrayList<>();
+ private int bufferIdx = 0;
+ private boolean done = false;
+
+ AzureFileIterator(String prefix) {
+ this.prefix = prefix;
+ }
+
+ @Override
+ public boolean hasNext() throws IOException {
+ if (bufferIdx < buffer.size()) {
+ return true;
+ }
+ if (done) {
+ return false;
+ }
+ fetchNextPage();
+ return bufferIdx < buffer.size();
+ }
+
+ private void fetchNextPage() throws IOException {
+ RemoteObjects page = objStorage.listObjects(prefix, continuationToken);
+ buffer = new ArrayList<>();
+ bufferIdx = 0;
+ for (RemoteObject obj : page.getObjectList()) {
+ Location loc = Location.of(rebuildUri(prefix, obj.getKey()));
+ buffer.add(new FileEntry(loc, obj.getSize(), false, List.of()));
+ }
+ if (page.isTruncated()) {
+ continuationToken = page.getContinuationToken();
+ } else {
+ done = true;
+ }
+ }
+
+ @Override
+ public FileEntry next() throws IOException {
+ return buffer.get(bufferIdx++);
+ }
+
+ @Override
+ public void close() throws IOException {
+ // no-op
+ }
+ }
+
+ /** Azure-backed DorisInputFile. */
+ private class AzureInputFile implements DorisInputFile {
+ private final Location location;
+
+ AzureInputFile(Location location) {
+ this.location = location;
+ }
+
+ @Override
+ public Location location() {
+ return location;
+ }
+
+ @Override
+ public long length() throws IOException {
+ return objStorage.headObject(location.uri()).getSize();
+ }
+
+ @Override
+ public InputStream newStream() throws IOException {
+ return ((AzureObjStorage) objStorage).openInputStream(location.uri());
+ }
+ }
+
+ /** Azure-backed DorisOutputFile (buffered in memory; uploads on close). */
+ private class AzureOutputFile implements DorisOutputFile {
+ private final Location location;
+
+ AzureOutputFile(Location location) {
+ this.location = location;
+ }
+
+ @Override
+ public Location location() {
+ return location;
+ }
+
+ @Override
+ public OutputStream create() throws IOException {
+ return createOrOverwrite();
+ }
+
+ @Override
+ public OutputStream createOrOverwrite() throws IOException {
+ return new AzureOutputStream(location.uri(), (AzureObjStorage) objStorage);
+ }
+ }
+
+ /** OutputStream that buffers writes in memory and uploads to Azure Blob on close. */
+ private static class AzureOutputStream extends OutputStream {
+ private final String remotePath;
+ private final AzureObjStorage storage;
+ private final ByteArrayOutputStream buffer = new ByteArrayOutputStream();
+ private boolean closed = false;
+
+ AzureOutputStream(String remotePath, AzureObjStorage storage) {
+ this.remotePath = remotePath;
+ this.storage = storage;
+ }
+
+ @Override
+ public void write(int b) throws IOException {
+ checkNotClosed();
+ buffer.write(b);
+ }
+
+ @Override
+ public void write(byte[] b, int off, int len) throws IOException {
+ checkNotClosed();
+ buffer.write(b, off, len);
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (closed) {
+ return;
+ }
+ closed = true;
+ byte[] data = buffer.toByteArray();
+ storage.putObject(remotePath, RequestBody.of(new ByteArrayInputStream(data), data.length));
+ }
+
+ private void checkNotClosed() throws IOException {
+ if (closed) {
+ throw new IOException("Stream already closed: " + remotePath);
+ }
+ }
+ }
+}
diff --git a/fe/fe-filesystem/fe-filesystem-azure/src/main/java/org/apache/doris/filesystem/azure/AzureFileSystemProvider.java b/fe/fe-filesystem/fe-filesystem-azure/src/main/java/org/apache/doris/filesystem/azure/AzureFileSystemProvider.java
new file mode 100644
index 00000000000..965c409e5fc
--- /dev/null
+++ b/fe/fe-filesystem/fe-filesystem-azure/src/main/java/org/apache/doris/filesystem/azure/AzureFileSystemProvider.java
@@ -0,0 +1,60 @@
+// 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.doris.filesystem.azure;
+
+import org.apache.doris.filesystem.spi.FileSystem;
+import org.apache.doris.filesystem.spi.FileSystemProvider;
+
+import java.io.IOException;
+import java.util.Map;
+
+/**
+ * SPI provider for Azure Blob Storage.
+ *
+ * <p>Registered via META-INF/services/org.apache.doris.filesystem.spi.FileSystemProvider.
+ *
+ * <p>Identified by the presence of {@code AZURE_ACCOUNT_NAME}, {@code azure.account_name},
+ * or an endpoint containing {@code blob.core.windows.net}.
+ */
+public class AzureFileSystemProvider implements FileSystemProvider {
+
+ @Override
+ public boolean supports(Map<String, String> properties) {
+ if (properties.containsKey(AzureObjStorage.PROP_ACCOUNT_NAME)) {
+ return true;
+ }
+ if (properties.containsKey(AzureObjStorage.PROP_ACCOUNT_NAME_ALT)) {
+ return true;
+ }
+ String endpoint = properties.get(AzureObjStorage.PROP_ENDPOINT);
+ if (endpoint == null) {
+ endpoint = properties.get(AzureObjStorage.PROP_ENDPOINT_ALT);
+ }
+ return endpoint != null && endpoint.contains("blob.core.windows.net");
+ }
+
+ @Override
+ public FileSystem create(Map<String, String> properties) throws IOException {
+ return new AzureFileSystem(new AzureObjStorage(properties));
+ }
+
+ @Override
+ public String name() {
+ return "AZURE";
+ }
+}
diff --git a/fe/fe-filesystem/fe-filesystem-azure/src/main/java/org/apache/doris/filesystem/azure/AzureObjStorage.java b/fe/fe-filesystem/fe-filesystem-azure/src/main/java/org/apache/doris/filesystem/azure/AzureObjStorage.java
new file mode 100644
index 00000000000..6d647b8594e
--- /dev/null
+++ b/fe/fe-filesystem/fe-filesystem-azure/src/main/java/org/apache/doris/filesystem/azure/AzureObjStorage.java
@@ -0,0 +1,342 @@
+// 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.doris.filesystem.azure;
+
+import org.apache.doris.filesystem.spi.ObjStorage;
+import org.apache.doris.filesystem.spi.RemoteObject;
+import org.apache.doris.filesystem.spi.RemoteObjects;
+import org.apache.doris.filesystem.spi.RequestBody;
+import org.apache.doris.filesystem.spi.UploadPartResult;
+
+import com.azure.core.http.rest.PagedIterable;
+import com.azure.core.http.rest.PagedResponse;
+import com.azure.identity.ClientSecretCredentialBuilder;
+import com.azure.identity.DefaultAzureCredentialBuilder;
+import com.azure.storage.blob.BlobClient;
+import com.azure.storage.blob.BlobContainerClient;
+import com.azure.storage.blob.BlobServiceClient;
+import com.azure.storage.blob.BlobServiceClientBuilder;
+import com.azure.storage.blob.models.BlobItem;
+import com.azure.storage.blob.models.BlobProperties;
+import com.azure.storage.blob.models.BlobStorageException;
+import com.azure.storage.blob.models.ListBlobsOptions;
+import com.azure.storage.blob.specialized.BlockBlobClient;
+import com.azure.storage.common.StorageSharedKeyCredential;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.ArrayList;
+import java.util.Base64;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Azure Blob Storage implementation of {@link ObjStorage}.
+ *
+ * <p>No dependency on fe-core, fe-common, or fe-catalog.
+ * Accepts a {@code Map<String, String>} of properties with the following keys:
+ * <ul>
+ * <li>{@code AZURE_ACCOUNT_NAME} / {@code azure.account_name} / {@code AWS_ACCESS_KEY}
+ * — storage account name</li>
+ * <li>{@code AZURE_ACCOUNT_KEY} / {@code azure.account_key} / {@code AWS_SECRET_KEY}
+ * — storage account key (shared key auth)</li>
+ * <li>{@code AZURE_ENDPOINT} / {@code AWS_ENDPOINT} — custom endpoint (optional)</li>
+ * <li>{@code AZURE_CLIENT_ID} / {@code azure.oauth2_client_id} — service principal client ID</li>
+ * <li>{@code AZURE_CLIENT_SECRET} / {@code azure.oauth2_client_secret} — service principal secret</li>
+ * <li>{@code AZURE_TENANT_ID} / {@code azure.oauth2_client_tenant_id} — AAD tenant ID</li>
+ * </ul>
+ */
+public class AzureObjStorage implements ObjStorage<BlobServiceClient> {
+
+ private static final Logger LOG = LogManager.getLogger(AzureObjStorage.class);
+
+ static final String PROP_ACCOUNT_NAME = "AZURE_ACCOUNT_NAME";
+ static final String PROP_ACCOUNT_NAME_ALT = "azure.account_name";
+ static final String PROP_ACCOUNT_KEY = "AZURE_ACCOUNT_KEY";
+ static final String PROP_ACCOUNT_KEY_ALT = "azure.account_key";
+ static final String PROP_ENDPOINT = "AZURE_ENDPOINT";
+ static final String PROP_ENDPOINT_ALT = "AWS_ENDPOINT";
+ static final String PROP_CLIENT_ID = "AZURE_CLIENT_ID";
+ static final String PROP_CLIENT_ID_ALT = "azure.oauth2_client_id";
+ static final String PROP_CLIENT_SECRET = "AZURE_CLIENT_SECRET";
+ static final String PROP_CLIENT_SECRET_ALT = "azure.oauth2_client_secret";
+ static final String PROP_TENANT_ID = "AZURE_TENANT_ID";
+ static final String PROP_TENANT_ID_ALT = "azure.oauth2_client_tenant_id";
+
+ private static final String DEFAULT_ENDPOINT_TEMPLATE = "https://%s.blob.core.windows.net";
+ private static final int HTTP_NOT_FOUND = 404;
+
+ private final Map<String, String> properties;
+ private volatile BlobServiceClient client;
+
+ public AzureObjStorage(Map<String, String> properties) {
+ this.properties = Collections.unmodifiableMap(properties);
+ }
+
+ @Override
+ public BlobServiceClient getClient() throws IOException {
+ if (client == null) {
+ synchronized (this) {
+ if (client == null) {
+ client = buildClient();
+ }
+ }
+ }
+ return client;
+ }
+
+ private BlobServiceClient buildClient() throws IOException {
+ String accountName = resolveAccountName();
+ String endpoint = resolveEndpoint(accountName);
+ String accountKey = resolve(PROP_ACCOUNT_KEY, PROP_ACCOUNT_KEY_ALT, null);
+ String clientId = resolve(PROP_CLIENT_ID, PROP_CLIENT_ID_ALT, null);
+ String clientSecret = resolve(PROP_CLIENT_SECRET, PROP_CLIENT_SECRET_ALT, null);
+ String tenantId = resolve(PROP_TENANT_ID, PROP_TENANT_ID_ALT, null);
+
+ BlobServiceClientBuilder builder = new BlobServiceClientBuilder().endpoint(endpoint);
+
+ if (accountKey != null && !accountKey.isEmpty()) {
+ builder.credential(new StorageSharedKeyCredential(accountName, accountKey));
+ } else if (clientId != null && !clientId.isEmpty()
+ && clientSecret != null && !clientSecret.isEmpty()
+ && tenantId != null && !tenantId.isEmpty()) {
+ builder.credential(new ClientSecretCredentialBuilder()
+ .tenantId(tenantId)
+ .clientId(clientId)
+ .clientSecret(clientSecret)
+ .build());
+ } else {
+ builder.credential(new DefaultAzureCredentialBuilder().build());
+ }
+ return builder.buildClient();
+ }
+
+ private String resolveAccountName() throws IOException {
+ String name = resolve(PROP_ACCOUNT_NAME, PROP_ACCOUNT_NAME_ALT, null);
+ if (name == null || name.isEmpty()) {
+ // Fall back to AWS_ACCESS_KEY for S3-compat configurations
+ name = properties.get("AWS_ACCESS_KEY");
+ }
+ if (name == null || name.isEmpty()) {
+ throw new IOException("Azure account name is required. Set " + PROP_ACCOUNT_NAME);
+ }
+ return name;
+ }
+
+ private String resolveEndpoint(String accountName) {
+ String endpoint = resolve(PROP_ENDPOINT, PROP_ENDPOINT_ALT, null);
+ if (endpoint != null && !endpoint.isEmpty()) {
+ if (!endpoint.startsWith("http")) {
+ endpoint = "https://" + endpoint;
+ }
+ return endpoint;
+ }
+ return String.format(DEFAULT_ENDPOINT_TEMPLATE, accountName);
+ }
+
+ private String resolve(String primaryKey, String altKey, String defaultValue) {
+ String value = properties.get(primaryKey);
+ if (value != null && !value.isEmpty()) {
+ return value;
+ }
+ if (altKey != null) {
+ value = properties.get(altKey);
+ if (value != null && !value.isEmpty()) {
+ return value;
+ }
+ }
+ return defaultValue;
+ }
+
+ @Override
+ public RemoteObjects listObjects(String remotePath, String continuationToken) throws IOException {
+ try {
+ AzureUri uri = AzureUri.parse(remotePath);
+ ListBlobsOptions options = new ListBlobsOptions().setPrefix(uri.key());
+ BlobContainerClient containerClient = getClient().getBlobContainerClient(uri.container());
+ PagedIterable<BlobItem> pagedBlobs = containerClient.listBlobs(options, continuationToken, null);
+ PagedResponse<BlobItem> page = pagedBlobs.iterableByPage().iterator().next();
+
+ List<RemoteObject> objects = new ArrayList<>();
+ for (BlobItem item : page.getElements()) {
+ objects.add(new RemoteObject(
+ item.getName(),
+ item.getName().startsWith(uri.key()) ? item.getName().substring(uri.key().length()) : "",
+ item.getProperties().getETag(),
+ item.getProperties().getContentLength() != null
+ ? item.getProperties().getContentLength() : 0L));
+ }
+ String nextToken = page.getContinuationToken();
+ return new RemoteObjects(objects, nextToken != null && !nextToken.isEmpty(), nextToken);
+ } catch (BlobStorageException e) {
+ throw new IOException("Failed to list Azure objects at " + remotePath + ": " + e.getMessage(), e);
+ }
+ }
+
+ @Override
+ public RemoteObject headObject(String remotePath) throws IOException {
+ try {
+ AzureUri uri = AzureUri.parse(remotePath);
+ BlobClient blobClient = getClient().getBlobContainerClient(uri.container())
+ .getBlobClient(uri.key());
+ BlobProperties props = blobClient.getProperties();
+ return new RemoteObject(uri.key(), "", props.getETag(), props.getBlobSize());
+ } catch (BlobStorageException e) {
+ if (e.getStatusCode() == HTTP_NOT_FOUND) {
+ throw new FileNotFoundException("404: Object not found: " + remotePath);
+ }
+ throw new IOException("headObject failed for " + remotePath + ": " + e.getMessage(), e);
+ }
+ }
+
+ @Override
+ public void putObject(String remotePath, RequestBody requestBody) throws IOException {
+ try {
+ AzureUri uri = AzureUri.parse(remotePath);
+ BlobClient blobClient = getClient().getBlobContainerClient(uri.container())
+ .getBlobClient(uri.key());
+ blobClient.upload(requestBody.content(), requestBody.contentLength(), true);
+ } catch (BlobStorageException e) {
+ throw new IOException("putObject failed for " + remotePath + ": " + e.getMessage(), e);
+ }
+ }
+
+ @Override
+ public void deleteObject(String remotePath) throws IOException {
+ try {
+ AzureUri uri = AzureUri.parse(remotePath);
+ BlobClient blobClient = getClient().getBlobContainerClient(uri.container())
+ .getBlobClient(uri.key());
+ blobClient.delete();
+ } catch (BlobStorageException e) {
+ if (e.getStatusCode() == HTTP_NOT_FOUND) {
+ return;
+ }
+ throw new IOException("deleteObject failed for " + remotePath + ": " + e.getMessage(), e);
+ }
+ }
+
+ @Override
+ public void copyObject(String srcPath, String dstPath) throws IOException {
+ try {
+ AzureUri srcUri = AzureUri.parse(srcPath);
+ AzureUri dstUri = AzureUri.parse(dstPath);
+ BlobClient srcClient = getClient().getBlobContainerClient(srcUri.container())
+ .getBlobClient(srcUri.key());
+ BlobClient dstClient = getClient().getBlobContainerClient(dstUri.container())
+ .getBlobClient(dstUri.key());
+ dstClient.beginCopy(srcClient.getBlobUrl(), null).waitForCompletion();
+ } catch (BlobStorageException e) {
+ throw new IOException("copyObject from " + srcPath + " to " + dstPath
+ + " failed: " + e.getMessage(), e);
+ }
+ }
+
+ @Override
+ public String initiateMultipartUpload(String remotePath) throws IOException {
+ // Azure block blobs don't have an explicit "initiate" API.
+ // Return the path itself as the upload ID; block IDs are derived from part numbers.
+ return remotePath;
+ }
+
+ @Override
+ public UploadPartResult uploadPart(String remotePath, String uploadId, int partNum,
+ RequestBody body) throws IOException {
+ try {
+ AzureUri uri = AzureUri.parse(remotePath);
+ BlockBlobClient blockBlobClient = getClient().getBlobContainerClient(uri.container())
+ .getBlobClient(uri.key()).getBlockBlobClient();
+ String blockId = toBlockId(partNum);
+ blockBlobClient.stageBlock(blockId, body.content(), body.contentLength());
+ return new UploadPartResult(partNum, blockId);
+ } catch (BlobStorageException e) {
+ throw new IOException("uploadPart failed for " + remotePath + " part " + partNum
+ + ": " + e.getMessage(), e);
+ }
+ }
+
+ @Override
+ public void completeMultipartUpload(String remotePath, String uploadId,
+ List<UploadPartResult> parts) throws IOException {
+ try {
+ AzureUri uri = AzureUri.parse(remotePath);
+ BlockBlobClient blockBlobClient = getClient().getBlobContainerClient(uri.container())
+ .getBlobClient(uri.key()).getBlockBlobClient();
+ List<String> blockIds = new ArrayList<>();
+ List<UploadPartResult> sorted = new ArrayList<>(parts);
+ sorted.sort((a, b) -> Integer.compare(a.partNumber(), b.partNumber()));
+ for (UploadPartResult part : sorted) {
+ blockIds.add(toBlockId(part.partNumber()));
+ }
+ blockBlobClient.commitBlockList(blockIds);
+ } catch (BlobStorageException e) {
+ throw new IOException("completeMultipartUpload failed for " + remotePath
+ + ": " + e.getMessage(), e);
+ }
+ }
+
+ @Override
+ public void abortMultipartUpload(String remotePath, String uploadId) throws IOException {
+ // Azure automatically discards uncommitted blocks; no explicit abort is needed.
+ LOG.warn("abortMultipartUpload called for {}; uncommitted blocks will expire automatically.",
+ remotePath);
+ }
+
+ /**
+ * Opens an InputStream to download the blob at the given path.
+ * Used by {@link AzureFileSystem} for read operations.
+ */
+ InputStream openInputStream(String remotePath) throws IOException {
+ try {
+ AzureUri uri = AzureUri.parse(remotePath);
+ BlobClient blobClient = getClient().getBlobContainerClient(uri.container())
+ .getBlobClient(uri.key());
+ return blobClient.openInputStream();
+ } catch (BlobStorageException e) {
+ if (e.getStatusCode() == HTTP_NOT_FOUND) {
+ throw new FileNotFoundException("Object not found: " + remotePath);
+ }
+ throw new IOException("openInputStream failed for " + remotePath + ": " + e.getMessage(), e);
+ }
+ }
+
+ @Override
+ public Map<String, String> getProperties() {
+ return properties;
+ }
+
+ @Override
+ public void close() throws IOException {
+ // BlobServiceClient is not Closeable and does not require explicit closing.
+ }
+
+ /**
+ * Converts a part number to a Base64-encoded block ID using little-endian byte order,
+ * consistent with the existing fe-core Azure multipart upload implementation.
+ */
+ private static String toBlockId(int partNum) {
+ byte[] bytes = ByteBuffer.allocate(4).order(ByteOrder.LITTLE_ENDIAN).putInt(partNum).array();
+ return Base64.getEncoder().encodeToString(bytes);
+ }
+}
diff --git a/fe/fe-filesystem/fe-filesystem-azure/src/main/java/org/apache/doris/filesystem/azure/AzureUri.java b/fe/fe-filesystem/fe-filesystem-azure/src/main/java/org/apache/doris/filesystem/azure/AzureUri.java
new file mode 100644
index 00000000000..845a14e90bc
--- /dev/null
+++ b/fe/fe-filesystem/fe-filesystem-azure/src/main/java/org/apache/doris/filesystem/azure/AzureUri.java
@@ -0,0 +1,146 @@
+// 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.doris.filesystem.azure;
+
+import java.io.IOException;
+
+/**
+ * Parsed Azure Blob Storage URI.
+ *
+ * <p>Supported formats:
+ * <ul>
+ * <li>wasb[s]://container@account.blob.core.windows.net/path</li>
+ * <li>abfs[s]://container@account.dfs.core.windows.net/path</li>
+ * <li>https://account.blob.core.windows.net/container/path</li>
+ * <li>s3[a|n]://container/key (S3-compatibility mode)</li>
+ * </ul>
+ */
+public final class AzureUri {
+
+ private final String scheme;
+ private final String accountName;
+ private final String container;
+ private final String key;
+
+ private AzureUri(String scheme, String accountName, String container, String key) {
+ this.scheme = scheme;
+ this.accountName = accountName;
+ this.container = container;
+ this.key = key;
+ }
+
+ /**
+ * Parses an Azure path into its components.
+ *
+ * @param path the Azure storage path
+ * @return parsed AzureUri
+ * @throws IOException if the path cannot be parsed
+ */
+ public static AzureUri parse(String path) throws IOException {
+ if (path == null || path.isEmpty()) {
+ throw new IOException("Azure path must not be null or empty");
+ }
+ int schemeEnd = path.indexOf("://");
+ if (schemeEnd < 0) {
+ throw new IOException("Cannot parse Azure URI without scheme: " + path);
+ }
+ String scheme = path.substring(0, schemeEnd).toLowerCase();
+ String rest = path.substring(schemeEnd + 3);
+
+ if (scheme.equals("wasb") || scheme.equals("wasbs")
+ || scheme.equals("abfs") || scheme.equals("abfss")) {
+ return parseWasbAbfs(scheme, rest, path);
+ } else if (scheme.equals("https") || scheme.equals("http")) {
+ return parseHttps(scheme, rest);
+ } else if (scheme.equals("s3") || scheme.equals("s3a") || scheme.equals("s3n")) {
+ return parseS3Compat(scheme, rest);
+ } else {
+ throw new IOException("Unsupported Azure URI scheme '" + scheme + "' in path: " + path);
+ }
+ }
+
+ private static AzureUri parseWasbAbfs(String scheme, String rest, String original) throws IOException {
+ // wasb://container@account.blob.core.windows.net/path
+ int atIdx = rest.indexOf('@');
+ if (atIdx < 0) {
+ throw new IOException("Invalid Azure URI format (missing '@'): " + original);
+ }
+ String container = rest.substring(0, atIdx);
+ String hostAndPath = rest.substring(atIdx + 1);
+ int slashIdx = hostAndPath.indexOf('/');
+ String host = slashIdx < 0 ? hostAndPath : hostAndPath.substring(0, slashIdx);
+ String key = slashIdx < 0 ? "" : hostAndPath.substring(slashIdx + 1);
+ String accountName = host.contains(".") ? host.substring(0, host.indexOf('.')) : host;
+ return new AzureUri(scheme, accountName, container, key);
+ }
+
+ private static AzureUri parseHttps(String scheme, String rest) {
+ // https://account.blob.core.windows.net/container/path
+ int slashIdx = rest.indexOf('/');
+ String host = slashIdx < 0 ? rest : rest.substring(0, slashIdx);
+ String pathAfterHost = slashIdx < 0 ? "" : rest.substring(slashIdx + 1);
+ String accountName = host.contains(".") ? host.substring(0, host.indexOf('.')) : host;
+ int containerEnd = pathAfterHost.indexOf('/');
+ String container;
+ String key;
+ if (containerEnd < 0) {
+ container = pathAfterHost;
+ key = "";
+ } else {
+ container = pathAfterHost.substring(0, containerEnd);
+ key = pathAfterHost.substring(containerEnd + 1);
+ }
+ return new AzureUri(scheme, accountName, container, key);
+ }
+
+ private static AzureUri parseS3Compat(String scheme, String rest) {
+ // s3://container/key — S3-compatibility mode; accountName from properties
+ int slashIdx = rest.indexOf('/');
+ String container;
+ String key;
+ if (slashIdx < 0) {
+ container = rest;
+ key = "";
+ } else {
+ container = rest.substring(0, slashIdx);
+ key = rest.substring(slashIdx + 1);
+ }
+ return new AzureUri(scheme, "", container, key);
+ }
+
+ public String scheme() {
+ return scheme;
+ }
+
+ public String accountName() {
+ return accountName;
+ }
+
+ public String container() {
+ return container;
+ }
+
+ public String key() {
+ return key;
+ }
+
+ @Override
+ public String toString() {
+ return scheme + "://" + container + "@" + accountName + "/" + key;
+ }
+}
diff --git a/fe/fe-filesystem/fe-filesystem-azure/src/main/resources/META-INF/services/org.apache.doris.filesystem.spi.FileSystemProvider b/fe/fe-filesystem/fe-filesystem-azure/src/main/resources/META-INF/services/org.apache.doris.filesystem.spi.FileSystemProvider
new file mode 100644
index 00000000000..e681d7b7985
--- /dev/null
+++ b/fe/fe-filesystem/fe-filesystem-azure/src/main/resources/META-INF/services/org.apache.doris.filesystem.spi.FileSystemProvider
@@ -0,0 +1 @@
+org.apache.doris.filesystem.azure.AzureFileSystemProvider
diff --git a/fe/fe-filesystem/fe-filesystem-cos/pom.xml b/fe/fe-filesystem/fe-filesystem-cos/pom.xml
new file mode 100644
index 00000000000..fd7b194e5b1
--- /dev/null
+++ b/fe/fe-filesystem/fe-filesystem-cos/pom.xml
@@ -0,0 +1,44 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.doris</groupId>
+ <artifactId>fe-filesystem</artifactId>
+ <version>${revision}</version>
+ <relativePath>../pom.xml</relativePath>
+ </parent>
+
+ <artifactId>fe-filesystem-cos</artifactId>
+ <packaging>jar</packaging>
+ <name>Doris FE Filesystem - COS</name>
+ <description>Tencent Cloud COS filesystem provider for Doris FE (delegates to S3FileSystem).</description>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.doris</groupId>
+ <artifactId>fe-filesystem-s3</artifactId>
+ <version>${revision}</version>
+ </dependency>
+ </dependencies>
+</project>
diff --git a/fe/fe-filesystem/fe-filesystem-cos/src/main/java/org/apache/doris/filesystem/cos/CosFileSystemProvider.java b/fe/fe-filesystem/fe-filesystem-cos/src/main/java/org/apache/doris/filesystem/cos/CosFileSystemProvider.java
new file mode 100644
index 00000000000..0bd6bebca16
--- /dev/null
+++ b/fe/fe-filesystem/fe-filesystem-cos/src/main/java/org/apache/doris/filesystem/cos/CosFileSystemProvider.java
@@ -0,0 +1,68 @@
+// 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.doris.filesystem.cos;
+
+import org.apache.doris.filesystem.s3.S3FileSystem;
+import org.apache.doris.filesystem.s3.S3ObjStorage;
+import org.apache.doris.filesystem.spi.FileSystem;
+import org.apache.doris.filesystem.spi.FileSystemProvider;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * SPI provider for Tencent Cloud COS.
+ *
+ * <p>Registered via META-INF/services/org.apache.doris.filesystem.spi.FileSystemProvider.
+ *
+ * <p>Identified by an endpoint containing {@code myqcloud.com}. Translates COS-specific
+ * property keys to S3-compatible keys and delegates to {@link S3FileSystem}.
+ */
+public class CosFileSystemProvider implements FileSystemProvider {
+
+ @Override
+ public boolean supports(Map<String, String> properties) {
+ String endpoint = properties.get("COS_ENDPOINT");
+ if (endpoint == null) {
+ endpoint = properties.get("AWS_ENDPOINT");
+ }
+ return endpoint != null && endpoint.contains("myqcloud.com");
+ }
+
+ @Override
+ public FileSystem create(Map<String, String> properties) throws IOException {
+ Map<String, String> s3Props = new HashMap<>(properties);
+ if (properties.containsKey("COS_ENDPOINT")) {
+ s3Props.put("AWS_ENDPOINT", properties.get("COS_ENDPOINT"));
+ }
+ if (properties.containsKey("COS_ACCESS_KEY")) {
+ s3Props.put("AWS_ACCESS_KEY", properties.get("COS_ACCESS_KEY"));
+ }
+ if (properties.containsKey("COS_SECRET_KEY")) {
+ s3Props.put("AWS_SECRET_KEY", properties.get("COS_SECRET_KEY"));
+ }
+ s3Props.put("use_path_style", "false");
+ return new S3FileSystem(new S3ObjStorage(s3Props));
+ }
+
+ @Override
+ public String name() {
+ return "COS";
+ }
+}
diff --git a/fe/fe-filesystem/fe-filesystem-cos/src/main/resources/META-INF/services/org.apache.doris.filesystem.spi.FileSystemProvider b/fe/fe-filesystem/fe-filesystem-cos/src/main/resources/META-INF/services/org.apache.doris.filesystem.spi.FileSystemProvider
new file mode 100644
index 00000000000..fbeb5a46fe9
--- /dev/null
+++ b/fe/fe-filesystem/fe-filesystem-cos/src/main/resources/META-INF/services/org.apache.doris.filesystem.spi.FileSystemProvider
@@ -0,0 +1 @@
+org.apache.doris.filesystem.cos.CosFileSystemProvider
diff --git a/fe/fe-filesystem/fe-filesystem-hdfs/pom.xml b/fe/fe-filesystem/fe-filesystem-hdfs/pom.xml
new file mode 100644
index 00000000000..a3c5c95cfdf
--- /dev/null
+++ b/fe/fe-filesystem/fe-filesystem-hdfs/pom.xml
@@ -0,0 +1,75 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.doris</groupId>
+ <artifactId>fe-filesystem</artifactId>
+ <version>${revision}</version>
+ <relativePath>../pom.xml</relativePath>
+ </parent>
+
+ <artifactId>fe-filesystem-hdfs</artifactId>
+ <packaging>jar</packaging>
+ <name>Doris FE Filesystem - HDFS / ViewFS / OSS-HDFS</name>
+ <description>HDFS, ViewFS, JFS, OFS, and OSSHdfs filesystem implementation for Doris FE.</description>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.doris</groupId>
+ <artifactId>fe-filesystem-spi</artifactId>
+ <version>${revision}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-client</artifactId>
+ <exclusions>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j12</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>log4j</groupId>
+ <artifactId>log4j</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-auth</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-api</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+</project>
diff --git a/fe/fe-filesystem/fe-filesystem-hdfs/src/main/java/org/apache/doris/filesystem/hdfs/DFSFileSystem.java b/fe/fe-filesystem/fe-filesystem-hdfs/src/main/java/org/apache/doris/filesystem/hdfs/DFSFileSystem.java
new file mode 100644
index 00000000000..bbc3bd45438
--- /dev/null
+++ b/fe/fe-filesystem/fe-filesystem-hdfs/src/main/java/org/apache/doris/filesystem/hdfs/DFSFileSystem.java
@@ -0,0 +1,149 @@
+// 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.doris.filesystem.hdfs;
+
+import org.apache.doris.filesystem.spi.DorisInputFile;
+import org.apache.doris.filesystem.spi.DorisOutputFile;
+import org.apache.doris.filesystem.spi.FileIterator;
+import org.apache.doris.filesystem.spi.HadoopAuthenticator;
+import org.apache.doris.filesystem.spi.Location;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.Path;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * SPI FileSystem implementation for HDFS, ViewFS, JFS, OFS, and OSSHdfs.
+ * Has zero dependency on fe-core; accepts {@code Map<String, String>} configuration.
+ */
+public class DFSFileSystem implements org.apache.doris.filesystem.spi.FileSystem {
+
+ private static final Logger LOG = LogManager.getLogger(DFSFileSystem.class);
+
+ private final Map<String, String> properties;
+ private final Configuration conf;
+ private final HadoopAuthenticator authenticator;
+ private volatile org.apache.hadoop.fs.FileSystem hadoopFs;
+ private final AtomicBoolean closed = new AtomicBoolean(false);
+
+ public DFSFileSystem(Map<String, String> properties) {
+ this.properties = Collections.unmodifiableMap(new HashMap<>(properties));
+ this.conf = HdfsConfigBuilder.build(properties);
+ if (HdfsConfigBuilder.isKerberosEnabled(properties)) {
+ this.authenticator = new KerberosHadoopAuthenticator(
+ properties.get(HdfsConfigBuilder.KEY_PRINCIPAL),
+ properties.get(HdfsConfigBuilder.KEY_KEYTAB),
+ conf);
+ } else {
+ this.authenticator = new SimpleHadoopAuthenticator();
+ }
+ }
+
+ private org.apache.hadoop.fs.FileSystem getHadoopFs(Path path) throws IOException {
+ if (closed.get()) {
+ throw new IOException("DFSFileSystem is closed.");
+ }
+ if (hadoopFs == null) {
+ synchronized (this) {
+ if (closed.get()) {
+ throw new IOException("DFSFileSystem is closed.");
+ }
+ if (hadoopFs == null) {
+ hadoopFs = authenticator.doAs(
+ () -> org.apache.hadoop.fs.FileSystem.get(path.toUri(), conf));
+ }
+ }
+ }
+ return hadoopFs;
+ }
+
+ /** Package-private accessor for HdfsInputFile and HdfsOutputFile. */
+ org.apache.hadoop.fs.FileSystem requireFs(Path path) throws IOException {
+ return getHadoopFs(path);
+ }
+
+ @Override
+ public boolean exists(Location location) throws IOException {
+ Path path = new Path(location.toString());
+ return authenticator.doAs(() -> getHadoopFs(path).exists(path));
+ }
+
+ @Override
+ public void mkdirs(Location location) throws IOException {
+ Path path = new Path(location.toString());
+ authenticator.doAs(() -> {
+ getHadoopFs(path).mkdirs(path);
+ return null;
+ });
+ }
+
+ @Override
+ public void delete(Location location, boolean recursive) throws IOException {
+ Path path = new Path(location.toString());
+ authenticator.doAs(() -> {
+ getHadoopFs(path).delete(path, recursive);
+ return null;
+ });
+ }
+
+ @Override
+ public void rename(Location src, Location dst) throws IOException {
+ Path srcPath = new Path(src.toString());
+ Path dstPath = new Path(dst.toString());
+ authenticator.doAs(() -> {
+ getHadoopFs(srcPath).rename(srcPath, dstPath);
+ return null;
+ });
+ }
+
+ @Override
+ public FileIterator list(Location location) throws IOException {
+ Path path = new Path(location.toString());
+ FileStatus[] statuses = authenticator.doAs(() -> getHadoopFs(path).listStatus(path));
+ return new HdfsFileIterator(statuses);
+ }
+
+ @Override
+ public DorisInputFile newInputFile(Location location) throws IOException {
+ Path path = new Path(location.toString());
+ return new HdfsInputFile(path, authenticator, this);
+ }
+
+ @Override
+ public DorisOutputFile newOutputFile(Location location) throws IOException {
+ Path path = new Path(location.toString());
+ return new HdfsOutputFile(path, authenticator, this);
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (closed.compareAndSet(false, true)) {
+ if (hadoopFs != null) {
+ hadoopFs.close();
+ }
+ }
+ }
+}
diff --git a/fe/fe-filesystem/fe-filesystem-hdfs/src/main/java/org/apache/doris/filesystem/hdfs/HdfsConfigBuilder.java b/fe/fe-filesystem/fe-filesystem-hdfs/src/main/java/org/apache/doris/filesystem/hdfs/HdfsConfigBuilder.java
new file mode 100644
index 00000000000..9c6a1bc3b99
--- /dev/null
+++ b/fe/fe-filesystem/fe-filesystem-hdfs/src/main/java/org/apache/doris/filesystem/hdfs/HdfsConfigBuilder.java
@@ -0,0 +1,58 @@
+// 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.doris.filesystem.hdfs;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdfs.HdfsConfiguration;
+
+import java.util.Map;
+
+/**
+ * Builds a Hadoop {@link Configuration} from a {@code Map<String, String>} of properties.
+ */
+public class HdfsConfigBuilder {
+
+ static final String KEY_PRINCIPAL = "hadoop.kerberos.principal";
+ static final String KEY_KEYTAB = "hadoop.kerberos.keytab";
+
+ private HdfsConfigBuilder() {
+ }
+
+ /**
+ * Builds a Hadoop Configuration from the given properties map.
+ * FS caching is disabled so that DFSFileSystem manages the lifecycle manually.
+ */
+ public static Configuration build(Map<String, String> properties) {
+ Configuration conf = new HdfsConfiguration();
+ conf.setBoolean("fs.hdfs.impl.disable.cache", true);
+ conf.setBoolean("fs.AbstractFileSystem.hdfs.impl.disable.cache", true);
+ properties.forEach((k, v) -> {
+ if (v != null && !v.isEmpty()) {
+ conf.set(k, v);
+ }
+ });
+ return conf;
+ }
+
+ /**
+ * Returns true if both Kerberos principal and keytab are present in the properties.
+ */
+ public static boolean isKerberosEnabled(Map<String, String> properties) {
+ return properties.containsKey(KEY_PRINCIPAL) && properties.containsKey(KEY_KEYTAB);
+ }
+}
diff --git a/fe/fe-filesystem/fe-filesystem-hdfs/src/main/java/org/apache/doris/filesystem/hdfs/HdfsFileIterator.java b/fe/fe-filesystem/fe-filesystem-hdfs/src/main/java/org/apache/doris/filesystem/hdfs/HdfsFileIterator.java
new file mode 100644
index 00000000000..8df248142ca
--- /dev/null
+++ b/fe/fe-filesystem/fe-filesystem-hdfs/src/main/java/org/apache/doris/filesystem/hdfs/HdfsFileIterator.java
@@ -0,0 +1,56 @@
+// 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.doris.filesystem.hdfs;
+
+import org.apache.doris.filesystem.spi.FileEntry;
+import org.apache.doris.filesystem.spi.FileIterator;
+import org.apache.doris.filesystem.spi.Location;
+
+import org.apache.hadoop.fs.FileStatus;
+
+import java.io.IOException;
+
+/**
+ * {@link FileIterator} backed by a Hadoop {@link FileStatus} array from a single listStatus call.
+ */
+class HdfsFileIterator implements FileIterator {
+
+ private final FileStatus[] statuses;
+ private int index = 0;
+
+ HdfsFileIterator(FileStatus[] statuses) {
+ this.statuses = statuses;
+ }
+
+ @Override
+ public boolean hasNext() throws IOException {
+ return index < statuses.length;
+ }
+
+ @Override
+ public FileEntry next() throws IOException {
+ FileStatus status = statuses[index++];
+ Location loc = Location.of(status.getPath().toString());
+ return new FileEntry(loc, status.getLen(), status.isDirectory(), null);
+ }
+
+ @Override
+ public void close() throws IOException {
+ // no-op: statuses array is already fully loaded
+ }
+}
diff --git a/fe/fe-filesystem/fe-filesystem-hdfs/src/main/java/org/apache/doris/filesystem/hdfs/HdfsFileSystemProvider.java b/fe/fe-filesystem/fe-filesystem-hdfs/src/main/java/org/apache/doris/filesystem/hdfs/HdfsFileSystemProvider.java
new file mode 100644
index 00000000000..b6785d52a9e
--- /dev/null
+++ b/fe/fe-filesystem/fe-filesystem-hdfs/src/main/java/org/apache/doris/filesystem/hdfs/HdfsFileSystemProvider.java
@@ -0,0 +1,61 @@
+// 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.doris.filesystem.hdfs;
+
+import org.apache.doris.filesystem.spi.FileSystem;
+import org.apache.doris.filesystem.spi.FileSystemProvider;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * SPI provider for HDFS-family filesystems: hdfs, viewfs, ofs, jfs, oss.
+ * Registered via META-INF/services for Java ServiceLoader discovery.
+ */
+public class HdfsFileSystemProvider implements FileSystemProvider {
+
+ private static final Set<String> SUPPORTED_SCHEMES = Set.of("hdfs", "viewfs", "ofs", "jfs", "oss");
+
+ @Override
+ public boolean supports(Map<String, String> properties) {
+ String uri = properties.get("HDFS_URI");
+ if (uri == null) {
+ uri = properties.get("fs.defaultFS");
+ }
+ if (uri != null) {
+ int schemeEnd = uri.indexOf("://");
+ if (schemeEnd > 0) {
+ String scheme = uri.substring(0, schemeEnd).toLowerCase();
+ return SUPPORTED_SCHEMES.contains(scheme);
+ }
+ }
+ return properties.containsKey("dfs.nameservices")
+ || properties.containsKey("hadoop.kerberos.principal");
+ }
+
+ @Override
+ public FileSystem create(Map<String, String> properties) throws IOException {
+ return new DFSFileSystem(properties);
+ }
+
+ @Override
+ public String name() {
+ return "HDFS";
+ }
+}
diff --git a/fe/fe-filesystem/fe-filesystem-hdfs/src/main/java/org/apache/doris/filesystem/hdfs/HdfsInputFile.java b/fe/fe-filesystem/fe-filesystem-hdfs/src/main/java/org/apache/doris/filesystem/hdfs/HdfsInputFile.java
new file mode 100644
index 00000000000..4ba3c2a477b
--- /dev/null
+++ b/fe/fe-filesystem/fe-filesystem-hdfs/src/main/java/org/apache/doris/filesystem/hdfs/HdfsInputFile.java
@@ -0,0 +1,60 @@
+// 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.doris.filesystem.hdfs;
+
+import org.apache.doris.filesystem.spi.DorisInputFile;
+import org.apache.doris.filesystem.spi.HadoopAuthenticator;
+import org.apache.doris.filesystem.spi.Location;
+
+import org.apache.hadoop.fs.Path;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+/**
+ * HDFS-backed {@link DorisInputFile} that opens an {@link InputStream} via Hadoop FSDataInputStream.
+ */
+class HdfsInputFile implements DorisInputFile {
+
+ private final Path path;
+ private final HadoopAuthenticator authenticator;
+ private final DFSFileSystem dfs;
+ private final Location location;
+
+ HdfsInputFile(Path path, HadoopAuthenticator authenticator, DFSFileSystem dfs) {
+ this.path = path;
+ this.authenticator = authenticator;
+ this.dfs = dfs;
+ this.location = Location.of(path.toString());
+ }
+
+ @Override
+ public Location location() {
+ return location;
+ }
+
+ @Override
+ public long length() throws IOException {
+ return authenticator.doAs(() -> dfs.requireFs(path).getFileStatus(path).getLen());
+ }
+
+ @Override
+ public InputStream newStream() throws IOException {
+ return authenticator.doAs(() -> dfs.requireFs(path).open(path));
+ }
+}
diff --git a/fe/fe-filesystem/fe-filesystem-hdfs/src/main/java/org/apache/doris/filesystem/hdfs/HdfsOutputFile.java b/fe/fe-filesystem/fe-filesystem-hdfs/src/main/java/org/apache/doris/filesystem/hdfs/HdfsOutputFile.java
new file mode 100644
index 00000000000..62991deaad9
--- /dev/null
+++ b/fe/fe-filesystem/fe-filesystem-hdfs/src/main/java/org/apache/doris/filesystem/hdfs/HdfsOutputFile.java
@@ -0,0 +1,60 @@
+// 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.doris.filesystem.hdfs;
+
+import org.apache.doris.filesystem.spi.DorisOutputFile;
+import org.apache.doris.filesystem.spi.HadoopAuthenticator;
+import org.apache.doris.filesystem.spi.Location;
+
+import org.apache.hadoop.fs.Path;
+
+import java.io.IOException;
+import java.io.OutputStream;
+
+/**
+ * HDFS-backed {@link DorisOutputFile} that creates an {@link OutputStream} via Hadoop FSDataOutputStream.
+ */
+class HdfsOutputFile implements DorisOutputFile {
+
+ private final Path path;
+ private final HadoopAuthenticator authenticator;
+ private final DFSFileSystem dfs;
+ private final Location location;
+
+ HdfsOutputFile(Path path, HadoopAuthenticator authenticator, DFSFileSystem dfs) {
+ this.path = path;
+ this.authenticator = authenticator;
+ this.dfs = dfs;
+ this.location = Location.of(path.toString());
+ }
+
+ @Override
+ public Location location() {
+ return location;
+ }
+
+ @Override
+ public OutputStream create() throws IOException {
+ return authenticator.doAs(() -> dfs.requireFs(path).create(path, false));
+ }
+
+ @Override
+ public OutputStream createOrOverwrite() throws IOException {
+ return authenticator.doAs(() -> dfs.requireFs(path).create(path, true));
+ }
+}
diff --git a/fe/fe-filesystem/fe-filesystem-hdfs/src/main/java/org/apache/doris/filesystem/hdfs/KerberosHadoopAuthenticator.java b/fe/fe-filesystem/fe-filesystem-hdfs/src/main/java/org/apache/doris/filesystem/hdfs/KerberosHadoopAuthenticator.java
new file mode 100644
index 00000000000..5af28e07414
--- /dev/null
+++ b/fe/fe-filesystem/fe-filesystem-hdfs/src/main/java/org/apache/doris/filesystem/hdfs/KerberosHadoopAuthenticator.java
@@ -0,0 +1,65 @@
+// 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.doris.filesystem.hdfs;
+
+import org.apache.doris.filesystem.spi.HadoopAuthenticator;
+import org.apache.doris.filesystem.spi.IOCallable;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.io.IOException;
+import java.security.PrivilegedExceptionAction;
+
+/**
+ * Kerberos-based implementation of {@link HadoopAuthenticator}.
+ * Logs in from a keytab and executes actions as the Kerberos principal via UGI.doAs().
+ */
+public class KerberosHadoopAuthenticator implements HadoopAuthenticator {
+
+ private static final Logger LOG = LogManager.getLogger(KerberosHadoopAuthenticator.class);
+
+ private final String principal;
+ private final String keytab;
+ private volatile UserGroupInformation ugi;
+
+ public KerberosHadoopAuthenticator(String principal, String keytab, Configuration conf) {
+ this.principal = principal;
+ this.keytab = keytab;
+ try {
+ UserGroupInformation.setConfiguration(conf);
+ this.ugi = UserGroupInformation.loginUserFromKeytabAndReturnUGI(principal, keytab);
+ LOG.info("Kerberos login succeeded for principal={}", principal);
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to login with Kerberos principal=" + principal
+ + ", keytab=" + keytab, e);
+ }
+ }
+
+ @Override
+ public <T> T doAs(IOCallable<T> action) throws IOException {
+ try {
+ return ugi.doAs((PrivilegedExceptionAction<T>) action::call);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IOException("Kerberos doAs interrupted for principal=" + principal, e);
+ }
+ }
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemProvider.java b/fe/fe-filesystem/fe-filesystem-hdfs/src/main/java/org/apache/doris/filesystem/hdfs/SimpleHadoopAuthenticator.java
similarity index 61%
copy from fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemProvider.java
copy to fe/fe-filesystem/fe-filesystem-hdfs/src/main/java/org/apache/doris/filesystem/hdfs/SimpleHadoopAuthenticator.java
index 1dd276fc12c..722a19fcdf6 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemProvider.java
+++ b/fe/fe-filesystem/fe-filesystem-hdfs/src/main/java/org/apache/doris/filesystem/hdfs/SimpleHadoopAuthenticator.java
@@ -15,10 +15,21 @@
// specific language governing permissions and limitations
// under the License.
-package org.apache.doris.fs;
+package org.apache.doris.filesystem.hdfs;
-import org.apache.doris.datasource.SessionContext;
+import org.apache.doris.filesystem.spi.HadoopAuthenticator;
+import org.apache.doris.filesystem.spi.IOCallable;
-public interface FileSystemProvider {
- LegacyFileSystemApi get(SessionContext ctx);
+import java.io.IOException;
+
+/**
+ * Simple (non-Kerberos) implementation of {@link HadoopAuthenticator}.
+ * Executes actions directly without any privilege switching.
+ */
+public class SimpleHadoopAuthenticator implements HadoopAuthenticator {
+
+ @Override
+ public <T> T doAs(IOCallable<T> action) throws IOException {
+ return action.call();
+ }
}
diff --git a/fe/fe-filesystem/fe-filesystem-hdfs/src/main/resources/META-INF/services/org.apache.doris.filesystem.spi.FileSystemProvider b/fe/fe-filesystem/fe-filesystem-hdfs/src/main/resources/META-INF/services/org.apache.doris.filesystem.spi.FileSystemProvider
new file mode 100644
index 00000000000..6d09e6fd7d4
--- /dev/null
+++ b/fe/fe-filesystem/fe-filesystem-hdfs/src/main/resources/META-INF/services/org.apache.doris.filesystem.spi.FileSystemProvider
@@ -0,0 +1 @@
+org.apache.doris.filesystem.hdfs.HdfsFileSystemProvider
diff --git a/fe/fe-filesystem/fe-filesystem-local/pom.xml b/fe/fe-filesystem/fe-filesystem-local/pom.xml
new file mode 100644
index 00000000000..89ed1c995dd
--- /dev/null
+++ b/fe/fe-filesystem/fe-filesystem-local/pom.xml
@@ -0,0 +1,44 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
+ http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.doris</groupId>
+ <artifactId>fe-filesystem</artifactId>
+ <version>${revision}</version>
+ <relativePath>../pom.xml</relativePath>
+ </parent>
+
+ <artifactId>fe-filesystem-local</artifactId>
+ <name>Doris FE Filesystem - Local (Testing Only)</name>
+ <description>Local filesystem implementation for unit tests. NOT for production use.</description>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.doris</groupId>
+ <artifactId>fe-filesystem-spi</artifactId>
+ <version>${revision}</version>
+ </dependency>
+ </dependencies>
+</project>
diff --git a/fe/fe-filesystem/fe-filesystem-local/src/main/java/org/apache/doris/filesystem/local/LocalFileSystem.java b/fe/fe-filesystem/fe-filesystem-local/src/main/java/org/apache/doris/filesystem/local/LocalFileSystem.java
new file mode 100644
index 00000000000..6ec90a01a6a
--- /dev/null
+++ b/fe/fe-filesystem/fe-filesystem-local/src/main/java/org/apache/doris/filesystem/local/LocalFileSystem.java
@@ -0,0 +1,180 @@
+// 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.doris.filesystem.local;
+
+import org.apache.doris.filesystem.spi.DorisInputFile;
+import org.apache.doris.filesystem.spi.DorisOutputFile;
+import org.apache.doris.filesystem.spi.FileEntry;
+import org.apache.doris.filesystem.spi.FileIterator;
+import org.apache.doris.filesystem.spi.FileSystem;
+import org.apache.doris.filesystem.spi.Location;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.file.DirectoryStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Local filesystem implementation for unit testing only. Not for production use.
+ */
+public class LocalFileSystem implements FileSystem {
+
+ private final Map<String, String> properties;
+
+ public LocalFileSystem(Map<String, String> properties) {
+ this.properties = properties;
+ }
+
+ private Path toPath(Location location) {
+ String uri = location.toString();
+ if (uri.startsWith("file://")) {
+ return Paths.get(uri.substring("file://".length()));
+ }
+ if (uri.startsWith("local://")) {
+ return Paths.get(uri.substring("local://".length()));
+ }
+ return Paths.get(uri);
+ }
+
+ @Override
+ public boolean exists(Location location) throws IOException {
+ return Files.exists(toPath(location));
+ }
+
+ @Override
+ public void mkdirs(Location location) throws IOException {
+ Files.createDirectories(toPath(location));
+ }
+
+ @Override
+ public void delete(Location location, boolean recursive) throws IOException {
+ Path path = toPath(location);
+ if (recursive && Files.isDirectory(path)) {
+ deleteRecursive(path);
+ } else {
+ Files.deleteIfExists(path);
+ }
+ }
+
+ private void deleteRecursive(Path path) throws IOException {
+ if (Files.isDirectory(path)) {
+ try (DirectoryStream<Path> stream = Files.newDirectoryStream(path)) {
+ for (Path child : stream) {
+ deleteRecursive(child);
+ }
+ }
+ }
+ Files.deleteIfExists(path);
+ }
+
+ @Override
+ public void rename(Location src, Location dst) throws IOException {
+ Files.move(toPath(src), toPath(dst));
+ }
+
+ @Override
+ public FileIterator list(Location location) throws IOException {
+ Path dirPath = toPath(location);
+ List<FileEntry> entries = new ArrayList<>();
+ try (DirectoryStream<Path> stream = Files.newDirectoryStream(dirPath)) {
+ for (Path child : stream) {
+ boolean isDir = Files.isDirectory(child);
+ long length = isDir ? 0L : Files.size(child);
+ entries.add(new FileEntry(Location.of(child.toUri().toString()), length, isDir, null));
+ }
+ }
+ Iterator<FileEntry> it = entries.iterator();
+ return new FileIterator() {
+ @Override
+ public boolean hasNext() {
+ return it.hasNext();
+ }
+
+ @Override
+ public FileEntry next() {
+ return it.next();
+ }
+
+ @Override
+ public void close() {
+ // no-op
+ }
+ };
+ }
+
+ @Override
+ public DorisInputFile newInputFile(Location location) throws IOException {
+ Path path = toPath(location);
+ return new DorisInputFile() {
+ @Override
+ public Location location() {
+ return location;
+ }
+
+ @Override
+ public long length() throws IOException {
+ return Files.size(path);
+ }
+
+ @Override
+ public InputStream newStream() throws IOException {
+ return Files.newInputStream(path);
+ }
+ };
+ }
+
+ @Override
+ public DorisOutputFile newOutputFile(Location location) throws IOException {
+ Path path = toPath(location);
+ return new DorisOutputFile() {
+ @Override
+ public Location location() {
+ return location;
+ }
+
+ @Override
+ public OutputStream create() throws IOException {
+ if (Files.exists(path)) {
+ throw new IOException("File already exists: " + path);
+ }
+ Files.createDirectories(path.getParent());
+ return Files.newOutputStream(path);
+ }
+
+ @Override
+ public OutputStream createOrOverwrite() throws IOException {
+ Files.createDirectories(path.getParent());
+ return Files.newOutputStream(path,
+ java.nio.file.StandardOpenOption.CREATE,
+ java.nio.file.StandardOpenOption.TRUNCATE_EXISTING);
+ }
+ };
+ }
+
+ @Override
+ public void close() throws IOException {
+ // no-op
+ }
+}
diff --git a/fe/fe-filesystem/fe-filesystem-local/src/main/java/org/apache/doris/filesystem/local/LocalFileSystemProvider.java b/fe/fe-filesystem/fe-filesystem-local/src/main/java/org/apache/doris/filesystem/local/LocalFileSystemProvider.java
new file mode 100644
index 00000000000..09fc4734902
--- /dev/null
+++ b/fe/fe-filesystem/fe-filesystem-local/src/main/java/org/apache/doris/filesystem/local/LocalFileSystemProvider.java
@@ -0,0 +1,47 @@
+// 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.doris.filesystem.local;
+
+import org.apache.doris.filesystem.spi.FileSystem;
+import org.apache.doris.filesystem.spi.FileSystemProvider;
+
+import java.io.IOException;
+import java.util.Map;
+
+/**
+ * SPI provider for the local filesystem (file:// and local:// URIs).
+ * For unit-testing only — not intended for production use.
+ */
+public class LocalFileSystemProvider implements FileSystemProvider {
+
+ @Override
+ public boolean supports(Map<String, String> properties) {
+ String uri = properties.getOrDefault("uri", "");
+ return uri.startsWith("file://") || uri.startsWith("local://");
+ }
+
+ @Override
+ public FileSystem create(Map<String, String> properties) throws IOException {
+ return new LocalFileSystem(properties);
+ }
+
+ @Override
+ public String name() {
+ return "LOCAL";
+ }
+}
diff --git a/fe/fe-filesystem/fe-filesystem-local/src/main/resources/META-INF/services/org.apache.doris.filesystem.spi.FileSystemProvider b/fe/fe-filesystem/fe-filesystem-local/src/main/resources/META-INF/services/org.apache.doris.filesystem.spi.FileSystemProvider
new file mode 100644
index 00000000000..6d2fa30065d
--- /dev/null
+++ b/fe/fe-filesystem/fe-filesystem-local/src/main/resources/META-INF/services/org.apache.doris.filesystem.spi.FileSystemProvider
@@ -0,0 +1 @@
+org.apache.doris.filesystem.local.LocalFileSystemProvider
diff --git a/fe/fe-filesystem/fe-filesystem-obs/pom.xml b/fe/fe-filesystem/fe-filesystem-obs/pom.xml
new file mode 100644
index 00000000000..5d24e886abc
--- /dev/null
+++ b/fe/fe-filesystem/fe-filesystem-obs/pom.xml
@@ -0,0 +1,44 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.doris</groupId>
+ <artifactId>fe-filesystem</artifactId>
+ <version>${revision}</version>
+ <relativePath>../pom.xml</relativePath>
+ </parent>
+
+ <artifactId>fe-filesystem-obs</artifactId>
+ <packaging>jar</packaging>
+ <name>Doris FE Filesystem - OBS</name>
+ <description>Huawei Cloud OBS filesystem provider for Doris FE (delegates to S3FileSystem).</description>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.doris</groupId>
+ <artifactId>fe-filesystem-s3</artifactId>
+ <version>${revision}</version>
+ </dependency>
+ </dependencies>
+</project>
diff --git a/fe/fe-filesystem/fe-filesystem-obs/src/main/java/org/apache/doris/filesystem/obs/ObsFileSystemProvider.java b/fe/fe-filesystem/fe-filesystem-obs/src/main/java/org/apache/doris/filesystem/obs/ObsFileSystemProvider.java
new file mode 100644
index 00000000000..fd1b7e9ebc7
--- /dev/null
+++ b/fe/fe-filesystem/fe-filesystem-obs/src/main/java/org/apache/doris/filesystem/obs/ObsFileSystemProvider.java
@@ -0,0 +1,68 @@
+// 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.doris.filesystem.obs;
+
+import org.apache.doris.filesystem.s3.S3FileSystem;
+import org.apache.doris.filesystem.s3.S3ObjStorage;
+import org.apache.doris.filesystem.spi.FileSystem;
+import org.apache.doris.filesystem.spi.FileSystemProvider;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * SPI provider for Huawei Cloud OBS.
+ *
+ * <p>Registered via META-INF/services/org.apache.doris.filesystem.spi.FileSystemProvider.
+ *
+ * <p>Identified by an endpoint containing {@code myhuaweicloud.com}. Translates OBS-specific
+ * property keys to S3-compatible keys and delegates to {@link S3FileSystem}.
+ */
+public class ObsFileSystemProvider implements FileSystemProvider {
+
+ @Override
+ public boolean supports(Map<String, String> properties) {
+ String endpoint = properties.get("OBS_ENDPOINT");
+ if (endpoint == null) {
+ endpoint = properties.get("AWS_ENDPOINT");
+ }
+ return endpoint != null && endpoint.contains("myhuaweicloud.com");
+ }
+
+ @Override
+ public FileSystem create(Map<String, String> properties) throws IOException {
+ Map<String, String> s3Props = new HashMap<>(properties);
+ if (properties.containsKey("OBS_ENDPOINT")) {
+ s3Props.put("AWS_ENDPOINT", properties.get("OBS_ENDPOINT"));
+ }
+ if (properties.containsKey("OBS_ACCESS_KEY")) {
+ s3Props.put("AWS_ACCESS_KEY", properties.get("OBS_ACCESS_KEY"));
+ }
+ if (properties.containsKey("OBS_SECRET_KEY")) {
+ s3Props.put("AWS_SECRET_KEY", properties.get("OBS_SECRET_KEY"));
+ }
+ s3Props.put("use_path_style", "false");
+ return new S3FileSystem(new S3ObjStorage(s3Props));
+ }
+
+ @Override
+ public String name() {
+ return "OBS";
+ }
+}
diff --git a/fe/fe-filesystem/fe-filesystem-obs/src/main/resources/META-INF/services/org.apache.doris.filesystem.spi.FileSystemProvider b/fe/fe-filesystem/fe-filesystem-obs/src/main/resources/META-INF/services/org.apache.doris.filesystem.spi.FileSystemProvider
new file mode 100644
index 00000000000..f25509bef1b
--- /dev/null
+++ b/fe/fe-filesystem/fe-filesystem-obs/src/main/resources/META-INF/services/org.apache.doris.filesystem.spi.FileSystemProvider
@@ -0,0 +1 @@
+org.apache.doris.filesystem.obs.ObsFileSystemProvider
diff --git a/fe/fe-filesystem/fe-filesystem-oss/pom.xml b/fe/fe-filesystem/fe-filesystem-oss/pom.xml
new file mode 100644
index 00000000000..a32a9f015bf
--- /dev/null
+++ b/fe/fe-filesystem/fe-filesystem-oss/pom.xml
@@ -0,0 +1,44 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.doris</groupId>
+ <artifactId>fe-filesystem</artifactId>
+ <version>${revision}</version>
+ <relativePath>../pom.xml</relativePath>
+ </parent>
+
+ <artifactId>fe-filesystem-oss</artifactId>
+ <packaging>jar</packaging>
+ <name>Doris FE Filesystem - OSS</name>
+ <description>Alibaba Cloud OSS filesystem provider for Doris FE (delegates to S3FileSystem).</description>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.doris</groupId>
+ <artifactId>fe-filesystem-s3</artifactId>
+ <version>${revision}</version>
+ </dependency>
+ </dependencies>
+</project>
diff --git a/fe/fe-filesystem/fe-filesystem-oss/src/main/java/org/apache/doris/filesystem/oss/OssFileSystemProvider.java b/fe/fe-filesystem/fe-filesystem-oss/src/main/java/org/apache/doris/filesystem/oss/OssFileSystemProvider.java
new file mode 100644
index 00000000000..90854e7ae93
--- /dev/null
+++ b/fe/fe-filesystem/fe-filesystem-oss/src/main/java/org/apache/doris/filesystem/oss/OssFileSystemProvider.java
@@ -0,0 +1,68 @@
+// 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.doris.filesystem.oss;
+
+import org.apache.doris.filesystem.s3.S3FileSystem;
+import org.apache.doris.filesystem.s3.S3ObjStorage;
+import org.apache.doris.filesystem.spi.FileSystem;
+import org.apache.doris.filesystem.spi.FileSystemProvider;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * SPI provider for Alibaba Cloud OSS.
+ *
+ * <p>Registered via META-INF/services/org.apache.doris.filesystem.spi.FileSystemProvider.
+ *
+ * <p>Identified by an endpoint containing {@code aliyuncs.com}. Translates OSS-specific
+ * property keys to S3-compatible keys and delegates to {@link S3FileSystem}.
+ */
+public class OssFileSystemProvider implements FileSystemProvider {
+
+ @Override
+ public boolean supports(Map<String, String> properties) {
+ String endpoint = properties.get("OSS_ENDPOINT");
+ if (endpoint == null) {
+ endpoint = properties.get("AWS_ENDPOINT");
+ }
+ return endpoint != null && endpoint.contains("aliyuncs.com");
+ }
+
+ @Override
+ public FileSystem create(Map<String, String> properties) throws IOException {
+ Map<String, String> s3Props = new HashMap<>(properties);
+ if (properties.containsKey("OSS_ENDPOINT")) {
+ s3Props.put("AWS_ENDPOINT", properties.get("OSS_ENDPOINT"));
+ }
+ if (properties.containsKey("OSS_ACCESS_KEY")) {
+ s3Props.put("AWS_ACCESS_KEY", properties.get("OSS_ACCESS_KEY"));
+ }
+ if (properties.containsKey("OSS_SECRET_KEY")) {
+ s3Props.put("AWS_SECRET_KEY", properties.get("OSS_SECRET_KEY"));
+ }
+ s3Props.put("use_path_style", "false");
+ return new S3FileSystem(new S3ObjStorage(s3Props));
+ }
+
+ @Override
+ public String name() {
+ return "OSS";
+ }
+}
diff --git a/fe/fe-filesystem/fe-filesystem-oss/src/main/resources/META-INF/services/org.apache.doris.filesystem.spi.FileSystemProvider b/fe/fe-filesystem/fe-filesystem-oss/src/main/resources/META-INF/services/org.apache.doris.filesystem.spi.FileSystemProvider
new file mode 100644
index 00000000000..d73035df232
--- /dev/null
+++ b/fe/fe-filesystem/fe-filesystem-oss/src/main/resources/META-INF/services/org.apache.doris.filesystem.spi.FileSystemProvider
@@ -0,0 +1 @@
+org.apache.doris.filesystem.oss.OssFileSystemProvider
diff --git a/fe/fe-filesystem/fe-filesystem-s3/pom.xml b/fe/fe-filesystem/fe-filesystem-s3/pom.xml
new file mode 100644
index 00000000000..d81de9fd811
--- /dev/null
+++ b/fe/fe-filesystem/fe-filesystem-s3/pom.xml
@@ -0,0 +1,73 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.doris</groupId>
+ <artifactId>fe-filesystem</artifactId>
+ <version>${revision}</version>
+ <relativePath>../pom.xml</relativePath>
+ </parent>
+
+ <artifactId>fe-filesystem-s3</artifactId>
+ <packaging>jar</packaging>
+ <name>Doris FE Filesystem - S3</name>
+ <description>AWS S3 and S3-compatible filesystem implementation for Doris FE.</description>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.doris</groupId>
+ <artifactId>fe-filesystem-spi</artifactId>
+ <version>${revision}</version>
+ </dependency>
+
+ <!-- AWS SDK v2 -->
+ <dependency>
+ <groupId>software.amazon.awssdk</groupId>
+ <artifactId>s3</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>software.amazon.awssdk</groupId>
+ <artifactId>sts</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>software.amazon.awssdk</groupId>
+ <artifactId>auth</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>software.amazon.awssdk</groupId>
+ <artifactId>url-connection-client</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-api</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+</project>
diff --git a/fe/fe-filesystem/fe-filesystem-s3/src/main/java/org/apache/doris/filesystem/s3/S3FileSystem.java b/fe/fe-filesystem/fe-filesystem-s3/src/main/java/org/apache/doris/filesystem/s3/S3FileSystem.java
new file mode 100644
index 00000000000..c6a9672ce4c
--- /dev/null
+++ b/fe/fe-filesystem/fe-filesystem-s3/src/main/java/org/apache/doris/filesystem/s3/S3FileSystem.java
@@ -0,0 +1,232 @@
+// 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.doris.filesystem.s3;
+
+import org.apache.doris.filesystem.spi.DorisInputFile;
+import org.apache.doris.filesystem.spi.DorisOutputFile;
+import org.apache.doris.filesystem.spi.FileEntry;
+import org.apache.doris.filesystem.spi.FileIterator;
+import org.apache.doris.filesystem.spi.Location;
+import org.apache.doris.filesystem.spi.ObjFileSystem;
+import org.apache.doris.filesystem.spi.RemoteObject;
+import org.apache.doris.filesystem.spi.RemoteObjects;
+import org.apache.doris.filesystem.spi.RequestBody;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * S3-backed FileSystem implementation for the Doris FE filesystem SPI.
+ * Does not depend on fe-core, fe-common, or fe-catalog.
+ */
+public class S3FileSystem extends ObjFileSystem {
+
+ private static final Logger LOG = LogManager.getLogger(S3FileSystem.class);
+
+ // S3 does not have real directories; use a zero-byte marker with trailing slash.
+ private static final String DIR_MARKER_SUFFIX = "/";
+
+ public S3FileSystem(S3ObjStorage objStorage) {
+ super("S3", objStorage);
+ }
+
+ @Override
+ protected boolean isNotFoundError(IOException e) {
+ return e instanceof java.io.FileNotFoundException
+ || (e.getMessage() != null && e.getMessage().contains("404"));
+ }
+
+ @Override
+ public void mkdirs(Location location) throws IOException {
+ // S3 is flat; create a zero-byte directory marker for compatibility
+ String path = location.uri().endsWith(DIR_MARKER_SUFFIX)
+ ? location.uri()
+ : location.uri() + DIR_MARKER_SUFFIX;
+ objStorage.putObject(path,
+ RequestBody.of(InputStream.nullInputStream(), 0));
+ }
+
+ @Override
+ public void delete(Location location, boolean recursive) throws IOException {
+ if (recursive) {
+ // Delete all objects under this prefix
+ String prefix = location.uri().endsWith(DIR_MARKER_SUFFIX)
+ ? location.uri() : location.uri() + DIR_MARKER_SUFFIX;
+ deleteRecursive(prefix);
+ }
+ // Always attempt to delete the exact object too
+ try {
+ objStorage.deleteObject(location.uri());
+ } catch (IOException e) {
+ if (!isNotFoundError(e)) {
+ throw e;
+ }
+ }
+ }
+
+ private void deleteRecursive(String prefix) throws IOException {
+ String continuationToken = null;
+ do {
+ RemoteObjects batch = objStorage.listObjects(prefix, continuationToken);
+ for (RemoteObject obj : batch.getObjectList()) {
+ objStorage.deleteObject(reconstructUri(prefix, obj.getKey()));
+ }
+ continuationToken = batch.isTruncated() ? batch.getContinuationToken() : null;
+ } while (continuationToken != null);
+ }
+
+ private static String reconstructUri(String prefix, String key) {
+ // prefix is like "s3://bucket/path/", key is the full object key
+ int schemeEnd = prefix.indexOf("://");
+ if (schemeEnd >= 0) {
+ String scheme = prefix.substring(0, schemeEnd);
+ int slashAfterScheme = prefix.indexOf('/', schemeEnd + 3);
+ if (slashAfterScheme >= 0) {
+ String bucket = prefix.substring(schemeEnd + 3, slashAfterScheme);
+ return scheme + "://" + bucket + "/" + key;
+ }
+ }
+ return prefix + key;
+ }
+
+ @Override
+ public void rename(Location src, Location dst) throws IOException {
+ objStorage.copyObject(src.uri(), dst.uri());
+ objStorage.deleteObject(src.uri());
+ }
+
+ @Override
+ public FileIterator list(Location location) throws IOException {
+ return new S3FileIterator(location.uri());
+ }
+
+ @Override
+ public DorisInputFile newInputFile(Location location) throws IOException {
+ return new S3InputFile(location);
+ }
+
+ @Override
+ public DorisOutputFile newOutputFile(Location location) throws IOException {
+ return new S3OutputFile(location);
+ }
+
+ /** Lazy-loading, paginating FileIterator over S3 list results. */
+ private class S3FileIterator implements FileIterator {
+ private final String prefix;
+ private String continuationToken;
+ private List<FileEntry> buffer = new ArrayList<>();
+ private int bufferIdx = 0;
+ private boolean done = false;
+
+ S3FileIterator(String prefix) {
+ this.prefix = prefix;
+ }
+
+ @Override
+ public boolean hasNext() throws IOException {
+ if (bufferIdx < buffer.size()) {
+ return true;
+ }
+ if (done) {
+ return false;
+ }
+ fetchNextPage();
+ return bufferIdx < buffer.size();
+ }
+
+ private void fetchNextPage() throws IOException {
+ RemoteObjects page = objStorage.listObjects(prefix, continuationToken);
+ buffer = new ArrayList<>();
+ bufferIdx = 0;
+ for (RemoteObject obj : page.getObjectList()) {
+ Location loc = Location.of(reconstructUri(prefix, obj.getKey()));
+ buffer.add(new FileEntry(loc, obj.getSize(), false, List.of()));
+ }
+ if (page.isTruncated()) {
+ continuationToken = page.getContinuationToken();
+ } else {
+ done = true;
+ }
+ }
+
+ @Override
+ public FileEntry next() throws IOException {
+ return buffer.get(bufferIdx++);
+ }
+
+ @Override
+ public void close() throws IOException {
+ // no-op
+ }
+ }
+
+ /** S3-backed DorisInputFile. */
+ private class S3InputFile implements DorisInputFile {
+ private final Location location;
+
+ S3InputFile(Location location) {
+ this.location = location;
+ }
+
+ @Override
+ public Location location() {
+ return location;
+ }
+
+ @Override
+ public long length() throws IOException {
+ return objStorage.headObject(location.uri()).getSize();
+ }
+
+ @Override
+ public InputStream newStream() throws IOException {
+ return ((S3ObjStorage) objStorage).openInputStream(location.uri());
+ }
+ }
+
+ /** S3-backed DorisOutputFile. */
+ private class S3OutputFile implements DorisOutputFile {
+ private final Location location;
+
+ S3OutputFile(Location location) {
+ this.location = location;
+ }
+
+ @Override
+ public Location location() {
+ return location;
+ }
+
+ @Override
+ public OutputStream create() throws IOException {
+ return createOrOverwrite();
+ }
+
+ @Override
+ public OutputStream createOrOverwrite() throws IOException {
+ // Use a buffered in-memory stream; flush triggers PutObject on close
+ return new S3OutputStream(location.uri(), (S3ObjStorage) objStorage);
+ }
+ }
+}
diff --git a/fe/fe-filesystem/fe-filesystem-s3/src/main/java/org/apache/doris/filesystem/s3/S3FileSystemProvider.java b/fe/fe-filesystem/fe-filesystem-s3/src/main/java/org/apache/doris/filesystem/s3/S3FileSystemProvider.java
new file mode 100644
index 00000000000..9adb8a54098
--- /dev/null
+++ b/fe/fe-filesystem/fe-filesystem-s3/src/main/java/org/apache/doris/filesystem/s3/S3FileSystemProvider.java
@@ -0,0 +1,57 @@
+// 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.doris.filesystem.s3;
+
+import org.apache.doris.filesystem.spi.FileSystem;
+import org.apache.doris.filesystem.spi.FileSystemProvider;
+
+import java.io.IOException;
+import java.util.Map;
+
+/**
+ * SPI provider for AWS S3 and S3-compatible storage (MinIO, etc.).
+ *
+ * <p>Registered via META-INF/services/org.apache.doris.filesystem.spi.FileSystemProvider.
+ *
+ * <p>Identified by presence of AWS_ACCESS_KEY with either AWS_ENDPOINT or AWS_REGION.
+ * S3 is intentionally the last-resort provider; cloud-specific providers (OSS, COS, OBS)
+ * should match first via their endpoint domain patterns.
+ */
+public class S3FileSystemProvider implements FileSystemProvider {
+
+ @Override
+ public boolean supports(Map<String, String> properties) {
+ String accessKey = properties.get(S3ObjStorage.PROP_ACCESS_KEY);
+ String endpoint = properties.get(S3ObjStorage.PROP_ENDPOINT);
+ String region = properties.get(S3ObjStorage.PROP_REGION);
+ // Require access key + (endpoint or region)
+ return accessKey != null && !accessKey.isEmpty()
+ && (endpoint != null || region != null);
+ }
+
+ @Override
+ public FileSystem create(Map<String, String> properties) throws IOException {
+ S3ObjStorage storage = new S3ObjStorage(properties);
+ return new S3FileSystem(storage);
+ }
+
+ @Override
+ public String name() {
+ return "S3";
+ }
+}
diff --git a/fe/fe-filesystem/fe-filesystem-s3/src/main/java/org/apache/doris/filesystem/s3/S3ObjStorage.java b/fe/fe-filesystem/fe-filesystem-s3/src/main/java/org/apache/doris/filesystem/s3/S3ObjStorage.java
new file mode 100644
index 00000000000..62d8ac71944
--- /dev/null
+++ b/fe/fe-filesystem/fe-filesystem-s3/src/main/java/org/apache/doris/filesystem/s3/S3ObjStorage.java
@@ -0,0 +1,331 @@
+// 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.doris.filesystem.s3;
+
+import org.apache.doris.filesystem.spi.ObjStorage;
+import org.apache.doris.filesystem.spi.RemoteObjects;
+import org.apache.doris.filesystem.spi.UploadPartResult;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
+import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
+import software.amazon.awssdk.auth.credentials.AwsSessionCredentials;
+import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider;
+import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
+import software.amazon.awssdk.http.urlconnection.UrlConnectionHttpClient;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.S3Configuration;
+import software.amazon.awssdk.services.s3.model.AbortMultipartUploadRequest;
+import software.amazon.awssdk.services.s3.model.CompleteMultipartUploadRequest;
+import software.amazon.awssdk.services.s3.model.CompletedMultipartUpload;
+import software.amazon.awssdk.services.s3.model.CompletedPart;
+import software.amazon.awssdk.services.s3.model.CopyObjectRequest;
+import software.amazon.awssdk.services.s3.model.CreateMultipartUploadRequest;
+import software.amazon.awssdk.services.s3.model.CreateMultipartUploadResponse;
+import software.amazon.awssdk.services.s3.model.DeleteObjectRequest;
+import software.amazon.awssdk.services.s3.model.GetObjectRequest;
+import software.amazon.awssdk.services.s3.model.HeadObjectRequest;
+import software.amazon.awssdk.services.s3.model.HeadObjectResponse;
+import software.amazon.awssdk.services.s3.model.ListObjectsV2Request;
+import software.amazon.awssdk.services.s3.model.ListObjectsV2Response;
+import software.amazon.awssdk.services.s3.model.NoSuchKeyException;
+import software.amazon.awssdk.services.s3.model.PutObjectRequest;
+import software.amazon.awssdk.services.s3.model.S3Exception;
+import software.amazon.awssdk.services.s3.model.UploadPartRequest;
+import software.amazon.awssdk.services.s3.model.UploadPartResponse;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.URI;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * Object storage implementation backed by AWS S3 SDK v2.
+ * Accepts only Map<String, String> in constructor; no dependency on fe-core or fe-common.
+ *
+ * <p>Recognized property keys:
+ * <ul>
+ * <li>AWS_ENDPOINT - S3 endpoint URL (required)</li>
+ * <li>AWS_REGION - AWS region identifier (required)</li>
+ * <li>AWS_ACCESS_KEY - AWS access key ID</li>
+ * <li>AWS_SECRET_KEY - AWS secret access key</li>
+ * <li>AWS_TOKEN - AWS session token (optional)</li>
+ * <li>use_path_style - "true" to enable path-style bucket access</li>
+ * </ul>
+ */
+public class S3ObjStorage implements ObjStorage<S3Client> {
+
+ private static final Logger LOG = LogManager.getLogger(S3ObjStorage.class);
+
+ static final String PROP_ENDPOINT = "AWS_ENDPOINT";
+ static final String PROP_REGION = "AWS_REGION";
+ static final String PROP_ACCESS_KEY = "AWS_ACCESS_KEY";
+ static final String PROP_SECRET_KEY = "AWS_SECRET_KEY";
+ static final String PROP_TOKEN = "AWS_TOKEN";
+ static final String PROP_PATH_STYLE = "use_path_style";
+
+ private final Map<String, String> properties;
+ private final boolean usePathStyle;
+ private volatile S3Client client;
+
+ public S3ObjStorage(Map<String, String> properties) {
+ this.properties = Collections.unmodifiableMap(properties);
+ this.usePathStyle = Boolean.parseBoolean(properties.getOrDefault(PROP_PATH_STYLE, "false"));
+ }
+
+ @Override
+ public S3Client getClient() throws IOException {
+ if (client == null) {
+ synchronized (this) {
+ if (client == null) {
+ client = buildClient();
+ }
+ }
+ }
+ return client;
+ }
+
+ private S3Client buildClient() throws IOException {
+ String endpointStr = properties.get(PROP_ENDPOINT);
+ if (endpointStr == null || endpointStr.isEmpty()) {
+ throw new IOException("S3 property " + PROP_ENDPOINT + " is required");
+ }
+ if (!endpointStr.contains("://")) {
+ endpointStr = "https://" + endpointStr;
+ }
+ String region = properties.getOrDefault(PROP_REGION, "us-east-1");
+ AwsCredentialsProvider credentialsProvider = buildCredentialsProvider();
+
+ return S3Client.builder()
+ .httpClient(UrlConnectionHttpClient.builder()
+ .socketTimeout(Duration.ofSeconds(30))
+ .connectionTimeout(Duration.ofSeconds(30))
+ .build())
+ .endpointOverride(URI.create(endpointStr))
+ .credentialsProvider(credentialsProvider)
+ .region(Region.of(region))
+ .serviceConfiguration(S3Configuration.builder()
+ .chunkedEncodingEnabled(false)
+ .pathStyleAccessEnabled(usePathStyle)
+ .build())
+ .build();
+ }
+
+ private AwsCredentialsProvider buildCredentialsProvider() {
+ String accessKey = properties.get(PROP_ACCESS_KEY);
+ String secretKey = properties.get(PROP_SECRET_KEY);
+ String token = properties.get(PROP_TOKEN);
+
+ if (accessKey != null && !accessKey.isEmpty() && secretKey != null && !secretKey.isEmpty()) {
+ if (token != null && !token.isEmpty()) {
+ return StaticCredentialsProvider.create(
+ AwsSessionCredentials.create(accessKey, secretKey, token));
+ }
+ return StaticCredentialsProvider.create(AwsBasicCredentials.create(accessKey, secretKey));
+ }
+ return DefaultCredentialsProvider.create();
+ }
+
+ @Override
+ public RemoteObjects listObjects(String remotePath, String continuationToken) throws IOException {
+ S3Uri uri = S3Uri.parse(remotePath, usePathStyle);
+ ListObjectsV2Request.Builder builder = ListObjectsV2Request.builder()
+ .bucket(uri.bucket())
+ .prefix(uri.key());
+ if (continuationToken != null && !continuationToken.isEmpty()) {
+ builder.continuationToken(continuationToken);
+ }
+ try {
+ ListObjectsV2Response response = getClient().listObjectsV2(builder.build());
+ List<org.apache.doris.filesystem.spi.RemoteObject> objects = response.contents().stream()
+ .map(s3Obj -> new org.apache.doris.filesystem.spi.RemoteObject(
+ s3Obj.key(),
+ getRelativePath(uri.key(), s3Obj.key()),
+ s3Obj.eTag(),
+ s3Obj.size()))
+ .collect(Collectors.toList());
+ return new RemoteObjects(objects, response.isTruncated(),
+ response.nextContinuationToken());
+ } catch (S3Exception e) {
+ throw new IOException("Failed to list objects at " + remotePath + ": " + e.getMessage(), e);
+ }
+ }
+
+ private static String getRelativePath(String prefix, String key) {
+ return key.startsWith(prefix) ? key.substring(prefix.length()) : key;
+ }
+
+ @Override
+ public org.apache.doris.filesystem.spi.RemoteObject headObject(String remotePath) throws IOException {
+ S3Uri uri = S3Uri.parse(remotePath, usePathStyle);
+ try {
+ HeadObjectResponse response = getClient().headObject(
+ HeadObjectRequest.builder().bucket(uri.bucket()).key(uri.key()).build());
+ return new org.apache.doris.filesystem.spi.RemoteObject(
+ uri.key(), uri.key(), response.eTag(), response.contentLength());
+ } catch (NoSuchKeyException e) {
+ throw new FileNotFoundException("Object not found: " + remotePath);
+ } catch (S3Exception e) {
+ if (e.statusCode() == 404) {
+ throw new FileNotFoundException("Object not found: " + remotePath);
+ }
+ throw new IOException("headObject failed for " + remotePath + ": " + e.getMessage(), e);
+ }
+ }
+
+ @Override
+ public void putObject(String remotePath, org.apache.doris.filesystem.spi.RequestBody requestBody)
+ throws IOException {
+ S3Uri uri = S3Uri.parse(remotePath, usePathStyle);
+ try (InputStream content = requestBody.content()) {
+ getClient().putObject(
+ PutObjectRequest.builder().bucket(uri.bucket()).key(uri.key()).build(),
+ software.amazon.awssdk.core.sync.RequestBody.fromInputStream(
+ content, requestBody.contentLength()));
+ } catch (S3Exception e) {
+ throw new IOException("putObject failed for " + remotePath + ": " + e.getMessage(), e);
+ }
+ }
+
+ @Override
+ public void deleteObject(String remotePath) throws IOException {
+ S3Uri uri = S3Uri.parse(remotePath, usePathStyle);
+ try {
+ getClient().deleteObject(DeleteObjectRequest.builder()
+ .bucket(uri.bucket()).key(uri.key()).build());
+ } catch (S3Exception e) {
+ if (e.statusCode() == 404) {
+ return; // already deleted
+ }
+ throw new IOException("deleteObject failed for " + remotePath + ": " + e.getMessage(), e);
+ }
+ }
+
+ @Override
+ public void copyObject(String srcPath, String dstPath) throws IOException {
+ S3Uri srcUri = S3Uri.parse(srcPath, usePathStyle);
+ S3Uri dstUri = S3Uri.parse(dstPath, usePathStyle);
+ try {
+ getClient().copyObject(CopyObjectRequest.builder()
+ .copySource(srcUri.bucket() + "/" + srcUri.key())
+ .destinationBucket(dstUri.bucket())
+ .destinationKey(dstUri.key())
+ .build());
+ } catch (S3Exception e) {
+ throw new IOException("copyObject from " + srcPath + " to " + dstPath
+ + " failed: " + e.getMessage(), e);
+ }
+ }
+
+ @Override
+ public String initiateMultipartUpload(String remotePath) throws IOException {
+ S3Uri uri = S3Uri.parse(remotePath, usePathStyle);
+ try {
+ CreateMultipartUploadResponse response = getClient().createMultipartUpload(
+ CreateMultipartUploadRequest.builder().bucket(uri.bucket()).key(uri.key()).build());
+ return response.uploadId();
+ } catch (S3Exception e) {
+ throw new IOException("initiateMultipartUpload failed for " + remotePath
+ + ": " + e.getMessage(), e);
+ }
+ }
+
+ @Override
+ public UploadPartResult uploadPart(String remotePath, String uploadId, int partNum,
+ org.apache.doris.filesystem.spi.RequestBody body) throws IOException {
+ S3Uri uri = S3Uri.parse(remotePath, usePathStyle);
+ try (InputStream content = body.content()) {
+ UploadPartResponse response = getClient().uploadPart(
+ UploadPartRequest.builder()
+ .bucket(uri.bucket()).key(uri.key())
+ .uploadId(uploadId).partNumber(partNum)
+ .contentLength(body.contentLength())
+ .build(),
+ software.amazon.awssdk.core.sync.RequestBody.fromInputStream(
+ content, body.contentLength()));
+ return new UploadPartResult(partNum, response.eTag());
+ } catch (S3Exception e) {
+ throw new IOException("uploadPart " + partNum + " failed for " + remotePath
+ + ": " + e.getMessage(), e);
+ }
+ }
+
+ @Override
+ public void completeMultipartUpload(String remotePath, String uploadId,
+ List<org.apache.doris.filesystem.spi.UploadPartResult> parts) throws IOException {
+ S3Uri uri = S3Uri.parse(remotePath, usePathStyle);
+ List<CompletedPart> completedParts = parts.stream()
+ .map(p -> CompletedPart.builder().partNumber(p.partNumber()).eTag(p.etag()).build())
+ .collect(Collectors.toList());
+ try {
+ getClient().completeMultipartUpload(CompleteMultipartUploadRequest.builder()
+ .bucket(uri.bucket()).key(uri.key()).uploadId(uploadId)
+ .multipartUpload(CompletedMultipartUpload.builder().parts(completedParts).build())
+ .build());
+ } catch (S3Exception e) {
+ throw new IOException("completeMultipartUpload failed for " + remotePath
+ + ": " + e.getMessage(), e);
+ }
+ }
+
+ @Override
+ public void abortMultipartUpload(String remotePath, String uploadId) throws IOException {
+ S3Uri uri = S3Uri.parse(remotePath, usePathStyle);
+ try {
+ getClient().abortMultipartUpload(AbortMultipartUploadRequest.builder()
+ .bucket(uri.bucket()).key(uri.key()).uploadId(uploadId).build());
+ } catch (S3Exception e) {
+ LOG.warn("abortMultipartUpload failed for {}: {}", remotePath, e.getMessage());
+ }
+ }
+
+ /**
+ * Open an InputStream for reading an S3 object.
+ */
+ InputStream openInputStream(String remotePath) throws IOException {
+ S3Uri uri = S3Uri.parse(remotePath, usePathStyle);
+ try {
+ return getClient().getObject(
+ GetObjectRequest.builder().bucket(uri.bucket()).key(uri.key()).build());
+ } catch (NoSuchKeyException e) {
+ throw new FileNotFoundException("Object not found: " + remotePath);
+ } catch (S3Exception e) {
+ throw new IOException("getObject failed for " + remotePath + ": " + e.getMessage(), e);
+ }
+ }
+
+ @Override
+ public Map<String, String> getProperties() {
+ return properties;
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (client != null) {
+ client.close();
+ client = null;
+ }
+ }
+}
diff --git a/fe/fe-filesystem/fe-filesystem-s3/src/main/java/org/apache/doris/filesystem/s3/S3OutputStream.java b/fe/fe-filesystem/fe-filesystem-s3/src/main/java/org/apache/doris/filesystem/s3/S3OutputStream.java
new file mode 100644
index 00000000000..ddd3768b112
--- /dev/null
+++ b/fe/fe-filesystem/fe-filesystem-s3/src/main/java/org/apache/doris/filesystem/s3/S3OutputStream.java
@@ -0,0 +1,71 @@
+// 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.doris.filesystem.s3;
+
+import org.apache.doris.filesystem.spi.RequestBody;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+
+/**
+ * OutputStream that buffers writes in memory and uploads to S3 on close.
+ * For production large-file writes, multipart upload should be used instead.
+ */
+class S3OutputStream extends OutputStream {
+
+ private final String remotePath;
+ private final S3ObjStorage objStorage;
+ private final ByteArrayOutputStream buffer = new ByteArrayOutputStream();
+ private boolean closed = false;
+
+ S3OutputStream(String remotePath, S3ObjStorage objStorage) {
+ this.remotePath = remotePath;
+ this.objStorage = objStorage;
+ }
+
+ @Override
+ public void write(int b) throws IOException {
+ checkNotClosed();
+ buffer.write(b);
+ }
+
+ @Override
+ public void write(byte[] b, int off, int len) throws IOException {
+ checkNotClosed();
+ buffer.write(b, off, len);
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (closed) {
+ return;
+ }
+ closed = true;
+ byte[] data = buffer.toByteArray();
+ objStorage.putObject(remotePath,
+ RequestBody.of(new ByteArrayInputStream(data), data.length));
+ }
+
+ private void checkNotClosed() throws IOException {
+ if (closed) {
+ throw new IOException("Stream already closed: " + remotePath);
+ }
+ }
+}
diff --git a/fe/fe-filesystem/fe-filesystem-s3/src/main/java/org/apache/doris/filesystem/s3/S3Uri.java b/fe/fe-filesystem/fe-filesystem-s3/src/main/java/org/apache/doris/filesystem/s3/S3Uri.java
new file mode 100644
index 00000000000..ec2e260de31
--- /dev/null
+++ b/fe/fe-filesystem/fe-filesystem-s3/src/main/java/org/apache/doris/filesystem/s3/S3Uri.java
@@ -0,0 +1,63 @@
+// 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.doris.filesystem.s3;
+
+/**
+ * Parsed S3/S3A URI. Extracts bucket and key without any fe-core dependency.
+ */
+public final class S3Uri {
+
+ private final String bucket;
+ private final String key;
+
+ private S3Uri(String bucket, String key) {
+ this.bucket = bucket;
+ this.key = key;
+ }
+
+ /**
+ * Parses URIs of the form: s3://bucket/key, s3a://bucket/key, s3n://bucket/key.
+ * Also handles path-style: https://endpoint/bucket/key (when pathStyle=true).
+ */
+ public static S3Uri parse(String path, boolean pathStyleAccess) {
+ if (path == null) {
+ throw new IllegalArgumentException("S3 path must not be null");
+ }
+ // Handle virtual-hosted style: s3://bucket/key, s3a://bucket/key
+ int schemeEnd = path.indexOf("://");
+ if (schemeEnd >= 0) {
+ String withoutScheme = path.substring(schemeEnd + 3);
+ int slashIdx = withoutScheme.indexOf('/');
+ if (slashIdx < 0) {
+ return new S3Uri(withoutScheme, "");
+ }
+ return new S3Uri(withoutScheme.substring(0, slashIdx),
+ withoutScheme.substring(slashIdx + 1));
+ }
+ // Treat bare string as key (shouldn't normally happen)
+ throw new IllegalArgumentException("Cannot parse S3 URI without scheme: " + path);
+ }
+
+ public String bucket() {
+ return bucket;
+ }
+
+ public String key() {
+ return key;
+ }
+}
diff --git a/fe/fe-filesystem/fe-filesystem-s3/src/main/resources/META-INF/services/org.apache.doris.filesystem.spi.FileSystemProvider b/fe/fe-filesystem/fe-filesystem-s3/src/main/resources/META-INF/services/org.apache.doris.filesystem.spi.FileSystemProvider
new file mode 100644
index 00000000000..ee68ae1adde
--- /dev/null
+++ b/fe/fe-filesystem/fe-filesystem-s3/src/main/resources/META-INF/services/org.apache.doris.filesystem.spi.FileSystemProvider
@@ -0,0 +1 @@
+org.apache.doris.filesystem.s3.S3FileSystemProvider
diff --git a/fe/fe-filesystem/fe-filesystem-spi/pom.xml b/fe/fe-filesystem/fe-filesystem-spi/pom.xml
new file mode 100644
index 00000000000..03f8d718521
--- /dev/null
+++ b/fe/fe-filesystem/fe-filesystem-spi/pom.xml
@@ -0,0 +1,46 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.doris</groupId>
+ <artifactId>fe-filesystem</artifactId>
+ <version>${revision}</version>
+ <relativePath>../pom.xml</relativePath>
+ </parent>
+
+ <artifactId>fe-filesystem-spi</artifactId>
+ <packaging>jar</packaging>
+ <name>Doris FE Filesystem SPI</name>
+ <description>
+ Service Provider Interface (SPI) for Doris FE filesystem abstraction.
+ Zero external dependencies - only JDK 11.
+ This is the ONLY filesystem artifact compiled into fe-core.
+ </description>
+
+ <!-- Intentionally no dependencies: only JDK 11 is allowed -->
+
+ <build>
+ <finalName>doris-fe-filesystem-spi</finalName>
+ </build>
+</project>
diff --git a/fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemProvider.java b/fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/BlockInfo.java
similarity index 56%
copy from fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemProvider.java
copy to fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/BlockInfo.java
index 1dd276fc12c..911b4e93d8f 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemProvider.java
+++ b/fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/BlockInfo.java
@@ -15,10 +15,32 @@
// specific language governing permissions and limitations
// under the License.
-package org.apache.doris.fs;
+package org.apache.doris.filesystem.spi;
-import org.apache.doris.datasource.SessionContext;
+/**
+ * Represents a block location within a file (e.g., HDFS block).
+ */
+public final class BlockInfo {
-public interface FileSystemProvider {
- LegacyFileSystemApi get(SessionContext ctx);
+ private final long offset;
+ private final long length;
+ private final String[] hosts;
+
+ public BlockInfo(long offset, long length, String[] hosts) {
+ this.offset = offset;
+ this.length = length;
+ this.hosts = hosts == null ? new String[0] : hosts.clone();
+ }
+
+ public long offset() {
+ return offset;
+ }
+
+ public long length() {
+ return length;
+ }
+
+ public String[] hosts() {
+ return hosts.clone();
+ }
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemProvider.java b/fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/DorisInputFile.java
similarity index 72%
copy from fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemProvider.java
copy to fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/DorisInputFile.java
index 1dd276fc12c..7eda9fd32fa 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemProvider.java
+++ b/fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/DorisInputFile.java
@@ -15,10 +15,19 @@
// specific language governing permissions and limitations
// under the License.
-package org.apache.doris.fs;
+package org.apache.doris.filesystem.spi;
-import org.apache.doris.datasource.SessionContext;
+import java.io.IOException;
+import java.io.InputStream;
-public interface FileSystemProvider {
- LegacyFileSystemApi get(SessionContext ctx);
+/**
+ * Represents a readable file in the filesystem.
+ */
+public interface DorisInputFile {
+
+ Location location();
+
+ long length() throws IOException;
+
+ InputStream newStream() throws IOException;
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemProvider.java b/fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/DorisOutputFile.java
similarity index 71%
copy from fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemProvider.java
copy to fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/DorisOutputFile.java
index 1dd276fc12c..4eb117c9764 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemProvider.java
+++ b/fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/DorisOutputFile.java
@@ -15,10 +15,19 @@
// specific language governing permissions and limitations
// under the License.
-package org.apache.doris.fs;
+package org.apache.doris.filesystem.spi;
-import org.apache.doris.datasource.SessionContext;
+import java.io.IOException;
+import java.io.OutputStream;
-public interface FileSystemProvider {
- LegacyFileSystemApi get(SessionContext ctx);
+/**
+ * Represents a writable file in the filesystem.
+ */
+public interface DorisOutputFile {
+
+ Location location();
+
+ OutputStream create() throws IOException;
+
+ OutputStream createOrOverwrite() throws IOException;
}
diff --git a/fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/FileEntry.java b/fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/FileEntry.java
new file mode 100644
index 00000000000..61a8ee74eea
--- /dev/null
+++ b/fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/FileEntry.java
@@ -0,0 +1,54 @@
+// 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.doris.filesystem.spi;
+
+import java.util.List;
+
+/**
+ * Immutable representation of a file or directory entry returned by FileSystem.list().
+ */
+public final class FileEntry {
+
+ private final Location location;
+ private final long length;
+ private final boolean isDirectory;
+ private final List<BlockInfo> blocks;
+
+ public FileEntry(Location location, long length, boolean isDirectory, List<BlockInfo> blocks) {
+ this.location = location;
+ this.length = length;
+ this.isDirectory = isDirectory;
+ this.blocks = blocks == null ? List.of() : List.copyOf(blocks);
+ }
+
+ public Location location() {
+ return location;
+ }
+
+ public long length() {
+ return length;
+ }
+
+ public boolean isDirectory() {
+ return isDirectory;
+ }
+
+ public List<BlockInfo> blocks() {
+ return blocks;
+ }
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemProvider.java b/fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/FileIterator.java
similarity index 68%
copy from fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemProvider.java
copy to fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/FileIterator.java
index 1dd276fc12c..9c61a954af9 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemProvider.java
+++ b/fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/FileIterator.java
@@ -15,10 +15,17 @@
// specific language governing permissions and limitations
// under the License.
-package org.apache.doris.fs;
+package org.apache.doris.filesystem.spi;
-import org.apache.doris.datasource.SessionContext;
+import java.io.Closeable;
+import java.io.IOException;
-public interface FileSystemProvider {
- LegacyFileSystemApi get(SessionContext ctx);
+/**
+ * Iterator for FileEntry that propagates IO errors via IOException.
+ * Differs from java.util.Iterator which only allows RuntimeException.
+ */
+public interface FileIterator extends Closeable {
+ boolean hasNext() throws IOException;
+
+ FileEntry next() throws IOException;
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemProvider.java b/fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/FileSystem.java
similarity index 51%
copy from fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemProvider.java
copy to fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/FileSystem.java
index 1dd276fc12c..29587cf75a1 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemProvider.java
+++ b/fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/FileSystem.java
@@ -15,10 +15,30 @@
// specific language governing permissions and limitations
// under the License.
-package org.apache.doris.fs;
+package org.apache.doris.filesystem.spi;
-import org.apache.doris.datasource.SessionContext;
+import java.io.IOException;
-public interface FileSystemProvider {
- LegacyFileSystemApi get(SessionContext ctx);
+/**
+ * Core filesystem abstraction.
+ * All methods throw IOException (not UserException or checked vendor exceptions).
+ */
+public interface FileSystem extends AutoCloseable {
+
+ boolean exists(Location location) throws IOException;
+
+ void mkdirs(Location location) throws IOException;
+
+ void delete(Location location, boolean recursive) throws IOException;
+
+ void rename(Location src, Location dst) throws IOException;
+
+ FileIterator list(Location location) throws IOException;
+
+ DorisInputFile newInputFile(Location location) throws IOException;
+
+ DorisOutputFile newOutputFile(Location location) throws IOException;
+
+ @Override
+ void close() throws IOException;
}
diff --git a/fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/FileSystemProvider.java b/fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/FileSystemProvider.java
new file mode 100644
index 00000000000..e254badf8f6
--- /dev/null
+++ b/fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/FileSystemProvider.java
@@ -0,0 +1,58 @@
+// 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.doris.filesystem.spi;
+
+import java.io.IOException;
+import java.util.Map;
+
+/**
+ * SPI interface for filesystem provider discovery via Java ServiceLoader.
+ *
+ * Implementations must:
+ * 1. Have a public no-arg constructor.
+ * 2. Register in META-INF/services/org.apache.doris.filesystem.spi.FileSystemProvider.
+ * 3. Have NO dependency on fe-core, fe-common, or fe-catalog.
+ */
+public interface FileSystemProvider {
+
+ /**
+ * Returns true if this provider can handle the given properties.
+ * Must be cheap (no network calls) and deterministic.
+ *
+ * @param properties key-value storage configuration
+ * @return true if this provider supports the configuration
+ */
+ boolean supports(Map<String, String> properties);
+
+ /**
+ * Creates a FileSystem instance from the given properties.
+ * Called only after {@link #supports(Map)} returns true.
+ *
+ * @param properties key-value storage configuration
+ * @return a ready-to-use FileSystem
+ * @throws IOException if the filesystem cannot be initialized
+ */
+ FileSystem create(Map<String, String> properties) throws IOException;
+
+ /**
+ * Human-readable name for logging/diagnostics (e.g., "S3", "HDFS", "Azure").
+ */
+ default String name() {
+ return getClass().getSimpleName().replace("FileSystemProvider", "");
+ }
+}
diff --git a/fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/HadoopAuthenticator.java b/fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/HadoopAuthenticator.java
new file mode 100644
index 00000000000..e8c0b488038
--- /dev/null
+++ b/fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/HadoopAuthenticator.java
@@ -0,0 +1,45 @@
+// 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.doris.filesystem.spi;
+
+import java.io.IOException;
+
+/**
+ * Abstraction for Hadoop-style privileged execution (Kerberos doAs).
+ *
+ * Defined here in fe-filesystem-spi with zero external dependencies, using
+ * {@link IOCallable} instead of {@code java.security.PrivilegedExceptionAction}
+ * to avoid Hadoop API dependency. Implementations live in fe-core (Kerberos/Simple
+ * authenticators) and are injected into DFSFileSystem at construction time.
+ *
+ * P3.0a: This interface replaces the Hadoop-dependent HadoopAuthenticator from
+ * fe-common for use in fe-filesystem-hdfs module.
+ */
+public interface HadoopAuthenticator {
+
+ /**
+ * Executes the given action under this authenticator's security context
+ * (e.g., as a specific Kerberos principal or simple user).
+ *
+ * @param action the IO operation to execute
+ * @param <T> the return type
+ * @return the result of the action
+ * @throws IOException if the action throws or authentication fails
+ */
+ <T> T doAs(IOCallable<T> action) throws IOException;
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemProvider.java b/fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/IOCallable.java
similarity index 66%
copy from fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemProvider.java
copy to fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/IOCallable.java
index 1dd276fc12c..fc211fed8e1 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemProvider.java
+++ b/fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/IOCallable.java
@@ -15,10 +15,18 @@
// specific language governing permissions and limitations
// under the License.
-package org.apache.doris.fs;
+package org.apache.doris.filesystem.spi;
-import org.apache.doris.datasource.SessionContext;
+import java.io.IOException;
-public interface FileSystemProvider {
- LegacyFileSystemApi get(SessionContext ctx);
+/**
+ * IO-throwing variant of {@link java.util.concurrent.Callable}.
+ * Used in place of {@code java.security.PrivilegedExceptionAction} to avoid
+ * Hadoop API dependency in fe-filesystem-spi.
+ *
+ * @param <T> the return type of the callable
+ */
+@FunctionalInterface
+public interface IOCallable<T> {
+ T call() throws IOException;
}
diff --git a/fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/Location.java b/fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/Location.java
new file mode 100644
index 00000000000..9ed8210dc26
--- /dev/null
+++ b/fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/Location.java
@@ -0,0 +1,71 @@
+// 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.doris.filesystem.spi;
+
+import java.util.Objects;
+
+/**
+ * Immutable, typed URI wrapper replacing raw String path parameters.
+ */
+public final class Location {
+
+ private final String uri;
+
+ private Location(String uri) {
+ this.uri = Objects.requireNonNull(uri, "uri must not be null");
+ }
+
+ public static Location of(String uri) {
+ return new Location(uri);
+ }
+
+ public String uri() {
+ return uri;
+ }
+
+ public String scheme() {
+ int idx = uri.indexOf("://");
+ return idx < 0 ? "" : uri.substring(0, idx);
+ }
+
+ public String withoutScheme() {
+ int idx = uri.indexOf("://");
+ return idx < 0 ? uri : uri.substring(idx + 3);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof Location)) {
+ return false;
+ }
+ return uri.equals(((Location) o).uri);
+ }
+
+ @Override
+ public int hashCode() {
+ return uri.hashCode();
+ }
+
+ @Override
+ public String toString() {
+ return uri;
+ }
+}
diff --git a/fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/ObjFileSystem.java b/fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/ObjFileSystem.java
new file mode 100644
index 00000000000..95d29f264cb
--- /dev/null
+++ b/fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/ObjFileSystem.java
@@ -0,0 +1,64 @@
+// 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.doris.filesystem.spi;
+
+import java.io.IOException;
+
+/**
+ * Abstract base class for object-storage-backed FileSystems.
+ * Delegates existence checks to the underlying ObjStorage instance.
+ */
+public abstract class ObjFileSystem implements FileSystem {
+
+ protected final ObjStorage<?> objStorage;
+ protected final String name;
+
+ protected ObjFileSystem(String name, ObjStorage<?> objStorage) {
+ this.name = name;
+ this.objStorage = objStorage;
+ }
+
+ public ObjStorage<?> getObjStorage() {
+ return objStorage;
+ }
+
+ @Override
+ public boolean exists(Location location) throws IOException {
+ try {
+ objStorage.headObject(location.withoutScheme());
+ return true;
+ } catch (IOException e) {
+ if (isNotFoundError(e)) {
+ return false;
+ }
+ throw e;
+ }
+ }
+
+ /**
+ * Subclasses may override to detect vendor-specific "not found" errors.
+ */
+ protected boolean isNotFoundError(IOException e) {
+ return e.getMessage() != null && e.getMessage().contains("404");
+ }
+
+ @Override
+ public void close() throws IOException {
+ // default no-op; subclasses may override
+ }
+}
diff --git a/fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/ObjStorage.java b/fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/ObjStorage.java
new file mode 100644
index 00000000000..797937fb3b9
--- /dev/null
+++ b/fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/ObjStorage.java
@@ -0,0 +1,58 @@
+// 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.doris.filesystem.spi;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Low-level object storage abstraction (S3-style API).
+ * Zero external dependencies - only JDK types.
+ *
+ * @param <C> the native client type (e.g., S3Client, BlobServiceClient)
+ */
+public interface ObjStorage<C> extends AutoCloseable {
+
+ C getClient() throws IOException;
+
+ RemoteObjects listObjects(String remotePath, String continuationToken) throws IOException;
+
+ RemoteObject headObject(String remotePath) throws IOException;
+
+ void putObject(String remotePath, RequestBody requestBody) throws IOException;
+
+ void deleteObject(String remotePath) throws IOException;
+
+ void copyObject(String srcPath, String dstPath) throws IOException;
+
+ String initiateMultipartUpload(String remotePath) throws IOException;
+
+ UploadPartResult uploadPart(String remotePath, String uploadId, int partNum,
+ RequestBody body) throws IOException;
+
+ void completeMultipartUpload(String remotePath, String uploadId,
+ List<UploadPartResult> parts) throws IOException;
+
+ void abortMultipartUpload(String remotePath, String uploadId) throws IOException;
+
+ Map<String, String> getProperties();
+
+ @Override
+ void close() throws IOException;
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemProvider.java b/fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/RemoteObject.java
similarity index 52%
copy from fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemProvider.java
copy to fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/RemoteObject.java
index 1dd276fc12c..309af604607 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemProvider.java
+++ b/fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/RemoteObject.java
@@ -15,10 +15,38 @@
// specific language governing permissions and limitations
// under the License.
-package org.apache.doris.fs;
+package org.apache.doris.filesystem.spi;
-import org.apache.doris.datasource.SessionContext;
+/**
+ * Metadata about a single remote object (S3-style).
+ */
+public final class RemoteObject {
-public interface FileSystemProvider {
- LegacyFileSystemApi get(SessionContext ctx);
+ private final String key;
+ private final String relativePath;
+ private final String etag;
+ private final long size;
+
+ public RemoteObject(String key, String relativePath, String etag, long size) {
+ this.key = key;
+ this.relativePath = relativePath;
+ this.etag = etag;
+ this.size = size;
+ }
+
+ public String getKey() {
+ return key;
+ }
+
+ public String getRelativePath() {
+ return relativePath;
+ }
+
+ public String getEtag() {
+ return etag;
+ }
+
+ public long getSize() {
+ return size;
+ }
}
diff --git a/fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/RemoteObjects.java b/fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/RemoteObjects.java
new file mode 100644
index 00000000000..6fe59b96cc1
--- /dev/null
+++ b/fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/RemoteObjects.java
@@ -0,0 +1,48 @@
+// 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.doris.filesystem.spi;
+
+import java.util.List;
+
+/**
+ * Result of a list-objects operation, with pagination support.
+ */
+public final class RemoteObjects {
+
+ private final List<RemoteObject> objectList;
+ private final boolean truncated;
+ private final String continuationToken;
+
+ public RemoteObjects(List<RemoteObject> objectList, boolean truncated, String continuationToken) {
+ this.objectList = objectList == null ? List.of() : List.copyOf(objectList);
+ this.truncated = truncated;
+ this.continuationToken = continuationToken;
+ }
+
+ public List<RemoteObject> getObjectList() {
+ return objectList;
+ }
+
+ public boolean isTruncated() {
+ return truncated;
+ }
+
+ public String getContinuationToken() {
+ return continuationToken;
+ }
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemProvider.java b/fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/RequestBody.java
similarity index 50%
copy from fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemProvider.java
copy to fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/RequestBody.java
index 1dd276fc12c..71232d209e3 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemProvider.java
+++ b/fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/RequestBody.java
@@ -15,10 +15,34 @@
// specific language governing permissions and limitations
// under the License.
-package org.apache.doris.fs;
+package org.apache.doris.filesystem.spi;
-import org.apache.doris.datasource.SessionContext;
+import java.io.IOException;
+import java.io.InputStream;
-public interface FileSystemProvider {
- LegacyFileSystemApi get(SessionContext ctx);
+/**
+ * Represents the body of a PUT or multipart upload request.
+ * Abstracts over SDK-specific request body types (S3RequestBody, etc.).
+ */
+public final class RequestBody {
+
+ private final InputStream content;
+ private final long contentLength;
+
+ public RequestBody(InputStream content, long contentLength) {
+ this.content = content;
+ this.contentLength = contentLength;
+ }
+
+ public InputStream content() throws IOException {
+ return content;
+ }
+
+ public long contentLength() {
+ return contentLength;
+ }
+
+ public static RequestBody of(InputStream content, long contentLength) {
+ return new RequestBody(content, contentLength);
+ }
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemProvider.java b/fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/UploadPartResult.java
similarity index 63%
rename from fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemProvider.java
rename to fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/UploadPartResult.java
index 1dd276fc12c..23f14e12c83 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemProvider.java
+++ b/fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/UploadPartResult.java
@@ -15,10 +15,26 @@
// specific language governing permissions and limitations
// under the License.
-package org.apache.doris.fs;
+package org.apache.doris.filesystem.spi;
-import org.apache.doris.datasource.SessionContext;
+/**
+ * Result of a single multipart upload part operation.
+ */
+public final class UploadPartResult {
-public interface FileSystemProvider {
- LegacyFileSystemApi get(SessionContext ctx);
+ private final int partNumber;
+ private final String etag;
+
+ public UploadPartResult(int partNumber, String etag) {
+ this.partNumber = partNumber;
+ this.etag = etag;
+ }
+
+ public int partNumber() {
+ return partNumber;
+ }
+
+ public String etag() {
+ return etag;
+ }
}
diff --git a/fe/fe-filesystem/pom.xml b/fe/fe-filesystem/pom.xml
new file mode 100644
index 00000000000..d7522789008
--- /dev/null
+++ b/fe/fe-filesystem/pom.xml
@@ -0,0 +1,50 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.doris</groupId>
+ <artifactId>fe</artifactId>
+ <version>${revision}</version>
+ <relativePath>../pom.xml</relativePath>
+ </parent>
+
+ <artifactId>fe-filesystem</artifactId>
+ <packaging>pom</packaging>
+ <name>Doris FE Filesystem (Aggregator)</name>
+ <description>
+ Aggregator POM for all Doris FE filesystem modules.
+ Contains NO Java code; only lists sub-modules.
+ </description>
+
+ <modules>
+ <module>fe-filesystem-spi</module>
+ <module>fe-filesystem-s3</module>
+ <module>fe-filesystem-oss</module>
+ <module>fe-filesystem-cos</module>
+ <module>fe-filesystem-obs</module>
+ <module>fe-filesystem-local</module>
+ <module>fe-filesystem-azure</module>
+ <module>fe-filesystem-hdfs</module>
+ </modules>
+</project>
diff --git a/fe/pom.xml b/fe/pom.xml
index b327978bbf6..e3effc96792 100644
--- a/fe/pom.xml
+++ b/fe/pom.xml
@@ -251,6 +251,7 @@ under the License.
<module>fe-foundation</module>
<module>fe-common</module>
<module>fe-catalog</module>
+ <module>fe-filesystem</module>
<module>fe-extension-spi</module>
<module>fe-extension-loader</module>
<module>fe-core</module>
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org