Function _infer_features_and_entities

sdk/python/feast/inference.py:203–277

Updates the specific feature in place with inferred features and entities. Args: fv: The feature view on which to run inference. join_keys: The set of join keys for the feature view's entities. run_inference_for_features: Whether to run inference for features.

(
    provider: Provider,
    fv: Union[FeatureView, OnDemandFeatureView],
    join_keys: Set[Optional[str]],
    run_inference_for_features,
    config,
)

def _infer_features_and_entities(
    provider: Provider,
    fv: Union[FeatureView, OnDemandFeatureView],
    join_keys: Set[Optional[str]],
    run_inference_for_features,
    config,
) -> None:
    """
    Updates the specific feature in place with inferred features and entities.

    Args:
        fv: The feature view on which to run inference.
        join_keys: The set of join keys for the feature view's entities.
        run_inference_for_features: Whether to run inference for features.
        config: The config for the current feature store.
    """
    if isinstance(fv, OnDemandFeatureView):
        return _infer_on_demand_features_and_entities(
            fv, join_keys, run_inference_for_features, config
        )

    if fv.batch_source is None:
        return

    entity_columns: List[Field] = fv.entity_columns if fv.entity_columns else []
    columns_to_exclude = {
        fv.batch_source.timestamp_field,
        fv.batch_source.created_timestamp_column,
    }
    for original_col, mapped_col in fv.batch_source.field_mapping.items():
        if mapped_col in columns_to_exclude:
            columns_to_exclude.remove(mapped_col)
            columns_to_exclude.add(original_col)

    table_column_names_and_types = (
        provider.get_table_column_names_and_types_from_data_source(
            config, fv.batch_source
        )
    )

    for col_name, col_datatype in table_column_names_and_types:
        if col_name in columns_to_exclude:
            continue
        elif col_name in join_keys:
            field = Field(
                name=col_name,
                dtype=from_value_type(
                    fv.batch_source.source_datatype_to_feast_value_type()(col_datatype)
                ),
            )
            if field.name not in [
                entity_column.name for entity_column in fv.entity_columns
            ]:
                entity_columns.append(field)
        elif not re.match(
            "^__|__$", col_name
        ):  # double underscores often signal an internal-use column
            if run_inference_for_features: