Method Append

cpp/src/arrow/compute/kernels/vector_hash.cc:455–488

  Status Reset() override { return indices_kernel_->Reset(); }

  Status Append(const ArraySpan& arr) override {
    auto arr_dict = arr.dictionary().ToArray();
    if (!first_dictionary_) {
      first_dictionary_ = arr_dict;
    } else if (!first_dictionary_->Equals(*arr_dict)) {
      // NOTE: This approach computes a new dictionary unification per chunk.
      // This is in effect O(n*k) where n is the total chunked array length and
      // k is the number of chunks (therefore O(n**2) if chunks have a fixed size).
      //
      // A better approach may be to run the kernel over each individual chunk,
      // and then hash-aggregate all results (for example sum-group-by for
      // the "value_counts" kernel).
      if (dictionary_unifier_ == nullptr) {
        ARROW_ASSIGN_OR_RAISE(dictionary_unifier_,
                              DictionaryUnifier::Make(first_dictionary_->type()));
        RETURN_NOT_OK(dictionary_unifier_->Unify(*first_dictionary_));
      }
      auto out_dict_type = first_dictionary_->type();
      std::shared_ptr<Buffer> transpose_map;

      RETURN_NOT_OK(dictionary_unifier_->Unify(*arr_dict, &transpose_map));

      auto transpose = reinterpret_cast<const int32_t*>(transpose_map->data());
      auto in_array = arr.ToArray();
      const auto& in_dict_array =
          arrow::internal::checked_cast<const DictionaryArray&>(*in_array);
      ARROW_ASSIGN_OR_RAISE(
          auto tmp, in_dict_array.Transpose(arr.type->GetSharedPtr(),
                                            in_dict_array.dictionary(), transpose));
      return indices_kernel_->Append(*tmp->data());
    }

    return indices_kernel_->Append(arr);
  }

  Status Flush(ExecResult* out) override { return indices_kernel_->Flush(out); }