Method AddBatchWork

be/src/runtime/krpc-data-stream-recvr.cc:428–477

}

Status KrpcDataStreamRecvr::SenderQueue::AddBatchWork(int64_t batch_size,
    const RowBatchHeaderPB& header, const kudu::Slice& tuple_offsets,
    const kudu::Slice& tuple_data, unique_lock<SpinLock>* lock,
    RpcContext* rpc_context) {
  DCHECK(lock != nullptr);
  DCHECK(lock->owns_lock());
  DCHECK(!is_cancelled_);

  // Reserve queue space before dropping the lock below.
  num_buffered_bytes_ += batch_size;
  // Bump 'num_pending_enqueue_' to avoid race with Close() when lock is dropped below.
  DCHECK_GE(num_pending_enqueue_, 0);
  ++num_pending_enqueue_;

  // Deserialization may take some time due to compression and memory allocation.
  // Drop the lock so we can deserialize multiple batches in parallel.
  lock->unlock();
  TRACE_TO(rpc_context->trace(), "Deserializing batch");
  unique_ptr<RowBatch> batch;
  Status status;
  {
    SCOPED_TIMER(recvr_->deserialize_row_batch_timer_);
    status = DebugAction(recvr_->runtime_state_.query_options(), "RECVR_ADD_BATCH");
    if (LIKELY(status.ok())) {
      // At this point, a row batch will be inserted into batch_queue_.
      // Close() will handle deleting any unconsumed batches from batch_queue_.
      // Close() cannot proceed until there are no pending insertion to batch_queue_.
      status = RowBatch::FromProtobuf(recvr_->row_desc(), header, tuple_offsets,
          tuple_data, recvr_->parent_tracker(), recvr_->buffer_pool_client(), &batch);
    }
  }
  lock->lock();

  DCHECK_GT(num_pending_enqueue_, 0);
  --num_pending_enqueue_;
  if (UNLIKELY(!status.ok())) {
    num_buffered_bytes_ -= batch_size;
    VLOG_QUERY << "Failed to deserialize batch for "
               << PrintId(recvr_->fragment_instance_id());
    TRACE_TO(rpc_context->trace(), "Failed to deserialize batch: $0", status.GetDetail());
    MarkErrorStatus(status, *lock);
    return status;
  }
  VLOG_ROW << "added #rows=" << batch->num_rows() << " batch_size=" << batch_size;
  TRACE_TO(rpc_context->trace(), "Enqueuing deserialized batch");
  COUNTER_ADD(recvr_->total_enqueued_batches_counter_, 1);
  batch_queue_.emplace_back(batch_size, move(batch));
  data_arrival_cv_.notify_one();
  return Status::OK();
}

void KrpcDataStreamRecvr::SenderQueue::AddBatch(const TransmitDataRequestPB* request,
    TransmitDataResponsePB* response, RpcContext* rpc_context) {