Method DecompressFileToBuffer

be/src/exec/hdfs-scanner.cc:777–818

}

Status HdfsScanner::DecompressFileToBuffer(uint8_t** buffer, int64_t* bytes_read) {
  // For other compressed file: attempt to read and decompress the entire file, point
  // to the decompressed buffer, and then continue normal processing.
  DCHECK(decompression_type_ != THdfsCompression::SNAPPY);
  const HdfsFileDesc* desc = scan_node_->GetFileDesc(
      context_->partition_descriptor()->id(), stream_->filename());
  int64_t file_size = desc->file_length;
  DCHECK_GT(file_size, 0);

  Status status;
  if (!stream_->GetBytes(file_size, buffer, bytes_read, &status)) {
    DCHECK(!status.ok());
    return status;
  }

  // If didn't read anything, return.
  if (*bytes_read == 0) return Status::OK();

  // Need to read the entire file.
  if (file_size > *bytes_read) {
    return Status(Substitute("Expected to read a compressed text file of size $0 bytes. "
        "But only read $1 bytes. This may indicate data file corruption. (file: $3).",
        file_size, *bytes_read, stream_->filename()));
  }

  // Decompress and adjust the buffer and bytes_read accordingly.
  int64_t decompressed_len = 0;
  uint8_t* decompressed_buffer = nullptr;
  SCOPED_TIMER(decompress_timer_);
  // TODO: Once the writers are in, add tests with very large compressed files (4GB)
  // that could overflow.
  RETURN_IF_ERROR(decompressor_->ProcessBlock(false, *bytes_read, *buffer,
      &decompressed_len, &decompressed_buffer));

  // Inform 'stream_' that the buffer with the compressed text can be released.
  context_->ReleaseCompletedResources(true);

  VLOG_FILE << "Decompressed " << *bytes_read << " to " << decompressed_len;
  *buffer = decompressed_buffer;
  *bytes_read = decompressed_len;
  return Status::OK();
}

Status HdfsScanner::DecompressStreamToBuffer(uint8_t** buffer, int64_t* bytes_read,
    MemPool* pool, bool* eosr) {