ARROW-12084: [C++][Compute] Add remainder and quotient compute::Function

cpp/src/arrow/array/builder_nested.h

@@ -371,14 +371,14 @@ class ARROW_EXPORT MapBuilder : public ArrayBuilder {
 /// \brief Builder class for fixed-length list array value types
 class ARROW_EXPORT FixedSizeListBuilder : public ArrayBuilder {
  public:
-  /// Use this constructor to define the built array's type explicitly. If value_builder
-  /// has indeterminate type, this builder will also.
+  /// Use this constructor to infer the built array's type. If value_builder has
+  /// indeterminate type, this builder will also infer it.
   FixedSizeListBuilder(MemoryPool* pool,
                        std::shared_ptr<ArrayBuilder> const& value_builder,
                        int32_t list_size);
 
-  /// Use this constructor to infer the built array's type. If value_builder has
-  /// indeterminate type, this builder will also.
+  /// Use this constructor to define the built array's type explicitly. If value_builder
+  /// has indeterminate type, this builder will also infer it.
   FixedSizeListBuilder(MemoryPool* pool,
                        std::shared_ptr<ArrayBuilder> const& value_builder,
                        const std::shared_ptr<DataType>& type);
@@ -401,7 +401,7 @@ class ARROW_EXPORT FixedSizeListBuilder : public ArrayBuilder {
 
   /// \brief Vector append
   ///
-  /// If passed, valid_bytes wil be read and any zero byte
+  /// If passed, valid_bytes will be read and any zero byte
   /// will cause the corresponding slot to be null
   ///
   /// This function affects only the validity bitmap; the child values must be appended

cpp/src/arrow/compute/api_scalar.cc

@@ -648,9 +648,9 @@ SCALAR_ARITHMETIC_UNARY(Ln, "ln", "ln_checked")
 SCALAR_ARITHMETIC_UNARY(Log10, "log10", "log10_checked")
 SCALAR_ARITHMETIC_UNARY(Log1p, "log1p", "log1p_checked")
 SCALAR_ARITHMETIC_UNARY(Log2, "log2", "log2_checked")
-SCALAR_ARITHMETIC_UNARY(Sqrt, "sqrt", "sqrt_checked")
 SCALAR_ARITHMETIC_UNARY(Negate, "negate", "negate_checked")
 SCALAR_ARITHMETIC_UNARY(Sin, "sin", "sin_checked")
+SCALAR_ARITHMETIC_UNARY(Sqrt, "sqrt", "sqrt_checked")
 SCALAR_ARITHMETIC_UNARY(Tan, "tan", "tan_checked")
 SCALAR_EAGER_UNARY(Atan, "atan")
 SCALAR_EAGER_UNARY(Sign, "sign")
@@ -673,15 +673,16 @@ Result<Datum> RoundToMultiple(const Datum& arg, RoundToMultipleOptions options,
 
 SCALAR_ARITHMETIC_BINARY(Add, "add", "add_checked")
 SCALAR_ARITHMETIC_BINARY(Divide, "divide", "divide_checked")
+SCALAR_ARITHMETIC_BINARY(Divmod, "divmod", "divmod_checked")
 SCALAR_ARITHMETIC_BINARY(Logb, "logb", "logb_checked")
 SCALAR_ARITHMETIC_BINARY(Multiply, "multiply", "multiply_checked")
 SCALAR_ARITHMETIC_BINARY(Power, "power", "power_checked")
 SCALAR_ARITHMETIC_BINARY(ShiftLeft, "shift_left", "shift_left_checked")
 SCALAR_ARITHMETIC_BINARY(ShiftRight, "shift_right", "shift_right_checked")
 SCALAR_ARITHMETIC_BINARY(Subtract, "subtract", "subtract_checked")
 SCALAR_EAGER_BINARY(Atan2, "atan2")
-SCALAR_EAGER_UNARY(Floor, "floor")
 SCALAR_EAGER_UNARY(Ceil, "ceil")
+SCALAR_EAGER_UNARY(Floor, "floor")
 SCALAR_EAGER_UNARY(Trunc, "trunc")
 
 Result<Datum> MaxElementWise(const std::vector<Datum>& args,

cpp/src/arrow/compute/api_scalar.h

@@ -584,6 +584,22 @@ Result<Datum> Divide(const Datum& left, const Datum& right,
                      ArithmeticOptions options = ArithmeticOptions(),
                      ExecContext* ctx = NULLPTR);
 
+/// \brief Calculate the quotient and remainder between two values.
+///
+/// Array values must be the same length. If either argument is null,
+/// then the result will be null. If divisor is zero, an error will be raised.
+///
+/// \param[in] dividend the dividend
+/// \param[in] divisor the divisor
+/// \param[in] options arithmetic options (enable/disable overflow checking), optional
+/// \param[in] ctx the function execution context, optional
+/// \return the elementwise quotient and remainder as an array of
+/// struct<quotient: T, remainder: T>
+ARROW_EXPORT
+Result<Datum> Divmod(const Datum& dividend, const Datum& divisor,
+                     ArithmeticOptions options = ArithmeticOptions(),
+                     ExecContext* ctx = NULLPTR);
+
 /// \brief Negate values.
 ///
 /// If argument is null the result will be null.

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

@@ -383,7 +383,7 @@ struct Divide {
   static enable_if_decimal_value<T> Call(KernelContext*, Arg0 left, Arg1 right,
                                          Status* st) {
     if (right == Arg1()) {
-      *st = Status::Invalid("Divide by zero");
+      *st = Status::Invalid("divide by zero");
       return T();
     } else {
       return left / right;

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

@@ -189,6 +189,11 @@ struct GetOutputType<Decimal256Type> {
   using T = Decimal256;
 };
 
+template <>
+struct GetOutputType<StructType> {
+  using T = StructScalar;
+};
+
 // ----------------------------------------------------------------------
 // enable_if helpers for C types
 
@@ -221,6 +226,9 @@ using enable_if_decimal_value =
     enable_if_t<std::is_same<Decimal128, T>::value || std::is_same<Decimal256, T>::value,
                 R>;
 
+template <typename T, typename R = T>
+using enable_if_c_number = enable_if_t<has_c_type<T>::value && !is_boolean_type<T>::value, R>;
+
 // ----------------------------------------------------------------------
 // Iteration / value access utilities
 
@@ -322,6 +330,24 @@ struct OutputArrayWriter<Type, enable_if_c_number_or_decimal<Type>> {
   }
 };
 
+template <typename Type>
+struct OutputArrayWriter<Type, enable_if_struct<Type>> {
+  using T = typename TypeTraits<Type>::ScalarType;
+  T* values;
+
+  explicit OutputArrayWriter(ArrayData* data) : values(data->GetMutableValues<T>(1)) {}
+
+  void Write(T value) { *values++ = value; }
+
+  // Note that this doesn't write the null bitmap, which should be consistent
+  // with Write / WriteNull calls
+  void WriteNull() { *values++ = T(null()); }
+
+  void WriteAllNull(int64_t length) {
+    std::memset(static_cast<void*>(values), 0, sizeof(T) * length);
+  }
+};
+
 // (Un)box Scalar to / from C++ value
 
 template <typename Type, typename Enable = void>
@@ -400,7 +426,16 @@ struct BoxScalar<Decimal256Type> {
   static void Box(T val, Scalar* out) { checked_cast<ScalarType*>(out)->value = val; }
 };
 
-// A VisitArraySpanInline variant that calls its visitor function with logical
+template <>
+struct BoxScalar<StructType> {
+  using T = StructScalar;
+  using ScalarType = StructScalar;
+  static void Box(T val, Scalar* out) {
+    checked_cast<ScalarType*>(out)->value = std::move(val.value);
+  }
+};
+
+// A VisitArrayDataInline variant that calls its visitor function with logical
 // values, such as Decimal128 rather than util::string_view.
 
 template <typename T, typename VisitFunc, typename NullFunc>
@@ -555,6 +590,21 @@ struct OutputAdapter<Type, enable_if_base_binary<Type>> {
   }
 };
 
+template <typename Type>
+struct OutputAdapter<Type, enable_if_struct<Type>> {
+  using T = typename TypeTraits<Type>::ScalarType;
+
+  template <typename Generator>
+  static Status Write(KernelContext*, Datum* out, Generator&& generator) {
+    ArrayData* out_arr = out->mutable_array();
+    auto out_data = out_arr->GetMutableValues<T>(1);
+    for (int64_t i = 0; i < out_arr->length; ++i) {
+      *out_data++ = generator();
+    }
+    return Status::OK();
+  }
+};
+
 // A kernel exec generator for unary functions that addresses both array and
 // scalar inputs and dispatches input iteration and output writing to other
 // templates
@@ -591,7 +641,7 @@ struct ScalarUnary {
   }
 };
 
-// An alternative to ScalarUnary that Applies a scalar operation with state on
+// An alternative to ScalarUnary that applies a scalar operation with state on
 // only the not-null values of a single array
 template <typename OutType, typename Arg0Type, typename Op>
 struct ScalarUnaryNotNullStateful {