1

Author: wyxxxcat

be/src/io/file_factory.h

@@ -107,7 +107,7 @@ class FileFactory {
     static Status create_pipe_reader(const TUniqueId& load_id, io::FileReaderSPtr* file_reader,
                                      RuntimeState* runtime_state, bool need_schema);
 
-    static TFileType::type convert_storage_type(TStorageBackendType::type type) {
+    static Result<TFileType::type> convert_storage_type(TStorageBackendType::type type) {
         switch (type) {
         case TStorageBackendType::LOCAL:
             return TFileType::FILE_LOCAL;
@@ -120,9 +120,8 @@ class FileFactory {
         case TStorageBackendType::HDFS:
             return TFileType::FILE_HDFS;
         default:
-            throw Exception(Status::FatalError("not match type to convert, from type:{}", type));
+            return ResultError(Status::FatalError("not match type to convert, from type:{}", type));
         }
-        throw Exception(Status::FatalError("__builtin_unreachable"));
     }
 
 private:

be/src/service/internal_service.cpp

@@ -725,14 +725,23 @@ void PInternalService::outfile_write_success(google::protobuf::RpcController* co
             }
         }
 
-        auto&& res = FileFactory::create_file_writer(
-                FileFactory::convert_storage_type(result_file_sink.storage_backend_type),
-                ExecEnv::GetInstance(), file_options.broker_addresses,
-                file_options.broker_properties, file_name,
-                {
-                        .write_file_cache = false,
-                        .sync_file_data = false,
-                });
+        auto file_type_res =
+                FileFactory::convert_storage_type(result_file_sink.storage_backend_type);
+        if (!file_type_res.has_value()) [[unlikely]] {
+            st = std::move(file_type_res).error();
+            st.to_protobuf(result->mutable_status());
+            LOG(WARNING) << "encounter unkonw type=" << result_file_sink.storage_backend_type
+                         << ", st=" << st;
+            return;
+        }
+
+        auto&& res = FileFactory::create_file_writer(file_type_res.value(), ExecEnv::GetInstance(),
+                                                     file_options.broker_addresses,
+                                                     file_options.broker_properties, file_name,
+                                                     {
+                                                             .write_file_cache = false,
+                                                             .sync_file_data = false,
+                                                     });
         using T = std::decay_t<decltype(res)>;
         if (!res.has_value()) [[unlikely]] {
             st = std::forward<T>(res).error();

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

@@ -118,9 +118,10 @@ Status VFileResultWriter::_create_next_file_writer() {
 }
 
 Status VFileResultWriter::_create_file_writer(const std::string& file_name) {
+    auto file_type = DORIS_TRY(FileFactory::convert_storage_type(_storage_type));
     _file_writer_impl = DORIS_TRY(FileFactory::create_file_writer(
-            FileFactory::convert_storage_type(_storage_type), _state->exec_env(),
-            _file_opts->broker_addresses, _file_opts->broker_properties, file_name,
+            file_type, _state->exec_env(), _file_opts->broker_addresses,
+            _file_opts->broker_properties, file_name,
             {
                     .write_file_cache = false,
                     .sync_file_data = false,