| 1530 | } |
| 1531 | |
| 1532 | Status ClientRequestState::FetchRowsInternal(const int32_t max_rows, |
| 1533 | QueryResultSet* fetched_rows, int64_t block_on_wait_time_us) { |
| 1534 | // Wait() guarantees that we've transitioned at least to FINISHED state (and any |
| 1535 | // state beyond that should have a non-OK query_status_ set). |
| 1536 | DCHECK(exec_state() == ExecState::FINISHED); |
| 1537 | |
| 1538 | if (eos_.Load()) return Status::OK(); |
| 1539 | |
| 1540 | if (request_result_set_ != NULL) { |
| 1541 | int num_rows = 0; |
| 1542 | const vector<TResultRow>& all_rows = (*(request_result_set_.get())); |
| 1543 | // max_rows <= 0 means no limit |
| 1544 | while ((num_rows < max_rows || max_rows <= 0) |
| 1545 | && num_rows_fetched_ < all_rows.size()) { |
| 1546 | RETURN_IF_ERROR(fetched_rows->AddOneRow(all_rows[num_rows_fetched_])); |
| 1547 | ++num_rows_fetched_; |
| 1548 | ++num_rows; |
| 1549 | } |
| 1550 | eos_.Store(num_rows_fetched_ == all_rows.size()); |
| 1551 | return Status::OK(); |
| 1552 | } |
| 1553 | |
| 1554 | Coordinator* coordinator = GetCoordinator(); |
| 1555 | if (coordinator == nullptr) { |
| 1556 | return Status("Client tried to fetch rows on a query that produces no results."); |
| 1557 | } |
| 1558 | |
| 1559 | int32_t num_rows_fetched_from_cache = 0; |
| 1560 | if (result_cache_max_size_ > 0 && result_cache_ != NULL) { |
| 1561 | // Satisfy the fetch from the result cache if possible. |
| 1562 | int cache_fetch_size = (max_rows <= 0) ? result_cache_->size() : max_rows; |
| 1563 | num_rows_fetched_from_cache = |
| 1564 | fetched_rows->AddRows(result_cache_.get(), num_rows_fetched_, cache_fetch_size); |
| 1565 | num_rows_fetched_ += num_rows_fetched_from_cache; |
| 1566 | COUNTER_ADD(num_rows_fetched_from_cache_counter_, num_rows_fetched_from_cache); |
| 1567 | if (num_rows_fetched_from_cache >= max_rows) return Status::OK(); |
| 1568 | } |
| 1569 | |
| 1570 | // Maximum number of rows to be fetched from the coord. |
| 1571 | int32_t max_coord_rows = max_rows; |
| 1572 | if (max_rows > 0) { |
| 1573 | DCHECK_LE(num_rows_fetched_from_cache, max_rows); |
| 1574 | max_coord_rows = max_rows - num_rows_fetched_from_cache; |
| 1575 | } |
| 1576 | { |
| 1577 | SCOPED_TIMER(row_materialization_timer_); |
| 1578 | size_t before = fetched_rows->size(); |
| 1579 | bool eos = false; |
| 1580 | |
| 1581 | // Temporarily release lock so calls to Cancel() are not blocked. fetch_rows_lock_ |
| 1582 | // (already held) ensures that we do not call coord_->GetNext() multiple times |
| 1583 | // concurrently. |
| 1584 | // TODO: Simplify this. |
| 1585 | lock_.unlock(); |
| 1586 | Status status = |
| 1587 | coordinator->GetNext(fetched_rows, max_coord_rows, &eos, block_on_wait_time_us); |
| 1588 | lock_.lock(); |
| 1589 |
nothing calls this directly
no test coverage detected