Method Deserialize

be/src/runtime/row-batch.cc:115–174

}

void RowBatch::Deserialize(const kudu::Slice& input_tuple_offsets,
    const kudu::Slice& input_tuple_data, int64_t uncompressed_size,
    bool is_compressed, uint8_t* tuple_data) {
  DCHECK(tuple_ptrs_ != nullptr);
  DCHECK(tuple_data != nullptr);
  if (is_compressed) {
    // Decompress tuple data into data pool
    const uint8_t* compressed_data = input_tuple_data.data();
    size_t compressed_size = input_tuple_data.size();

    Lz4Decompressor decompressor(nullptr, false);
    Status status = decompressor.Init();
    DCHECK(status.ok()) << status.GetDetail();
    auto compressor_cleanup =
        MakeScopeExitTrigger([&decompressor]() { decompressor.Close(); });

    status = decompressor.ProcessBlock(
        true, compressed_size, compressed_data, &uncompressed_size, &tuple_data);
    DCHECK_NE(uncompressed_size, -1) << "RowBatch decompression failed";
    DCHECK(status.ok()) << "RowBatch decompression failed.";
  } else {
    // Tuple data uncompressed, copy directly into data pool
    DCHECK_EQ(uncompressed_size, input_tuple_data.size());
    memcpy(tuple_data, input_tuple_data.data(), input_tuple_data.size());
  }

  // Convert input_batch.tuple_offsets into pointers
  const int32_t* tuple_offsets =
      reinterpret_cast<const int32_t*>(input_tuple_offsets.data());
  DCHECK_EQ(input_tuple_offsets.size() % sizeof(int32_t), 0);
  int num_tuples = input_tuple_offsets.size() / sizeof(int32_t);
  for (int tuple_idx = 0; tuple_idx < num_tuples; ++tuple_idx) {
    int32_t offset = tuple_offsets[tuple_idx];
    if (offset == -1) {
      tuple_ptrs_[tuple_idx] = nullptr;
    } else {
      tuple_ptrs_[tuple_idx] = reinterpret_cast<Tuple*>(tuple_data + offset);
    }
  }

  // Check whether we have slots that require offset-to-pointer conversion.
  if (!row_desc_->HasVarlenSlots()) return;

  // For every unique tuple, convert string offsets contained in tuple data into
  // pointers. Tuples were serialized in the order we are deserializing them in,
  // so the first occurrence of a tuple will always have a higher offset than any
  // tuple we already converted.
  Tuple* last_converted = nullptr;
  for (int i = 0; i < num_rows_; ++i) {
    for (int j = 0; j < num_tuples_per_row_; ++j) {
      const TupleDescriptor* desc = row_desc_->tuple_descriptors()[j];
      if (!desc->HasVarlenSlots()) continue;
      Tuple* tuple = GetRow(i)->GetTuple(j);
      // Handle NULL or already converted tuples with one check.
      if (tuple <= last_converted) continue;
      last_converted = tuple;
      tuple->ConvertOffsetsToPointers(*desc, tuple_data);
    }