Function broadcast_join

sdk/python/feast/infra/ray_shared_utils.py:329–465

(
    entity_ds: Dataset,
    feature_df: pd.DataFrame,
    join_keys: List[str],
    timestamp_field: str,
    requested_feats: List[str],
    full_feature_names: bool = False,
    feature_view_name: Optional[str] = None,
    original_join_keys: Optional[List[str]] = None,
)

def broadcast_join(
    entity_ds: Dataset,
    feature_df: pd.DataFrame,
    join_keys: List[str],
    timestamp_field: str,
    requested_feats: List[str],
    full_feature_names: bool = False,
    feature_view_name: Optional[str] = None,
    original_join_keys: Optional[List[str]] = None,
) -> Dataset:
    import ray

    def join_batch_with_features(batch: pd.DataFrame) -> pd.DataFrame:
        features = ray.get(feature_ref)
        if original_join_keys:
            feature_join_keys = original_join_keys
            entity_join_keys = join_keys
        else:
            feature_join_keys = join_keys
            entity_join_keys = join_keys
        feature_cols = [timestamp_field] + feature_join_keys + requested_feats
        available_feature_cols = [
            col for col in feature_cols if col in features.columns
        ]
        features_filtered = features[available_feature_cols].copy()

        batch = normalize_timestamp_columns(batch, timestamp_field, inplace=True)
        features_filtered = normalize_timestamp_columns(
            features_filtered, timestamp_field, inplace=True
        )
        if not entity_join_keys:
            batch_sorted = batch.sort_values(timestamp_field).reset_index(drop=True)
            features_sorted = features_filtered.sort_values(
                timestamp_field
            ).reset_index(drop=True)
            result = pd.merge_asof(
                batch_sorted,
                features_sorted,
                on=timestamp_field,
                direction="backward",
            )
        else:
            for key in entity_join_keys:
                if key not in batch.columns:
                    batch[key] = np.nan
            for key in feature_join_keys:
                if key not in features_filtered.columns:
                    features_filtered[key] = np.nan
            batch_clean = batch.dropna(
                subset=entity_join_keys + [timestamp_field]
            ).copy()
            features_clean = features_filtered.dropna(
                subset=feature_join_keys + [timestamp_field]
            ).copy()
            if batch_clean.empty or features_clean.empty:
                return batch.head(0)
            if timestamp_field in batch_clean.columns:
                batch_sorted = batch_clean.sort_values(