GH-36831: [C++] DictionaryArray support for MinMax Function

cpp/src/arrow/compute/kernel.cc

@@ -349,6 +349,37 @@ std::shared_ptr<TypeMatcher> RunEndEncoded(
                                                 std::move(value_type_matcher));
 }
 
+class NotMatcher : public TypeMatcher {
+ public:
+  explicit NotMatcher(std::shared_ptr<TypeMatcher> base_matcher)
+      : base_matcher{std::move(base_matcher)} {}
+
+  ~NotMatcher() override = default;
+
+  bool Matches(const DataType& type) const override {
+    return !base_matcher->Matches(type);
+  }
+
+  bool Equals(const TypeMatcher& other) const override {
+    if (this == &other) {
+      return true;
+    }
+    const auto* casted = dynamic_cast<const NotMatcher*>(&other);
+    return casted != nullptr && base_matcher->Equals(*casted->base_matcher);
+  }
+
+  std::string ToString() const override {
+    return "not(" + base_matcher->ToString() + ")";
+  };
+
+ private:
+  std::shared_ptr<TypeMatcher> base_matcher;
+};
+
+std::shared_ptr<TypeMatcher> Not(std::shared_ptr<TypeMatcher> base_matcher) {
+  return std::make_shared<NotMatcher>(std::move(base_matcher));
+}
+
 }  // namespace match
 
 // ----------------------------------------------------------------------

cpp/src/arrow/compute/kernel.h

@@ -172,6 +172,11 @@ ARROW_EXPORT std::shared_ptr<TypeMatcher> RunEndEncoded(
     std::shared_ptr<TypeMatcher> run_end_type_matcher,
     std::shared_ptr<TypeMatcher> value_type_matcher);
 
+/// \brief Match types that the base_matcher doesn't match
+///
+/// @param[in] base_matcher a matcher used to negation match
+ARROW_EXPORT std::shared_ptr<TypeMatcher> Not(std::shared_ptr<TypeMatcher> base_matcher);
+
 }  // namespace match
 
 /// \brief An object used for type-checking arguments to be passed to a kernel

cpp/src/arrow/compute/kernels/aggregate_basic.cc

