Function _apply_aggregations_to_response

sdk/python/feast/utils.py:670–723

Apply aggregations using PandasBackend. Args: response_data: Either a pyarrow.Table or dict of lists containing the data aggregations: List of Aggregation objects to apply group_keys: List of column names to group by (optional) mode: Transformation mode ("py

(
    response_data: Union[pyarrow.Table, Dict[str, List[Any]]],
    aggregations,
    group_keys: Optional[List[str]],
    mode: str,
)

def _apply_aggregations_to_response(
    response_data: Union[pyarrow.Table, Dict[str, List[Any]]],
    aggregations,
    group_keys: Optional[List[str]],
    mode: str,
) -> Union[pyarrow.Table, Dict[str, List[Any]]]:
    """
    Apply aggregations using PandasBackend.

    Args:
        response_data: Either a pyarrow.Table or dict of lists containing the data
        aggregations: List of Aggregation objects to apply
        group_keys: List of column names to group by (optional)
        mode: Transformation mode ("python", "pandas", or "substrait")

    Returns:
        Aggregated data in the same format as input

    TODO: Consider refactoring to support backends other than pandas in the future.
    """
    if not aggregations:
        return response_data

    backend = PandasBackend()

    # Convert to pandas DataFrame
    if isinstance(response_data, dict):
        df = pd.DataFrame(response_data)
    else:  # pyarrow.Table
        df = backend.from_arrow(response_data)

    if df.empty:
        return response_data

    # Convert aggregations to agg_ops format
    agg_ops = aggregation_specs_to_agg_ops(
        aggregations,
        time_window_unsupported_error_message=(
            "Time window aggregation is not supported in online serving."
        ),
    )

    # Apply aggregations using PandasBackend
    if group_keys:
        result_df = backend.groupby_agg(df, group_keys, agg_ops)
    else:
        # No grouping - aggregate over entire dataset
        result_df = backend.groupby_agg(df, [], agg_ops)

    # Convert back to original format
    if mode == "python":
        return {col: result_df[col].tolist() for col in result_df.columns}
    else:  # pandas or substrait
        return backend.to_arrow(result_df)


def _augment_response_with_on_demand_transforms(