[feat](param-refactor)Rename FS v2 to v1, remove legacy FS, and integrate new Storage/Azure features (#51539)

#50238 
### What problem does this PR solve?
#### Rename FS v2 to v1

The current FS v2 implementation has been renamed to FS v1, making it
the default and primary file system implementation.

Remove Legacy FS

All code and dependencies related to the legacy FS (the old v1
implementation) have been completely removed.


####  Repository Cleanup

Dropped all compatibility logic for the old FS.

The Repository layer now exclusively uses the new FS and Storage
interfaces.

Deprecated configuration entries, injection code, and transitional hooks
have been removed.


####  New Storage and Azure Feature Integration

Added support for new Azure file system features
Improved integration between the Storage layer and Repository for better
scalability.

Author: CalvinKirs

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

@@ -30,7 +30,7 @@
 import org.apache.doris.common.util.PrintableMap;
 import org.apache.doris.common.util.TimeUtils;
 import org.apache.doris.datasource.property.storage.ObjectStorageProperties;
-import org.apache.doris.fsv2.FileSystemFactory;
+import org.apache.doris.fs.FileSystemFactory;
 import org.apache.doris.load.EtlJobType;
 import org.apache.doris.load.loadv2.LoadTask;
 import org.apache.doris.mysql.privilege.PrivPredicate;

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

@@ -44,17 +44,16 @@
 import org.apache.doris.common.ErrorCode;
 import org.apache.doris.common.ErrorReport;
 import org.apache.doris.common.Pair;
-import org.apache.doris.common.UserException;
 import org.apache.doris.common.io.Writable;
 import org.apache.doris.common.util.DebugPointUtil;
 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.fsv2.FileSystemFactory;
-import org.apache.doris.fsv2.remote.AzureFileSystem;
-import org.apache.doris.fsv2.remote.RemoteFileSystem;
-import org.apache.doris.fsv2.remote.S3FileSystem;
+import org.apache.doris.fs.FileSystemFactory;
+import org.apache.doris.fs.remote.AzureFileSystem;
+import org.apache.doris.fs.remote.RemoteFileSystem;
+import org.apache.doris.fs.remote.S3FileSystem;
 import org.apache.doris.nereids.trees.plans.commands.CancelBackupCommand;
 import org.apache.doris.nereids.trees.plans.commands.CreateRepositoryCommand;
 import org.apache.doris.persist.BarrierLog;
@@ -218,17 +217,10 @@ public void createRepository(CreateRepositoryCommand command) throws DdlExceptio
         }
 
         RemoteFileSystem fileSystem;
