diff --git a/AGENTS.md b/AGENTS.md index 34a7f1921..41e7b4665 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -548,6 +548,17 @@ CLI. include MkDocs. The documentation task is available in the `docs` environment. **Recommendation**: Run `pixi run -e docs build-docs` to build documentation. +### [2026-08-26] Include performance evidence in optimization PRs and issues + +**Context**: Performance changes are coordinated with implementation changes +in janitor-rs. +**Learning**: Benchmark results are part of the performance change's review +record, not merely local investigation notes. +**Recommendation**: Every performance PR and its tracking issue must include a +properly formatted comparison with the old implementation, covering runtime +and memory for tiny, large, very-large, and super-large cases with duplicate +and unique label distributions. State benchmark limitations explicitly. + --- ## Version History diff --git a/janitor/functions/_conditional_join/_agg_functions.py b/janitor/functions/_conditional_join/_agg_functions.py index 96fa0b7a2..35910f331 100644 --- a/janitor/functions/_conditional_join/_agg_functions.py +++ b/janitor/functions/_conditional_join/_agg_functions.py @@ -1,7 +1,53 @@ +"""Conditional-join aggregation adapters. + +The reverse match kernels consume a flattened candidate tape. The Rust API +requires that tape to be non-empty and exactly as wide as the supplied ranges; +the comparison stage owns the invariant that its values are 0 or 1. A batch +whose every range is zero-width is filtered before these adapters are called, +so it produces the normal empty result without sending an empty tape to Rust. +Integer reverse sum/product kernels use deterministic wrapping arithmetic; +floating-point aggregation is unchanged. + +ELI5: Rust receives one long roll of candidate tickets, while ``starts`` and +``ends`` say which tickets belong to each row. Python builds the roll and +checks its yes/no flags; Rust checks that the roll has the right shape. +""" + import janitor_rs import numpy as np +def _call_rev_starts_matches(func, kwargs, length: int) -> tuple: + """Call a starts+matches kernel across old and new janitor-rs releases. + + New kernels derive their right-hand length from ``index`` and do not need + the legacy ``length`` argument. Keeping this compatibility shim here + lets pyjanitor support an older installed wheel during the rollout without + weakening the new Rust input contract. + """ + try: + return func(**kwargs) + except TypeError as exc: + if "missing 1 required positional argument: 'length'" not in str(exc): + raise + return func(**kwargs, length=length) + + +def _call_rev_positions(func, kwargs, length: int) -> tuple: + """Call a reverse-positions kernel across old and new Rust releases. + + Current positions kernels derive their capacity inputs from the arrays and + no longer accept the redundant ``length`` keyword. Older wheels still + require it, so retry only for that specific legacy-signature error. + """ + try: + return func(**kwargs) + except TypeError as exc: + if "missing 1 required positional argument: 'length'" not in str(exc): + raise + return func(**kwargs, length=length) + + def _sum_starts( arr: np.ndarray, starts: np.ndarray, @@ -112,13 +158,14 @@ def _size_rev_starts_matches( starts: np.ndarray, index: np.ndarray, matches: np.ndarray, - length: int, ) -> tuple: """ Compute size_rev """ - return janitor_rs.compute_size_rev_start_matches( - starts=starts, index=index, matches=matches, length=length + return _call_rev_starts_matches( + janitor_rs.compute_size_rev_start_matches, + dict(starts=starts, index=index, matches=matches), + index.size, ) @@ -144,15 +191,17 @@ def _size_rev_positions( positions: np.ndarray, length: int, ) -> tuple: + """Compute reverse size over the indirect positions tape. + + ``length`` is retained only for the legacy Rust fallback. It is + ``index.size`` because the number of distinct output labels cannot exceed + the right index length; the new Rust kernel derives its own capacity and + never receives this redundant value. """ - Compute size_rev - """ - return janitor_rs.compute_size_rev_positions( - starts=starts, - ends=ends, - index=index, - positions=positions, - length=length, + return _call_rev_positions( + janitor_rs.compute_size_rev_positions, + dict(starts=starts, ends=ends, index=index, positions=positions), + length, ) @@ -1096,7 +1145,6 @@ def _prod_rev_starts_matches( index: np.ndarray, matches: np.ndarray, booleans: np.ndarray, - length: int, ) -> tuple: """ Compute prod @@ -1118,14 +1166,17 @@ def _prod_rev_starts_matches( func = mapping[dtype_name] except KeyError: raise KeyError(f"Unsupported data type -> {dtype_name}") - return func( - arr=arr, - starts=starts, - counts=counts, - index=index, - matches=matches, - booleans=booleans, - length=length, + return _call_rev_starts_matches( + func, + dict( + arr=arr, + starts=starts, + counts=counts, + index=index, + matches=matches, + booleans=booleans, + ), + index.size, ) @@ -1198,14 +1249,17 @@ def _prod_rev_positions( func = mapping[dtype_name] except KeyError: raise KeyError(f"Unsupported data type -> {dtype_name}") - return func( - arr=arr, - starts=starts, - ends=ends, - index=index, - positions=positions, - booleans=booleans, - length=length, + return _call_rev_positions( + func, + dict( + arr=arr, + starts=starts, + ends=ends, + index=index, + positions=positions, + booleans=booleans, + ), + length, ) @@ -1356,7 +1410,6 @@ def _min_rev_starts_matches( index: np.ndarray, matches: np.ndarray, booleans: np.ndarray, - length: int, ) -> tuple: """ Compute min @@ -1378,14 +1431,17 @@ def _min_rev_starts_matches( func = mapping[dtype_name] except KeyError: raise KeyError(f"Unsupported data type -> {dtype_name}") - return func( - arr=arr, - starts=starts, - counts=counts, - index=index, - matches=matches, - booleans=booleans, - length=length, + return _call_rev_starts_matches( + func, + dict( + arr=arr, + starts=starts, + counts=counts, + index=index, + matches=matches, + booleans=booleans, + ), + index.size, ) @@ -1458,14 +1514,17 @@ def _min_rev_positions( func = mapping[dtype_name] except KeyError: raise KeyError(f"Unsupported data type -> {dtype_name}") - return func( - arr=arr, - starts=starts, - ends=ends, - index=index, - positions=positions, - booleans=booleans, - length=length, + return _call_rev_positions( + func, + dict( + arr=arr, + starts=starts, + ends=ends, + index=index, + positions=positions, + booleans=booleans, + ), + length, ) @@ -1616,7 +1675,6 @@ def _max_rev_starts_matches( index: np.ndarray, matches: np.ndarray, booleans: np.ndarray, - length: int, ) -> tuple: """ Compute max @@ -1638,14 +1696,17 @@ def _max_rev_starts_matches( func = mapping[dtype_name] except KeyError: raise KeyError(f"Unsupported data type -> {dtype_name}") - return func( - arr=arr, - starts=starts, - counts=counts, - index=index, - matches=matches, - booleans=booleans, - length=length, + return _call_rev_starts_matches( + func, + dict( + arr=arr, + starts=starts, + counts=counts, + index=index, + matches=matches, + booleans=booleans, + ), + index.size, ) @@ -1718,14 +1779,17 @@ def _max_rev_positions( func = mapping[dtype_name] except KeyError: raise KeyError(f"Unsupported data type -> {dtype_name}") - return func( - arr=arr, - starts=starts, - ends=ends, - index=index, - positions=positions, - booleans=booleans, - length=length, + return _call_rev_positions( + func, + dict( + arr=arr, + starts=starts, + ends=ends, + index=index, + positions=positions, + booleans=booleans, + ), + length, ) @@ -2020,7 +2084,6 @@ def _sum_rev_starts_matches( index: np.ndarray, matches: np.ndarray, booleans: np.ndarray, - length: int, ) -> tuple: """ Compute sum @@ -2042,14 +2105,17 @@ def _sum_rev_starts_matches( func = mapping[dtype_name] except KeyError: raise KeyError(f"Unsupported data type -> {dtype_name}") - return func( - arr=arr, - starts=starts, - counts=counts, - index=index, - matches=matches, - booleans=booleans, - length=length, + return _call_rev_starts_matches( + func, + dict( + arr=arr, + starts=starts, + counts=counts, + index=index, + matches=matches, + booleans=booleans, + ), + index.size, ) @@ -2122,14 +2188,17 @@ def _sum_rev_positions( func = mapping[dtype_name] except KeyError: raise KeyError(f"Unsupported data type -> {dtype_name}") - return func( - arr=arr, - starts=starts, - ends=ends, - index=index, - positions=positions, - booleans=booleans, - length=length, + return _call_rev_positions( + func, + dict( + arr=arr, + starts=starts, + ends=ends, + index=index, + positions=positions, + booleans=booleans, + ), + length, ) diff --git a/janitor/functions/_conditional_join/_get_join_aggs.py b/janitor/functions/_conditional_join/_get_join_aggs.py index d757c6f41..5f702cd0f 100644 --- a/janitor/functions/_conditional_join/_get_join_aggs.py +++ b/janitor/functions/_conditional_join/_get_join_aggs.py @@ -8,22 +8,42 @@ from janitor.functions._conditional_join import _agg_functions, _helpers +def _empty_agg_result( + dtypes: pd.Series, aggfunc: list[tuple[Hashable, str]] +) -> pd.DataFrame: + """Build the typed empty result used when no candidate rows exist. + + The Rust reverse match and positions kernels intentionally reject an + empty tape. Python handles the valid user-facing no-candidate case here, + before dispatch, so an all-zero-width batch still returns an empty frame. + """ + aggs = {} + for column_name, agg in aggfunc: + if agg == "size": + _dtype = "int64" + else: + _dtype = dtypes.loc[column_name] + out = pd.array([], dtype=_dtype, copy=False) + new_label = _build_agg_label(column_name=column_name, agg_name=agg) + aggs[new_label] = out + return pd.DataFrame(aggs, copy=False) + + +def _has_empty_candidate_tape(indices: dict) -> bool: + """Return whether a strict Rust kernel would receive no tape entries.""" + for name in ("matches", "positions"): + tape = indices.get(name) + if tape is not None and not tape.size: + return True + return False + + def _agg_join_left(df: pd.DataFrame, aggfunc: list, indices: dict) -> pd.DataFrame: """ Compute aggregation for multiple joins """ - if not indices["left_index"].size: - dtypes = df.dtypes - aggs = {} - for column_name, agg in aggfunc: - if agg == "size": - _dtype = "int64" - else: - _dtype = dtypes.loc[column_name] - out = pd.array([], dtype=_dtype, copy=False) - new_label = _build_agg_label(column_name=column_name, agg_name=agg) - aggs[new_label] = out - return pd.DataFrame(aggs, copy=False) + if not indices["left_index"].size or _has_empty_candidate_tape(indices): + return _empty_agg_result(df.dtypes, aggfunc) # single join - less than if (indices.get("starts") is not None) and isinstance(indices.get("ends"), int): aggs = {} @@ -172,7 +192,6 @@ def _agg_join_left(df: pd.DataFrame, aggfunc: list, indices: dict) -> pd.DataFra starts=indices["starts"], index=indices["right_index"], matches=indices["matches"], - length=indices["right_index"].size, ) else: ser = df.loc[indices["left_index"], column_name] @@ -180,15 +199,15 @@ def _agg_join_left(df: pd.DataFrame, aggfunc: list, indices: dict) -> pd.DataFra booleans = pd.isna(arr) arr = _helpers._convert_array_to_numpy(array=arr) func = mapping[agg] - _index, out = func( + kwargs = dict( arr=arr, starts=indices["starts"], index=indices["right_index"], matches=indices["matches"], counts=indices["counts_array"], booleans=booleans, - length=indices["right_index"].size, ) + _index, out = func(**kwargs) if agg in { "sum", "prod", @@ -263,6 +282,10 @@ def _agg_join_left(df: pd.DataFrame, aggfunc: list, indices: dict) -> pd.DataFra for column_name, agg in aggfunc: if agg == "size": func = mapping[agg] + # Legacy Rust wheels use length only as an output-slot upper + # bound. The right index is the tightest safe bound: there + # cannot be more distinct labels than right-index entries, + # while the positions tape may be much larger with repeats. _index, out = func( starts=indices["starts"], ends=indices["ends"], @@ -407,18 +430,8 @@ def _agg_join_right(right: pd.DataFrame, aggfunc: list, indices: dict) -> pd.Dat """ Compute aggregation for multiple joins """ - if not indices["left_index"].size: - dtypes = right.dtypes - aggs = {} - for column_name, agg in aggfunc: - if agg == "size": - _dtype = "int64" - else: - _dtype = dtypes.loc[column_name] - out = pd.array([], dtype=_dtype, copy=False) - new_label = _build_agg_label(column_name=column_name, agg_name=agg) - aggs[new_label] = out - return pd.DataFrame(aggs, copy=False) + if not indices["left_index"].size or _has_empty_candidate_tape(indices): + return _empty_agg_result(right.dtypes, aggfunc) # single join - less than if (indices.get("starts") is not None) and isinstance(indices.get("ends"), int): aggs = {} diff --git a/tests/functions/test_conditional_join.py b/tests/functions/test_conditional_join.py index 94fb70ac5..107aa0189 100644 --- a/tests/functions/test_conditional_join.py +++ b/tests/functions/test_conditional_join.py @@ -5333,6 +5333,30 @@ def test_gt_ne_agg(df, right): assert_frame_equal(expected, actual) +def test_range_only_agg_rev_all_zero_width_ranges_skip_rust_tape(): + """An all-zero-width batch returns no rows without calling Rust.""" + left = pd.DataFrame({"value": [0], "payload": [3]}) + right = pd.DataFrame({"value": [1, 2]}) + + actual = left.join_agg( + right, + ("value", "value", ">"), + ("value", "value", "<"), + aggfunc=[("payload", "sum"), ("payload", "size")], + reverse=True, + ) + + expected = pd.DataFrame( + { + ("payload", "sum"): pd.Series([], dtype="int64"), + ("payload", "size"): pd.Series([], dtype="int64"), + } + ) + expected.columns = pd.MultiIndex.from_tuples(expected.columns) + + assert_frame_equal(actual, expected) + + @pytest.mark.turtle @settings(deadline=None, max_examples=10) @given(df=conditional_df(), right=conditional_right())