Method DecodeBuffer

be/src/util/char-codec.cc:39–98

const int CharCodec::MAX_SYMBOL = 4;

Status CharCodec::DecodeBuffer(uint8_t** buffer, int64_t* bytes_read, MemPool* pool,
    bool eosr, bool decompress, ScannerContext* context) {
  std::string result_prefix;
  std::string result_core;
  std::string result_suffix;

  // We're about to create a new decoding buffer (if we can't reuse). Attach the
  // memory from previous decoding rounds to 'pool'. In case of streaming decompression
  // this is already done in DecompressStreamToBuffer().
  if (!decompress && !reuse_buffer_) {
    if (pool != nullptr) {
      pool->AcquireData(memory_pool_, false);
    } else {
      memory_pool_->FreeAll();
    }
    out_buffer_ = nullptr;
  }

  uint8_t* buf_start = *buffer;
  uint8_t* buf_end = buf_start + *bytes_read;

  // Allocate memory twice the size of the input buffer to handle the worst case
  ScopedMemTracker scoped_mem_tracker(memory_pool_->mem_tracker());
  RETURN_IF_ERROR(scoped_mem_tracker.TryConsume((*bytes_read) * 2));

  RETURN_IF_ERROR(HandlePrefix(&buf_start, buf_end, &result_prefix));
  RETURN_IF_ERROR(HandleCore(&buf_start, buf_end, &result_core));
  RETURN_IF_ERROR(HandleSuffix(&buf_start, buf_end, &result_suffix));

  if (eosr && !partial_symbol_.empty()) {
    return Status(TErrorCode::CHARSET_CONVERSION_ERROR,
        "End of stream reached with partial symbol.");
  }

  // In case of decompression, decompressed data can be freed up after decoding
  if (decompress) {
    memory_pool_->FreeAll();
  } else if (eosr) {
    context->ReleaseCompletedResources(false);
  }

  // Concat the results onto the output buffer
  *bytes_read = result_prefix.size() + result_core.size() + result_suffix.size();
  if (out_buffer_ == nullptr || buffer_length_ < *bytes_read) {
    buffer_length_ =  *bytes_read;
    out_buffer_ = memory_pool_->TryAllocate(buffer_length_);
    if (UNLIKELY(out_buffer_ == nullptr)) {
      string details = Substitute(
          "HdfsTextScanner::DecodeBuffer() failed to allocate $1 bytes.", *bytes_read);
      return memory_pool_->mem_tracker()->MemLimitExceeded(nullptr, details, *bytes_read);
    }
  }
  *buffer = out_buffer_;
  memcpy(*buffer, result_prefix.data(), result_prefix.size());
  memcpy(*buffer + result_prefix.size(), result_core.data(), result_core.size());
  memcpy(*buffer + result_prefix.size() + result_core.size(),
      result_suffix.data(), result_suffix.size());