branch-3.0: [fix](hive/iceberg) rectify the fs name if path already contains fs #49998

be/src/io/file_factory.cpp

@@ -88,20 +88,36 @@ Result<io::FileSystemSPtr> FileFactory::create_fs(const io::FSPropertiesRef& fs_
                 *fs_properties.properties, s3_uri, &s3_conf));
         return io::S3FileSystem::create(std::move(s3_conf), io::FileSystem::TMP_FS_ID);
     }
-    case TFileType::FILE_HDFS:
-        return fs_properties.hdfs_params
-                       ? io::HdfsFileSystem::create(*fs_properties.hdfs_params,
-                                                    file_description.fs_name,
-                                                    io::FileSystem::TMP_FS_ID, nullptr)
-                       : io::HdfsFileSystem::create(*fs_properties.properties,
-                                                    file_description.fs_name,
-                                                    io::FileSystem::TMP_FS_ID, nullptr);
+    case TFileType::FILE_HDFS: {
+        std::string fs_name = _get_fs_name(file_description);
+        return io::HdfsFileSystem::create(*fs_properties.properties, fs_name,
+                                          io::FileSystem::TMP_FS_ID, nullptr);
+    }
     default:
         return ResultError(Status::InternalError("unsupported fs type: {}",
                                                  std::to_string(fs_properties.type)));
     }
 }
 
+std::string FileFactory::_get_fs_name(const io::FileDescription& file_description) {
+    // If the destination path contains a schema, use the schema directly.
+    // If not, use origin file_description.fs_name
+    // Because the default fsname in file_description.fs_name maybe different from
+    // file's.
+    // example:
+    //    hdfs://host:port/path1/path2  --> hdfs://host:port
+    //    hdfs://nameservice/path1/path2 --> hdfs://nameservice
+    std::string fs_name = file_description.fs_name;
+    string::size_type idx = file_description.path.find("://");
+    if (idx != string::npos) {
+        idx = file_description.path.find("/", idx + 3);
+        if (idx != string::npos) {
+            fs_name = file_description.path.substr(0, idx);
+        }
+    }
+    return fs_name;
+}
+
 Result<io::FileWriterPtr> FileFactory::create_file_writer(
         TFileType::type type, ExecEnv* env, const std::vector<TNetworkAddress>& broker_addresses,
         const std::map<std::string, std::string>& properties, const std::string& path,

be/src/io/file_factory.h

@@ -123,6 +123,9 @@ class FileFactory {
         LOG(FATAL) << "__builtin_unreachable";
         __builtin_unreachable();
     }
+
+private:
+    static std::string _get_fs_name(const io::FileDescription& file_description);
 };
 
 } // namespace doris

be/src/vec/sink/writer/vhive_partition_writer.cpp

@@ -61,19 +61,6 @@ Status VHivePartitionWriter::open(RuntimeState* state, RuntimeProfile* profile)
     io::FileDescription file_description = {
             .path = fmt::format("{}/{}", _write_info.write_path, _get_target_file_name()),
             .fs_name {}};
-    // If the destination path contains a schema, use the schema directly.
-    // If not, use defaultFS.
-    // Otherwise a write error will occur.
-    // example:
-    //    hdfs://host:port/path1/path2  --> hdfs://host:port
-    //    hdfs://nameservice/path1/path2 --> hdfs://nameservice
-    string::size_type idx = file_description.path.find("://");
-    if (idx != string::npos) {
-        idx = file_description.path.find("/", idx + 3);
-        if (idx != string::npos) {
-            file_description.fs_name = file_description.path.substr(0, idx);
-        }
-    }
     _fs = DORIS_TRY(FileFactory::create_fs(fs_properties, file_description));
     io::FileWriterOptions file_writer_options = {.used_by_s3_committer = true};
     RETURN_IF_ERROR(_fs->create_file(file_description.path, &_file_writer, &file_writer_options));