-        try {
-            fileSystem = FileSystemFactory.get(command.getStorageType(), command.getProperties());
-        } catch (UserException e) {
-            throw new DdlException("Failed to initialize remote file system: " + e.getMessage());
-        }
-        org.apache.doris.fs.remote.RemoteFileSystem oldfs = org.apache.doris.fs.FileSystemFactory
-                .get(command.getBrokerName(), command.getStorageType(),
-                command.getProperties());
+        fileSystem = FileSystemFactory.get(command.getStorageType(), command.getBrokerName(), command.getProperties());
         long repoId = env.getNextId();
         Repository repo = new Repository(repoId, command.getName(), command.isReadOnly(), command.getLocation(),
-                fileSystem, oldfs);
+                fileSystem);
 
         Status st = repoMgr.addAndInitRepoIfNotExist(repo, false);
         if (!st.ok()) {
@@ -250,17 +242,10 @@ public void createRepository(CreateRepositoryStmt stmt) throws DdlException {
         }
 
         RemoteFileSystem fileSystem;
-        try {
-            fileSystem = FileSystemFactory.get(stmt.getStorageType(), stmt.getProperties());
-        } catch (UserException e) {
-            throw new DdlException("Failed to initialize remote file system: " + e.getMessage());
-        }
-        org.apache.doris.fs.remote.RemoteFileSystem oldfs = org.apache.doris.fs.FileSystemFactory
-                .get(stmt.getBrokerName(), stmt.getStorageType(),
-                        stmt.getProperties());
+        fileSystem = FileSystemFactory.get(stmt.getStorageType(), stmt.getBrokerName(), stmt.getProperties());
         long repoId = env.getNextId();
         Repository repo = new Repository(repoId, stmt.getName(), stmt.isReadOnly(), stmt.getLocation(),
-                fileSystem, oldfs);
+                fileSystem);
 
         Status st = repoMgr.addAndInitRepoIfNotExist(repo, false);
         if (!st.ok()) {
@@ -295,18 +280,10 @@ public void alterRepository(String repoName, Map<String, String> newProps, boole
             Map<String, String> mergedProps = mergeProperties(oldRepo, newProps, strictCheck);
             // Create new remote file system with merged properties
             RemoteFileSystem fileSystem = FileSystemFactory.get(StorageProperties.createPrimary(mergedProps));
-            org.apache.doris.fs.remote.RemoteFileSystem oldfs = null;
-            if (oldRepo.getRemoteFileSystem() instanceof S3FileSystem) {
-                oldfs = org.apache.doris.fs.FileSystemFactory.get(oldRepo.getRemoteFileSystem().getName(),
-                        StorageBackend.StorageType.S3, mergedProps);
-            } else if (oldRepo.getRemoteFileSystem() instanceof AzureFileSystem) {
-                oldfs = org.apache.doris.fs.FileSystemFactory.get(oldRepo.getRemoteFileSystem().getName(),
-                        StorageBackend.StorageType.AZURE, mergedProps);
-            }
             // Create new Repository instance with updated file system
             Repository newRepo = new Repository(
                     oldRepo.getId(), oldRepo.getName(), oldRepo.isReadOnly(),
-                    oldRepo.getLocation(), fileSystem, oldfs
+                    oldRepo.getLocation(), fileSystem
             );
             // Verify the repository can be connected with new settings
             if (!newRepo.ping()) {

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

@@ -32,12 +32,12 @@
 import org.apache.doris.common.util.TimeUtils;
 import org.apache.doris.datasource.property.constants.S3Properties;
 import org.apache.doris.datasource.property.storage.StorageProperties;
-import org.apache.doris.fsv2.FileSystemFactory;
-import org.apache.doris.fsv2.PersistentFileSystem;
-import org.apache.doris.fsv2.remote.BrokerFileSystem;
-import org.apache.doris.fsv2.remote.RemoteFile;
-import org.apache.doris.fsv2.remote.RemoteFileSystem;
-import org.apache.doris.fsv2.remote.S3FileSystem;
+import org.apache.doris.fs.FileSystemFactory;
+import org.apache.doris.fs.PersistentFileSystem;
+import org.apache.doris.fs.remote.BrokerFileSystem;
+import org.apache.doris.fs.remote.RemoteFile;
+import org.apache.doris.fs.remote.RemoteFileSystem;
+import org.apache.doris.fs.remote.S3FileSystem;
 import org.apache.doris.persist.gson.GsonPostProcessable;
 import org.apache.doris.persist.gson.GsonUtils;
 import org.apache.doris.system.Backend;
@@ -132,28 +132,22 @@ public class Repository implements Writable, GsonPostProcessable {
     private String location;
 
     @SerializedName("fs")
-    private org.apache.doris.fs.PersistentFileSystem oldfs;
-
-    // Temporary field: currently still using the legacy fs config (oldfs).
-    // This field can be removed once the new fs configuration is fully enabled.
     private PersistentFileSystem fileSystem;
 
-    public org.apache.doris.fs.PersistentFileSystem getOldfs() {
-        return oldfs;
+    public PersistentFileSystem getFileSystem() {
+        return fileSystem;
     }
 
     private Repository() {
         // for persist
     }
 
-    public Repository(long id, String name, boolean isReadOnly, String location, RemoteFileSystem fileSystem,
-                      org.apache.doris.fs.PersistentFileSystem oldFs) {
+    public Repository(long id, String name, boolean isReadOnly, String location, RemoteFileSystem fileSystem) {
         this.id = id;
         this.name = name;
         this.isReadOnly = isReadOnly;
         this.location = location;
         this.fileSystem = fileSystem;
-        this.oldfs = oldFs;
         this.createTime = System.currentTimeMillis();
     }
 
@@ -248,16 +242,8 @@ public Status alterRepositoryS3Properties(Map<String, String> properties) {
     @Override
     public void gsonPostProcess() {
         StorageBackend.StorageType type = StorageBackend.StorageType.BROKER;
-        if (this.oldfs.properties.containsKey(org.apache.doris.fs.PersistentFileSystem.STORAGE_TYPE)) {
-            type = StorageBackend.StorageType.valueOf(
-                    this.oldfs.properties.get(org.apache.doris.fs.PersistentFileSystem.STORAGE_TYPE));
-            this.oldfs.properties.remove(org.apache.doris.fs.PersistentFileSystem.STORAGE_TYPE);
-        }
-        this.oldfs = org.apache.doris.fs.FileSystemFactory.get(this.oldfs.getName(),
-                type,
-                this.oldfs.getProperties());
         if (!type.equals(StorageBackend.StorageType.BROKER)) {
-            StorageProperties storageProperties = StorageProperties.createPrimary(this.oldfs.properties);
+            StorageProperties storageProperties = StorageProperties.createPrimary(this.fileSystem.properties);
             this.fileSystem = FileSystemFactory.get(storageProperties);
         }
     }
@@ -867,13 +853,7 @@ public void readFields(DataInput in) throws IOException {
         name = Text.readString(in);
         isReadOnly = in.readBoolean();
         location = Text.readString(in);
-        oldfs = org.apache.doris.fs.PersistentFileSystem.read(in);
-        try {
-            fileSystem = FileSystemFactory.get(oldfs.getStorageType(), oldfs.getProperties());
-        } catch (UserException e) {
-            // do we ignore this exception?
-            throw new IOException("Failed to create file system: " + e.getMessage());
-        }
+        fileSystem = PersistentFileSystem.read(in);
         createTime = in.readLong();
     }
 }

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

@@ -23,8 +23,8 @@
 import org.apache.doris.common.io.Text;
 import org.apache.doris.common.io.Writable;
 import org.apache.doris.common.util.Daemon;
-import org.apache.doris.fsv2.remote.AzureFileSystem;
-import org.apache.doris.fsv2.remote.S3FileSystem;
+import org.apache.doris.fs.remote.AzureFileSystem;
+import org.apache.doris.fs.remote.S3FileSystem;
 import org.apache.doris.persist.gson.GsonPostProcessable;
 import org.apache.doris.persist.gson.GsonUtils;
 

fe/fe-core/src/main/java/org/apache/doris/catalog/AzureResource.java

@@ -22,6 +22,8 @@
 import org.apache.doris.common.proc.BaseProcResult;
 import org.apache.doris.common.util.PrintableMap;
 import org.apache.doris.datasource.property.constants.S3Properties;
+import org.apache.doris.datasource.property.storage.AzureProperties;
+import org.apache.doris.datasource.property.storage.StorageProperties;
 import org.apache.doris.fs.obj.AzureObjStorage;
 import org.apache.doris.fs.obj.ObjStorage;
 import org.apache.doris.fs.obj.RemoteObjects;
@@ -90,7 +92,8 @@ protected static void pingAzure(String bucketName, String rootPath,
 
         byte[] contentData = new byte[2 * ObjStorage.CHUNK_SIZE];
         Arrays.fill(contentData, (byte) 'A');
-        AzureObjStorage azureObjStorage = new AzureObjStorage(newProperties);
+        AzureProperties azureProperties = (AzureProperties) StorageProperties.createPrimary(newProperties);
+        AzureObjStorage azureObjStorage = new AzureObjStorage(azureProperties);
 
         Status status = azureObjStorage.putObject(testObj, new ByteArrayInputStream(contentData), contentData.length);
         if (!Status.OK.equals(status)) {

fe/fe-core/src/main/java/org/apache/doris/catalog/HdfsStorageVault.java

@@ -23,6 +23,8 @@
 import org.apache.doris.common.security.authentication.AuthenticationConfig;
 import org.apache.doris.common.util.PrintableMap;
 import org.apache.doris.datasource.property.constants.S3Properties;
+import org.apache.doris.datasource.property.storage.HdfsCompatibleProperties;
+import org.apache.doris.datasource.property.storage.StorageProperties;
 import org.apache.doris.fs.remote.dfs.DFSFileSystem;
 
 import com.google.common.base.Preconditions;
@@ -130,7 +132,8 @@ public static void checkConnectivity(Map<String, String> newProperties) throws D
         Preconditions.checkArgument(
                 !Strings.isNullOrEmpty(pathPrefix), "%s is null or empty", PropertyKey.VAULT_PATH_PREFIX);
 
-        try (DFSFileSystem dfsFileSystem = new DFSFileSystem(newProperties)) {
+        try (DFSFileSystem dfsFileSystem = new DFSFileSystem((HdfsCompatibleProperties) StorageProperties
+                .createPrimary(newProperties))) {
             Long timestamp = System.currentTimeMillis();
             String remotePath = hadoopFsName + "/" + pathPrefix + "/doris-check-connectivity" + timestamp.toString();
 

fe/fe-core/src/main/java/org/apache/doris/catalog/S3Resource.java

@@ -23,6 +23,8 @@
 import org.apache.doris.common.proc.BaseProcResult;
 import org.apache.doris.common.util.PrintableMap;
 import org.apache.doris.datasource.property.constants.S3Properties;
+import org.apache.doris.datasource.property.storage.AbstractS3CompatibleProperties;
+import org.apache.doris.datasource.property.storage.StorageProperties;
 import org.apache.doris.fs.obj.ObjStorage;
 import org.apache.doris.fs.obj.RemoteObjects;
 import org.apache.doris.fs.obj.S3ObjStorage;
@@ -126,7 +128,8 @@ protected static void pingS3(String bucketName, String rootPath, Map<String, Str
 
         byte[] contentData = new byte[2 * ObjStorage.CHUNK_SIZE];
         Arrays.fill(contentData, (byte) 'A');
-        S3ObjStorage s3ObjStorage = new S3ObjStorage(newProperties);
+        S3ObjStorage s3ObjStorage = new S3ObjStorage((AbstractS3CompatibleProperties) StorageProperties
+                .createPrimary(newProperties));
 
         Status status = s3ObjStorage.putObject(testObj, new ByteArrayInputStream(contentData), contentData.length);
         if (!Status.OK.equals(status)) {

fe/fe-core/src/main/java/org/apache/doris/common/util/BrokerUtil.java

@@ -28,9 +28,9 @@
 import org.apache.doris.common.Pair;
 import org.apache.doris.common.UserException;
 import org.apache.doris.datasource.hive.HiveMetaStoreCache;
-import org.apache.doris.fsv2.FileSystemFactory;
-import org.apache.doris.fsv2.remote.RemoteFile;
-import org.apache.doris.fsv2.remote.RemoteFileSystem;
+import org.apache.doris.fs.FileSystemFactory;
+import org.apache.doris.fs.remote.RemoteFile;
+import org.apache.doris.fs.remote.RemoteFileSystem;
 import org.apache.doris.service.FrontendOptions;
 import org.apache.doris.thrift.TBrokerCheckPathExistRequest;
 import org.apache.doris.thrift.TBrokerCheckPathExistResponse;

fe/fe-core/src/main/java/org/apache/doris/common/util/LocationPath.java

@@ -26,7 +26,7 @@
 import org.apache.doris.datasource.property.constants.ObsProperties;
 import org.apache.doris.datasource.property.constants.OssProperties;
 import org.apache.doris.datasource.property.constants.S3Properties;
-import org.apache.doris.fsv2.FileSystemType;
+import org.apache.doris.fs.FileSystemType;
 import org.apache.doris.thrift.TFileType;
 
 import com.google.common.annotations.VisibleForTesting;

fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalMetaCacheMgr.java

@@ -37,7 +37,7 @@
 import org.apache.doris.datasource.mvcc.MvccUtil;
 import org.apache.doris.datasource.paimon.PaimonMetadataCache;
 import org.apache.doris.datasource.paimon.PaimonMetadataCacheMgr;
-import org.apache.doris.fsv2.FileSystemCache;
+import org.apache.doris.fs.FileSystemCache;
 import org.apache.doris.nereids.exceptions.NotSupportedException;
 
 import com.github.benmanes.caffeine.cache.CacheLoader;

fe/fe-core/src/main/java/org/apache/doris/datasource/hive/AcidUtil.java

@@ -21,8 +21,8 @@
 import org.apache.doris.common.util.LocationPath;
 import org.apache.doris.datasource.hive.AcidInfo.DeleteDeltaInfo;
 import org.apache.doris.datasource.hive.HiveMetaStoreCache.FileCacheValue;
-import org.apache.doris.fsv2.FileSystem;
-import org.apache.doris.fsv2.remote.RemoteFile;
+import org.apache.doris.fs.FileSystem;
+import org.apache.doris.fs.remote.RemoteFile;
 
 import lombok.EqualsAndHashCode;
 import lombok.Getter;

fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HMSExternalCatalog.java

@@ -37,9 +37,9 @@
 import org.apache.doris.datasource.operations.ExternalMetadataOperations;
 import org.apache.doris.datasource.property.PropertyConverter;
 import org.apache.doris.datasource.property.constants.HMSProperties;
-import org.apache.doris.fsv2.FileSystemProvider;
-import org.apache.doris.fsv2.FileSystemProviderImpl;
-import org.apache.doris.fsv2.remote.dfs.DFSFileSystem;
+import org.apache.doris.fs.FileSystemProvider;
+import org.apache.doris.fs.FileSystemProviderImpl;
+import org.apache.doris.fs.remote.dfs.DFSFileSystem;
 import org.apache.doris.transaction.TransactionManagerFactory;
 
 import com.google.common.annotations.VisibleForTesting;

fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HMSExternalTable.java

@@ -50,7 +50,7 @@
 import org.apache.doris.datasource.mvcc.MvccUtil;
 import org.apache.doris.datasource.systable.SupportedSysTables;
 import org.apache.doris.datasource.systable.SysTable;
-import org.apache.doris.fsv2.FileSystemDirectoryLister;
+import org.apache.doris.fs.FileSystemDirectoryLister;
 import org.apache.doris.mtmv.MTMVBaseTableIf;
 import org.apache.doris.mtmv.MTMVRefreshContext;
 import org.apache.doris.mtmv.MTMVRelatedTableIf;

fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HMSTransaction.java

@@ -27,12 +27,12 @@
 import org.apache.doris.common.info.SimpleTableInfo;
 import org.apache.doris.common.profile.SummaryProfile;
 import org.apache.doris.datasource.statistics.CommonStatistics;
-import org.apache.doris.fsv2.FileSystem;
-import org.apache.doris.fsv2.FileSystemProvider;
-import org.apache.doris.fsv2.FileSystemUtil;
-import org.apache.doris.fsv2.remote.RemoteFile;
-import org.apache.doris.fsv2.remote.S3FileSystem;
-import org.apache.doris.fsv2.remote.SwitchingFileSystem;
+import org.apache.doris.fs.FileSystem;
+import org.apache.doris.fs.FileSystemProvider;
+import org.apache.doris.fs.FileSystemUtil;
+import org.apache.doris.fs.remote.RemoteFile;
+import org.apache.doris.fs.remote.S3FileSystem;
+import org.apache.doris.fs.remote.SwitchingFileSystem;
 import org.apache.doris.nereids.trees.plans.commands.insert.HiveInsertCommandContext;
 import org.apache.doris.qe.ConnectContext;
 import org.apache.doris.thrift.TFileType;

fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HiveMetaStoreCache.java

@@ -40,14 +40,14 @@
 import org.apache.doris.datasource.ExternalCatalog;
 import org.apache.doris.datasource.ExternalMetaCacheMgr;
 import org.apache.doris.datasource.property.PropertyConverter;
-import org.apache.doris.fsv2.DirectoryLister;
-import org.apache.doris.fsv2.FileSystemCache;
-import org.apache.doris.fsv2.FileSystemDirectoryLister;
-import org.apache.doris.fsv2.FileSystemIOException;
-import org.apache.doris.fsv2.RemoteIterator;
-import org.apache.doris.fsv2.remote.RemoteFile;
-import org.apache.doris.fsv2.remote.RemoteFileSystem;
-import org.apache.doris.fsv2.remote.dfs.DFSFileSystem;
+import org.apache.doris.fs.DirectoryLister;
+import org.apache.doris.fs.FileSystemCache;
+import org.apache.doris.fs.FileSystemDirectoryLister;
+import org.apache.doris.fs.FileSystemIOException;
+import org.apache.doris.fs.RemoteIterator;
+import org.apache.doris.fs.remote.RemoteFile;
+import org.apache.doris.fs.remote.RemoteFileSystem;
+import org.apache.doris.fs.remote.dfs.DFSFileSystem;
 import org.apache.doris.metric.GaugeMetric;
 import org.apache.doris.metric.Metric;
 import org.apache.doris.metric.MetricLabel;

fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HiveUtil.java

@@ -22,8 +22,8 @@
 import org.apache.doris.common.UserException;
 import org.apache.doris.datasource.ExternalCatalog;
 import org.apache.doris.datasource.statistics.CommonStatistics;
-import org.apache.doris.fsv2.remote.BrokerFileSystem;
-import org.apache.doris.fsv2.remote.RemoteFileSystem;
+import org.apache.doris.fs.remote.BrokerFileSystem;
+import org.apache.doris.fs.remote.RemoteFileSystem;
 import org.apache.doris.nereids.exceptions.AnalysisException;
 import org.apache.doris.qe.ConnectContext;
 

fe/fe-core/src/main/java/org/apache/doris/datasource/hive/source/HiveScanNode.java

@@ -43,7 +43,7 @@
 import org.apache.doris.datasource.hive.HiveTransaction;
 import org.apache.doris.datasource.hive.source.HiveSplit.HiveSplitCreator;
 import org.apache.doris.datasource.mvcc.MvccUtil;
-import org.apache.doris.fsv2.DirectoryLister;
+import org.apache.doris.fs.DirectoryLister;
 import org.apache.doris.nereids.trees.plans.logical.LogicalFileScan.SelectedPartitions;
 import org.apache.doris.planner.PlanNodeId;
 import org.apache.doris.qe.ConnectContext;

fe/fe-core/src/main/java/org/apache/doris/datasource/hudi/source/HudiScanNode.java

@@ -36,7 +36,7 @@
 import org.apache.doris.datasource.hudi.HudiSchemaCacheValue;
 import org.apache.doris.datasource.hudi.HudiUtils;
 import org.apache.doris.datasource.mvcc.MvccUtil;
-import org.apache.doris.fsv2.DirectoryLister;
+import org.apache.doris.fs.DirectoryLister;
 import org.apache.doris.planner.PlanNodeId;
 import org.apache.doris.qe.SessionVariable;
 import org.apache.doris.spi.Split;