Method FetchInternal

be/src/service/impala-hs2-server.cc:200–253

}

Status ImpalaServer::FetchInternal(const TUniqueId& query_id, SessionState* session,
    int32_t fetch_size, bool fetch_first, TFetchResultsResp* fetch_results,
    int32_t* num_results) {
  bool timed_out = false;
  int64_t block_on_wait_time_us = 0;
  QueryHandle query_handle;
  RETURN_IF_ERROR(
      WaitForResults(query_id, &query_handle, &block_on_wait_time_us, &timed_out));
  if (timed_out) {
    fetch_results->status.__set_statusCode(thrift::TStatusCode::STILL_EXECUTING_STATUS);
    fetch_results->__set_hasMoreRows(true);
    fetch_results->__isset.results = false;
    return Status::OK();
  }

  int64_t start_time_ns = MonotonicNanos();
  lock_guard<mutex> frl(*query_handle->fetch_rows_lock());
  lock_guard<mutex> l(*query_handle->lock());
  int64_t lock_wait_time_ns = MonotonicNanos() - start_time_ns;
  query_handle->AddClientFetchLockWaitTime(lock_wait_time_ns);

  // Check for cancellation or an error.
  RETURN_IF_ERROR(query_handle->query_status());

  if (query_handle->num_rows_fetched() == 0) {
    query_handle->set_fetched_rows();
  }

  if (fetch_first) RETURN_IF_ERROR(query_handle->RestartFetch());

  fetch_results->results.__set_startRowOffset(query_handle->num_rows_fetched());

  // Child queries should always return their results in row-major format, rather than
  // inheriting the parent session's setting.
  bool is_child_query = query_handle->parent_query_id() != TUniqueId();
  TProtocolVersion::type version = is_child_query ?
      TProtocolVersion::HIVE_CLI_SERVICE_PROTOCOL_V1 : session->hs2_version;

  // In the first fetch, expect 0 results to avoid reserving unnecessarily large result
  // vectors for small queries. If there are more fetches (so there are more rows than
  // num_rows_fetched), then expect subsequent fetches to be fully filled.
  int expected_result_count = query_handle->num_rows_fetched() == 0 ? 0
      : fetch_size;

  scoped_ptr<QueryResultSet> result_set(QueryResultSet::CreateHS2ResultSet(
      version, *(query_handle->result_metadata()), &(fetch_results->results),
      query_handle->query_options().stringify_map_keys, expected_result_count));
  RETURN_IF_ERROR(
      query_handle->FetchRows(fetch_size, result_set.get(), block_on_wait_time_us));
  *num_results = result_set->size();
  fetch_results->__isset.results = true;
  fetch_results->__set_hasMoreRows(!query_handle->eos());
  return Status::OK();
}

Status ImpalaServer::TExecuteStatementReqToTQueryContext(
    const TExecuteStatementReq execute_request, TQueryCtx* query_ctx) {