[refactor](fs) Migrate fe-core storage consumers onto the fe-filesystem SPI facade

fe/fe-common/src/main/java/org/apache/doris/common/security/authentication/ExecutionAuthenticator.java

@@ -26,8 +26,15 @@
  * or {@link Runnable}) with authentication logic, for example using Kerberos or other mechanisms.
  * If no authentication is needed, the default implementations simply execute the tasks directly.
  * </p>
+ * <p>
+ * Migration bridge: extends the fe-foundation {@code ExecutionAuthenticator} (the single doAs
+ * abstraction shared with fe-filesystem), so every fe-common authenticator IS-A foundation
+ * authenticator and consumers can be re-typed to the foundation interface incrementally. This
+ * interface keeps the passthrough default that the foundation interface deliberately leaves
+ * abstract, preserving legacy behavior for existing implementors.
+ * </p>
  */
-public interface ExecutionAuthenticator {
+public interface ExecutionAuthenticator extends org.apache.doris.foundation.security.ExecutionAuthenticator {
 
     /**
      * Executes the given task, optionally within an authenticated context.

fe/fe-core/pom.xml

@@ -104,6 +104,111 @@ under the License.
             <version>${project.version}</version>
             <scope>test</scope>
         </dependency>
+        <!-- All remaining fe-filesystem impl modules, TEST scope only: gives the SPI-vs-fe-core
+             bind-parity golden tests real providers via ServiceLoader on the test classpath.
+             Production still loads providers from the plugins/filesystem/ directory. -->
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>fe-filesystem-hdfs</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>fe-filesystem-oss-hdfs</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>fe-filesystem-jfs</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>fe-filesystem-s3</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>fe-filesystem-oss</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>fe-filesystem-obs</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+            <!-- The cloud SDK's okhttp (3.x/5.x) must not shadow fe-core's own okhttp on the
+                 test classpath; the bind-parity tests never open SDK http connections. -->
+            <exclusions>
+                <exclusion>
+                    <groupId>com.squareup.okhttp3</groupId>
+                    <artifactId>*</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>com.squareup.okio</groupId>
+                    <artifactId>*</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>fe-filesystem-cos</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+            <!-- The cloud SDK's okhttp (3.x/5.x) must not shadow fe-core's own okhttp on the
+                 test classpath; the bind-parity tests never open SDK http connections. -->
+            <exclusions>
+                <exclusion>
+                    <groupId>com.squareup.okhttp3</groupId>
+                    <artifactId>*</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>com.squareup.okio</groupId>
+                    <artifactId>*</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>fe-filesystem-gcs</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>fe-filesystem-minio</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>fe-filesystem-ozone</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>fe-filesystem-azure</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>fe-filesystem-http</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>fe-filesystem-broker</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>${project.groupId}</groupId>
             <artifactId>fe-thrift</artifactId>

fe/fe-core/src/main/java/org/apache/doris/analysis/BrokerDesc.java

@@ -21,8 +21,7 @@
 import org.apache.doris.common.UserException;
 import org.apache.doris.common.io.Text;
 import org.apache.doris.common.io.Writable;
-import org.apache.doris.datasource.property.storage.BrokerProperties;
-import org.apache.doris.datasource.property.storage.StorageProperties;
+import org.apache.doris.datasource.storage.StorageAdapter;
 import org.apache.doris.foundation.property.StoragePropertiesException;
 import org.apache.doris.persist.gson.GsonUtils;
 import org.apache.doris.thrift.TFileType;
@@ -84,14 +83,14 @@ public BrokerDesc(String name, Map<String, String> properties) {
         // Try to determine the actual storage type from properties if available
         if (MapUtils.isNotEmpty(this.properties)) {
             try {
-                // Create primary storage properties from the given configuration
-                this.storageProperties = StorageProperties.createPrimary(this.properties);
+                // Bind the primary storage adapter from the given configuration
+                this.storageAdapter = StorageAdapter.of(this.properties);
                 // Override the storage type based on property configuration
-                this.storageType = StorageBackend.StorageType.valueOfIgnoreCase(storageProperties.getStorageName());
+                this.storageType = StorageBackend.StorageType.valueOfIgnoreCase(storageAdapter.getStorageName());
             } catch (StoragePropertiesException e) {
                 // Currently ignored: these properties might be broker-specific.
-                // Just keep the storage type as BROKER, and try to create BrokerProperties
-                this.storageProperties = BrokerProperties.of(name, properties);
+                // Just keep the storage type as BROKER, and bind a broker adapter
+                this.storageAdapter = StorageAdapter.ofBroker(name, this.properties);
                 this.storageType = StorageBackend.StorageType.BROKER;
             }
         }
@@ -109,17 +108,17 @@ public BrokerDesc(String name, StorageBackend.StorageType storageType, Map<Strin
             this.properties.putAll(properties);
         }
         if (StorageType.BROKER.equals(storageType)) {
-            this.storageProperties = BrokerProperties.of(name, properties);
+            this.storageAdapter = StorageAdapter.ofBroker(name, properties);
             return;
         }
         if (MapUtils.isNotEmpty(this.properties) && StorageType.REFACTOR_STORAGE_TYPES.contains(storageType)) {
-            this.storageProperties = StorageProperties.createPrimary(properties);
+            this.storageAdapter = StorageAdapter.of(properties);
         }
 
     }
 
     public String getFileLocation(String location) throws UserException {
-        return (null != storageProperties) ? storageProperties.validateAndNormalizeUri(location) : location;
+        return (null != storageAdapter) ? storageAdapter.validateAndNormalizeUri(location) : location;
     }
 
     public static BrokerDesc createForStreamLoad() {

fe/fe-core/src/main/java/org/apache/doris/analysis/OutFileClause.java

@@ -34,8 +34,6 @@
 import org.apache.doris.common.util.ParseUtil;
 import org.apache.doris.datasource.property.fileformat.CsvFileFormatProperties;
 import org.apache.doris.datasource.property.fileformat.FileFormatProperties;
-import org.apache.doris.datasource.property.storage.HdfsProperties;
-import org.apache.doris.datasource.property.storage.HdfsPropertiesUtils;
 import org.apache.doris.thrift.TFileFormatType;
 import org.apache.doris.thrift.TParquetDataType;
 import org.apache.doris.thrift.TParquetRepetitionType;
@@ -47,6 +45,7 @@
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 import com.google.common.collect.Sets;
+import org.apache.commons.lang3.StringUtils;
 
 import java.net.URI;
 import java.net.URISyntaxException;
@@ -613,8 +612,9 @@ private void analyzeBrokerDesc(Map<String, String> copiedProps) throws UserExcep
         if (null != brokerDesc.getStorageType() && (brokerDesc.getStorageType()
                 .equals(StorageBackend.StorageType.HDFS)
                 || brokerDesc.getStorageType().equals(StorageBackend.StorageType.JFS))) {
-            String defaultFs = HdfsPropertiesUtils.extractDefaultFsFromPath(filePath);
-            brokerDesc.getBackendConfigProperties().put(HdfsProperties.HDFS_DEFAULT_FS_NAME, defaultFs);
+            String defaultFs = extractDefaultFsFromPath(filePath);
+            // "fs.defaultFS" is the exact literal of the legacy HdfsProperties.HDFS_DEFAULT_FS_NAME.
+            brokerDesc.getBackendConfigProperties().put("fs.defaultFS", defaultFs);
         }
     }
 
@@ -768,4 +768,13 @@ public TResultFileSinkOptions toSinkOptions() {
         }
         return sinkOptions;
     }
+
+    /** Direct copy of the legacy {@code HdfsPropertiesUtils.extractDefaultFsFromPath} pure function. */
+    private static String extractDefaultFsFromPath(String filePath) {
+        if (StringUtils.isBlank(filePath)) {
+            return null;
+        }
+        URI uri = URI.create(filePath);
+        return uri.getScheme() + "://" + uri.getAuthority();
+    }
 }

fe/fe-core/src/main/java/org/apache/doris/analysis/StorageBackend.java

@@ -139,7 +139,7 @@ public TStorageBackendType toThrift() {
          * Includes: S3 (referring to all systems compatible with the S3 protocol),
          * HDFS, OFS, JFS, and AZURE. For S3, this is a generalized type that matches
          * any system whose storage type name is returned as "s3" (or compatible)
-         * by {@link org.apache.doris.datasource.property.storage.StorageProperties#getStorageName()}.
+         * by {@link org.apache.doris.datasource.storage.StorageAdapter#getStorageName()}.
          * <p>
          * This set is a temporary solution. Once parameter refactoring is fully supported
          * across all storage systems, this class can be removed.

fe/fe-core/src/main/java/org/apache/doris/analysis/StorageDesc.java

@@ -17,12 +17,10 @@
 
 package org.apache.doris.analysis;
 
-import org.apache.doris.datasource.property.storage.BrokerProperties;
-import org.apache.doris.datasource.property.storage.StorageProperties;
+import org.apache.doris.datasource.storage.StorageAdapter;
 import org.apache.doris.persist.gson.GsonPostProcessable;
 
 import com.google.gson.annotations.SerializedName;
-import lombok.Getter;
 
 import java.io.IOException;
 import java.util.HashMap;
@@ -44,8 +42,8 @@ public class StorageDesc extends ResourceDesc implements GsonPostProcessable {
     @SerializedName("st")
     protected StorageBackend.StorageType storageType;
 
-    @Getter
-    protected StorageProperties storageProperties;
+    /** SPI facade binding; lazily bound from the raw properties. Not Gson-serialized. */
+    protected StorageAdapter storageAdapter;
 
     public StorageDesc() {
     }
@@ -54,22 +52,22 @@ public StorageDesc(String name, StorageBackend.StorageType storageType, Map<Stri
         this.name = name;
         this.storageType = storageType;
         this.properties = properties;
-        initStorageProperties();
+        initStorageAdapter();
     }
 
-    protected void initStorageProperties() {
-        if (storageProperties != null) {
+    protected void initStorageAdapter() {
+        if (storageAdapter != null) {
             return;
         }
         if (properties == null) {
             properties = new HashMap<>();
         }
         if (null != storageType && storageType.equals(StorageBackend.StorageType.BROKER)) {
-            this.storageProperties = BrokerProperties.of(name, properties);
+            this.storageAdapter = StorageAdapter.ofBroker(name, properties);
             return;
         }
         if (!properties.isEmpty()) {
-            this.storageProperties = StorageProperties.createPrimary(properties);
+            this.storageAdapter = StorageAdapter.of(properties);
         }
     }
 
@@ -98,20 +96,20 @@ public Map<String, String> getProperties() {
     }
 
     public Map<String, String> getBackendConfigProperties() {
-        initStorageProperties();
-        if (null == storageProperties) {
+        initStorageAdapter();
+        if (null == storageAdapter) {
             return properties;
         }
-        return storageProperties.getBackendConfigProperties();
+        return storageAdapter.getBackendConfigProperties();
     }
 
-    public StorageProperties getStorageProperties() {
-        initStorageProperties();
-        return storageProperties;
+    public StorageAdapter getStorageAdapter() {
+        initStorageAdapter();
+        return storageAdapter;
     }
 
     @Override
     public void gsonPostProcess() throws IOException {
-        initStorageProperties();
+        initStorageAdapter();
     }
 }

fe/fe-core/src/main/java/org/apache/doris/backup/BackupHandler.java

@@ -39,7 +39,7 @@
 import org.apache.doris.common.util.MasterDaemon;
 import org.apache.doris.common.util.TimeUtils;
 import org.apache.doris.common.util.Util;
-import org.apache.doris.datasource.property.storage.StorageProperties;
+import org.apache.doris.datasource.storage.StorageAdapter;
 import org.apache.doris.info.TableRefInfo;
 import org.apache.doris.nereids.trees.plans.commands.BackupCommand;
 import org.apache.doris.nereids.trees.plans.commands.CancelBackupCommand;
@@ -207,7 +207,7 @@ public void createRepository(CreateRepositoryCommand command) throws DdlExceptio
 
         long repoId = env.getNextId();
         Repository repo = new Repository(repoId, command.getName(), command.isReadOnly(), command.getLocation(),
-                command.getStorageProperties());
+                command.getStorageAdapter());
 
         Status st = repoMgr.addAndInitRepoIfNotExist(repo, false);
         if (!st.ok()) {
@@ -242,7 +242,7 @@ public void alterRepository(String repoName, Map<String, String> newProps)
             // Create new Repository instance with merged properties
             Repository newRepo = new Repository(
                     oldRepo.getId(), oldRepo.getName(), oldRepo.isReadOnly(),
-                    oldRepo.getLocation(), StorageProperties.createPrimary(mergedProps)
+                    oldRepo.getLocation(), StorageAdapter.of(mergedProps)
             );
             // Verify the repository can be connected with new settings
             if (!newRepo.ping()) {