| 1012 | } |
| 1013 | |
| 1014 | Status Coordinator::Wait() { |
| 1015 | DCHECK(exec_rpcs_complete_.Load()) << "Exec() must be called first"; |
| 1016 | lock_guard<SpinLock> l(wait_lock_); |
| 1017 | SCOPED_TIMER(query_profile_->total_time_counter()); |
| 1018 | if (has_called_wait_.Load()) return Status::OK(); |
| 1019 | has_called_wait_.Store(true); |
| 1020 | |
| 1021 | if (stmt_type_ == TStmtType::QUERY && !exec_params_.HasResultSink()) { |
| 1022 | DCHECK(coord_instance_ != nullptr); |
| 1023 | RETURN_IF_ERROR(UpdateExecState(coord_instance_->WaitForOpen(), |
| 1024 | &coord_instance_->runtime_state()->fragment_instance_id(), FLAGS_hostname)); |
| 1025 | if (query_state_->query_options().retry_failed_queries |
| 1026 | && query_state_->query_options().spool_query_results |
| 1027 | && query_state_->query_options().spool_all_results_for_retries) { |
| 1028 | // Wait until the BufferedPlanRootSink spooled all results or any errors stopping |
| 1029 | // it, e.g. batch queue full, cancellation or failures. |
| 1030 | auto sink = static_cast<BufferedPlanRootSink*>(coord_sink_); |
| 1031 | if (sink->WaitForAllResultsSpooled()) { |
| 1032 | VLOG_QUERY << "Cannot spool all results in the allocated result spooling space." |
| 1033 | " Query retry will be skipped if any results have been returned."; |
| 1034 | } |
| 1035 | } |
| 1036 | return Status::OK(); |
| 1037 | } |
| 1038 | DCHECK(stmt_type_ == TStmtType::DML || |
| 1039 | (stmt_type_ == TStmtType::QUERY && exec_params_.HasResultSink())); |
| 1040 | // DML finalization can only happen when all backends have completed all side-effects |
| 1041 | // and reported relevant state. We also wait for backends to complete if we are writing |
| 1042 | // results to a filesystem. |
| 1043 | WaitForBackends(); |
| 1044 | |
| 1045 | if (finalize_params() != nullptr) { |
| 1046 | RETURN_IF_ERROR(UpdateExecState(FinalizeHdfsDml(), nullptr, FLAGS_hostname)); |
| 1047 | } else if (exec_params_.HasResultSink()) { |
| 1048 | RETURN_IF_ERROR(UpdateExecState(FinalizeResultSink(), nullptr, FLAGS_hostname)); |
| 1049 | } |
| 1050 | |
| 1051 | // DML queries are finished at this point. |
| 1052 | RETURN_IF_ERROR(SetNonErrorTerminalState(ExecState::RETURNED_RESULTS)); |
| 1053 | query_profile_->AddInfoString( |
| 1054 | "DML Stats", dml_exec_state_.OutputPartitionStats("\n")); |
| 1055 | if (parent_request_state_->otel_trace_query()) { |
| 1056 | parent_request_state_->otel_trace_manager()->AddChildSpanEvent("LastRowFetched"); |
| 1057 | } |
| 1058 | return Status::OK(); |
| 1059 | } |
| 1060 | |
| 1061 | Status Coordinator::GetNext(QueryResultSet* results, int max_rows, bool* eos, |
| 1062 | int64_t block_on_wait_time_us) { |