| 1823 | } |
| 1824 | |
| 1825 | void ImpalaServer::CloseClientRequestState(const QueryHandle& query_handle) { |
| 1826 | int64_t duration_us = query_handle->end_time_us() - query_handle->start_time_us(); |
| 1827 | int64_t duration_ms = duration_us / MICROS_PER_MILLI; |
| 1828 | |
| 1829 | // duration_ms can be negative when the local timezone changes during query execution. |
| 1830 | if (duration_ms >= 0) { |
| 1831 | if (query_handle->stmt_type() == TStmtType::DDL) { |
| 1832 | ImpaladMetrics::DDL_DURATIONS->Update(duration_ms); |
| 1833 | } else { |
| 1834 | ImpaladMetrics::QUERY_DURATIONS->Update(duration_ms); |
| 1835 | } |
| 1836 | } |
| 1837 | |
| 1838 | // Final attempt to capture completed RPC stats |
| 1839 | query_handle->UnRegisterCompletedRPCs(); |
| 1840 | // Unregister any remaining RPC and discard stats |
| 1841 | query_handle->UnRegisterRemainingRPCs(); |
| 1842 | |
| 1843 | { |
| 1844 | lock_guard<mutex> l(query_handle->session()->lock); |
| 1845 | if (query_handle->session()->inflight_queries.erase(query_handle->query_id()) == 0 |
| 1846 | && query_handle->IsSetRetriedId()) { |
| 1847 | // Closing a ClientRequestState but the query is retrying with a new state and the |
| 1848 | // original ID was not yet inflight: skip adding it later. We don't want to track |
| 1849 | // other scenarios because they happen when the query was started and then errored |
| 1850 | // before a call to SetQueryInflight; we don't expect it to ever be called. |
| 1851 | auto prestopped_it = |
| 1852 | query_handle->session()->prestopped_queries.insert(query_handle->query_id()); |
| 1853 | if (UNLIKELY(!prestopped_it.second)) { |
| 1854 | LOG(WARNING) << "Query " << PrintId(query_handle->query_id()) |
| 1855 | << " closed again before in-flight."; |
| 1856 | DCHECK(false) << "CloseClientRequestState called twice for query_id=" |
| 1857 | << PrintId(query_handle->query_id()); |
| 1858 | } else { |
| 1859 | VLOG_QUERY << "Query " << PrintId(query_handle->query_id()) |
| 1860 | << " closed before in-flight."; |
| 1861 | } |
| 1862 | } |
| 1863 | } |
| 1864 | |
| 1865 | if (query_handle->GetCoordinator() != nullptr) { |
| 1866 | UpdateExecSummary(query_handle); |
| 1867 | discard_result(StoreExecutionStats(query_handle)); |
| 1868 | } |
| 1869 | |
| 1870 | if (query_handle->schedule() != nullptr) { |
| 1871 | const RepeatedPtrField<BackendExecParamsPB>& backend_exec_params = |
| 1872 | query_handle->schedule()->backend_exec_params(); |
| 1873 | if (!backend_exec_params.empty()) { |
| 1874 | lock_guard<mutex> l(query_locations_lock_); |
| 1875 | for (const BackendExecParamsPB& param : backend_exec_params) { |
| 1876 | // Query may have been removed already by cancellation path. In particular, if |
| 1877 | // node to fail was last sender to an exchange, the coordinator will realise and |
| 1878 | // fail the query at the same time the failure detection path does the same |
| 1879 | // thing. They will harmlessly race to remove the query from this map. |
| 1880 | auto it = query_locations_.find(param.backend_id()); |
| 1881 | if (it != query_locations_.end()) { |
| 1882 | it->second.query_ids.erase(query_handle->query_id()); |
no test coverage detected