MCPcopy Create free account
hub / github.com/apache/impala / FetchRowsInternal

Method FetchRowsInternal

be/src/service/client-request-state.cc:1532–1660  ·  view source on GitHub ↗

Source from the content-addressed store, hash-verified

1530}
1531
1532Status 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

Callers

nothing calls this directly

Calls 15

OKFunction · 0.85
SubstituteFunction · 0.85
AddOneRowMethod · 0.80
getMethod · 0.65
StatusClass · 0.50
LoadMethod · 0.45
sizeMethod · 0.45
StoreMethod · 0.45
AddRowsMethod · 0.45
unlockMethod · 0.45
GetNextMethod · 0.45
lockMethod · 0.45

Tested by

no test coverage detected