Skip to content

Commit 4b0cf57

Browse files
authored
chore: allow PRECEDING and FOLLOWING to appear on both side of BETWEEN when windowing (#1507)
* chore: allow PRECEDING and FOLLOWING to appear on both side of BETWEEN when windowing * fix lint * Simplify the code by using the sign of the value for PRECEDING/FOLLOWING * fix lint * fix mypy * polish doc * remove float support for range rolling because Pandas does not support that
1 parent d553fa2 commit 4b0cf57

File tree

8 files changed

+128
-59
lines changed

8 files changed

+128
-59
lines changed

bigframes/core/block_transforms.py

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -213,8 +213,8 @@ def _interpolate_column(
213213
if interpolate_method not in ["linear", "nearest", "ffill"]:
214214
raise ValueError("interpolate method not supported")
215215
window_ordering = (ordering.OrderingExpression(ex.deref(x_values)),)
216-
backwards_window = windows.rows(following=0, ordering=window_ordering)
217-
forwards_window = windows.rows(preceding=0, ordering=window_ordering)
216+
backwards_window = windows.rows(end=0, ordering=window_ordering)
217+
forwards_window = windows.rows(start=0, ordering=window_ordering)
218218

219219
# Note, this method may
220220
block, notnull = block.apply_unary_op(column, ops.notnull_op)
@@ -450,7 +450,7 @@ def rank(
450450
)
451451
if method == "dense"
452452
else windows.rows(
453-
following=0, ordering=window_ordering, grouping_keys=grouping_cols
453+
end=0, ordering=window_ordering, grouping_keys=grouping_cols
454454
),
455455
skip_reproject_unsafe=(col != columns[-1]),
456456
)

bigframes/core/compile/compiled.py

Lines changed: 34 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,9 @@
2121
import bigframes_vendored.ibis
2222
import bigframes_vendored.ibis.backends.bigquery.backend as ibis_bigquery
2323
import bigframes_vendored.ibis.common.deferred as ibis_deferred # type: ignore
24+
from bigframes_vendored.ibis.expr import builders as ibis_expr_builders
2425
import bigframes_vendored.ibis.expr.datatypes as ibis_dtypes
26+
from bigframes_vendored.ibis.expr.operations import window as ibis_expr_window
2527
import bigframes_vendored.ibis.expr.operations as ibis_ops
2628
import bigframes_vendored.ibis.expr.types as ibis_types
2729
import pandas
@@ -551,20 +553,9 @@ def _ibis_window_from_spec(self, window_spec: WindowSpec):
551553
# Unbound grouping window. Suitable for aggregations but not for analytic function application.
552554
order_by = None
553555

554-
bounds = window_spec.bounds
555556
window = bigframes_vendored.ibis.window(order_by=order_by, group_by=group_by)
556-
if bounds is not None:
557-
if isinstance(bounds, RangeWindowBounds):
558-
window = window.preceding_following(
559-
bounds.preceding, bounds.following, how="range"
560-
)
561-
if isinstance(bounds, RowsWindowBounds):
562-
if bounds.preceding is not None or bounds.following is not None:
563-
window = window.preceding_following(
564-
bounds.preceding, bounds.following, how="rows"
565-
)
566-
else:
567-
raise ValueError(f"unrecognized window bounds {bounds}")
557+
if window_spec.bounds is not None:
558+
return _add_boundary(window_spec.bounds, window)
568559
return window
569560

570561

@@ -681,3 +672,33 @@ def _as_groupable(value: ibis_types.Value):
681672
return scalar_op_compiler.to_json_string(value)
682673
else:
683674
return value
675+
676+
677+
def _to_ibis_boundary(
678+
boundary: Optional[int],
679+
) -> Optional[ibis_expr_window.WindowBoundary]:
680+
if boundary is None:
681+
return None
682+
return ibis_expr_window.WindowBoundary(
683+
abs(boundary), preceding=boundary <= 0 # type:ignore
684+
)
685+
686+
687+
def _add_boundary(
688+
bounds: typing.Union[RowsWindowBounds, RangeWindowBounds],
689+
ibis_window: ibis_expr_builders.LegacyWindowBuilder,
690+
) -> ibis_expr_builders.LegacyWindowBuilder:
691+
if isinstance(bounds, RangeWindowBounds):
692+
return ibis_window.range(
693+
start=_to_ibis_boundary(bounds.start),
694+
end=_to_ibis_boundary(bounds.end),
695+
)
696+
if isinstance(bounds, RowsWindowBounds):
697+
if bounds.start is not None or bounds.end is not None:
698+
return ibis_window.rows(
699+
start=_to_ibis_boundary(bounds.start),
700+
end=_to_ibis_boundary(bounds.end),
701+
)
702+
return ibis_window
703+
else:
704+
raise ValueError(f"unrecognized window bounds {bounds}")

bigframes/core/compile/polars/compiler.py

Lines changed: 15 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -16,9 +16,10 @@
1616
import dataclasses
1717
import functools
1818
import itertools
19-
from typing import cast, Sequence, Tuple, TYPE_CHECKING
19+
from typing import cast, Optional, Sequence, Tuple, TYPE_CHECKING, Union
2020

2121
import bigframes.core
22+
from bigframes.core import window_spec
2223
import bigframes.core.expression as ex
2324
import bigframes.core.guid as guid
2425
import bigframes.core.nodes as nodes
@@ -366,23 +367,8 @@ def compile_window(self, node: nodes.WindowOpNode):
366367
indexed_df = df.with_row_index(index_col_name)
367368
if len(window.grouping_keys) == 0: # rolling-only window
368369
# https://docs.pola.rs/api/python/stable/reference/dataframe/api/polars.DataFrame.rolling.html
369-
finite = (
370-
window.bounds.preceding is not None
371-
and window.bounds.following is not None
372-
)
373-
offset_n = (
374-
None
375-
if window.bounds.preceding is None
376-
else -window.bounds.preceding
377-
)
378-
# collecting height is a massive kludge
379-
period_n = (
380-
df.collect().height
381-
if not finite
382-
else cast(int, window.bounds.preceding)
383-
+ cast(int, window.bounds.following)
384-
+ 1
385-
)
370+
offset_n = window.bounds.start
371+
period_n = _get_period(window.bounds) or df.collect().height
386372
results = indexed_df.rolling(
387373
index_column=index_col_name,
388374
period=f"{period_n}i",
@@ -395,3 +381,14 @@ def compile_window(self, node: nodes.WindowOpNode):
395381
# polars is columnar, so this is efficient
396382
# TODO: why can't just add columns?
397383
return pl.concat([df, results], how="horizontal")
384+
385+
386+
def _get_period(
387+
bounds: Union[window_spec.RowsWindowBounds, window_spec.RangeWindowBounds]
388+
) -> Optional[int]:
389+
"""Returns None if the boundary is infinite."""
390+
if bounds.start is None or bounds.end is None:
391+
return None
392+
393+
# collecting height is a massive kludge
394+
return bounds.end - bounds.start + 1

bigframes/core/groupby/__init__.py

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -309,8 +309,8 @@ def rolling(self, window: int, min_periods=None) -> windows.Window:
309309
# To get n size window, need current row and n-1 preceding rows.
310310
window_spec = window_specs.rows(
311311
grouping_keys=tuple(self._by_col_ids),
312-
preceding=window - 1,
313-
following=0,
312+
start=-(window - 1),
313+
end=0,
314314
min_periods=min_periods or window,
315315
)
316316
block = self._block.order_by(
@@ -742,8 +742,8 @@ def rolling(self, window: int, min_periods=None) -> windows.Window:
742742
# To get n size window, need current row and n-1 preceding rows.
743743
window_spec = window_specs.rows(
744744
grouping_keys=tuple(self._by_col_ids),
745-
preceding=window - 1,
746-
following=0,
745+
start=-(window - 1),
746+
end=0,
747747
min_periods=min_periods or window,
748748
)
749749
block = self._block.order_by(

bigframes/core/window_spec.py

Lines changed: 37 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -52,8 +52,8 @@ def unbound(
5252
### Rows-based Windows
5353
def rows(
5454
grouping_keys: Tuple[str, ...] = (),
55-
preceding: Optional[int] = None,
56-
following: Optional[int] = None,
55+
start: Optional[int] = None,
56+
end: Optional[int] = None,
5757
min_periods: int = 0,
5858
ordering: Tuple[orderings.OrderingExpression, ...] = (),
5959
) -> WindowSpec:
@@ -63,18 +63,23 @@ def rows(
6363
Args:
6464
grouping_keys:
6565
Columns ids of grouping keys
66-
preceding:
67-
number of preceding rows to include. If None, include all preceding rows
66+
start:
67+
The window's starting boundary relative to the current row. For example, "-1" means one row prior
68+
"1" means one row after, and "0" means the current row. If None, the window is unbounded from the start.
6869
following:
69-
number of following rows to include. If None, include all following rows
70+
The window's ending boundary relative to the current row. For example, "-1" means one row prior
71+
"1" means one row after, and "0" means the current row. If None, the window is unbounded until the end.
7072
min_periods (int, default 0):
7173
Minimum number of input rows to generate output.
7274
ordering:
7375
Ordering to apply on top of based dataframe ordering
7476
Returns:
7577
WindowSpec
7678
"""
77-
bounds = RowsWindowBounds(preceding=preceding, following=following)
79+
bounds = RowsWindowBounds(
80+
start=start,
81+
end=end,
82+
)
7883
return WindowSpec(
7984
grouping_keys=tuple(map(ex.deref, grouping_keys)),
8085
bounds=bounds,
@@ -97,7 +102,7 @@ def cumulative_rows(
97102
Returns:
98103
WindowSpec
99104
"""
100-
bounds = RowsWindowBounds(following=0)
105+
bounds = RowsWindowBounds(end=0)
101106
return WindowSpec(
102107
grouping_keys=tuple(map(ex.deref, grouping_keys)),
103108
bounds=bounds,
@@ -119,7 +124,7 @@ def inverse_cumulative_rows(
119124
Returns:
120125
WindowSpec
121126
"""
122-
bounds = RowsWindowBounds(preceding=0)
127+
bounds = RowsWindowBounds(start=0)
123128
return WindowSpec(
124129
grouping_keys=tuple(map(ex.deref, grouping_keys)),
125130
bounds=bounds,
@@ -132,18 +137,35 @@ def inverse_cumulative_rows(
132137

133138
@dataclass(frozen=True)
134139
class RowsWindowBounds:
135-
preceding: Optional[int] = None
136-
following: Optional[int] = None
137-
140+
start: Optional[int] = None
141+
end: Optional[int] = None
138142

139-
# TODO: Expand to datetime offsets
140-
OffsetType = Union[float, int]
143+
def __post_init__(self):
144+
if self.start is None:
145+
return
146+
if self.end is None:
147+
return
148+
if self.start > self.end:
149+
raise ValueError(
150+
f"Invalid window: start({self.start}) is greater than end({self.end})"
151+
)
141152

142153

143154
@dataclass(frozen=True)
144155
class RangeWindowBounds:
145-
preceding: Optional[OffsetType] = None
146-
following: Optional[OffsetType] = None
156+
# TODO(b/388916840) Support range rolling on timeseries with timedeltas.
157+
start: Optional[int] = None
158+
end: Optional[int] = None
159+
160+
def __post_init__(self):
161+
if self.start is None:
162+
return
163+
if self.end is None:
164+
return
165+
if self.start > self.end:
166+
raise ValueError(
167+
f"Invalid window: start({self.start}) is greater than end({self.end})"
168+
)
147169

148170

149171
@dataclass(frozen=True)

bigframes/dataframe.py

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -2428,12 +2428,12 @@ def replace(
24282428

24292429
@validations.requires_ordering()
24302430
def ffill(self, *, limit: typing.Optional[int] = None) -> DataFrame:
2431-
window = windows.rows(preceding=limit, following=0)
2431+
window = windows.rows(start=None if limit is None else -limit, end=0)
24322432
return self._apply_window_op(agg_ops.LastNonNullOp(), window)
24332433

24342434
@validations.requires_ordering()
24352435
def bfill(self, *, limit: typing.Optional[int] = None) -> DataFrame:
2436-
window = windows.rows(preceding=0, following=limit)
2436+
window = windows.rows(start=0, end=limit)
24372437
return self._apply_window_op(agg_ops.FirstNonNullOp(), window)
24382438

24392439
def isin(self, values) -> DataFrame:
@@ -3310,7 +3310,7 @@ def _perform_join_by_index(
33103310
def rolling(self, window: int, min_periods=None) -> bigframes.core.window.Window:
33113311
# To get n size window, need current row and n-1 preceding rows.
33123312
window_def = windows.rows(
3313-
preceding=window - 1, following=0, min_periods=min_periods or window
3313+
start=-(window - 1), end=0, min_periods=min_periods or window
33143314
)
33153315
return bigframes.core.window.Window(
33163316
self._block, window_def, self._block.value_columns

bigframes/series.py

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -544,15 +544,15 @@ def cumsum(self) -> Series:
544544

545545
@validations.requires_ordering()
546546
def ffill(self, *, limit: typing.Optional[int] = None) -> Series:
547-
window = windows.rows(preceding=limit, following=0)
547+
window = windows.rows(start=None if limit is None else -limit, end=0)
548548
return self._apply_window_op(agg_ops.LastNonNullOp(), window)
549549

550550
pad = ffill
551551
pad.__doc__ = inspect.getdoc(vendored_pandas_series.Series.ffill)
552552

553553
@validations.requires_ordering()
554554
def bfill(self, *, limit: typing.Optional[int] = None) -> Series:
555-
window = windows.rows(preceding=0, following=limit)
555+
window = windows.rows(start=0, end=limit)
556556
return self._apply_window_op(agg_ops.FirstNonNullOp(), window)
557557

558558
@validations.requires_ordering()
@@ -1441,7 +1441,7 @@ def sort_index(self, *, axis=0, ascending=True, na_position="last") -> Series:
14411441
def rolling(self, window: int, min_periods=None) -> bigframes.core.window.Window:
14421442
# To get n size window, need current row and n-1 preceding rows.
14431443
window_spec = windows.rows(
1444-
preceding=window - 1, following=0, min_periods=min_periods or window
1444+
start=-(window - 1), end=0, min_periods=min_periods or window
14451445
)
14461446
return bigframes.core.window.Window(
14471447
self._block, window_spec, self._block.value_columns, is_series=True

tests/unit/core/test_windowspec.py

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,29 @@
1+
# Copyright 2025 Google LLC
2+
#
3+
# Licensed under the Apache License, Version 2.0 (the "License");
4+
# you may not use this file except in compliance with the License.
5+
# You may obtain a copy of the License at
6+
#
7+
# http://www.apache.org/licenses/LICENSE-2.0
8+
#
9+
# Unless required by applicable law or agreed to in writing, software
10+
# distributed under the License is distributed on an "AS IS" BASIS,
11+
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
# See the License for the specific language governing permissions and
13+
# limitations under the License.
14+
15+
import pytest
16+
17+
from bigframes.core import window_spec
18+
19+
20+
@pytest.mark.parametrize(("start", "end"), [(-1, -2), (1, -2), (2, 1)])
21+
def test_invalid_rows_window_boundary_raise_error(start, end):
22+
with pytest.raises(ValueError):
23+
window_spec.RowsWindowBounds(start, end)
24+
25+
26+
@pytest.mark.parametrize(("start", "end"), [(-1, -2), (1, -2), (2, 1)])
27+
def test_invalid_range_window_boundary_raise_error(start, end):
28+
with pytest.raises(ValueError):
29+
window_spec.RangeWindowBounds(start, end)

0 commit comments

Comments
 (0)