Method ProcessBlock

be/src/util/decompress.cc:149–245

}

Status GzipDecompressor::ProcessBlock(bool output_preallocated, int64_t input_length,
    const uint8_t* input, int64_t* output_length, uint8_t** output) {
  int64_t output_length_local = *output_length;
  *output_length = 0;
  if (UNLIKELY(output_preallocated && output_length_local == 0)) {
    // The zlib library does not allow *output to be nullptr, even when output_length is 0
    // (inflate() will return Z_STREAM_ERROR). We don't consider this an error, so bail
    // early if no output is expected. Note that we don't signal an error if the input
    // actually contains compressed data.
    return Status::OK();
  }

  bool use_temp = false;
  if (!output_preallocated) {
    if (!reuse_buffer_ || out_buffer_ == nullptr) {
      // guess that we will need 2x the input length.
      buffer_length_ = input_length * 2;
      out_buffer_ = temp_memory_pool_->TryAllocate(buffer_length_);
      if (UNLIKELY(out_buffer_ == nullptr)) {
        string details = Substitute(DECOMPRESSOR_MEM_LIMIT_EXCEEDED, "Gzip",
            buffer_length_);
        return temp_memory_pool_->mem_tracker()->MemLimitExceeded(
            nullptr, details, buffer_length_);
      }
    }
    use_temp = true;
    *output = out_buffer_;
    output_length_local = buffer_length_;
  }

  // Reset the stream for this block
  int ret = inflateReset(&stream_);
  if (ret != Z_OK) {
    return Status(TErrorCode::COMPRESSED_FILE_DECOMPRESSOR_ERROR, "Gzip",
        "inflateReset()", ret);
  }

  // We only support the non-streaming use case where we present it the entire
  // compressed input and a buffer big enough to contain the entire decompressed
  // output.  In the case where we don't know the output, we just make a bigger
  // buffer and try the non-streaming mode from the beginning again.
  // TODO: IMPALA-3073 Verify if compressed block could be multistream. If yes, we need
  // to support it and shouldn't stop decompressing while ret == Z_STREAM_END.
  while (ret != Z_STREAM_END) {
    stream_.next_in = const_cast<Bytef*>(reinterpret_cast<const Bytef*>(input));
    stream_.avail_in = input_length;
    stream_.next_out = reinterpret_cast<Bytef*>(*output);
    stream_.avail_out = output_length_local;

    if (use_temp) {
      // We don't know the output size, so this might fail.
      ret = inflate(&stream_, Z_PARTIAL_FLUSH);
    } else {
      // We know the output size.  In this case, we can use Z_FINISH
      // which is more efficient.
      ret = inflate(&stream_, Z_FINISH);
    }
    if (ret == Z_STREAM_END || ret != Z_OK) break;