GH-46901: [C++][Compute] Add remainder and mod kernels

cpp/src/arrow/compute/api_scalar.cc

@@ -799,8 +799,10 @@ 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(Logb, "logb", "logb_checked")
+SCALAR_ARITHMETIC_BINARY(Mod, "mod", "mod_checked")
 SCALAR_ARITHMETIC_BINARY(Multiply, "multiply", "multiply_checked")
 SCALAR_ARITHMETIC_BINARY(Power, "power", "power_checked")
+SCALAR_ARITHMETIC_BINARY(Remainder, "remainder", "remainder_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")

cpp/src/arrow/compute/api_scalar.h

@@ -671,6 +671,40 @@ Result<Datum> Divide(const Datum& left, const Datum& right,
                      ArithmeticOptions options = ArithmeticOptions(),
                      ExecContext* ctx = NULLPTR);
 
+/// \brief Compute the remainder (truncated division) of two values.
+/// Array values must be the same length. If either argument is null the result
+/// will be null. For integer types, if there is a zero divisor, an error will be
+/// raised.
+///
+/// The result has the same sign as the dividend (C/C++ semantics).
+///
+/// \param[in] left the dividend
+/// \param[in] right the divisor
+/// \param[in] options arithmetic options (enable/disable overflow checking), optional
+/// \param[in] ctx the function execution context, optional
+/// \return the elementwise remainder
+ARROW_EXPORT
+Result<Datum> Remainder(const Datum& left, const Datum& right,
+                        ArithmeticOptions options = ArithmeticOptions(),
+                        ExecContext* ctx = NULLPTR);
+
+/// \brief Compute the modulo (floored division) of two values.
+/// Array values must be the same length. If either argument is null the result
+/// will be null. For integer types, if there is a zero divisor, an error will be
+/// raised.
+///
+/// The result has the same sign as the divisor (Python semantics).
+///
+/// \param[in] left the dividend
+/// \param[in] right the divisor
+/// \param[in] options arithmetic options (enable/disable overflow checking), optional
+/// \param[in] ctx the function execution context, optional
+/// \return the elementwise modulo
+ARROW_EXPORT
+Result<Datum> Mod(const Datum& left, const Datum& right,
+                  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

@@ -34,6 +34,7 @@ namespace arrow {
 
 using internal::AddWithOverflow;
 using internal::DivideWithOverflow;
+using internal::ModuloWithOverflow;
 using internal::MultiplyWithOverflow;
 using internal::NegateWithOverflow;
 using internal::SubtractWithOverflow;
@@ -468,6 +469,172 @@ struct FloatingDivideChecked {
   // TODO: Add decimal
 };
 
+// Remainder (truncated): result has same sign as dividend (C/C++ semantics)
+struct Remainder {
+  template <typename T, typename Arg0, typename Arg1>
+  static enable_if_floating_value<T> Call(KernelContext*, Arg0 left, Arg1 right,
+                                          Status*) {
+    return std::fmod(left, right);
+  }
+
+  template <typename T, typename Arg0, typename Arg1>
+  static enable_if_integer_value<T> Call(KernelContext*, Arg0 left, Arg1 right,
+                                         Status* st) {
+    T result;
+    if (ARROW_PREDICT_FALSE(ModuloWithOverflow(left, right, &result))) {
+      if (right == 0) {
+        *st = Status::Invalid("divide by zero");
+      } else {
+        // INT_MIN % -1 overflow case, result is 0
+        result = 0;
+      }
+    }
+    return result;
+  }
+
+  template <typename T, typename Arg0, typename Arg1>
+  static enable_if_decimal_value<T> Call(KernelContext*, Arg0 left, Arg1 right,
+                                         Status* st) {
+    if (right == Arg1()) {
+      *st = Status::Invalid("divide by zero");
+      return T();
+    }
+    return left % right;
+  }
+};
+
+struct RemainderChecked {
+  template <typename T, typename Arg0, typename Arg1>
+  static enable_if_floating_value<T> Call(KernelContext*, Arg0 left, Arg1 right,
+                                          Status* st) {
+    static_assert(std::is_same<T, Arg0>::value && std::is_same<T, Arg1>::value, "");
+    if (ARROW_PREDICT_FALSE(right == 0)) {
+      *st = Status::Invalid("divide by zero");
+      return 0;
+    }
+    return std::fmod(left, right);
+  }
+
+  template <typename T, typename Arg0, typename Arg1>
+  static enable_if_integer_value<T> Call(KernelContext*, Arg0 left, Arg1 right,
+                                         Status* st) {
+    static_assert(std::is_same<T, Arg0>::value && std::is_same<T, Arg1>::value, "");
+    T result;
+    if (ARROW_PREDICT_FALSE(ModuloWithOverflow(left, right, &result))) {
+      if (right == 0) {
+        *st = Status::Invalid("divide by zero");
+      } else {
+        *st = Status::Invalid("overflow");
+      }
+    }
+    return result;
+  }
+
+  template <typename T, typename Arg0, typename Arg1>
+  static enable_if_decimal_value<T> Call(KernelContext* ctx, Arg0 left, Arg1 right,
+                                         Status* st) {
+    return Remainder::Call<T>(ctx, left, right, st);
+  }
+};
+
+// Helper: Convert truncated remainder to floored modulo for signed types.
+// Floored modulo has the same sign as the divisor (Python semantics).
+template <typename T>
+T AdjustRemainderToFloored(T rem, T right) {
+  if constexpr (std::is_signed_v<T>) {
+    if ((rem > 0 && right < 0) || (rem < 0 && right > 0)) {
+      rem += right;
+    }
+  }
+  return rem;
+}
+
+// Mod (floored): result has same sign as divisor (Python semantics)
+struct Mod {
+  template <typename T, typename Arg0, typename Arg1>
+  static enable_if_floating_value<T> Call(KernelContext*, Arg0 left, Arg1 right,
+                                          Status*) {
+    T rem = std::fmod(left, right);
+    if (rem == 0) {
+      // Preserve the sign based on divisor for zero results
+      return std::copysign(rem, right);
+    }
+    return AdjustRemainderToFloored(rem, right);
+  }
+
+  template <typename T, typename Arg0, typename Arg1>
+  static enable_if_integer_value<T> Call(KernelContext*, Arg0 left, Arg1 right,
+                                         Status* st) {
+    T result;
+    if (ARROW_PREDICT_FALSE(ModuloWithOverflow(left, right, &result))) {
+      if (right == 0) {
+        *st = Status::Invalid("divide by zero");
+      } else {
+        // INT_MIN % -1 overflow case, result is 0
+        result = 0;
+      }
+      return result;
+    }
+    return AdjustRemainderToFloored(result, right);
+  }
+
+  template <typename T, typename Arg0, typename Arg1>
+  static enable_if_decimal_value<T> Call(KernelContext*, Arg0 left, Arg1 right,
+                                         Status* st) {
+    static const T kZero{};
+    if (right == kZero) {
+      *st = Status::Invalid("divide by zero");
+      return T();
+    }
+    T rem = left % right;
+    // Convert truncated to floored: adjust if signs differ
+    if ((rem > kZero && right < kZero) || (rem < kZero && right > kZero)) {
+      rem = rem + right;
+    }
+    return rem;
+  }
+};
+
+struct ModChecked {
+  template <typename T, typename Arg0, typename Arg1>
+  static enable_if_floating_value<T> Call(KernelContext*, Arg0 left, Arg1 right,
+                                          Status* st) {
+    static_assert(std::is_same<T, Arg0>::value && std::is_same<T, Arg1>::value, "");
+    if (ARROW_PREDICT_FALSE(right == 0)) {
+      *st = Status::Invalid("divide by zero");
+      return 0;
+    }
+    T rem = std::fmod(left, right);
+    if (rem == 0) {
+      // Preserve the sign based on divisor for zero results
+      return std::copysign(rem, right);
+    }
+    return AdjustRemainderToFloored(rem, right);
+  }
+
+  template <typename T, typename Arg0, typename Arg1>
+  static enable_if_integer_value<T> Call(KernelContext*, Arg0 left, Arg1 right,
+                                         Status* st) {
+    static_assert(std::is_same<T, Arg0>::value && std::is_same<T, Arg1>::value, "");
+    T result;
+    if (ARROW_PREDICT_FALSE(ModuloWithOverflow(left, right, &result))) {
+      if (right == 0) {
+        *st = Status::Invalid("divide by zero");
+      } else {
+        *st = Status::Invalid("overflow");
+      }
+      return result;
+    }
+    return AdjustRemainderToFloored(result, right);
+  }
+
+  template <typename T, typename Arg0, typename Arg1>
+  static enable_if_decimal_value<T> Call(KernelContext* ctx, Arg0 left, Arg1 right,
+                                         Status* st) {
+    return Mod::Call<T>(ctx, left, right, st);
+  }
+};
+
 struct Negate {
   template <typename T, typename Arg>
   static constexpr enable_if_floating_value<T> Call(KernelContext*, Arg arg, Status*) {

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

@@ -663,7 +663,7 @@ void AddDecimalBinaryKernels(const std::string& name, ScalarFunction* func) {
   OutputType out_type(null());
   std::shared_ptr<MatchConstraint> constraint = nullptr;
   const std::string op = name.substr(0, name.find("_"));
-  if (op == "add" || op == "subtract") {
+  if (op == "add" || op == "subtract" || op == "remainder" || op == "mod") {
     out_type = OutputType(ResolveDecimalAdditionOrSubtractionOutput);
     constraint = DecimalsHaveSameScale();
   } else if (op == "multiply") {
@@ -776,7 +776,7 @@ struct ArithmeticFunction : ScalarFunction {
       // "add_checked" -> "add"
       const auto func_name = name();
       const std::string op = func_name.substr(0, func_name.find("_"));
-      if (op == "add" || op == "subtract") {
+      if (op == "add" || op == "subtract" || op == "remainder" || op == "mod") {
         return CastBinaryDecimalArgs(DecimalPromotion::kAdd, types);
       } else if (op == "multiply") {
         return CastBinaryDecimalArgs(DecimalPromotion::kMultiply, types);
@@ -1165,6 +1165,40 @@ const FunctionDoc div_checked_doc{
      "integer overflow is encountered."),
     {"dividend", "divisor"}};
 
+const FunctionDoc remainder_doc{
+    "Compute the remainder after integer division (truncated)",
+    ("Returns the remainder after dividing the dividend by the divisor.\n"
+     "The result has the same sign as the dividend (truncated division).\n"
+     "This is equivalent to the C/C++ '%' operator.\n"
+     "Integer division by zero returns an error."),
+    {"dividend", "divisor"}};
+
+const FunctionDoc remainder_checked_doc{
+    "Compute the remainder after integer division (truncated)",
+    ("Returns the remainder after dividing the dividend by the divisor.\n"
+     "The result has the same sign as the dividend (truncated division).\n"
+     "This is equivalent to the C/C++ '%' operator.\n"
+     "An error is returned when trying to divide by zero, or when\n"
+     "integer overflow is encountered."),
+    {"dividend", "divisor"}};
+
+const FunctionDoc mod_doc{
+    "Compute the modulo (floored)",
+    ("Returns the modulo after floored division.\n"
+     "The result has the same sign as the divisor (floored division).\n"
+     "This is equivalent to Python's '%' operator.\n"
+     "Integer division by zero returns an error."),
+    {"dividend", "divisor"}};
+
+const FunctionDoc mod_checked_doc{
+    "Compute the modulo (floored)",
+    ("Returns the modulo after floored division.\n"
+     "The result has the same sign as the divisor (floored division).\n"
+     "This is equivalent to Python's '%' operator.\n"
+     "An error is returned when trying to divide by zero, or when\n"
+     "integer overflow is encountered."),
+    {"dividend", "divisor"}};
+
 const FunctionDoc negate_doc{"Negate the argument element-wise",
                              ("Results will wrap around on integer overflow.\n"
                               "Use function \"negate_checked\" if you want overflow\n"
@@ -1708,6 +1742,28 @@ void RegisterScalarArithmetic(FunctionRegistry* registry) {
 
   DCHECK_OK(registry->AddFunction(std::move(divide_checked)));
 
+  // ----------------------------------------------------------------------
+  auto remainder = MakeArithmeticFunctionNotNull<Remainder>("remainder", remainder_doc);
+  AddDecimalBinaryKernels<Remainder>("remainder", remainder.get());
+  DCHECK_OK(registry->AddFunction(std::move(remainder)));
+
+  // ----------------------------------------------------------------------
+  auto remainder_checked = MakeArithmeticFunctionNotNull<RemainderChecked>(
+      "remainder_checked", remainder_checked_doc);
+  AddDecimalBinaryKernels<RemainderChecked>("remainder_checked", remainder_checked.get());
+  DCHECK_OK(registry->AddFunction(std::move(remainder_checked)));
+
+  // ----------------------------------------------------------------------
+  auto mod = MakeArithmeticFunctionNotNull<Mod>("mod", mod_doc);
+  AddDecimalBinaryKernels<Mod>("mod", mod.get());
+  DCHECK_OK(registry->AddFunction(std::move(mod)));
+
+  // ----------------------------------------------------------------------
+  auto mod_checked =
+      MakeArithmeticFunctionNotNull<ModChecked>("mod_checked", mod_checked_doc);
+  AddDecimalBinaryKernels<ModChecked>("mod_checked", mod_checked.get());
+  DCHECK_OK(registry->AddFunction(std::move(mod_checked)));
+
   // ----------------------------------------------------------------------
   auto negate = MakeUnaryArithmeticFunction<Negate>("negate", negate_doc);
   AddDecimalUnaryKernels<Negate>(negate.get());