Merge remote-tracking branch 'github/main' into defer_reproject

Author: TrevorBergeron

CHANGELOG.md

@@ -4,6 +4,26 @@
 
 [1]: https://pypi.org/project/bigframes/#history
 
+## [1.13.0](https://github.com/googleapis/python-bigquery-dataframes/compare/v1.12.0...v1.13.0) (2024-08-05)
+
+
+### Features
+
+* `df.apply(axis=1)` to support remote function with mutiple params ([#851](https://github.com/googleapis/python-bigquery-dataframes/issues/851)) ([2158818](https://github.com/googleapis/python-bigquery-dataframes/commit/2158818e53e09e55c87ffd574e3ebc2e201285fb))
+* Allow windowing in 'partial' ordering mode ([#861](https://github.com/googleapis/python-bigquery-dataframes/issues/861)) ([ca26fe5](https://github.com/googleapis/python-bigquery-dataframes/commit/ca26fe5f9edec519788c276a09eaff33ecd87434))
+* Create a separate OrderingModePartialPreviewWarning for more fine-grained warning filters ([#879](https://github.com/googleapis/python-bigquery-dataframes/issues/879)) ([8753bdd](https://github.com/googleapis/python-bigquery-dataframes/commit/8753bdd1e44701e56eae914ebc0e91d9b1a6adf1))
+
+
+### Bug Fixes
+
+* Fix issue with invalid sql generated by ml distance functions ([#865](https://github.com/googleapis/python-bigquery-dataframes/issues/865)) ([9959fc8](https://github.com/googleapis/python-bigquery-dataframes/commit/9959fc8fcba93441fdd3d9c17e8fdbe6e6a7b504))
+
+
+### Documentation
+
+* Create sample notebook using `ordering_mode="partial"` ([#880](https://github.com/googleapis/python-bigquery-dataframes/issues/880)) ([c415eb9](https://github.com/googleapis/python-bigquery-dataframes/commit/c415eb91eb71dea53d245ba2bce416062e3f02f8))
+* Update streaming notebook ([#875](https://github.com/googleapis/python-bigquery-dataframes/issues/875)) ([e9b0557](https://github.com/googleapis/python-bigquery-dataframes/commit/e9b05571123cf13079772856317ca3cd3d564c5a))
+
 ## [1.12.0](https://github.com/googleapis/python-bigquery-dataframes/compare/v1.11.1...v1.12.0) (2024-07-31)
 
 

bigframes/_config/bigquery_options.py

@@ -16,7 +16,6 @@
 
 from __future__ import annotations
 
-from enum import Enum
 from typing import Literal, Optional
 import warnings
 
@@ -25,14 +24,9 @@
 import jellyfish
 
 import bigframes.constants
+import bigframes.enums
 import bigframes.exceptions
 
-
-class OrderingMode(Enum):
-    STRICT = "strict"
-    PARTIAL = "partial"
-
-
 SESSION_STARTED_MESSAGE = (
     "Cannot change '{attribute}' once a session has started. "
     "Call bigframes.pandas.close_session() first, if you are using the bigframes.pandas API."
@@ -64,11 +58,11 @@ def _validate_location(value: Optional[str]):
         )
 
 
-def _validate_ordering_mode(value: str) -> OrderingMode:
-    if value.casefold() == OrderingMode.STRICT.value.casefold():
-        return OrderingMode.STRICT
-    if value.casefold() == OrderingMode.PARTIAL.value.casefold():
-        return OrderingMode.PARTIAL
+def _validate_ordering_mode(value: str) -> bigframes.enums.OrderingMode:
+    if value.casefold() == bigframes.enums.OrderingMode.STRICT.value.casefold():
+        return bigframes.enums.OrderingMode.STRICT
+    if value.casefold() == bigframes.enums.OrderingMode.PARTIAL.value.casefold():
+        return bigframes.enums.OrderingMode.PARTIAL
     raise ValueError("Ordering mode must be one of 'strict' or 'partial'.")
 
 

bigframes/core/__init__.py

@@ -194,8 +194,17 @@ def promote_offsets(self, col_id: str) -> ArrayValue:
         """
         Convenience function to promote copy of column offsets to a value column. Can be used to reset index.
         """
-        if self.node.order_ambiguous and not self.session._strictly_ordered:
-            raise ValueError("Generating offsets not supported in unordered mode")
+        if self.node.order_ambiguous and not (self.session._strictly_ordered):
+            if not self.session._allows_ambiguity:
+                raise ValueError(
+                    "Generating offsets not supported in partial ordering mode"
+                )
+            else:
+                warnings.warn(
+                    "Window ordering may be ambiguous, this can cause unstable results.",
+                    bigframes.exceptions.AmbiguousWindowWarning,
+                )
+
         return ArrayValue(nodes.PromoteOffsetsNode(child=self.node, col_id=col_id))
 
     def concat(self, other: typing.Sequence[ArrayValue]) -> ArrayValue:
@@ -347,9 +356,16 @@ def project_window_op(
         # TODO: Support non-deterministic windowing
         if window_spec.row_bounded or not op.order_independent:
             if self.node.order_ambiguous and not self.session._strictly_ordered:
-                raise ValueError(
-                    "Order-dependent windowed ops not supported in unordered mode"
-                )
+                if not self.session._allows_ambiguity:
+                    raise ValueError(
+                        "Generating offsets not supported in partial ordering mode"
+                    )
+                else:
+                    warnings.warn(
+                        "Window ordering may be ambiguous, this can cause unstable results.",
+                        bigframes.exceptions.AmbiguousWindowWarning,
+                    )
+
         return ArrayValue(
             nodes.WindowOpNode(
                 child=self.node,

bigframes/core/blocks.py

@@ -280,6 +280,10 @@ def index_name_to_col_id(self) -> typing.Mapping[Label, typing.Sequence[str]]:
             mapping[label] = (*mapping.get(label, ()), id)
         return mapping
 
+    @property
+    def explicitly_ordered(self) -> bool:
+        return self.expr.node.explicitly_ordered
+
     def cols_matching_label(self, partial_label: Label) -> typing.Sequence[str]:
         """
         Unlike label_to_col_id, this works with partial labels for multi-index.

bigframes/core/compile/compiled.py

@@ -256,7 +256,7 @@ def to_sql(
         ordered: bool = False,
     ) -> str:
         if offset_column or ordered:
-            raise ValueError("Cannot produce sorted sql in unordered mode")
+            raise ValueError("Cannot produce sorted sql in partial ordering mode")
         sql = ibis_bigquery.Backend().compile(
             self._to_ibis_expr(
                 col_id_overrides=col_id_overrides,

bigframes/core/compile/scalar_op_compiler.py

@@ -191,19 +191,27 @@ def normalized_impl(args: typing.Sequence[ibis_types.Value], op: ops.RowOp):
 
         return decorator
 
-    def register_nary_op(self, op_ref: typing.Union[ops.NaryOp, type[ops.NaryOp]]):
+    def register_nary_op(
+        self, op_ref: typing.Union[ops.NaryOp, type[ops.NaryOp]], pass_op: bool = False
+    ):
         """
         Decorator to register a nary op implementation.
 
         Args:
             op_ref (NaryOp or NaryOp type):
                 Class or instance of operator that is implemented by the decorated function.
+            pass_op (bool):
+                Set to true if implementation takes the operator object as the last argument.
+                This is needed for parameterized ops where parameters are part of op object.
         """
         key = typing.cast(str, op_ref.name)
 
         def decorator(impl: typing.Callable[..., ibis_types.Value]):
             def normalized_impl(args: typing.Sequence[ibis_types.Value], op: ops.RowOp):
-                return impl(*args)
+                if pass_op:
+                    return impl(*args, op=op)
+                else:
+                    return impl(*args)
 
             self._register(key, normalized_impl)
             return impl
@@ -1468,6 +1476,7 @@ def clip_op(
         )
 
 
+# N-ary Operations
 @scalar_op_compiler.register_nary_op(ops.case_when_op)
 def case_when_op(*cases_and_outputs: ibis_types.Value) -> ibis_types.Value:
     # ibis can handle most type coercions, but we need to force bool -> int
@@ -1487,6 +1496,19 @@ def case_when_op(*cases_and_outputs: ibis_types.Value) -> ibis_types.Value:
     return case_val.end()
 
 
+@scalar_op_compiler.register_nary_op(ops.NaryRemoteFunctionOp, pass_op=True)
+def nary_remote_function_op_impl(
+    *operands: ibis_types.Value, op: ops.NaryRemoteFunctionOp
+):
+    ibis_node = getattr(op.func, "ibis_node", None)
+    if ibis_node is None:
+        raise TypeError(
+            f"only a bigframes remote function is supported as a callable. {constants.FEEDBACK_LINK}"
+        )
+    result = ibis_node(*operands)
+    return result
+
+
 # Helpers
 def is_null(value) -> bool:
     # float NaN/inf should be treated as distinct from 'true' null values

bigframes/core/groupby/__init__.py

@@ -109,7 +109,7 @@ def __getitem__(
                 dropna=self._dropna,
             )
 
-    @validations.requires_strict_ordering()
+    @validations.requires_ordering()
     def head(self, n: int = 5) -> df.DataFrame:
         block = self._block
         if self._dropna:
@@ -235,25 +235,25 @@ def count(self) -> df.DataFrame:
     def nunique(self) -> df.DataFrame:
         return self._aggregate_all(agg_ops.nunique_op)
 
-    @validations.requires_strict_ordering()
+    @validations.requires_ordering()
     def cumsum(self, *args, numeric_only: bool = False, **kwargs) -> df.DataFrame:
         if not numeric_only:
             self._raise_on_non_numeric("cumsum")
         return self._apply_window_op(agg_ops.sum_op, numeric_only=True)
 
-    @validations.requires_strict_ordering()
+    @validations.requires_ordering()
     def cummin(self, *args, numeric_only: bool = False, **kwargs) -> df.DataFrame:
         return self._apply_window_op(agg_ops.min_op, numeric_only=numeric_only)
 
-    @validations.requires_strict_ordering()
+    @validations.requires_ordering()
     def cummax(self, *args, numeric_only: bool = False, **kwargs) -> df.DataFrame:
         return self._apply_window_op(agg_ops.max_op, numeric_only=numeric_only)
 
-    @validations.requires_strict_ordering()
+    @validations.requires_ordering()
     def cumprod(self, *args, **kwargs) -> df.DataFrame:
         return self._apply_window_op(agg_ops.product_op, numeric_only=True)
 
-    @validations.requires_strict_ordering()
+    @validations.requires_ordering()
     def shift(self, periods=1) -> series.Series:
         window = window_specs.rows(
             grouping_keys=tuple(self._by_col_ids),
@@ -262,7 +262,7 @@ def shift(self, periods=1) -> series.Series:
         )
         return self._apply_window_op(agg_ops.ShiftOp(periods), window=window)
 
-    @validations.requires_strict_ordering()
+    @validations.requires_ordering()
     def diff(self, periods=1) -> series.Series:
         window = window_specs.rows(
             grouping_keys=tuple(self._by_col_ids),
@@ -271,7 +271,7 @@ def diff(self, periods=1) -> series.Series:
         )
         return self._apply_window_op(agg_ops.DiffOp(periods), window=window)
 
-    @validations.requires_strict_ordering()
+    @validations.requires_ordering()
     def rolling(self, window: int, min_periods=None) -> windows.Window:
         # To get n size window, need current row and n-1 preceding rows.
         window_spec = window_specs.rows(
@@ -287,7 +287,7 @@ def rolling(self, window: int, min_periods=None) -> windows.Window:
             block, window_spec, self._selected_cols, drop_null_groups=self._dropna
         )
 
-    @validations.requires_strict_ordering()
+    @validations.requires_ordering()
     def expanding(self, min_periods: int = 1) -> windows.Window:
         window_spec = window_specs.cumulative_rows(
             grouping_keys=tuple(self._by_col_ids),
@@ -532,7 +532,7 @@ def __init__(
     def _session(self) -> core.Session:
         return self._block.session
 
-    @validations.requires_strict_ordering()
+    @validations.requires_ordering()
     def head(self, n: int = 5) -> series.Series:
         block = self._block
         if self._dropna:
@@ -650,31 +650,31 @@ def agg(self, func=None) -> typing.Union[df.DataFrame, series.Series]:
 
     aggregate = agg
 
-    @validations.requires_strict_ordering()
+    @validations.requires_ordering()
     def cumsum(self, *args, **kwargs) -> series.Series:
         return self._apply_window_op(
             agg_ops.sum_op,
         )
 
-    @validations.requires_strict_ordering()
+    @validations.requires_ordering()
     def cumprod(self, *args, **kwargs) -> series.Series:
         return self._apply_window_op(
             agg_ops.product_op,
         )
 
-    @validations.requires_strict_ordering()
+    @validations.requires_ordering()
     def cummax(self, *args, **kwargs) -> series.Series:
         return self._apply_window_op(
             agg_ops.max_op,
         )
 
-    @validations.requires_strict_ordering()
+    @validations.requires_ordering()
     def cummin(self, *args, **kwargs) -> series.Series:
         return self._apply_window_op(
             agg_ops.min_op,
         )
 
-    @validations.requires_strict_ordering()
+    @validations.requires_ordering()
     def cumcount(self, *args, **kwargs) -> series.Series:
         return (
             self._apply_window_op(
@@ -684,7 +684,7 @@ def cumcount(self, *args, **kwargs) -> series.Series:
             - 1
         )
 
-    @validations.requires_strict_ordering()
+    @validations.requires_ordering()
     def shift(self, periods=1) -> series.Series:
         """Shift index by desired number of periods."""
         window = window_specs.rows(
@@ -694,7 +694,7 @@ def shift(self, periods=1) -> series.Series:
         )
         return self._apply_window_op(agg_ops.ShiftOp(periods), window=window)
 
-    @validations.requires_strict_ordering()
+    @validations.requires_ordering()
     def diff(self, periods=1) -> series.Series:
         window = window_specs.rows(
             grouping_keys=tuple(self._by_col_ids),
@@ -703,7 +703,7 @@ def diff(self, periods=1) -> series.Series:
         )
         return self._apply_window_op(agg_ops.DiffOp(periods), window=window)
 
-    @validations.requires_strict_ordering()
+    @validations.requires_ordering()
     def rolling(self, window: int, min_periods=None) -> windows.Window:
         # To get n size window, need current row and n-1 preceding rows.
         window_spec = window_specs.rows(
@@ -723,7 +723,7 @@ def rolling(self, window: int, min_periods=None) -> windows.Window:
             is_series=True,
         )
 
-    @validations.requires_strict_ordering()
+    @validations.requires_ordering()
     def expanding(self, min_periods: int = 1) -> windows.Window:
         window_spec = window_specs.cumulative_rows(
             grouping_keys=tuple(self._by_col_ids),

bigframes/core/indexes/base.py

@@ -184,7 +184,7 @@ def empty(self) -> bool:
         return self.shape[0] == 0
 
     @property
-    @validations.requires_strict_ordering()
+    @validations.requires_ordering()
     def is_monotonic_increasing(self) -> bool:
         """
         Return a boolean if the values are equal or increasing.
@@ -198,7 +198,7 @@ def is_monotonic_increasing(self) -> bool:
         )
 
     @property
-    @validations.requires_strict_ordering()
+    @validations.requires_ordering()
     def is_monotonic_decreasing(self) -> bool:
         """
         Return a boolean if the values are equal or decreasing.
@@ -348,7 +348,7 @@ def max(self) -> typing.Any:
     def min(self) -> typing.Any:
         return self._apply_aggregation(agg_ops.min_op)
 
-    @validations.requires_strict_ordering()
+    @validations.requires_ordering()
     def argmax(self) -> int:
         block, row_nums = self._block.promote_offsets()
         block = block.order_by(
@@ -361,7 +361,7 @@ def argmax(self) -> int:
 
         return typing.cast(int, series.Series(block.select_column(row_nums)).iloc[0])
 
-    @validations.requires_strict_ordering()
+    @validations.requires_ordering()
     def argmin(self) -> int:
         block, row_nums = self._block.promote_offsets()
         block = block.order_by(