Implement groupby aggregations M2 and MERGE_M2

cpp/CMakeLists.txt

@@ -199,6 +199,7 @@ add_library(cudf
     src/groupby/sort/aggregate.cpp
     src/groupby/sort/group_collect.cu
     src/groupby/sort/group_merge_lists.cu
+    src/groupby/sort/group_merge_variances.cu
     src/groupby/sort/group_count.cu
     src/groupby/sort/group_max.cu
     src/groupby/sort/group_min.cu

cpp/include/cudf/aggregation.hpp

@@ -57,33 +57,36 @@ class aggregation {
    * @brief Possible aggregation operations
    */
   enum Kind {
-    SUM,             ///< sum reduction
-    PRODUCT,         ///< product reduction
-    MIN,             ///< min reduction
-    MAX,             ///< max reduction
-    COUNT_VALID,     ///< count number of valid elements
-    COUNT_ALL,       ///< count number of elements
-    ANY,             ///< any reduction
-    ALL,             ///< all reduction
-    SUM_OF_SQUARES,  ///< sum of squares reduction
-    MEAN,            ///< arithmetic mean reduction
-    VARIANCE,        ///< groupwise variance
-    STD,             ///< groupwise standard deviation
-    MEDIAN,          ///< median reduction
-    QUANTILE,        ///< compute specified quantile(s)
-    ARGMAX,          ///< Index of max element
-    ARGMIN,          ///< Index of min element
-    NUNIQUE,         ///< count number of unique elements
-    NTH_ELEMENT,     ///< get the nth element
-    ROW_NUMBER,      ///< get row-number of current index (relative to rolling window)
-    COLLECT_LIST,    ///< collect values into a list
-    COLLECT_SET,     ///< collect values into a list without duplicate entries
-    MERGE_LISTS,     ///< merge multiple lists values into one list
-    MERGE_SETS,      ///< merge multiple lists values into one list then drop duplicate entries
-    LEAD,            ///< window function, accesses row at specified offset following current row
-    LAG,             ///< window function, accesses row at specified offset preceding current row
-    PTX,             ///< PTX  UDF based reduction
-    CUDA             ///< CUDA UDF based reduction
+    SUM,              ///< sum reduction
+    PRODUCT,          ///< product reduction
+    MIN,              ///< min reduction
+    MAX,              ///< max reduction
+    COUNT_VALID,      ///< count number of valid elements
+    COUNT_ALL,        ///< count number of elements
+    ANY,              ///< any reduction
+    ALL,              ///< all reduction
+    SUM_OF_SQUARES,   ///< sum of squares reduction
+    MEAN,             ///< arithmetic mean reduction
+    M2,               ///< groupwise sum of squares of differences from the current mean
+    VARIANCE,         ///< groupwise variance
+    STD,              ///< groupwise standard deviation
+    MEDIAN,           ///< median reduction
+    QUANTILE,         ///< compute specified quantile(s)
+    ARGMAX,           ///< Index of max element
+    ARGMIN,           ///< Index of min element
+    NUNIQUE,          ///< count number of unique elements
+    NTH_ELEMENT,      ///< get the nth element
+    ROW_NUMBER,       ///< get row-number of current index (relative to rolling window)
+    COLLECT_LIST,     ///< collect values into a list
+    COLLECT_SET,      ///< collect values into a list without duplicate entries
+    LEAD,             ///< window function, accesses row at specified offset following current row
+    LAG,              ///< window function, accesses row at specified offset preceding current row
+    PTX,              ///< PTX  UDF based reduction
+    CUDA,             ///< CUDA UDF based reduction
+    MERGE_LISTS,      ///< merge multiple lists values into one list
+    MERGE_SETS,       ///< merge multiple lists values into one list then drop duplicate entries
+    MERGE_VARIANCES,  ///< merge partial variance values
+    MERGE_STDS        ///< merge partial standard deviation values
   };
 
   aggregation() = delete;
@@ -159,6 +162,16 @@ std::unique_ptr<Base> make_sum_of_squares_aggregation();
 template <typename Base = aggregation>
 std::unique_ptr<Base> make_mean_aggregation();
 
+/**
+ * @brief Factory to create a M2 aggregation
+ *
+ * A M2 aggregation is groupwise sum of squares of differences from the current mean. From this,
+ * a `VARIANCE` aggregation can be computed as `M2 / (N - ddof)`, where `N` is the population size
+ * and `ddof` is the delta degrees of freedom.
+ */
+template <typename Base = aggregation>
+std::unique_ptr<Base> make_m2_aggregation();
+
 /**
  * @brief Factory to create a VARIANCE aggregation
  *
@@ -271,6 +284,28 @@ std::unique_ptr<Base> make_collect_set_aggregation(null_policy null_handling = n
                                                    null_equality nulls_equal = null_equality::EQUAL,
                                                    nan_equality nans_equal = nan_equality::UNEQUAL);
 
+/// Factory to create a LAG aggregation
+template <typename Base = aggregation>
+std::unique_ptr<Base> make_lag_aggregation(size_type offset);
+
+/// Factory to create a LEAD aggregation
+template <typename Base = aggregation>
+std::unique_ptr<Base> make_lead_aggregation(size_type offset);
+
+/**
+ * @brief Factory to create an aggregation base on UDF for PTX or CUDA
+ *
+ * @param[in] type: either udf_type::PTX or udf_type::CUDA
+ * @param[in] user_defined_aggregator A string containing the aggregator code
+ * @param[in] output_type expected output type
+ *
+ * @return aggregation unique pointer housing user_defined_aggregator string.
+ */
+template <typename Base = aggregation>
+std::unique_ptr<Base> make_udf_aggregation(udf_type type,
+                                           std::string const& user_defined_aggregator,
+                                           data_type output_type);
+
 /**
  * @brief Factory to create a MERGE_LISTS aggregation.
  *
@@ -308,27 +343,44 @@ template <typename Base = aggregation>
 std::unique_ptr<Base> make_merge_sets_aggregation(null_equality nulls_equal = null_equality::EQUAL,
                                                   nan_equality nans_equal = nan_equality::UNEQUAL);
 
-/// Factory to create a LAG aggregation
-template <typename Base = aggregation>
-std::unique_ptr<Base> make_lag_aggregation(size_type offset);
-
-/// Factory to create a LEAD aggregation
+/**
+ * @brief Factory to create a MERGE_VARIANCES aggregation
+ *
+ * This aggregation is designed specificly to perform distributed computing of `VARIANCE`
+ * aggregation. The partial results input to this aggregation is generated by two groupby
+ * aggregations: `VARIANCE` and `COUNT_VALID`.
+ *
+ * In order to use this aggregation, the `aggregation_request` array input to `groupby::aggregate`
+ * must contain at least two requests:
+ *  - A request for `COLLECT_LIST` aggregation to collect the partial results of `COUNT_VALID`
+ *  - This `MERGE_VARIANCES` request, which must be given AFTER the request above so that it can
+ *    access the cached results generated by that request
+ *
+ * For a merging operation that is not a final merge (i.e., its outputs will be used as input to
+ * perform another `MERGE_VARIANCES` aggregation), a `SUM` aggregation must also be added to the
+ * same request for `COLLECT_LIST` above to produce the merged values for `COUNT_VALID`.
+ *
+ * Since the partial results output from `VARIANCE` and `COUNT_VALID` do not contain nulls, the
+ * input values columns to those two requests must be non-nullable.
+ *
+ * @param ddof Delta degrees of freedom. The divisor used in calculation of `variance` is
+ *        `N - ddof`, where `N` is the population size.
+ */
 template <typename Base = aggregation>
-std::unique_ptr<Base> make_lead_aggregation(size_type offset);
+std::unique_ptr<Base> make_merge_variances_aggregation(size_type ddof = 1);
 
 /**
- * @brief Factory to create an aggregation base on UDF for PTX or CUDA
+ * @brief Factory to create a MERGE_STDS aggregation
  *
- * @param[in] type: either udf_type::PTX or udf_type::CUDA
- * @param[in] user_defined_aggregator A string containing the aggregator code
- * @param[in] output_type expected output type
+ * This aggregation is designed specificly to perform distributed computing of `STD`
+ * aggregation. The partial results input to this aggregation and its usage are the same as of
+ * `MERGE_VARIANCES` aggregation.
  *
- * @return aggregation unique pointer housing user_defined_aggregator string.
+ * @param ddof Delta degrees of freedom. The divisor used in calculation of `variance` is
+ *        `N - ddof`, where `N` is the population size.
  */
 template <typename Base = aggregation>
-std::unique_ptr<Base> make_udf_aggregation(udf_type type,
-                                           std::string const& user_defined_aggregator,
-                                           data_type output_type);
+std::unique_ptr<Base> make_merge_stds_aggregation(size_type ddof = 1);
 
 /** @} */  // end of group
 }  // namespace cudf