| 510 | } |
| 511 | |
| 512 | Status BufferedTupleStream::NextReadPage(ReadIterator* read_iter) { |
| 513 | DCHECK(read_iter->is_valid()); |
| 514 | DCHECK(!closed_); |
| 515 | DCHECK(read_iter == &read_it_ || (pinned_ && !read_iter->attach_on_read_)) |
| 516 | << "External read iterators only support pinned streams with no attach on read " |
| 517 | << read_iter->DebugString(pages_); |
| 518 | CHECK_CONSISTENCY_FAST(*read_iter); |
| 519 | |
| 520 | if (read_iter->read_page_ == pages_.end()) { |
| 521 | // No rows read yet - start reading at first page. If the stream is unpinned, we can |
| 522 | // use the reservation saved in PrepareForReadWrite() to pin the first page. |
| 523 | read_iter->SetReadPage(pages_.begin()); |
| 524 | if (NeedReadReservation(pinned_, num_pages_, true, false) |
| 525 | && !NeedReadReservation(pinned_, num_pages_, true, true)) { |
| 526 | buffer_pool_client_->RestoreReservation(&read_page_reservation_, default_page_len_); |
| 527 | } |
| 528 | } else if (read_iter->attach_on_read_) { |
| 529 | DCHECK(read_iter->read_page_ == pages_.begin()) |
| 530 | << read_iter->read_page_->DebugString() << " " << DebugString(); |
| 531 | DCHECK_NE(&*read_iter->read_page_, write_page_); |
| 532 | DCHECK(read_iter->read_page_->attached_to_output_batch); |
| 533 | pages_.pop_front(); |
| 534 | --num_pages_; |
| 535 | read_iter->SetReadPage(pages_.begin()); |
| 536 | } else { |
| 537 | // Unpin pages after reading them if needed. |
| 538 | Page* prev_read_page = &*read_iter->read_page_; |
| 539 | read_iter->AdvanceReadPage(pages_); |
| 540 | UnpinPageIfNeeded(prev_read_page, pinned_); |
| 541 | } |
| 542 | |
| 543 | if (read_iter->read_page_ == pages_.end()) { |
| 544 | CHECK_CONSISTENCY_FULL(*read_iter); |
| 545 | return Status::OK(); |
| 546 | } |
| 547 | |
| 548 | int64_t read_page_len = read_iter->read_page_->len(); |
| 549 | if (!pinned_ && read_page_len > default_page_len_) { |
| 550 | // If we are iterating over an unpinned stream and encounter a page that is larger |
| 551 | // than the default page length, then unpinning the previous page may not have |
| 552 | // freed up enough reservation to pin the next one. Try to restore some extra saved |
| 553 | // reservation for reading a large page. |
| 554 | int64_t needed_reservation = read_page_len - default_page_len_; |
| 555 | if (large_read_page_reservation_.GetReservation() >= needed_reservation) { |
| 556 | buffer_pool_client_->RestoreReservation(&large_read_page_reservation_, |
| 557 | needed_reservation); |
| 558 | } |
| 559 | if (buffer_pool_client_->GetUnusedReservation() < read_page_len) { |
| 560 | // Still failed to get enough unused reservation. The client is responsible for |
| 561 | // ensuring the reservation is available, so this indicates a bug. |
| 562 | return Status(TErrorCode::INTERNAL_ERROR, Substitute("Internal error: couldn't pin " |
| 563 | "large page of $0 bytes, client only had $1 bytes of unused reservation:\n$2", |
| 564 | read_page_len, buffer_pool_client_->GetUnusedReservation(), |
| 565 | buffer_pool_client_->DebugString())); |
| 566 | } |
| 567 | } |
| 568 | // Ensure the next page is pinned for reading. By this point we should have enough |
| 569 | // reservation to pin the page. If the stream is pinned, the page is already pinned. |
nothing calls this directly
no test coverage detected