add windowizable prop

Author: TrevorBergeron

bigframes/core/array_value.py

@@ -306,6 +306,9 @@ def compute_general_reduction(
         multiple input rows into a single scalar value per group. If grouping keys are
         provided, the operation is performed per group; otherwise, it is a global reduction.
 
+        Note: Intermediate aggregations (those that are inputs to further aggregations)
+        must be windowizable. Notably excluded are approx quantile, top count ops.
+
         Args:
             assignments (Sequence[ex.Expression]): A sequence of aggregation expressions
                 to be calculated.

bigframes/core/expression_factoring.py

@@ -134,6 +134,8 @@ def windowize(
 ) -> expression.Expression:
     def windowize_local(expr: expression.Expression):
         if isinstance(expr, agg_expressions.Aggregation):
+            if expr.op.can_be_windowized:
+                raise ValueError(f"Op: {expr.op} cannot be windowized.")
             return agg_expressions.WindowExpression(expr, window)
         if isinstance(expr, agg_expressions.WindowExpression):
             raise ValueError(f"Expression {expr} already windowed!")

bigframes/operations/aggregations.py

@@ -68,6 +68,11 @@ def order_independent(self):
     def output_type(self, *input_types: dtypes.ExpressionType) -> dtypes.ExpressionType:
         ...
 
+    @property
+    def can_be_windowized(self):
+        # this is more of an engine property, but will treat feasibility in bigquery sql as source of truth
+        return True
+
 
 @dataclasses.dataclass(frozen=True)
 class NullaryWindowOp(WindowOp):
@@ -257,6 +262,10 @@ def output_type(self, *input_types: dtypes.ExpressionType) -> dtypes.ExpressionT
             raise TypeError(f"Type {input_types[0]} is not orderable")
         return input_types[0]
 
+    @property
+    def can_be_windowized(self):
+        return False
+
 
 @dataclasses.dataclass(frozen=True)
 class ApproxTopCountOp(UnaryAggregateOp):
@@ -274,6 +283,10 @@ def output_type(self, *input_types: dtypes.ExpressionType) -> dtypes.ExpressionT
         ]
         return pd.ArrowDtype(pa.list_(pa.struct(fields)))
 
+    @property
+    def can_be_windowized(self):
+        return False
+
 
 @dataclasses.dataclass(frozen=True)
 class MeanOp(UnaryAggregateOp):