[feat](refactor-param)Integrate New Storage System Support for BACKUP/RESTORE/LOAD/TVF (#50849)

Issue Number: #50238
#### Key Changes
Enhanced core functionalities (BACKUP, RESTORE, LOAD, TVF) to support
new storage parameters

Unified storage path and parameter parsing logic across modules to
enable compatibility with multiple storage backends (S3, OSS, COS,
etc.).

#### Details
Refactored the storage parameter handling logic to automatically detect
the schema and route to the corresponding storage system implementation.

Each operation (BACKUP/RESTORE/LOAD/TVF) now uses a unified file system
interface to ensure consistent behavior and extensibility.

Maintains full backward compatibility with existing storage formats such
as HDFS and local file systems.

#### Tests
Added comprehensive unit and integration tests covering:

Storage parameter parsing across different systems

Execution flow for each operation under new storage systems

Edge cases including invalid parameters, permission errors, and
non-existent paths

Author: CalvinKirs

fe/fe-core/pom.xml

@@ -102,6 +102,10 @@ under the License.
             <groupId>commons-pool</groupId>
             <artifactId>commons-pool</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.google.re2j</groupId>
+            <artifactId>re2j</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.apache.commons</groupId>
             <artifactId>commons-text</artifactId>

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

@@ -20,17 +20,19 @@
 import org.apache.doris.analysis.StorageBackend.StorageType;
 import org.apache.doris.catalog.Env;
 import org.apache.doris.common.FeMetaVersion;
+import org.apache.doris.common.UserException;
 import org.apache.doris.common.io.Text;
 import org.apache.doris.common.io.Writable;
 import org.apache.doris.common.util.PrintableMap;
-import org.apache.doris.datasource.property.S3ClientBEProperties;
-import org.apache.doris.datasource.property.constants.BosProperties;
-import org.apache.doris.fs.PersistentFileSystem;
+import org.apache.doris.datasource.property.storage.StorageProperties;
+import org.apache.doris.datasource.property.storage.exception.StoragePropertiesException;
 import org.apache.doris.persist.gson.GsonUtils;
 import org.apache.doris.thrift.TFileType;
 
 import com.google.common.collect.Maps;
 import com.google.gson.annotations.SerializedName;
+import org.apache.commons.collections.MapUtils;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
@@ -53,6 +55,7 @@ public class BrokerDesc extends StorageDesc implements Writable {
     // just for multi load
     public static final String MULTI_LOAD_BROKER = "__DORIS_MULTI_LOAD_BROKER__";
     public static final String MULTI_LOAD_BROKER_BACKEND_KEY = "__DORIS_MULTI_LOAD_BROKER_BACKEND__";
+    @Deprecated
     @SerializedName("cts3")
     private boolean convertedToS3 = false;
 
@@ -75,42 +78,56 @@ public BrokerDesc(String name, Map<String, String> properties) {
         if (properties != null) {
             this.properties.putAll(properties);
         }
+        // Assume the storage type is BROKER by default
+        // If it's a multi-load broker, override the storage type to LOCAL
         if (isMultiLoadBroker()) {
             this.storageType = StorageBackend.StorageType.LOCAL;
         } else {
             this.storageType = StorageBackend.StorageType.BROKER;
         }
-        this.properties.putAll(S3ClientBEProperties.getBeFSProperties(this.properties));
-        this.convertedToS3 = BosProperties.tryConvertBosToS3(this.properties, this.storageType);
-        if (this.convertedToS3) {
-            this.storageType = StorageBackend.StorageType.S3;
+
+        // 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);
+                // Override the storage type based on property configuration
+                this.storageType = StorageBackend.StorageType.valueOf(storageProperties.getStorageName());
+            } catch (StoragePropertiesException e) {
+                // Currently ignored: these properties might be broker-specific.
+                // Support for broker properties will be added in the future.
+                LOG.info("Failed to create storage properties for broker: {}, properties: {}", name, properties, e);
+            }
+        }
+        //only storage type is broker
+        if (StringUtils.isBlank(this.name) && (this.getStorageType() != StorageType.BROKER)) {
+            this.name = this.storageType().name();
         }
     }
 
     public BrokerDesc(String name, StorageBackend.StorageType storageType, Map<String, String> properties) {
         this.name = name;
         this.properties = Maps.newHashMap();
+        this.storageType = storageType;
         if (properties != null) {
             this.properties.putAll(properties);
         }
-        this.storageType = storageType;
-        this.properties.putAll(S3ClientBEProperties.getBeFSProperties(this.properties));
-        this.convertedToS3 = BosProperties.tryConvertBosToS3(this.properties, this.storageType);
-        if (this.convertedToS3) {
-            this.storageType = StorageBackend.StorageType.S3;
+        if (MapUtils.isNotEmpty(this.properties) && StorageType.REFACTOR_STORAGE_TYPES.contains(storageType)) {
+            this.storageProperties = StorageProperties.createPrimary(properties);
         }
+
     }
 
-    public String getFileLocation(String location) {
-        return this.convertedToS3 ? BosProperties.convertPathToS3(location) : location;
+    public String getFileLocation(String location) throws UserException {
+        return (null != storageProperties) ? storageProperties.validateAndNormalizeUri(location) : location;
     }
 
     public static BrokerDesc createForStreamLoad() {
         return new BrokerDesc("", StorageType.STREAM, null);
     }
 
     public boolean isMultiLoadBroker() {
-        return this.name.equalsIgnoreCase(MULTI_LOAD_BROKER);
+        return StringUtils.isNotBlank(this.name) && this.name.equalsIgnoreCase(MULTI_LOAD_BROKER);
     }
 
     public TFileType getFileType() {
@@ -150,16 +167,18 @@ public void readFields(DataInput in) throws IOException {
             final String val = Text.readString(in);
             properties.put(key, val);
         }
-        StorageBackend.StorageType st = StorageBackend.StorageType.BROKER;
-        String typeStr = properties.remove(PersistentFileSystem.STORAGE_TYPE);
-        if (typeStr != null) {
+        if (MapUtils.isNotEmpty(properties)) {
             try {
-                st = StorageBackend.StorageType.valueOf(typeStr);
-            }  catch (IllegalArgumentException e) {
-                LOG.warn("set to BROKER, because of exception", e);
+                this.storageProperties = StorageProperties.createPrimary(properties);
+                this.storageType = StorageBackend.StorageType.valueOf(storageProperties.getStorageName());
+            } catch (RuntimeException e) {
+                // Currently ignored: these properties might be broker-specific.
+                // Support for broker properties will be added in the future.
+                LOG.warn("Failed to create storage properties for broker: {}, properties: {}", name, properties, e);
+                this.storageType = StorageBackend.StorageType.BROKER;
             }
+
         }
-        storageType = st;
     }
 
     public static BrokerDesc read(DataInput in) throws IOException {

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

@@ -21,23 +21,21 @@
 import org.apache.doris.catalog.Env;
 import org.apache.doris.catalog.KeysType;
 import org.apache.doris.catalog.OlapTable;
-import org.apache.doris.cloud.proto.Cloud.ObjectStoreInfoPB;
 import org.apache.doris.cloud.security.SecurityChecker;
-import org.apache.doris.cloud.storage.RemoteBase;
-import org.apache.doris.cloud.storage.RemoteBase.ObjectInfo;
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.Config;
 import org.apache.doris.common.DdlException;
 import org.apache.doris.common.InternalErrorCode;
 import org.apache.doris.common.UserException;
 import org.apache.doris.common.util.PrintableMap;
 import org.apache.doris.common.util.TimeUtils;
-import org.apache.doris.datasource.property.constants.AzureProperties;
-import org.apache.doris.datasource.property.constants.S3Properties;
+import org.apache.doris.datasource.property.storage.ObjectStorageProperties;
+import org.apache.doris.fsv2.FileSystemFactory;
 import org.apache.doris.load.EtlJobType;
 import org.apache.doris.load.loadv2.LoadTask;
 import org.apache.doris.mysql.privilege.PrivPredicate;
 import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.thrift.TFileType;
 
 import com.google.common.base.Function;
 import com.google.common.base.Joiner;
@@ -102,15 +100,11 @@ public class LoadStmt extends DdlStmt implements NotFallbackInParser {
     // deprecated, keeping this property to make LoadStmt#checkProperties() happy
     public static final String USE_NEW_LOAD_SCAN_NODE = "use_new_load_scan_node";
 
-    // for load data from Baidu Object Store(BOS)
+    // for load data from Baidu Object Store(BOS) todo wait new property support
     public static final String BOS_ENDPOINT = "bos_endpoint";
     public static final String BOS_ACCESSKEY = "bos_accesskey";
     public static final String BOS_SECRET_ACCESSKEY = "bos_secret_accesskey";
 
-    // for S3 load check
-    public static final List<String> PROVIDERS =
-            new ArrayList<>(Arrays.asList("cos", "oss", "s3", "obs", "bos", "azure"));
-
     // mini load params
     public static final String KEY_IN_PARAM_COLUMNS = "columns";
     public static final String KEY_IN_PARAM_SET = "set";
@@ -454,8 +448,6 @@ public void analyze(Analyzer analyzer) throws UserException {
                 for (int i = 0; i < dataDescription.getFilePaths().size(); i++) {
                     String location = brokerDesc.getFileLocation(dataDescription.getFilePaths().get(i));
                     dataDescription.getFilePaths().set(i, location);
-                    StorageBackend.checkPath(dataDescription.getFilePaths().get(i),
-                            brokerDesc.getStorageType(), "DATA INFILE must be specified.");
                     dataDescription.getFilePaths().set(i, dataDescription.getFilePaths().get(i));
                 }
             }
@@ -522,27 +514,6 @@ public void analyze(Analyzer analyzer) throws UserException {
         user = ConnectContext.get().getQualifiedUser();
     }
 
-
-    private String getProviderFromEndpoint() {
-        Map<String, String> properties = brokerDesc.getProperties();
-        for (Map.Entry<String, String> entry : properties.entrySet()) {
-            if (entry.getKey().equalsIgnoreCase(S3Properties.PROVIDER)) {
-                // S3 Provider properties should be case insensitive.
-                return entry.getValue().toUpperCase();
-            }
-        }
-        return S3Properties.S3_PROVIDER;
-    }
-
-    private String getBucketFromFilePath(String filePath) throws Exception {
-        String[] parts = filePath.split("\\/\\/");
-        if (parts.length < 2) {
-            throw new Exception("filePath is not valid");
-        }
-        String buckt = parts[1].split("\\/")[0];
-        return buckt;
-    }
-
     public String getComment() {
         return comment;
     }
@@ -630,21 +601,17 @@ private void checkEndpoint(String endpoint) throws UserException {
     }
 
     public void checkS3Param() throws UserException {
-        Map<String, String> brokerDescProperties = brokerDesc.getProperties();
-        if (brokerDescProperties.containsKey(S3Properties.Env.ENDPOINT)
-                && brokerDescProperties.containsKey(S3Properties.Env.ACCESS_KEY)
-                && brokerDescProperties.containsKey(S3Properties.Env.SECRET_KEY)
-                && brokerDescProperties.containsKey(S3Properties.Env.REGION)) {
-            String endpoint = brokerDescProperties.get(S3Properties.Env.ENDPOINT);
-            endpoint = endpoint.replaceFirst("^http://", "");
-            endpoint = endpoint.replaceFirst("^https://", "");
-            brokerDescProperties.put(S3Properties.Env.ENDPOINT, endpoint);
+        if (brokerDesc.getFileType() != null && brokerDesc.getFileType().equals(TFileType.FILE_S3)) {
+
+            ObjectStorageProperties storageProperties = (ObjectStorageProperties) brokerDesc.getStorageProperties();
+            String endpoint = storageProperties.getEndpoint();
+            checkEndpoint(endpoint);
             checkWhiteList(endpoint);
-            if (AzureProperties.checkAzureProviderPropertyExist(brokerDescProperties)) {
-                return;
+            //should add connectivity test
+            boolean connectivityTest = FileSystemFactory.get(brokerDesc.getStorageProperties()).connectivityTest();
+            if (!connectivityTest) {
+                throw new UserException("Failed to access object storage, message=connectivity test failed");
             }
-            checkEndpoint(endpoint);
-            checkAkSk();
         }
     }
 
@@ -657,47 +624,6 @@ public void checkWhiteList(String endpoint) throws UserException {
         }
     }
 
-    private void checkAkSk() throws UserException {
-        RemoteBase remote = null;
-        ObjectInfo objectInfo = null;
-        String curFile = null;
-        try {
-            Map<String, String> brokerDescProperties = brokerDesc.getProperties();
-            String provider = getProviderFromEndpoint();
-            for (DataDescription dataDescription : dataDescriptions) {
-                for (String filePath : dataDescription.getFilePaths()) {
-                    curFile = filePath;
-                    String bucket = getBucketFromFilePath(filePath);
-                    objectInfo = new ObjectInfo(ObjectStoreInfoPB.Provider.valueOf(provider.toUpperCase()),
-                            brokerDescProperties.get(S3Properties.Env.ACCESS_KEY),
-                            brokerDescProperties.get(S3Properties.Env.SECRET_KEY),
-                            bucket, brokerDescProperties.get(S3Properties.Env.ENDPOINT),
-                            brokerDescProperties.get(S3Properties.Env.REGION), "");
-                    remote = RemoteBase.newInstance(objectInfo);
-                    // RemoteBase#headObject does not throw exception if key does not exist.
-                    remote.headObject("1");
-                    remote.listObjects(null);
-                    remote.close();
-                }
-            }
-        } catch (Exception e) {
-            LOG.warn("Failed to access object storage, file={}, proto={}, err={}", curFile, objectInfo, e.toString());
-            String msg;
-            if (e instanceof UserException) {
-                msg = ((UserException) e).getDetailMessage();
-            } else {
-                msg = e.getMessage();
-            }
-            throw new UserException(InternalErrorCode.GET_REMOTE_DATA_ERROR,
-                    "Failed to access object storage, message=" + msg, e);
-        } finally {
-            if (remote != null) {
-                remote.close();
-            }
-        }
-
-    }
-
     @Override
     public StmtType stmtType() {
         return StmtType.LOAD;