Function _get_unique_entities

sdk/python/feast/utils.py:958–1017

Return the set of unique composite Entities for a Feature View and the indexes at which they appear. This method allows us to query the OnlineStore for data we need only once rather than requesting and processing data for the same combination of Entities multiple times.

(
    table: "FeatureView",
    join_key_values: Dict[str, List[ValueProto]],
    entity_name_to_join_key_map: Dict[str, str],
)

def _get_unique_entities(
    table: "FeatureView",
    join_key_values: Dict[str, List[ValueProto]],
    entity_name_to_join_key_map: Dict[str, str],
) -> Tuple[Tuple[Dict[str, ValueProto], ...], Tuple[List[int], ...], int]:
    """Return the set of unique composite Entities for a Feature View and the indexes at which they appear.

    This method allows us to query the OnlineStore for data we need only once
    rather than requesting and processing data for the same combination of
    Entities multiple times.
    """
    # Get the correct set of entity values with the correct join keys.
    table_entity_values = _get_table_entity_values(
        table,
        entity_name_to_join_key_map,
        join_key_values,
    )
    # Validate that all expected join keys exist and have non-empty values.
    expected_keys = set(entity_name_to_join_key_map.values())
    expected_keys.discard("__dummy_id")
    missing_keys = sorted(
        list(set([key for key in expected_keys if key not in table_entity_values]))
    )
    empty_keys = sorted(
        list(set([key for key in expected_keys if not table_entity_values.get(key)]))
    )

    if missing_keys or empty_keys:
        if not any(table_entity_values.values()):
            raise KeyError(
                f"Missing join key values for keys: {missing_keys}. "
                f"No values provided for keys: {empty_keys}. "
                f"Provided join_key_values: {list(join_key_values.keys())}"
            )

    # Convert the column-oriented table_entity_values into row-wise data.
    keys = list(table_entity_values.keys())
    # Each row is a tuple of ValueProto objects corresponding to the join keys.
    rowise = list(enumerate(zip(*table_entity_values.values())))

    # If there are no rows, return empty tuples.
    if not rowise:
        return (), (), 0

    # Sort rowise so that rows with the same join key values are adjacent.
    rowise.sort(key=lambda row: tuple(getattr(x, x.WhichOneof("val")) for x in row[1]))

    # Group rows by their composite join key value.
    groups = [
        (dict(zip(keys, key_tuple)), [idx for idx, _ in group])
        for key_tuple, group in itertools.groupby(rowise, key=lambda row: row[1])
    ]

    # If no groups were formed (should not happen for valid input), return empty tuples.
    if not groups:
        return (), (), 0

    # Unpack the unique entities and their original row indexes.