[fix](be) return correct canceled status from scanner

be/src/pipeline/exec/scan_operator.cpp

@@ -1461,7 +1461,7 @@ Status ScanOperatorX<LocalStateType>::get_block(RuntimeState* state, vectorized:
         if (local_state._scanner_ctx) {
             local_state._scanner_ctx->stop_scanners(state);
         }
-        return Status::Cancelled("Query cancelled in ScanOperator");
+        return state->cancel_reason();
     }
 
     if (local_state._eos) {

be/src/runtime/runtime_state.cpp

@@ -345,7 +345,15 @@ bool RuntimeState::is_cancelled() const {
 }
 
 Status RuntimeState::cancel_reason() const {
-    return _exec_status.status();
+    if (!_exec_status.ok()) {
+        return _exec_status.status();
+    }
+
+    if (_query_ctx) {
+        return _query_ctx->exec_status();
+    }
+
+    return Status::Cancelled("Query cancelled");
 }
 
 const int64_t MAX_ERROR_NUM = 50;

be/src/vec/exec/scan/scanner_context.cpp

@@ -215,7 +215,7 @@ Status ScannerContext::get_block_from_queue(RuntimeState* state, vectorized::Blo
                                             bool* eos, int id) {
     if (state->is_cancelled()) {
         _set_scanner_done();
-        return Status::Cancelled("Query cancelled in ScannerContext");
+        return state->cancel_reason();
     }
     std::unique_lock l(_transfer_lock);