Method ReadCompressedBlock

be/src/exec/sequence/hdfs-sequence-scanner.cc:523–570

}

Status HdfsSequenceScanner::ReadCompressedBlock() {
  int64_t num_buffered_records;
  RETURN_IF_FALSE(stream_->ReadVLong(
      &num_buffered_records, &parse_status_));
  if (num_buffered_records < 0) {
    if (state_->LogHasSpace()) {
      stringstream ss;
      ss << stream_->filename()
         << " Bad compressed block record count: " << num_buffered_records;
      state_->LogError(ErrorMsg(TErrorCode::GENERAL, ss.str()));
    }
    return Status("bad record count");
  }

  // Skip the compressed key length and key buffers, we don't need them.
  RETURN_IF_FALSE(stream_->SkipText(&parse_status_));
  RETURN_IF_FALSE(stream_->SkipText(&parse_status_));

  // Skip the compressed value length buffer. We don't need these either since the
  // records are in Text format with length included.
  RETURN_IF_FALSE(stream_->SkipText(&parse_status_));

  // Read the compressed value buffer from the unbuffered stream.
  int64_t block_size = 0;
  RETURN_IF_FALSE(stream_->ReadVLong(&block_size, &parse_status_));
  // Check for a reasonable size
  if (block_size > MAX_BLOCK_SIZE || block_size < 0) {
    stringstream ss;
    ss << stream_->filename() << " Compressed block size is: " << block_size;
    return Status(ss.str());
  }

  uint8_t* compressed_data = nullptr;
  RETURN_IF_FALSE(stream_->ReadBytes(block_size, &compressed_data, &parse_status_));

  {
    int64_t len;
    SCOPED_TIMER(decompress_timer_);
    RETURN_IF_ERROR(decompressor_->ProcessBlock(false, block_size, compressed_data,
                                                &len, &unparsed_data_buffer_));
    VLOG_FILE << "Decompressed " << block_size << " to " << len;
    next_record_in_compressed_block_ = unparsed_data_buffer_;
    next_record_in_compressed_block_len_ = len;
    data_buffer_end_ = unparsed_data_buffer_ + len;
  }
  num_buffered_records_in_compressed_block_ = num_buffered_records;
  return Status::OK();
}