Method AddBatch

be/src/runtime/spillable-row-batch-queue.cc:69–111

}

Status SpillableRowBatchQueue::AddBatch(RowBatch* batch) {
  DCHECK(!IsFull()) << "Cannot AddBatch on a full SpillableRowBatchQueue";
  DCHECK(!closed_) << "Cannot AddBatch on a closed SpillableRowBatchQueue";
  Status status;
  FOREACH_ROW(batch, 0, batch_itr) {
    // AddRow should only return false if there was not enough unused reservation to
    // allocate a page for the given row. If a row cannot be added to the batch_queue_
    // then start spilling to disk by unpining the stream. Once the stream is unpinned,
    // adding the row to the stream should succeed unless the unpinned pages needed to
    // be spilled and either (1) there was an error (e.g. IO error) when writing to disk,
    // (2) there is no more scratch space left to write to disk, or (3) spilling to disk
    // is disabled.
    if (UNLIKELY(!batch_queue_->AddRow(batch_itr.Get(), &status))) {
      RETURN_IF_ERROR(status);
      // StartSpilling checks if spilling is disabled and returns an error if it is not.
      RETURN_IF_ERROR(state_->StartSpilling(mem_tracker_));

      // The pin should be stream at this point.
      DCHECK(batch_queue_->is_pinned());
      DCHECK_EQ(batch_queue_->bytes_unpinned(), 0);

      // Unpin the stream and then add the row.
      RETURN_IF_ERROR(
          batch_queue_->UnpinStream(BufferedTupleStream::UNPIN_ALL_EXCEPT_CURRENT));

      // Append "Spilled" to the "ExecOption" info string in the runtime profile.
      profile_->AppendExecOption("Spilled");

      if (!batch_queue_->AddRow(batch_itr.Get(), &status)) {
        RETURN_IF_ERROR(status);
        // If the row could not be added after the stream was unpinned, an error should
        // have been set.
        DCHECK(false) << Substitute("Row with a size of $0 should be added successfully "
                                    "in unpinned mode unless an error occurred. "
                                    "batch_queue_: $1",
            PrettyPrinter::PrintBytes(
                batch_queue_->ComputeRowSize(batch_itr.Get())),
            batch_queue_->DebugString());
      }
    }
  }
  return Status::OK();
}

Status SpillableRowBatchQueue::GetBatch(RowBatch* batch) {
  DCHECK(!IsEmpty()) << "Cannot GetBatch on an empty SpillableRowBatchQueue";