address comments

Author: sycai

bigframes/core/compile/compiled.py

@@ -29,6 +29,7 @@
 from google.cloud import bigquery
 import pyarrow as pa
 
+from bigframes.core import utils
 import bigframes.core.compile.aggregate_compiler as agg_compiler
 import bigframes.core.compile.googlesql
 import bigframes.core.compile.ibis_types
@@ -597,6 +598,17 @@ def _convert_range_ordering_to_table_value(
     value_lookup: typing.Mapping[str, ibis_types.Value],
     ordering_column: OrderingExpression,
 ) -> ibis_types.Value:
+    """Converts the ordering for range windows to Ibis references.
+
+    Note that this method is different from `_convert_row_ordering_to_table_values` in
+    that it does not arrange null values. There are two reasons:
+    1. Manipulating null positions requires more than one ordering key, which is forbidden
+       by SQL window syntax for range rolling.
+    2. Pandas does not allow range rolling on timeseries with nulls.
+
+    Therefore, we opt for the simplest approach here: generate the simplest SQL and follow
+    the BigQuery engine behavior.
+    """
     expr = op_compiler.compile_expression(
         ordering_column.scalar_expression, value_lookup
     )
@@ -695,8 +707,14 @@ def _add_boundary(
 ) -> ibis_expr_builders.LegacyWindowBuilder:
     if isinstance(bounds, RangeWindowBounds):
         return ibis_window.range(
-            start=_to_ibis_boundary(bounds.start_micros),
-            end=_to_ibis_boundary(bounds.end_micros),
+            start=_to_ibis_boundary(
+                None
+                if bounds.start is None
+                else utils.timedelta_to_micros(bounds.start)
+            ),
+            end=_to_ibis_boundary(
+                None if bounds.end is None else utils.timedelta_to_micros(bounds.end)
+            ),
         )
     if isinstance(bounds, RowsWindowBounds):
         if bounds.start is not None or bounds.end is not None:

bigframes/core/window/ordering.py

@@ -0,0 +1,74 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from functools import singledispatch
+
+from bigframes.core import expression as ex
+from bigframes.core import nodes, ordering
+
+
+@singledispatch
+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,
+    or the ordering information is not available, return None.
+    """
+    return None
+
+
+@find_order_direction.register
+def _(root: nodes.OrderByNode, column_id: str):
+    if len(root.by) == 0:
+        # This is a no-op
+        return find_order_direction(root.child, column_id)
+
+    # Make sure the window key is the prefix of sorting keys.
+    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_order_direction.register
+def _(root: nodes.ReversedNode, column_id: str):
+    direction = find_order_direction(root.child, column_id)
+
+    if direction is None:
+        return None
+    return direction.reverse()
+
+
+@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_order_direction(root.child, alias_ref.ref.id.name)
+
+
+@find_order_direction.register
+def _(root: nodes.FilterNode, column_id: str):
+    return find_order_direction(root.child, column_id)
+
+
+@find_order_direction.register
+def _(root: nodes.InNode, column_id: str):
+    return find_order_direction(root.left_child, column_id)
+
+
+@find_order_direction.register
+def _(root: nodes.WindowOpNode, column_id: str):
+    return find_order_direction(root.child, column_id)

bigframes/core/window/rolling.py

@@ -15,7 +15,6 @@
 from __future__ import annotations
 
 import datetime
-from functools import singledispatch
 import typing
 
 import bigframes_vendored.pandas.core.window.rolling as vendored_pandas_rolling
@@ -24,8 +23,9 @@
 
 from bigframes import dtypes
 from bigframes.core import expression as ex
-from bigframes.core import log_adapter, nodes, ordering, window_spec
+from bigframes.core import log_adapter, ordering, window_spec
 import bigframes.core.blocks as blocks
+from bigframes.core.window import ordering as window_ordering
 import bigframes.operations.aggregations as agg_ops
 
 
@@ -140,7 +140,9 @@ def create_range_window(
     if index_dtypes[0] != dtypes.TIMESTAMP_DTYPE:
         raise ValueError("Index type should be timestamps with timezones")
 
-    order_direction = _find_order_direction(block.expr.node, block.index_columns[0])
+    order_direction = window_ordering.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."
@@ -157,48 +159,3 @@ def create_range_window(
         ),
     )
     return Window(block, spec, block.value_columns, is_series=is_series)
-
-
-@singledispatch
-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,
-    or the ordering information is not available, return None.
-    """
-    return None
-
-
-@_find_order_direction.register
-def _(root: nodes.OrderByNode, column_id: str):
-    if len(root.by) == 0:
-        return None
-
-    # Make sure the window key is the prefix of sorting keys.
-    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_order_direction.register
-def _(root: nodes.ReversedNode, column_id: str):
-    direction = _find_order_direction(root.child, column_id)
-
-    if direction is None:
-        return None
-    return direction.reverse()
-
-
-@_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_order_direction(root.child, alias_ref.ref.id.name)
-
-
-@_find_order_direction.register
-def _(root: nodes.FilterNode, column_id: str):
-    return _find_order_direction(root.child, column_id)

bigframes/core/window_spec.py

@@ -21,7 +21,6 @@
 import numpy as np
 import pandas as pd
 
-from bigframes.core import utils
 import bigframes.core.expression as ex
 import bigframes.core.identifiers as ids
 import bigframes.core.ordering as orderings
@@ -199,18 +198,6 @@ def from_timedelta_window(
         else:
             raise ValueError(f"Unsupported value for 'closed' parameter: {closed}")
 
-    @property
-    def start_micros(self) -> int | None:
-        if self.start is None:
-            return None
-        return utils.timedelta_to_micros(self.start)
-
-    @property
-    def end_micros(self) -> int | None:
-        if self.end is None:
-            return None
-        return utils.timedelta_to_micros(self.end)
-
     def __post_init__(self):
         if self.start is None:
             return

tests/system/small/test_window.py

@@ -260,16 +260,18 @@ def test_range_rolling_order_info_lookup(range_rolling_dfs):
     actual_result = (
         bf_df.set_index("ts_col")
         .sort_index(ascending=False)["int_col"]
+        .isin(bf_df["int_col"])
         .rolling(window="3s")
-        .min()
+        .count()
         .to_pandas()
     )
 
     expected_result = (
         pd_df.set_index("ts_col")
         .sort_index(ascending=False)["int_col"]
+        .isin(pd_df["int_col"])
         .rolling(window="3s")
-        .min()
+        .count()
     )
     pd.testing.assert_series_equal(
         actual_result, expected_result, check_dtype=False, check_index=False