Method Consume

cpp/src/arrow/compute/kernels/aggregate_basic.cc:167–190

      : options(std::move(options)), memo_table_(new MemoTable(memory_pool, 0)) {}

  Status Consume(KernelContext*, const ExecSpan& batch) override {
    if (batch[0].is_array()) {
      const ArraySpan& arr = batch[0].array;
      this->has_nulls = arr.GetNullCount() > 0;

      auto visit_null = []() { return Status::OK(); };
      auto visit_value = [&](VisitorArgType arg) {
        int32_t y;
        return memo_table_->GetOrInsert(arg, &y);
      };
      RETURN_NOT_OK(VisitArraySpanInline<Type>(arr, visit_value, visit_null));
    } else {
      const Scalar& input = *batch[0].scalar;
      this->has_nulls = !input.is_valid;

      if (input.is_valid) {
        int32_t unused;
        RETURN_NOT_OK(memo_table_->GetOrInsert(UnboxScalar<Type>::Unbox(input), &unused));
      }
    }

    this->non_nulls = memo_table_->size();
    return Status::OK();
  }

  Status MergeFrom(KernelContext*, KernelState&& src) override {
    const auto& other_state = checked_cast<const CountDistinctImpl&>(src);