Method SortInternal

cpp/src/arrow/compute/kernels/vector_sort.cc:80–175

 private:
  template <typename Type>
  Status SortInternal() {
    using ArrayType = typename TypeTraits<Type>::ArrayType;
    ArraySortOptions options(order_, null_placement_);
    const auto num_chunks = static_cast<int>(physical_chunks_.size());
    if (num_chunks == 0) {
      *output_ = {indices_end_, indices_end_, indices_end_, indices_end_};
      return Status::OK();
    }
    const int64_t num_indices = static_cast<int64_t>(indices_end_ - indices_begin_);
    const auto arrays = GetArrayPointers(physical_chunks_);

    // Sort each chunk independently and merge to sorted indices.
    // This is a serial implementation.
    std::vector<NullPartitionResult> sorted(num_chunks);

    // First sort all individual chunks
    int64_t begin_offset = 0;
    int64_t end_offset = 0;
    int64_t null_count = 0;
    for (int i = 0; i < num_chunks; ++i) {
      const auto array = checked_cast<const ArrayType*>(arrays[i]);
      end_offset += array->length();
      null_count += array->null_count();
      ARROW_ASSIGN_OR_RAISE(sorted[i], array_sorter_(indices_begin_ + begin_offset,
                                                     indices_begin_ + end_offset, *array,
                                                     begin_offset, options, ctx_));
      begin_offset = end_offset;
    }
    DCHECK_EQ(end_offset, num_indices);

    // Then merge them by pairs, recursively
    if (sorted.size() > 1) {
      ChunkedIndexMapper chunked_mapper(arrays, indices_begin_, indices_end_);
      ARROW_ASSIGN_OR_RAISE(auto chunked_indices_pair,
                            chunked_mapper.LogicalToPhysical());
      auto [chunked_indices_begin, chunked_indices_end] = chunked_indices_pair;

      std::vector<ChunkedNullPartitionResult> chunk_sorted(num_chunks);
      for (int i = 0; i < num_chunks; ++i) {
        chunk_sorted[i] = sorted[i].TranslateTo(indices_begin_, chunked_indices_begin);
      }

      auto merge_nulls = [&](CompressedChunkLocation* nulls_begin,
                             CompressedChunkLocation* nulls_middle,
                             CompressedChunkLocation* nulls_end,
                             CompressedChunkLocation* temp_indices, int64_t null_count) {
        if (has_null_like_values<typename ArrayType::TypeClass>()) {
          PartitionNullsOnly<StablePartitioner>(nulls_begin, nulls_end, arrays,
                                                null_count, null_placement_);
        }
      };
      auto merge_non_nulls =
          [&](CompressedChunkLocation* range_begin, CompressedChunkLocation* range_middle,
              CompressedChunkLocation* range_end, CompressedChunkLocation* temp_indices) {
            MergeNonNulls<ArrayType>(range_begin, range_middle, range_end, arrays,
                                     temp_indices);
          };