[fix](Outfile) fix bug that it will core dump if the _schema fails to…

… build in the open phase (#39770)

## Proposed changes

refer to: #25920

*** SIGSEGV address not mapped to object (@0x0) received by PID 73639
(TID 0x7f16ef041700) from PID 0; stack trace: ***
0# doris::signal::(anonymous namespace)::FailureSignalHandler(int,
siginfo_t*, void*) in /usr/local/service/doris/lib/be/doris_be
1# os::Linux::chained_handler(int, siginfo*, void*) in
/usr/local/jdk/jre/lib/amd64/server/libjvm.so
2# JVM_handle_linux_signal in
/usr/local/jdk/jre/lib/amd64/server/libjvm.so
3# signalHandler(int, siginfo*, void*) in
/usr/local/jdk/jre/lib/amd64/server/libjvm.so
 4# 0x00007F18DFB65400 in /lib64/libc.so.6
5# doris::vectorized::VOrcWriterWrapper::written_len() in
/usr/local/service/doris/lib/be/doris_be
6# doris::vectorized::VFileResultWriter::_close_file_writer(bool, bool)
in /usr/local/service/doris/lib/be/doris_be
7# doris::vectorized::VFileResultWriter::close() in
/usr/local/service/doris/lib/be/doris_be
8# doris::vectorized::VResultFileSink::close(doris::RuntimeState*,
doris::Status) in /usr/local/service/doris/lib/be/doris_be
9# doris::PlanFragmentExecutor::close() in
/usr/local/service/doris/lib/be/doris_be
10# doris::PlanFragmentExecutor::~PlanFragmentExecutor() in
/usr/local/service/doris/lib/be/doris_be
11# std::_Sp_counted_ptr<doris::FragmentExecState*,
(__gnu_cxx::_Lock_policy)2>::_M_dispose() in
/usr/local/service/doris/lib/be/doris_be
12# std::_Sp_counted_base<(__gnu_cxx::_Lock_policy)2>::_M_release() in
/usr/local/service/doris/lib/be/doris_be
13#
doris::FragmentMgr::exec_plan_fragment(doris::TExecPlanFragmentParams
const&, std::function<void (doris::PlanFragmentExecutor*)>) in
/usr/local/s
ervice/doris/lib/be/doris_be
14#
doris::FragmentMgr::exec_plan_fragment(doris::TExecPlanFragmentParams
const&) in /usr/local/service/doris/lib/be/doris_be
15#
doris::PInternalServiceImpl::_exec_plan_fragment_impl(std::__cxx11::basic_string<char,
std::char_traits<char>, std::allocator<char> > const&, dor
is::PFragmentRequestVersion, bool) in
/usr/local/service/doris/lib/be/doris_be
16#
doris::PInternalServiceImpl::_exec_plan_fragment_in_pthread(google::protobuf::RpcController*,
doris::PExecPlanFragmentRequest const*, doris::PExe
cPlanFragmentResult*, google::protobuf::Closure*) in
/usr/local/service/doris/lib/be/doris_be
17# doris::PriorityThreadPool::work_thread(int) in
/usr/local/service/doris/lib/be/doris_be
18# execute_native_thread_routine in
/usr/local/service/doris/lib/be/doris_be
19# start_thread in /lib64/libpthread.so.0
20# __clone in /lib64/libc.so.6

<!--Describe your changes.-->

Co-authored-by: derenli <[email protected]>

be/src/vec/runtime/vfile_result_writer.cpp

@@ -434,12 +434,12 @@ Status VFileResultWriter::_create_new_file_if_exceed_size() {
 
 Status VFileResultWriter::_close_file_writer(bool done, bool only_close) {
     if (_vfile_writer) {
-        _vfile_writer->close();
         // we can not use _current_written_bytes to COUNTER_UPDATE(_written_data_bytes, _current_written_bytes)
         // because it will call `write()` function of orc/parquet function in `_vfile_writer->close()`
         // and the real written_len will increase
         // and _current_written_bytes will less than _vfile_writer->written_len()
         COUNTER_UPDATE(_written_data_bytes, _vfile_writer->written_len());
+        _vfile_writer->close();
         _vfile_writer.reset(nullptr);
     } else if (_file_writer_impl) {
         _file_writer_impl->close();

be/src/vec/runtime/vorc_writer.cpp

@@ -92,13 +92,22 @@ std::unique_ptr<orc::ColumnVectorBatch> VOrcWriterWrapper::_create_row_batch(siz
 }
 
 int64_t VOrcWriterWrapper::written_len() {
-    return _output_stream->getLength();
+    // written_len() will be called in VFileResultWriter::_close_file_writer
+    // but _output_stream may be nullptr
+    // because the failure built by _schema in open()
+    if (_output_stream) {
+        return _output_stream->getLength();
+    }
+    return 0;
 }
 
 void VOrcWriterWrapper::close() {
     if (_writer != nullptr) {
         _writer->close();
     }
+    if (_output_stream) {
+        _output_stream->close();
+    }
 }
 
 #define RETURN_WRONG_TYPE \