GH-45733: [C++][Python] Add biased/unbiased toggle to skew and kurtosis functions

Author: raulcd

cpp/src/arrow/compute/api_aggregate.cc

@@ -111,6 +111,7 @@ static auto kVarianceOptionsType = GetFunctionOptionsType<VarianceOptions>(
     DataMember("min_count", &VarianceOptions::min_count));
 static auto kSkewOptionsType = GetFunctionOptionsType<SkewOptions>(
     DataMember("skip_nulls", &SkewOptions::skip_nulls),
+    DataMember("bias", &SkewOptions::bias),
     DataMember("min_count", &SkewOptions::min_count));
 static auto kQuantileOptionsType = GetFunctionOptionsType<QuantileOptions>(
     DataMember("q", &QuantileOptions::q),
@@ -154,9 +155,10 @@ VarianceOptions::VarianceOptions(int ddof, bool skip_nulls, uint32_t min_count)
       min_count(min_count) {}
 constexpr char VarianceOptions::kTypeName[];
 
-SkewOptions::SkewOptions(bool skip_nulls, uint32_t min_count)
+SkewOptions::SkewOptions(bool skip_nulls, bool bias, uint32_t min_count)
     : FunctionOptions(internal::kSkewOptionsType),
       skip_nulls(skip_nulls),
+      bias(bias),
       min_count(min_count) {}
 
 QuantileOptions::QuantileOptions(double q, enum Interpolation interpolation,

cpp/src/arrow/compute/api_aggregate.h

@@ -117,13 +117,14 @@ class ARROW_EXPORT VarianceOptions : public FunctionOptions {
 /// \brief Control Skew and Kurtosis kernel behavior
 class ARROW_EXPORT SkewOptions : public FunctionOptions {
  public:
-  explicit SkewOptions(bool skip_nulls = true, uint32_t min_count = 0);
+  explicit SkewOptions(bool skip_nulls = true, bool bias = true, uint32_t min_count = 0);
   static constexpr char const kTypeName[] = "SkewOptions";
   static SkewOptions Defaults() { return SkewOptions{}; }
 
   /// If true (the default), null values are ignored. Otherwise, if any value is null,
   /// emit null.
   bool skip_nulls;
+  bool bias;
   /// If less than this many non-null values are observed, emit null.
   uint32_t min_count;
 };

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

@@ -176,6 +176,7 @@ struct StatisticImpl : public ScalarAggregator {
       : out_type(out_type),
         stat_type(stat_type),
         skip_nulls(options.skip_nulls),
+        bias(options.bias),
         min_count(options.min_count),
         ddof(0),
         state(moments_level_for_statistic(stat_type), decimal_scale, skip_nulls) {}
@@ -208,7 +209,7 @@ struct StatisticImpl : public ScalarAggregator {
           out->value = std::make_shared<DoubleScalar>(state.moments.Variance(ddof));
           break;
         case StatisticType::Skew:
-          out->value = std::make_shared<DoubleScalar>(state.moments.Skew());
+          out->value = std::make_shared<DoubleScalar>(state.moments.Skew(bias));
           break;
         case StatisticType::Kurtosis:
           out->value = std::make_shared<DoubleScalar>(state.moments.Kurtosis());
@@ -224,6 +225,7 @@ struct StatisticImpl : public ScalarAggregator {
   std::shared_ptr<DataType> out_type;
   StatisticType stat_type;
   bool skip_nulls;
+  bool bias;
   uint32_t min_count;
   int ddof = 0;
   MomentsState<ArrowType> state;

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

@@ -84,9 +84,16 @@ struct Moments {
 
   double Stddev(int ddof) const { return sqrt(Variance(ddof)); }
 
-  double Skew() const {
+  double Skew(bool bias = true) const {
+    double result;
     // This may return NaN for m2 == 0 and m3 == 0, which is expected
-    return sqrt(count) * m3 / sqrt(m2 * m2 * m2);
+    if (bias) {
+      result = sqrt(count) * m3 / sqrt(m2 * m2 * m2);
+    } else {
+      result =
+          sqrt(count * (count - 1)) / (count - 2) * (m3 / count) / pow((m2 / count), 1.5);
+    }
+    return result;
   }
 
   double Kurtosis() const {

python/pyarrow/_compute.pyx

@@ -1909,8 +1909,8 @@ class VarianceOptions(_VarianceOptions):
 
 
 cdef class _SkewOptions(FunctionOptions):
-    def _set_options(self, skip_nulls, min_count):
-        self.wrapped.reset(new CSkewOptions(skip_nulls, min_count))
+    def _set_options(self, skip_nulls, bias, min_count):
+        self.wrapped.reset(new CSkewOptions(skip_nulls, bias, min_count))
 
 
 class SkewOptions(_SkewOptions):
@@ -1923,8 +1923,8 @@ class SkewOptions(_SkewOptions):
     {_min_count_doc(default=0)}
     """
 
-    def __init__(self, *, skip_nulls=True, min_count=0):
-        self._set_options(skip_nulls, min_count)
+    def __init__(self, *, skip_nulls=True, bias=True, min_count=0):
+        self._set_options(skip_nulls, bias, min_count)
 
 
 cdef class _SplitOptions(FunctionOptions):

python/pyarrow/includes/libarrow.pxd

@@ -2628,8 +2628,9 @@ cdef extern from "arrow/compute/api.h" namespace "arrow::compute" nogil:
 
     cdef cppclass CSkewOptions \
             "arrow::compute::SkewOptions"(CFunctionOptions):
-        CSkewOptions(c_bool skip_nulls, uint32_t min_count)
+        CSkewOptions(c_bool skip_nulls, c_bool bias, uint32_t min_count)
         c_bool skip_nulls
+        c_bool bias
         uint32_t min_count
 
     cdef cppclass CScalarAggregateOptions \

python/pyarrow/tests/test_compute.py

@@ -3838,3 +3838,10 @@ def test_pivot_wider():
     with pytest.raises(ValueError, match="Encountered more than one non-null value"):
         result = pc.pivot_wider(["height", "width", "height"], [10, None, 11],
                                 key_names=key_names)
+
+
+@pytest.mark.pandas
+def test_biased_skew():
+    arrow_skew = pc.skew([1.0, 2.0, 3.0, 40.0, None], skip_nulls=True, bias=False)
+    pandas_skew = pd.Series(np.array([1.0, 2.0, 3.0, 40.0, np.nan])).skew(skipna=True)
+    assert arrow_skew == pa.scalar(pandas_skew)