Added AggFormula in io.deephaven.api.agg.Aggregation, fixes #1699

pyintegration/deephaven2/agg.py

@@ -108,6 +108,21 @@ def first(cols: List[str]) -> Aggregation:
     return Aggregation(j_aggregation=_JAggregation.AggFirst(*cols))
 
 
+def formula(formula: str, formula_param: str, cols: List[str]) -> Aggregation:
+    """ Create a user defined formula aggregation.
+
+    Args:
+        formula (str): the user defined formula to apply to each group
+        formula_param (str): the parameter name within the formula
+        cols (List[str]): the columns to aggregate on, can be renaming expressions, i.e. "new_col = col"
+
+    Returns:
+        an aggregation
+    """
+
+    return Aggregation(j_aggregation=_JAggregation.AggFormula(formula, formula_param, *cols))
+
+
 def last(cols: List[str]) -> Aggregation:
     """ Create Last aggregation.
 

pyintegration/tests/test_table.py

@@ -5,7 +5,7 @@
 
 from deephaven2 import DHError, read_csv, empty_table, SortDirection
 from deephaven2.agg import sum_, weighted_avg, avg, pct, group, count_, first, last, max_, median, min_, std, abs_sum, \
-    var
+    var, formula
 from deephaven2.table import Table
 from tests.testbase import BaseTestCase
 
@@ -258,14 +258,15 @@ def test_agg_by(self):
         aggs = [sum_(cols=["SumC=c"]),
                 avg(cols=["AvgB = b", "AvgD = d"]),
                 pct(percentile=0.5, cols=["PctC = c"]),
-                weighted_avg(wcol="d", cols=["WavGD = d"])]
+                weighted_avg(wcol="d", cols=["WavGD = d"]),
+                formula(formula="min(each)", formula_param="each", cols=["MinA=a", "MinD=d"])]
 
         result_table = self.test_table.agg_by(aggs=aggs, by=["a"])
         self.assertEqual(result_table.size, num_distinct_a)
 
     def test_agg_by_2(self):
         test_table = empty_table(10)
-        test_table = test_table.update(["dumb=(int)(i/5)", "var=(int)i", "weights=(double)1.0/(i+1)"])
+        test_table = test_table.update(["grp_id=(int)(i/5)", "var=(int)i", "weights=(double)1.0/(i+1)"])
 
         aggs = [group(["aggGroup=var"]),
                 avg(["aggAvg=var"]),
@@ -282,8 +283,7 @@ def test_agg_by_2(self):
                 var(["aggVar=var"]),
                 weighted_avg("var", ["weights"])]
 
-        result_table = test_table.agg_by(aggs, ["dumb"])
-        # TODO: AggFormula - this is terrible
+        result_table = test_table.agg_by(aggs, ["grp_id"])
 
         self.assertGreaterEqual(result_table.size, 1)
 

table-api/src/main/java/io/deephaven/api/agg/Aggregation.java

@@ -2,6 +2,7 @@
 
 import io.deephaven.api.agg.NormalAggregations.Builder;
 import io.deephaven.api.agg.spec.AggSpec;
+import io.deephaven.api.agg.spec.AggSpecFormula;
 
 import java.io.Serializable;
 import java.util.Collection;
@@ -63,6 +64,10 @@ static Aggregation AggFirst(String... pairs) {
         return of(AggSpec.first(), pairs);
     }
 
+    static Aggregation AggFormula(String formula, String formulaParam, String... pairs) {
+        return of(AggSpec.formula(formula, formulaParam), pairs);
+    }
+
     static Aggregation AggGroup(String... pairs) {
         return of(AggSpec.group(), pairs);
     }