differentiate range bound and unbounded window

Author: sycai

bigframes/core/array_value.py

@@ -412,7 +412,7 @@ def project_window_op(
         skip_reproject_unsafe: skips the reprojection step, can be used when performing many non-dependent window operations, user responsible for not nesting window expressions, or using outputs as join, filter or aggregation keys before a reprojection
         """
         # TODO: Support non-deterministic windowing
-        if window_spec.row_bounded or not op.order_independent:
+        if window_spec.is_row_bounded or not op.order_independent:
             if self.node.order_ambiguous and not self.session._strictly_ordered:
                 if not self.session._allows_ambiguity:
                     raise ValueError(

bigframes/core/compile/compiled.py

@@ -463,6 +463,9 @@ def project_window_op(
                 never_skip_nulls=never_skip_nulls,
             )
 
+        if expression.op.order_independent and window_spec.is_unbounded:
+            # notably percentile_cont does not support ordering clause
+            window_spec = window_spec.without_order()
         window = self._ibis_window_from_spec(window_spec)
         bindings = {col: self._get_ibis_column(col) for col in self.column_ids}
 
@@ -538,21 +541,29 @@ def _ibis_window_from_spec(self, window_spec: WindowSpec):
         # 1. Order-independent op (aggregation, cut, rank) with unbound window - no ordering clause needed
         # 2. Order-independent op (aggregation, cut, rank) with range window - use ordering clause, ties allowed
         # 3. Order-depedenpent op (navigation functions, array_agg) or rows bounds - use total row order to break ties.
-        if not window_spec.ordering:
-            # If window spec has following or preceding bounds, we need to apply an unambiguous ordering.
-            raise ValueError("No ordering provided for ordered analytic function")
+        if window_spec.is_row_bounded:
+            if not window_spec.ordering:
+                # If window spec has following or preceding bounds, we need to apply an unambiguous ordering.
+                raise ValueError("No ordering provided for ordered analytic function")
+            order_by = _convert_row_ordering_to_table_values(
+                self._column_names,
+                window_spec.ordering,
+            )
 
-        if window_spec.row_bounded:
+        elif window_spec.is_range_bounded:
+            order_by = _convert_range_ordering_to_table_value(
+                self._column_names,
+                window_spec.ordering[0],
+            )
+        # The rest if branches are for unbounded windows
+        elif window_spec.ordering:
+            # Unbound grouping window. Suitable for aggregations but not for analytic function application.
             order_by = _convert_row_ordering_to_table_values(
                 self._column_names,
                 window_spec.ordering,
             )
         else:
-            order_by = [
-                _convert_range_ordering_to_table_value(
-                    self._column_names, window_spec.ordering[0]
-                )
-            ]
+            order_by = None
 
         window = bigframes_vendored.ibis.window(order_by=order_by, group_by=group_by)
         if window_spec.bounds is not None:

bigframes/core/nodes.py

@@ -1358,7 +1358,7 @@ def _validate(self):
         """Validate the local data in the node."""
         # Since inner order and row bounds are coupled, rank ops can't be row bounded
         assert (
-            not self.window_spec.row_bounded
+            not self.window_spec.is_row_bounded
         ) or self.expression.op.implicitly_inherits_order
         assert all(ref in self.child.ids for ref in self.expression.column_references)
 
@@ -1420,7 +1420,7 @@ def inherits_order(self) -> bool:
         op_inherits_order = (
             not self.expression.op.order_independent
         ) and self.expression.op.implicitly_inherits_order
-        return op_inherits_order or self.window_spec.row_bounded
+        return op_inherits_order or self.window_spec.is_row_bounded
 
     @property
     def additive_base(self) -> BigFrameNode:

bigframes/core/rewrite/windows.py

@@ -24,7 +24,7 @@ def rewrite_range_rolling(node: nodes.BigFrameNode) -> nodes.BigFrameNode:
     if not isinstance(node, nodes.WindowOpNode):
         return node
 
-    if node.window_spec.row_bounded:
+    if not node.window_spec.is_range_bounded:
         return node
 
     if len(node.window_spec.ordering) != 1:

bigframes/core/window/rolling.py

@@ -140,7 +140,7 @@ def create_range_window(
     if index_dtypes[0] != dtypes.TIMESTAMP_DTYPE:
         raise ValueError("Index type should be timestamps with timezones")
 
-    order_direction = _find_ordering(block.expr.node, block.index_columns[0])
+    order_direction = _find_order_direction(block.expr.node, block.index_columns[0])
     if order_direction is None:
         raise ValueError(
             "The index might not be in a monotonic order. Please sort the index before rolling."
@@ -160,7 +160,7 @@ def create_range_window(
 
 
 @singledispatch
-def _find_ordering(
+def _find_order_direction(
     root: nodes.BigFrameNode, column_id: str
 ) -> ordering.OrderingDirection | None:
     """Returns the order of the given column with tree traversal. If the column cannot be found,
@@ -169,32 +169,37 @@ def _find_ordering(
     return None
 
 
-@_find_ordering.register
+@_find_order_direction.register
 def _(root: nodes.OrderByNode, column_id: str):
-    for order_expr in root.by:
-        scalar_expr = order_expr.scalar_expression
-        if isinstance(scalar_expr, ex.DerefOp) and scalar_expr.id.name == column_id:
-            return order_expr.direction
+    if len(root.by) == 0:
+        return None
+
+    # Only when the column is used as the first ordering key
+    # does it guarantee that its values are in a monotonic order.
+    order_expr = root.by[0]
+    scalar_expr = order_expr.scalar_expression
+    if isinstance(scalar_expr, ex.DerefOp) and scalar_expr.id.name == column_id:
+        return order_expr.direction
 
     return None
 
 
-@_find_ordering.register
+@_find_order_direction.register
 def _(root: nodes.ReversedNode, column_id: str):
-    direction = _find_ordering(root.child, column_id)
+    direction = _find_order_direction(root.child, column_id)
 
     if direction is None:
         return None
     return direction.reverse()
 
 
-@_find_ordering.register
+@_find_order_direction.register
 def _(root: nodes.SelectionNode, column_id: str):
     for alias_ref in root.input_output_pairs:
         if alias_ref.id.name == column_id:
-            return _find_ordering(root.child, alias_ref.ref.id.name)
+            return _find_order_direction(root.child, alias_ref.ref.id.name)
 
 
-@_find_ordering.register
+@_find_order_direction.register
 def _(root: nodes.FilterNode, column_id: str):
-    return _find_ordering(root.child, column_id)
+    return _find_order_direction(root.child, column_id)

bigframes/core/window_spec.py

@@ -240,7 +240,7 @@ class WindowSpec:
     min_periods: int = 0
 
     @property
-    def row_bounded(self):
+    def is_row_bounded(self):
         """
         Whether the window is bounded by row offsets.
 
@@ -249,6 +249,26 @@ def row_bounded(self):
         """
         return isinstance(self.bounds, RowsWindowBounds)
 
+    @property
+    def is_range_bounded(self):
+        """
+        Whether the window is bounded by range offsets.
+
+        This is relevant for determining whether the window requires a total order
+        to calculate deterministically.
+        """
+        return isinstance(self.bounds, RangeWindowBounds)
+
+    @property
+    def is_unbounded(self):
+        """
+        Whether the window is unbounded.
+
+        This is relevant for determining whether the window requires a total order
+        to calculate deterministically.
+        """
+        return self.bounds is None
+
     @property
     def all_referenced_columns(self) -> Set[ids.ColumnId]:
         """
@@ -261,7 +281,7 @@ def all_referenced_columns(self) -> Set[ids.ColumnId]:
 
     def without_order(self) -> WindowSpec:
         """Removes ordering clause if ordering isn't required to define bounds."""
-        if self.row_bounded:
+        if self.is_row_bounded:
             raise ValueError("Cannot remove order from row-bounded window")
         return replace(self, ordering=())
 

tests/system/small/test_window.py

@@ -228,7 +228,7 @@ def test_dataframe_window_agg_ops(scalars_dfs, windowing, agg_op):
 
 @pytest.mark.parametrize("closed", ["left", "right", "both", "neither"])
 @pytest.mark.parametrize(
-    "window",  # skipped numpy because Pandas does not support it.
+    "window",  # skipped numpy timedelta because Pandas does not support it.
     [pd.Timedelta("3s"), datetime.timedelta(seconds=3), "3s"],
 )
 @pytest.mark.parametrize("ascending", [True, False])