Method CombineChunks

cpp/src/arrow/table.cc:599–635

}

Result<std::shared_ptr<Table>> Table::CombineChunks(MemoryPool* pool) const {
  const int ncolumns = num_columns();
  std::vector<std::shared_ptr<ChunkedArray>> compacted_columns(ncolumns);
  for (int i = 0; i < ncolumns; ++i) {
    const auto& col = column(i);
    if (col->num_chunks() <= 1) {
      compacted_columns[i] = col;
      continue;
    }

    if (is_binary_like(col->type()->id())) {
      // ARROW-5744 Allow binary columns to be combined into multiple chunks to avoid
      // buffer overflow
      ArrayVector chunks;
      int chunk_i = 0;
      while (chunk_i < col->num_chunks()) {
        ArrayVector safe_chunks;
        int64_t data_length = 0;
        for (; chunk_i < col->num_chunks(); ++chunk_i) {
          const auto& chunk = col->chunk(chunk_i);
          data_length += checked_cast<const BinaryArray&>(*chunk).total_values_length();
          if (data_length >= kBinaryMemoryLimit) {
            break;
          }
          safe_chunks.push_back(chunk);
        }
        chunks.emplace_back();
        ARROW_ASSIGN_OR_RAISE(chunks.back(), Concatenate(safe_chunks, pool));
      }
      compacted_columns[i] = std::make_shared<ChunkedArray>(std::move(chunks));
    } else {
      ARROW_ASSIGN_OR_RAISE(auto compacted, Concatenate(col->chunks(), pool));
      compacted_columns[i] = std::make_shared<ChunkedArray>(compacted);
    }
  }
  return Table::Make(schema(), std::move(compacted_columns), num_rows_);
}

Result<std::shared_ptr<RecordBatch>> Table::CombineChunksToBatch(MemoryPool* pool) const {
  ARROW_ASSIGN_OR_RAISE(std::shared_ptr<Table> combined, CombineChunks(pool));