Merge pull request #1476 from finos/stddev

Add standard deviation and variance aggregates

Author: texodus

cpp/perspective/src/cpp/aggspec.cpp

@@ -193,6 +193,12 @@ t_aggspec::agg_str() const {
         case AGGTYPE_PCT_SUM_GRAND_TOTAL: {
             return "pct_sum_grand_total";
         }
+        case AGGTYPE_VARIANCE: {
+            return "variance";
+        }
+        case AGGTYPE_STANDARD_DEVIATION: {
+            return "stddev";
+        }
         default: {
             PSP_COMPLAIN_AND_ABORT("Unknown agg type");
             return "unknown";
@@ -315,10 +321,8 @@ t_aggspec::get_output_specs(const t_schema& schema) const {
         case AGGTYPE_COUNT: {
             return mk_col_name_type_vec(name(), DTYPE_INT64);
         }
+        case AGGTYPE_MEAN:
         case AGGTYPE_MEAN_BY_COUNT:
-        case AGGTYPE_MEAN: {
-            return mk_col_name_type_vec(name(), DTYPE_F64PAIR);
-        }
         case AGGTYPE_WEIGHTED_MEAN: {
             return mk_col_name_type_vec(name(), DTYPE_F64PAIR);
         }
@@ -327,7 +331,9 @@ t_aggspec::get_output_specs(const t_schema& schema) const {
         }
         case AGGTYPE_SCALED_DIV:
         case AGGTYPE_SCALED_ADD:
-        case AGGTYPE_SCALED_MUL: {
+        case AGGTYPE_SCALED_MUL:
+        case AGGTYPE_VARIANCE:
+        case AGGTYPE_STANDARD_DEVIATION: {
             return mk_col_name_type_vec(name(), DTYPE_FLOAT64);
         }
         case AGGTYPE_UDF_COMBINER:

cpp/perspective/src/cpp/base.cpp

@@ -481,8 +481,14 @@ str_to_aggtype(const std::string& str) {
         return t_aggtype::AGGTYPE_UDF_COMBINER;
     } else if (str.find("udf_reducer_") != std::string::npos) {
         return t_aggtype::AGGTYPE_UDF_REDUCER;
+    } else if (str =="var" || str == "variance") {
+        return t_aggtype::AGGTYPE_VARIANCE;
+    } else if (str == "stddev" || str == "standard deviation") {
+        return t_aggtype::AGGTYPE_STANDARD_DEVIATION;
     } else {
-        PSP_COMPLAIN_AND_ABORT("Encountered unknown aggregate operation.");
+        std::stringstream ss;
+        ss << "Encountered unknown aggregate operation: '" << str << "'" << std::endl;
+        PSP_COMPLAIN_AND_ABORT(ss.str());
         // use any as default
         return t_aggtype::AGGTYPE_ANY;
     }

cpp/perspective/src/cpp/config.cpp

@@ -195,6 +195,8 @@ t_config::setup(const std::vector<std::string>& detail_columns,
             case AGGTYPE_MUL:
             case AGGTYPE_DISTINCT_COUNT:
             case AGGTYPE_DISTINCT_LEAF:
+            case AGGTYPE_VARIANCE:
+            case AGGTYPE_STANDARD_DEVIATION:
                 m_has_pkey_agg = true;
                 break;
             default:

cpp/perspective/src/cpp/extract_aggregate.cpp

@@ -68,7 +68,9 @@ extract_aggregate(
         case AGGTYPE_JOIN:
         case AGGTYPE_IDENTITY:
         case AGGTYPE_DISTINCT_COUNT:
-        case AGGTYPE_DISTINCT_LEAF: {
+        case AGGTYPE_DISTINCT_LEAF:
+        case AGGTYPE_VARIANCE:
+        case AGGTYPE_STANDARD_DEVIATION: {
             t_tscalar rval = aggcol->get_scalar(ridx);
             return rval;
         } break;

cpp/perspective/src/cpp/sparse_tree.cpp

@@ -1397,6 +1397,43 @@ t_stree::update_agg_table(
                 if (!skip)
                     dst->set_scalar(dst_ridx, new_value);
             } break;
+            case AGGTYPE_VARIANCE:
+            case AGGTYPE_STANDARD_DEVIATION: {
+                old_value.set(dst->get_scalar(dst_ridx));
+
+                auto pkeys = get_pkeys(nidx);
+                std::vector<double> values;
+
+                read_column_from_gstate(
+                    gstate, expression_master_table, spec.get_dependencies()[0].name(), pkeys, values, false);
+
+                // Calculate the count, rolling mean, and sum of squares of
+                // differences from the current mean at each iteration.
+                double count = 0, mean = 0, m2 = 0;
+
+                for (double num : values) {
+                    count++;
+                    double next_mean = mean + (num - mean) / count;
+                    m2 += (num - mean) * (num - next_mean);
+                    mean = next_mean;
+                }
+
+                // Only calculate stddev for more than 1 element in the group.
+                if (count >= 2) {
+                    double value = m2 / count;
+
+                    if (spec.agg() == AGGTYPE_STANDARD_DEVIATION) {
+                        value = std::sqrt(value);
+                    }
+
+                    new_value.set(value);
+                    dst->set_scalar(dst_ridx, new_value);
+                    dst->set_valid(dst_ridx, true);
+                } else {
+                    dst->set_valid(dst_ridx, false);
+                }
+
+            } break;
             default: { PSP_COMPLAIN_AND_ABORT("Not implemented"); }
         } // end switch
 

cpp/perspective/src/cpp/view.cpp

@@ -554,6 +554,16 @@ View<CTX_T>::data_slice_to_arrow(
     for (auto cidx = start_col; cidx < end_col; ++cidx) {
         std::vector<t_tscalar> col_path = names.at(cidx);
         t_dtype dtype = get_column_dtype(cidx);
+
+        // mean and weighted mean uses DTYPE_F64PAIR on the aggtable, which is
+        // the dtype returned by get_column_dtype. However, in the output data
+        // slice they are DTYPE_FLOAT64 and the f64 pair is not exposed outside
+        // of the sparse tree. Thus, treat f64 pair as DTYPE_FLOAT64 for arrow
+        // serialization.
+        if (dtype == DTYPE_F64PAIR) {
+            dtype = DTYPE_FLOAT64;
+        }
+
         std::string name;
 
         if (sides() > 1) {
@@ -918,7 +928,9 @@ View<CTX_T>::_map_aggregate_types(
                 case AGGTYPE_MEAN_BY_COUNT:
                 case AGGTYPE_WEIGHTED_MEAN:
                 case AGGTYPE_PCT_SUM_PARENT:
-                case AGGTYPE_PCT_SUM_GRAND_TOTAL: {
+                case AGGTYPE_PCT_SUM_GRAND_TOTAL:
+                case AGGTYPE_VARIANCE:
+                case AGGTYPE_STANDARD_DEVIATION: {
                     return "float";
                 } break;
                 default: { return typestring; } break;

cpp/perspective/src/include/perspective/base.h

@@ -256,7 +256,9 @@ enum t_aggtype {
     AGGTYPE_DISTINCT_COUNT,
     AGGTYPE_DISTINCT_LEAF,
     AGGTYPE_PCT_SUM_PARENT,
-    AGGTYPE_PCT_SUM_GRAND_TOTAL
+    AGGTYPE_PCT_SUM_GRAND_TOTAL,
+    AGGTYPE_VARIANCE,
+    AGGTYPE_STANDARD_DEVIATION
 };
 
 PERSPECTIVE_EXPORT t_aggtype str_to_aggtype(const std::string& str);

examples/workspace-editing-python/package.json

@@ -7,7 +7,7 @@
         "start": "yarn webpack && yarn start:server",
         "start:client": "webpack-dev-server --open",
         "start:server": "PYTHONPATH=../../python/perspective python3 src/server.py",
-        "webpack": "webpack --colour"
+        "webpack": "webpack --color"
     },
     "keywords": [],
     "license": "Apache-2.0",

examples/workspace-editing-python/src/index.js

@@ -33,7 +33,7 @@ const worker = perspective.shared_worker();
  * tables just as you would on a locally created table.
  */
 const server_table = websocket.open_table("data_source_one");
-const server_view = await server_table.view();
+let server_view;
 
 // All viewers are based on the same table, which then feed edits back to a
 // table on the server with a schema.
@@ -55,6 +55,7 @@ const PORTS = [];
  */
 const datasource = async function() {
     const load_start = performance.now();
+    server_view = await server_table.view();
 
     // The API of the remote table/view are symmetric.
     const arrow = await server_view.to_arrow();

packages/perspective/index.d.ts

@@ -43,10 +43,12 @@ declare module "@finos/perspective" {
         MEDIAN = "median",
         PCT_SUM_PARENT = "pct sum parent",
         PCT_SUM_TOTAL = "pct sum grand total",
+        STANDARD_DEVIATION = "stddev",
         SUM = "sum",
         SUM_ABS = "sum abs",
         SUM_NOT_NULL = "sum not null",
-        UNIQUE = "unique"
+        UNIQUE = "unique",
+        VARIANCE = "var"
     }
 
     enum STRING_AGGREGATES {