Method GetBytesInternal

be/src/exec/scanner-context.cc:263–324

}

Status ScannerContext::Stream::GetBytesInternal(int64_t requested_len,
    uint8_t** out_buffer, bool peek, int64_t* out_len) {
  DCHECK_GT(requested_len, boundary_buffer_bytes_left_);
  DCHECK(output_buffer_bytes_left_ != &io_buffer_bytes_left_
      || requested_len > io_buffer_bytes_left_) << "All bytes in output buffer "
      << requested_len << " " << io_buffer_bytes_left_;
  *out_buffer = nullptr;

  if (boundary_buffer_bytes_left_ == 0) boundary_buffer_->Clear();
  DCHECK(ValidateBufferPointers());

  // First this loop ensures, by reading I/O buffers one-by-one, that we've got all of
  // the requested bytes in 'boundary_buffer_', 'io_buffer_', or split between the two.
  // We may not be able to get all of the bytes if we hit eof.
  while (boundary_buffer_bytes_left_ + io_buffer_bytes_left_ < requested_len) {
    if (io_buffer_bytes_left_ > 0) {
      // Copy the remainder of 'io_buffer_' to 'boundary_buffer_' before getting the next
      // 'io_buffer_'. Preallocate 'boundary_buffer_' to avoid unnecessary resizes for
      // large reads.
      RETURN_IF_ERROR(boundary_buffer_->GrowBuffer(requested_len));
      RETURN_IF_ERROR(CopyIoToBoundary(io_buffer_bytes_left_));
    }
    int64_t remaining_requested_len = requested_len - boundary_buffer_bytes_left_;
    RETURN_IF_ERROR(GetNextBuffer(remaining_requested_len));
    if (UNLIKELY(parent_->cancelled())) return CONTEXT_CANCELLED;
    // No more bytes (i.e. EOF).
    if (io_buffer_bytes_left_ == 0) break;
  }

  // We have read the full 'requested_len' bytes or hit eof.
  // We can assemble the contiguous bytes in two ways:
  // 1. if the the read range falls entirely with an I/O buffer, we return a pointer into
  //    that I/O buffer.
  // 2. if the read straddles I/O buffers, we append the data to 'boundary_buffer_'.
  //    'boundary_buffer_' may already contain some of the data that we need if we did a
  //    "peek" earlier.
  int64_t requested_bytes_left = requested_len - boundary_buffer_bytes_left_;
  DCHECK_GE(requested_bytes_left, 0);
  int64_t num_bytes_left_to_copy = min(io_buffer_bytes_left_, requested_bytes_left);
  *out_len = boundary_buffer_bytes_left_ + num_bytes_left_to_copy;
  DCHECK_LE(*out_len, requested_len);
  if (boundary_buffer_bytes_left_ == 0) {
    // Case 1: return a pointer into the I/O buffer.
    output_buffer_pos_ = &io_buffer_pos_;
    output_buffer_bytes_left_ = &io_buffer_bytes_left_;
  } else {
    // Case 2: return a pointer into the boundary buffer, after copying any required
    // data from the I/O buffer.
    DCHECK_EQ(output_buffer_pos_, &boundary_buffer_pos_);
    DCHECK_EQ(output_buffer_bytes_left_, &boundary_buffer_bytes_left_);
    if (io_buffer_bytes_left_ > 0) {
      RETURN_IF_ERROR(CopyIoToBoundary(num_bytes_left_to_copy));
    }
  }
  *out_buffer = *output_buffer_pos_;
  if (!peek) {
    total_bytes_returned_ += *out_len;
    AdvanceBufferPos(*out_len, output_buffer_pos_, output_buffer_bytes_left_);