@@ -482,6 +482,9 @@ void AddFirstOrLastAggKernel(ScalarAggregateFunction* func,
 // ----------------------------------------------------------------------
 // MinMax implementation
 
+using arrow::compute::match::Not;
+using arrow::compute::match::SameTypeId;
+
 Result<std::unique_ptr<KernelState>> MinMaxInit(KernelContext* ctx,
                                                 const KernelInitArgs& args) {
   ARROW_ASSIGN_OR_RAISE(TypeHolder out_type,
@@ -494,9 +497,10 @@ Result<std::unique_ptr<KernelState>> MinMaxInit(KernelContext* ctx,
 
 // For "min" and "max" functions: override finalize and return the actual value
 template <MinOrMax min_or_max>
-void AddMinOrMaxAggKernel(ScalarAggregateFunction* func,
-                          ScalarAggregateFunction* min_max_func) {
-  auto sig = KernelSignature::Make({InputType::Any()}, FirstType);
+void AddMinOrMaxAggKernels(ScalarAggregateFunction* func,
+                           ScalarAggregateFunction* min_max_func) {
+  std::shared_ptr<arrow::compute::KernelSignature> sig =
+      KernelSignature::Make({InputType(Not(SameTypeId(Type::DICTIONARY)))}, FirstType);
   auto init = [min_max_func](
                   KernelContext* ctx,
                   const KernelInitArgs& args) -> Result<std::unique_ptr<KernelState>> {
@@ -516,6 +520,9 @@ void AddMinOrMaxAggKernel(ScalarAggregateFunction* func,
 
   // Note SIMD level is always NONE, but the convenience kernel will
   // dispatch to an appropriate implementation
+  AddAggKernel(std::move(sig), init, finalize, func);
+
+  sig = KernelSignature::Make({InputType(Type::DICTIONARY)}, DictionaryValueType);
   AddAggKernel(std::move(sig), std::move(init), std::move(finalize), func);
 }
 
@@ -873,6 +880,15 @@ Result<TypeHolder> MinMaxType(KernelContext*, const std::vector<TypeHolder>& typ
   return struct_({field("min", ty), field("max", ty)});
 }
 
+Result<TypeHolder> DictionaryMinMaxType(KernelContext*,
+                                        const std::vector<TypeHolder>& types) {
+  // T -> struct<min: T.value_type, max: T.value_type>
+  auto ty = types.front();
+  const DictionaryType& ty_dict = checked_cast<const DictionaryType&>(*ty);
+  return struct_(
+      {field("min", ty_dict.value_type()), field("max", ty_dict.value_type())});
+}
+
 }  // namespace
 
 Result<TypeHolder> FirstLastType(KernelContext*, const std::vector<TypeHolder>& types) {
@@ -896,7 +912,12 @@ void AddFirstLastKernels(KernelInit init,
 
 void AddMinMaxKernel(KernelInit init, internal::detail::GetTypeId get_id,
                      ScalarAggregateFunction* func, SimdLevel::type simd_level) {
-  auto sig = KernelSignature::Make({InputType(get_id.id)}, MinMaxType);
+  std::shared_ptr<arrow::compute::KernelSignature> sig;
+  if (get_id.id == Type::DICTIONARY) {
+    sig = KernelSignature::Make({InputType(get_id.id)}, DictionaryMinMaxType);
+  } else {
+    sig = KernelSignature::Make({InputType(get_id.id)}, MinMaxType);
+  }
   AddAggKernel(std::move(sig), init, func, simd_level);
 }
 
@@ -1118,6 +1139,7 @@ void RegisterScalarAggregateBasic(FunctionRegistry* registry) {
   AddMinMaxKernels(MinMaxInit, NumericTypes(), func.get());
   AddMinMaxKernels(MinMaxInit, TemporalTypes(), func.get());
   AddMinMaxKernels(MinMaxInit, BaseBinaryTypes(), func.get());
+  AddMinMaxKernel(MinMaxInit, Type::DICTIONARY, func.get());
   AddMinMaxKernel(MinMaxInit, Type::FIXED_SIZE_BINARY, func.get());
   AddMinMaxKernel(MinMaxInit, Type::INTERVAL_MONTHS, func.get());
   AddMinMaxKernel(MinMaxInit, Type::DECIMAL128, func.get());
@@ -1140,12 +1162,12 @@ void RegisterScalarAggregateBasic(FunctionRegistry* registry) {
   // Add min/max as convenience functions
   func = std::make_shared<ScalarAggregateFunction>("min", Arity::Unary(), min_or_max_doc,
                                                    &default_scalar_aggregate_options);
-  AddMinOrMaxAggKernel<MinOrMax::Min>(func.get(), min_max_func);
+  AddMinOrMaxAggKernels<MinOrMax::Min>(func.get(), min_max_func);
   DCHECK_OK(registry->AddFunction(std::move(func)));
 
   func = std::make_shared<ScalarAggregateFunction>("max", Arity::Unary(), min_or_max_doc,
                                                    &default_scalar_aggregate_options);
-  AddMinOrMaxAggKernel<MinOrMax::Max>(func.get(), min_max_func);
+  AddMinOrMaxAggKernels<MinOrMax::Max>(func.get(), min_max_func);
   DCHECK_OK(registry->AddFunction(std::move(func)));
 
   func = std::make_shared<ScalarAggregateFunction>("product", Arity::Unary(), product_doc,

cpp/src/arrow/compute/kernels/aggregate_basic_internal.h

@@ -971,6 +971,11 @@ struct FirstLastInitState {
   }
 };
 
+template <SimdLevel::type SimdLevel>
+std::unique_ptr<KernelState> DictionaryMinMaxImplFunc(const DataType& in_type,
+                                                      std::shared_ptr<DataType> out_type,
+                                                      ScalarAggregateOptions options);
+
 template <SimdLevel::type SimdLevel>
 struct MinMaxInitState {
   std::unique_ptr<KernelState> state;
@@ -1002,6 +1007,11 @@ struct MinMaxInitState {
     return Status::OK();
   }
 
+  Status Visit(const DictionaryType&) {
+    state = DictionaryMinMaxImplFunc<SimdLevel>(in_type, out_type, options);
+    return Status::OK();
+  }
+
   template <typename Type>
   enable_if_physical_integer<Type, Status> Visit(const Type&) {
     using PhysicalType = typename Type::PhysicalType;
@@ -1033,4 +1043,112 @@ struct MinMaxInitState {
   }
 };
 
+template <SimdLevel::type SimdLevel>
+struct DictionaryMinMaxImpl : public ScalarAggregator {
+  using ThisType = DictionaryMinMaxImpl<SimdLevel>;
+
+  DictionaryMinMaxImpl(const DataType& in_type, std::shared_ptr<DataType> out_type,
+                       ScalarAggregateOptions options)
+      : options(std::move(options)),
+        out_type(std::move(out_type)),
+        has_nulls(false),
+        count(0),
+        value_type(checked_cast<const DictionaryType&>(in_type).value_type()),
+        value_state(nullptr) {
+    this->options.min_count = std::max<uint32_t>(1, this->options.min_count);
+  }
+
+  Status Consume(KernelContext* ctx, const ExecSpan& batch) override {
+    if (batch[0].is_scalar()) {
+      return Status::NotImplemented("No min/max implemented for DictionaryScalar");
+    }
+    RETURN_NOT_OK(this->InitValueState());
+
+    // The minmax is computed from dictionay values, in case some values are not
+    // referenced by indices, a compaction needs to be excuted here.
+    DictionaryArray dict_arr(batch[0].array.ToArrayData());
+    ARROW_ASSIGN_OR_RAISE(auto compacted_arr, dict_arr.Compact(ctx->memory_pool()));
+    const DictionaryArray& compacted_dict_arr =
+        checked_cast<const DictionaryArray&>(*compacted_arr);
+    const int64_t null_count = compacted_dict_arr.ComputeLogicalNullCount();
+    const int64_t non_null_count = compacted_dict_arr.length() - null_count;
+
+    this->has_nulls |= null_count > 0;
+    this->count += non_null_count;
+    if ((this->has_nulls && !options.skip_nulls) || (non_null_count == 0)) {
+      return Status::OK();
+    }
+
+    const ArrayData& dict_data =
+        checked_cast<const ArrayData&>(*compacted_dict_arr.dictionary()->data());
+    RETURN_NOT_OK(
+        checked_cast<ScalarAggregator*>(this->value_state.get())
+            ->Consume(nullptr, ExecSpan(std::vector({ExecValue(dict_data)}), 1)));
+    return Status::OK();
+  }
+
+  Status MergeFrom(KernelContext*, KernelState&& src) override {
+    auto&& other = checked_cast<ThisType&&>(src);
+    this->has_nulls |= other.has_nulls;
+    this->count += other.count;
+    if ((this->has_nulls && !options.skip_nulls) || other.value_state == nullptr) {
+      return Status::OK();
+    }
+
+    if (this->value_state == nullptr) {
+      this->value_state.reset(other.value_state.release());
+    } else {
+      RETURN_NOT_OK(checked_cast<ScalarAggregator*>(this->value_state.get())
+                        ->MergeFrom(nullptr, std::move(*other.value_state)));
+    }
+    return Status::OK();
+  }
+
+  Status Finalize(KernelContext*, Datum* out) override {
+    if ((this->has_nulls && !options.skip_nulls) || (this->count < options.min_count) ||
+        this->value_state.get() == nullptr) {
+      const auto& struct_type = checked_cast<const StructType&>(*out_type);
+      const auto& child_type = struct_type.field(0)->type();
+
+      std::shared_ptr<Scalar> null_scalar = MakeNullScalar(child_type);
+      std::vector<std::shared_ptr<Scalar>> values = {null_scalar, null_scalar};
+      out->value = std::make_shared<StructScalar>(std::move(values), this->out_type);
+    } else {
+      Datum temp;
+      RETURN_NOT_OK(checked_cast<ScalarAggregator*>(this->value_state.get())
+                        ->Finalize(nullptr, &temp));
+      const auto& result = temp.scalar_as<StructScalar>();
+      DCHECK(result.is_valid);
+      out->value = result.GetSharedPtr();
+    }
+    return Status::OK();
+  }
+
+  ScalarAggregateOptions options;
+  std::shared_ptr<DataType> out_type;
+  bool has_nulls;
+  int64_t count;
+  std::shared_ptr<DataType> value_type;
+  std::unique_ptr<KernelState> value_state;
+
+ private:
+  inline Status InitValueState() {
+    if (this->value_state == nullptr) {
+      const DataType& value_type_ref = checked_cast<const DataType&>(*this->value_type);
+      ScalarAggregateOptions options = ScalarAggregateOptions::Defaults();
+      MinMaxInitState<SimdLevel::NONE> valueMinMaxInitState(nullptr, value_type_ref,
+                                                            out_type, options);
+      ARROW_ASSIGN_OR_RAISE(this->value_state, valueMinMaxInitState.Create());
+    }
+    return Status::OK();
+  }
+};
+
+template <SimdLevel::type SimdLevel>
+std::unique_ptr<KernelState> DictionaryMinMaxImplFunc(const DataType& in_type,
+                                                      std::shared_ptr<DataType> out_type,
+                                                      ScalarAggregateOptions options) {
+  return std::make_unique<DictionaryMinMaxImpl<SimdLevel>>(in_type, out_type, options);
+}
+
 }  // namespace arrow::compute::internal