GH-37059: [C++][Python] Add rolling_* functions

cpp/src/arrow/CMakeLists.txt

@@ -483,6 +483,7 @@ if(ARROW_COMPUTE)
        compute/kernels/vector_nested.cc
        compute/kernels/vector_rank.cc
        compute/kernels/vector_replace.cc
+       compute/kernels/vector_rolling.cc
        compute/kernels/vector_run_end_encode.cc
        compute/kernels/vector_select_k.cc
        compute/kernels/vector_sort.cc)

cpp/src/arrow/array/builder_primitive.h

@@ -312,6 +312,11 @@ class NumericBuilder
     data_builder_.UnsafeAppend(value_type{});  // zero
   }
 
+  void UnsafeAppendNulls(int64_t length) {
+    ArrayBuilder::UnsafeAppendToBitmap(length, false);
+    data_builder_.UnsafeAppend(length, value_type{});  // zero
+  }
+
   std::shared_ptr<DataType> type() const override { return type_; }
 
  protected:

cpp/src/arrow/compute/api_vector.cc

@@ -48,6 +48,7 @@ using compute::DictionaryEncodeOptions;
 using compute::FilterOptions;
 using compute::NullPlacement;
 using compute::RankOptions;
+using compute::RollingOptions;
 
 template <>
 struct EnumTraits<FilterOptions::NullSelectionBehavior>
@@ -153,6 +154,10 @@ static auto kRankOptionsType = GetFunctionOptionsType<RankOptions>(
     DataMember("tiebreaker", &RankOptions::tiebreaker));
 static auto kPairwiseOptionsType = GetFunctionOptionsType<PairwiseOptions>(
     DataMember("periods", &PairwiseOptions::periods));
+static auto kRollingOptionsType = GetFunctionOptionsType<RollingOptions>(
+    DataMember("window_length", &RollingOptions::window_length),
+    DataMember("min_periods", &RollingOptions::min_periods),
+    DataMember("ignore_nulls", &RollingOptions::ignore_nulls));
 }  // namespace
 }  // namespace internal
 
@@ -224,6 +229,14 @@ PairwiseOptions::PairwiseOptions(int64_t periods)
     : FunctionOptions(internal::kPairwiseOptionsType), periods(periods) {}
 constexpr char PairwiseOptions::kTypeName[];
 
