feat: implement AggCountWhere to support generic counting using Filter

engine/table/src/main/java/io/deephaven/engine/table/impl/by/CountWhereOperator.java

@@ -28,7 +28,7 @@
 
 /**
  * Implements a counting operator that counts the number of rows that pass a set of filters. Chunk data is accessed
- * through {@link RecordingInternalOperator recorder> instances.
+ * through {@link RecordingInternalOperator recorder} instances.
  */
 
 public class CountWhereOperator implements IterativeChunkedAggregationOperator {

engine/table/src/test/java/io/deephaven/engine/table/impl/select/FormulaKernelSample.java

@@ -5,6 +5,7 @@
 import java.lang.*;
 import java.util.*;
 import io.deephaven.base.string.cache.CompressedString;
+import io.deephaven.chunk.BooleanChunk;
 import io.deephaven.chunk.ByteChunk;
 import io.deephaven.chunk.CharChunk;
 import io.deephaven.chunk.Chunk;
@@ -14,6 +15,7 @@
 import io.deephaven.chunk.LongChunk;
 import io.deephaven.chunk.ObjectChunk;
 import io.deephaven.chunk.ShortChunk;
+import io.deephaven.chunk.WritableBooleanChunk;
 import io.deephaven.chunk.WritableByteChunk;
 import io.deephaven.chunk.WritableCharChunk;
 import io.deephaven.chunk.WritableChunk;

engine/table/src/test/java/io/deephaven/engine/table/impl/select/FormulaSample.java

@@ -5,6 +5,7 @@
 import java.lang.*;
 import java.util.*;
 import io.deephaven.base.string.cache.CompressedString;
+import io.deephaven.chunk.BooleanChunk;
 import io.deephaven.chunk.ByteChunk;
 import io.deephaven.chunk.CharChunk;
 import io.deephaven.chunk.Chunk;
@@ -14,6 +15,7 @@
 import io.deephaven.chunk.LongChunk;
 import io.deephaven.chunk.ObjectChunk;
 import io.deephaven.chunk.ShortChunk;
+import io.deephaven.chunk.WritableBooleanChunk;
 import io.deephaven.chunk.WritableByteChunk;
 import io.deephaven.chunk.WritableCharChunk;
 import io.deephaven.chunk.WritableChunk;

proto/proto-backplane-grpc/src/main/proto/deephaven_core/proto/table.proto

@@ -1059,7 +1059,7 @@ message Aggregation {
     // The output column name
     string column_name = 1;
 
-    repeated Condition filters = 2;
+    repeated string filters = 2;
   }
 
   message AggregationRowKey {

py/client/pydeephaven/agg.py

@@ -16,6 +16,7 @@
 _GrpcAggregation = table_pb2.Aggregation
 _GrpcAggregationColumns = _GrpcAggregation.AggregationColumns
 _GrpcAggregationCount = _GrpcAggregation.AggregationCount
+_GrpcAggregationCountWhere = _GrpcAggregation.AggregationCountWhere
 _GrpcAggregationFormula = _GrpcAggregation.AggregationFormula
 _GrpcAggregationPartition = _GrpcAggregation.AggregationPartition
 _GrpcAggSpec = table_pb2.AggSpec
@@ -51,6 +52,17 @@ def make_grpc_message(self) -> _GrpcAggregation:
         agg_count = _GrpcAggregationCount(column_name=self.col)
         return _GrpcAggregation(count=agg_count)
 
+
+@dataclass
+class _AggregationCountWhere(Aggregation):
+    col: str
+    filters: Union[str, List[str]]
+
+    def make_grpc_message(self) -> _GrpcAggregation:
+        agg_count_where = _GrpcAggregationCountWhere(column_name=self.col, filters=to_list(self.filters))
+        return _GrpcAggregation(count_where=agg_count_where)
+
+
 @dataclass
 class _AggregationFormula(Aggregation):
     selectable: _GrpcSelectable
@@ -59,6 +71,7 @@ def make_grpc_message(self) -> _GrpcAggregation:
         agg_formula = _GrpcAggregationFormula(selectable=self.selectable)
         return _GrpcAggregation(formula=agg_formula)
 
+
 @dataclass
 class _AggregationPartition(Aggregation):
     col: str
@@ -137,6 +150,18 @@ def count_(col: str) -> Aggregation:
     return _AggregationCount(col=col)
 
 
+def count_where(col: str, filters: Union[str, List[str]]) -> Aggregation:
+    """Creates a Count aggregation. This is not supported in 'Table.agg_all_by'.
+
+    Args:
+        col (str): the column to hold the counts of each distinct group
+
+    Returns:
+        an aggregation
+    """
+    return _AggregationCountWhere(col=col, filters=to_list(filters))
+
+
 def partition(col: str, include_by_columns: bool = True) -> Aggregation:
     """Creates a Partition aggregation. This is not supported in 'Table.agg_all_by'.
 

py/client/tests/test_table.py

@@ -10,7 +10,7 @@
 
 from pydeephaven import DHError
 from pydeephaven import SortDirection
-from pydeephaven.agg import sum_, avg, pct, weighted_avg, count_, partition, median, unique, count_distinct, distinct, formula
+from pydeephaven.agg import sum_, avg, pct, weighted_avg, count_, count_where, partition, median, unique, count_distinct, distinct, formula
 from pydeephaven.table import Table
 from tests.testbase import BaseTestCase
 
@@ -247,6 +247,10 @@ def test_agg_by(self):
                 pct(percentile=0.5, cols=["PctC = c"]),
                 weighted_avg(wcol="d", cols=["WavGD = d"]),
                 count_(col="ca"),
+                count_where(col="count_where1", filters="a > 5"),
+                count_where("agg_count_where_1", "a > 100"),
+                count_where("agg_count_where_2", ["a > 100", "b < 250"]),
+                count_where("agg_count_where_3", "a <= 100 || b >= 250"),
                 partition(col="aggPartition"),
                 formula(formula="min(x)", formula_param="x", cols=["min_a=a", "min_b=b"]),
                 formula(formula="avg(x)", formula_param="x", cols=["avg_c=c", "avg_d=d"]),

server/src/main/java/io/deephaven/server/table/ops/AggregationAdapter.java

@@ -86,7 +86,7 @@ public static Count adapt(AggregationCount count) {
     }
 
     public static CountWhere adapt(AggregationCountWhere count) {
-        return Aggregation.AggCountWhere(count.getColumnName());
+        return Aggregation.AggCountWhere(count.getColumnName(), count.getFiltersList().toArray(String[]::new));
     }
 
     public static Formula adapt(AggregationFormula formula) {
@@ -180,7 +180,7 @@ public void visit(Count count) {
 
         public void visit(CountWhere countWhere) {
             add(
-                    TypeCase.COUNT,
+                    TypeCase.COUNT_WHERE,
                     AggregationCountWhere.class,
                     CountWhere.class,
                     GrpcErrorHelper::checkHasNoUnknownFieldsRecursive,

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

@@ -733,6 +733,7 @@ static void visitAll(Visitor visitor) {
         visitor.visit((ColumnAggregation) null);
         visitor.visit((ColumnAggregations) null);
         visitor.visit((Count) null);
+        visitor.visit((CountWhere) null);
         visitor.visit((FirstRowKey) null);
         visitor.visit((LastRowKey) null);
         visitor.visit((Partition) null);

table-api/src/test/java/io/deephaven/api/agg/AggregationTest.java

@@ -56,6 +56,11 @@ public void visit(Count c) {
             ++count;
         }
 
+        @Override
+        public void visit(CountWhere countWhere) {
+            ++count;
+        }
+
         @Override
         public void visit(FirstRowKey firstRowKey) {
             ++count;