[fix](sort) fix coredump by uncaught exception。 DO NOT MERGE

be/src/vec/core/sort_cursor.h

@@ -251,7 +251,7 @@ struct BlockSupplierSortCursorImpl : public MergeSortCursorImpl {
         } else if (!status.ok()) {
             // Currently, a known error status is emitted when sender
             // close recei
-            throw std::runtime_error(status.msg());
+            throw Exception(status.code(), status.msg());
         }
         return false;
     }

be/src/vec/runtime/vsorted_run_merger.cpp

@@ -97,7 +97,7 @@ Status VSortedRunMerger::prepare(const vector<BlockSupplier>& input_runs) {
     return Status::OK();
 }
 
-Status VSortedRunMerger::get_next(Block* output_block, bool* eos) {
+Status VSortedRunMerger::_get_next_internal(Block* output_block, bool* eos) {
     ScopedTimer<MonotonicStopWatch> timer(_get_next_timer);
     // Only have one receive data queue of data, no need to do merge and
     // copy the data of block.
@@ -213,6 +213,10 @@ Status VSortedRunMerger::get_next(Block* output_block, bool* eos) {
     return Status::OK();
 }
 
+Status VSortedRunMerger::get_next(Block* output_block, bool* eos) {
+    RETURN_IF_CATCH_EXCEPTION(return _get_next_internal(output_block, eos));
+}
+
 bool VSortedRunMerger::next_heap(MergeSortCursor& current) {
     if (!current->isLast()) {
         current->next();

be/src/vec/runtime/vsorted_run_merger.h

@@ -99,6 +99,8 @@ class VSortedRunMerger {
     /// In pipeline engine, return false if need to read one more block from sender.
     bool next_heap(MergeSortCursor& current);
     bool has_next_block(MergeSortCursor& current);
+
+    Status _get_next_internal(Block* output_block, bool* eos);
 };
 
 } // namespace vectorized