| 407 | } |
| 408 | |
| 409 | Status BufferedTupleStream::AdvanceWritePage( |
| 410 | int64_t row_size, bool* got_reservation) noexcept { |
| 411 | DCHECK(has_write_iterator()); |
| 412 | CHECK_CONSISTENCY_FAST(read_it_); |
| 413 | |
| 414 | int64_t page_len; |
| 415 | RETURN_IF_ERROR(CalcPageLenForRow(row_size, &page_len)); |
| 416 | |
| 417 | // Reservation may have been saved for the next write page, e.g. by PrepareForWrite() |
| 418 | // if the stream is empty. |
| 419 | int64_t write_reservation_to_restore = 0, read_reservation_to_restore = 0; |
| 420 | if (NeedWriteReservation( |
| 421 | pinned_, num_pages_, true, write_page_ != nullptr, has_read_write_page()) |
| 422 | && !NeedWriteReservation(pinned_, num_pages_ + 1, true, true, false)) { |
| 423 | write_reservation_to_restore = default_page_len_; |
| 424 | } |
| 425 | // If the stream is pinned, we need to keep the previous write page pinned for reading. |
| 426 | // Check if we saved reservation for this case. |
| 427 | if (NeedReadReservation(pinned_, num_pages_, has_read_iterator(), |
| 428 | read_it_.read_page_ != pages_.end(), true, write_page_ != nullptr) |
| 429 | && !NeedReadReservation(pinned_, num_pages_ + 1, has_read_iterator(), |
| 430 | read_it_.read_page_ != pages_.end(), true, true)) { |
| 431 | read_reservation_to_restore = default_page_len_; |
| 432 | } |
| 433 | |
| 434 | // We may reclaim reservation by unpinning a page that was pinned for writing. |
| 435 | int64_t write_page_reservation_to_reclaim = |
| 436 | (write_page_ != nullptr && !pinned_ && !has_read_write_page()) ? |
| 437 | write_page_->len() : 0; |
| 438 | // Check to see if we can get the reservation before changing the state of the stream. |
| 439 | if (!buffer_pool_client_->IncreaseReservationToFit(page_len |
| 440 | - write_reservation_to_restore - read_reservation_to_restore |
| 441 | - write_page_reservation_to_reclaim)) { |
| 442 | DCHECK(pinned_ || page_len > default_page_len_) |
| 443 | << "If the stream is unpinned, this should only fail for large pages"; |
| 444 | CHECK_CONSISTENCY_FAST(read_it_); |
| 445 | *got_reservation = false; |
| 446 | return Status::OK(); |
| 447 | } |
| 448 | if (write_reservation_to_restore > 0) { |
| 449 | buffer_pool_client_->RestoreReservation( |
| 450 | &write_page_reservation_, write_reservation_to_restore); |
| 451 | } |
| 452 | if (read_reservation_to_restore > 0) { |
| 453 | buffer_pool_client_->RestoreReservation( |
| 454 | &read_page_reservation_, read_reservation_to_restore); |
| 455 | } |
| 456 | ResetWritePage(); |
| 457 | RETURN_IF_ERROR(NewWritePage(page_len)); |
| 458 | *got_reservation = true; |
| 459 | return Status::OK(); |
| 460 | } |
| 461 | |
| 462 | void BufferedTupleStream::ResetWritePage() { |
| 463 | if (write_page_ == nullptr) return; |
nothing calls this directly
no test coverage detected