+RollingOptions::RollingOptions(int64_t window_length, int64_t min_periods,
+                               bool ignore_nulls)
+    : FunctionOptions(internal::kRollingOptionsType),
+      window_length(window_length),
+      min_periods(min_periods),
+      ignore_nulls(ignore_nulls) {}
+constexpr char RollingOptions::kTypeName[];
+
 namespace internal {
 void RegisterVectorOptions(FunctionRegistry* registry) {
   DCHECK_OK(registry->AddFunctionOptionsType(kFilterOptionsType));
@@ -237,6 +250,7 @@ void RegisterVectorOptions(FunctionRegistry* registry) {
   DCHECK_OK(registry->AddFunctionOptionsType(kCumulativeOptionsType));
   DCHECK_OK(registry->AddFunctionOptionsType(kRankOptionsType));
   DCHECK_OK(registry->AddFunctionOptionsType(kPairwiseOptionsType));
+  DCHECK_OK(registry->AddFunctionOptionsType(kRollingOptionsType));
 }
 }  // namespace internal
 

cpp/src/arrow/compute/api_vector.h

@@ -26,8 +26,7 @@
 #include "arrow/result.h"
 #include "arrow/type_fwd.h"
 
-namespace arrow {
-namespace compute {
+namespace arrow::compute {
 
 class ExecContext;
 
@@ -246,6 +245,25 @@ class ARROW_EXPORT PairwiseOptions : public FunctionOptions {
   int64_t periods = 1;
 };
 
+/// \brief Options for rolling functions
+class ARROW_EXPORT RollingOptions : public FunctionOptions {
+ public:
+  explicit RollingOptions(int64_t window_length, int64_t min_periods = 1,
+                          bool ignore_nulls = true);
+  RollingOptions() : RollingOptions(1) {}
+  static constexpr char const kTypeName[] = "RollingOptions";
+
+  /// Length of the window, must be positive
+  int64_t window_length;
+  /// Minimum number of elements in window required to have a value
+  int64_t min_periods = 1;
+  /// Whether to ignore null values in the window
+  /// True: Only valid values are used to compute the result, produce valid result
+  /// if there are >= min_periods valid values in the window
+  /// False: The result is null if any value in the window is null
+  bool ignore_nulls = true;
+};
+
 /// @}
 
 /// \brief Filter with a boolean selection filter
@@ -694,5 +712,4 @@ Result<std::shared_ptr<Array>> PairwiseDiff(const Array& array,
                                             bool check_overflow = false,
                                             ExecContext* ctx = NULLPTR);
 
-}  // namespace compute
-}  // namespace arrow
+}  // namespace arrow::compute

cpp/src/arrow/compute/kernels/CMakeLists.txt

@@ -93,6 +93,7 @@ add_arrow_compute_test(vector_test
                        vector_hash_test.cc
                        vector_nested_test.cc
                        vector_replace_test.cc
+                       vector_rolling_test.cc
                        vector_run_end_encode_test.cc
                        select_k_test.cc
                        EXTRA_LINK_LIBS

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

@@ -59,7 +59,10 @@ using internal::FirstTimeBitmapWriter;
 using internal::GenerateBitsUnrolled;
 using internal::VisitBitBlocks;
 using internal::VisitBitBlocksVoid;
+using internal::VisitTwoBitBlocks;
+using internal::VisitTwoBitBlocksAllCases;
 using internal::VisitTwoBitBlocksVoid;
+using internal::VisitTwoBitBlocksVoidAllCases;
 
 namespace compute {
 namespace internal {
@@ -455,6 +458,41 @@ static void VisitTwoArrayValuesInline(const ArraySpan& arr0, const ArraySpan& ar
                         std::move(visit_null));
 }
 
+template <typename Arg0Type, typename Arg1Type, typename VisitBothNotNull,
+          typename VisitLeftNull, typename VisitRightNull, typename VisitBothNull>
+static void VisitTwoArrayValuesInlineAllCases(const ArraySpan& arr0,
+                                              const ArraySpan& arr1,
+                                              VisitBothNotNull&& visit_both_not_null,
+                                              VisitLeftNull&& visit_left_null,
+                                              VisitRightNull&& visit_right_null,
+                                              VisitBothNull&& visit_both_null) {
+  ArrayIterator<Arg0Type> arr0_it(arr0);
+  ArrayIterator<Arg1Type> arr1_it(arr1);
+
+  auto forward_visit_both_not_null = [&](int64_t i) {
+    visit_both_not_null(GetViewType<Arg0Type>::LogicalValue(arr0_it()),
+                        GetViewType<Arg1Type>::LogicalValue(arr1_it()));
+  };
+  auto forward_visit_left_null = [&](int64_t i) {
+    arr0_it();
+    visit_left_null(GetViewType<Arg1Type>::LogicalValue(arr1_it()));
+  };
+  auto forward_visit_right_null = [&](int64_t i) {
+    visit_right_null(GetViewType<Arg0Type>::LogicalValue(arr0_it()));
+    arr1_it();
+  };
+  auto forward_visit_both_null = [&]() {
+    arr0_it();
+    arr1_it();
+    visit_both_null();
+  };
+
+  VisitTwoBitBlocksVoidAllCases(
+      arr0.buffers[0].data, arr0.offset, arr1.buffers[0].data, arr1.offset, arr0.length,
+      std::move(forward_visit_both_not_null), std::move(forward_visit_left_null),
+      std::move(forward_visit_right_null), std::move(forward_visit_both_null));
+}
+
 // ----------------------------------------------------------------------
 // Reusable type resolvers
 

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

@@ -331,7 +331,7 @@ struct MakeVectorCumulativeBinaryOpFunction {
 
 }  // namespace
 
-void RegisterVectorCumulativeSum(FunctionRegistry* registry) {
+void RegisterVectorCumulative(FunctionRegistry* registry) {
   MakeVectorCumulativeBinaryOpFunction<Add, CumulativeOptions>::Call(
       registry, "cumulative_sum", cumulative_sum_doc);
   MakeVectorCumulativeBinaryOpFunction<AddChecked, CumulativeOptions>::Call(