Method FromProtobuf

be/src/runtime/row-batch.cc:176–207

}

Status RowBatch::FromProtobuf(const RowDescriptor* row_desc,
    const RowBatchHeaderPB& header, const kudu::Slice& input_tuple_offsets,
    const kudu::Slice& input_tuple_data, MemTracker* mem_tracker,
    BufferPool::ClientHandle* client, unique_ptr<RowBatch>* row_batch_ptr) {
  unique_ptr<RowBatch> row_batch(new RowBatch(row_desc, header, mem_tracker));

  DCHECK(client != nullptr);
  row_batch->tuple_ptrs_info_.reset(new BufferInfo());
  row_batch->tuple_ptrs_info_->client = client;
  BufferPool::BufferHandle* tuple_ptrs_buffer = &(row_batch->tuple_ptrs_info_->buffer);
  RETURN_IF_ERROR(
      row_batch->AllocateBuffer(client, row_batch->tuple_ptrs_size_, tuple_ptrs_buffer));
  row_batch->tuple_ptrs_ = reinterpret_cast<Tuple**>(tuple_ptrs_buffer->data());

  const int64_t uncompressed_size = header.uncompressed_size();
  BufferPool::BufferHandle tuple_data_buffer;
  RETURN_IF_ERROR(
      row_batch->AllocateBuffer(client, uncompressed_size, &tuple_data_buffer));
  uint8_t* tuple_data = tuple_data_buffer.data();
  row_batch->AddBuffer(client, move(tuple_data_buffer), FlushMode::NO_FLUSH_RESOURCES);

  row_batch->num_rows_ = header.num_rows();
  row_batch->capacity_ = header.num_rows();
  const CompressionTypePB& compression_type = header.compression_type();
  DCHECK(compression_type == CompressionTypePB::NONE ||
      compression_type == CompressionTypePB::LZ4)
      << "Unexpected compression type: " << compression_type;
  row_batch->Deserialize(input_tuple_offsets, input_tuple_data, uncompressed_size,
      compression_type == CompressionTypePB::LZ4, tuple_data);
  *row_batch_ptr = std::move(row_batch);
  return Status::OK();
}

RowBatch::~RowBatch() {
  tuple_data_pool_.FreeAll();