Method DecompressStreamToBuffer

be/src/exec/hdfs-scanner.cc:820–862

}

Status HdfsScanner::DecompressStreamToBuffer(uint8_t** buffer, int64_t* bytes_read,
    MemPool* pool, bool* eosr) {
  // We're about to create a new decompression buffer (if we can't reuse). Attach the
  // memory from previous decompression rounds to 'pool'.
  if (!decompressor_->reuse_output_buffer()) {
    if (pool != nullptr) {
      pool->AcquireData(data_buffer_pool_.get(), false);
    } else {
      data_buffer_pool_->FreeAll();
    }
  }

  uint8_t* decompressed_buffer = nullptr;
  int64_t decompressed_len = 0;
  // Set bytes_to_read = -1 because we don't know how much data decompressor need.
  // Just read the first available buffer within the scan range.
  Status status = DecompressStream(-1, &decompressed_buffer, &decompressed_len,
      eosr);
  if (status.code() == TErrorCode::COMPRESSED_FILE_DECOMPRESSOR_NO_PROGRESS) {
    // It's possible (but very unlikely) that ProcessBlockStreaming() wasn't able to
    // make progress if the compressed buffer returned by GetBytes() is too small.
    // (Note that this did not even occur in simple experiments where the input buffer
    // is always 1 byte, but we need to handle this case to be defensive.) In this
    // case, try again with a reasonably large fixed size buffer. If we still did not
    // make progress, then return an error.
    LOG(INFO) << status.GetDetail();
    // Number of bytes to read when the previous attempt to streaming decompress did not
    // make progress.
    constexpr int64_t COMPRESSED_DATA_FIXED_READ_SIZE = 1 * 1024 * 1024;
    status = DecompressStream(COMPRESSED_DATA_FIXED_READ_SIZE, &decompressed_buffer,
        &decompressed_len, eosr);
  }
  RETURN_IF_ERROR(status);
  *buffer = decompressed_buffer;
  *bytes_read = decompressed_len;

  if (*eosr) {
    DCHECK(stream_->eosr());
    context_->ReleaseCompletedResources(true);
  }

  return Status::OK();
}

Status HdfsScanner::DecompressStream(int64_t bytes_to_read,
    uint8_t** decompressed_buffer, int64_t* decompressed_len, bool *eosr) {