diff --git a/docs/getting-started/faq.md b/docs/getting-started/faq.md index b790d6dd719..af545acd5e8 100644 --- a/docs/getting-started/faq.md +++ b/docs/getting-started/faq.md @@ -39,7 +39,16 @@ Yes, this is possible. For example, you can use BigQuery as an offline store and ### How do I run `get_historical_features` without providing an entity dataframe? -Feast does not provide a way to do this right now. This is an area we're actively interested in contributions for. See [GitHub issue](https://github.com/feast-dev/feast/issues/1611) +Feast does supports fetching historical features without passing an entity dataframe with the request. +- As of today, only `postgres offline feature store` is supported for entity dataframe less retrieval. Remaining offline stores would be gradually updated to support the entity df less retrieval. The stores would be selected based on priorities and user base/request. +- The retrieval is based on `start_date` and `end_date` parameters to the function. Here are some combinations supported. + - Both params are given, Returns data during the given start to end timerange. + - Only start_date param is given, Returns data from the start date to `now` time. + - Only end_date param is given, Returns data during the end_date minus TTL time in feature view. + - Both params are `not` given, Returns data during the TTL time in feature view to now time. +- When multiple features are requested from multiple feature-views it is required to have entity ids in both of them for `JOIN` so that + +This is an area we're actively interested in contributions for. See [GitHub issue](https://github.com/feast-dev/feast/issues/1611) ### Does Feast provide security or access control? diff --git a/sdk/python/feast/cli/features.py b/sdk/python/feast/cli/features.py index 6228fbacb6a..403a98e4f27 100644 --- a/sdk/python/feast/cli/features.py +++ b/sdk/python/feast/cli/features.py @@ -1,4 +1,5 @@ import json +from datetime import datetime from typing import List import click @@ -140,37 +141,69 @@ def get_online_features(ctx: click.Context, entities: List[str], features: List[ "--dataframe", "-d", type=str, - required=True, help='JSON string containing entities and timestamps. Example: \'[{"event_timestamp": "2025-03-29T12:00:00", "driver_id": 1001}]\'', ) @click.option( "--features", "-f", multiple=True, - required=True, help="Features to retrieve. feature-view:feature-name ex: driver_hourly_stats:conv_rate", ) +@click.option( + "--start-date", + "-s", + type=str, + help="Start date for historical feature retrieval. Format: YYYY-MM-DD HH:MM:SS", +) +@click.option( + "--end-date", + "-e", + type=str, + help="End date for historical feature retrieval. Format: YYYY-MM-DD HH:MM:SS", +) @click.pass_context -def get_historical_features(ctx: click.Context, dataframe: str, features: List[str]): +def get_historical_features( + ctx: click.Context, + dataframe: str, + features: List[str], + start_date: str, + end_date: str, +): """ Fetch historical feature values for a given entity ID """ store = create_feature_store(ctx) - try: - entity_list = json.loads(dataframe) - if not isinstance(entity_list, list): - raise ValueError("Entities must be a list of dictionaries.") - - entity_df = pd.DataFrame(entity_list) - entity_df["event_timestamp"] = pd.to_datetime(entity_df["event_timestamp"]) + if not dataframe and not start_date and not end_date: + click.echo( + "Either --dataframe or --start-date and/or --end-date must be provided." + ) + return - except Exception as e: - click.echo(f"Error parsing entities JSON: {e}", err=True) + if dataframe and (start_date or end_date): + click.echo("Cannot specify both --dataframe and --start-date/--end-date.") return + entity_df = None + if dataframe: + try: + entity_list = json.loads(dataframe) + if not isinstance(entity_list, list): + raise ValueError("Entities must be a list of dictionaries.") + + entity_df = pd.DataFrame(entity_list) + entity_df["event_timestamp"] = pd.to_datetime(entity_df["event_timestamp"]) + + except Exception as e: + click.echo(f"Error parsing entities JSON: {e}", err=True) + return + feature_vector = store.get_historical_features( entity_df=entity_df, features=list(features), + start_date=datetime.strptime(start_date, "%Y-%m-%d %H:%M:%S") + if start_date + else None, + end_date=datetime.strptime(end_date, "%Y-%m-%d %H:%M:%S") if end_date else None, ).to_df() click.echo(feature_vector.to_json(orient="records", indent=4)) diff --git a/sdk/python/feast/feature_store.py b/sdk/python/feast/feature_store.py index d8229ea1e56..7aff6485a29 100644 --- a/sdk/python/feast/feature_store.py +++ b/sdk/python/feast/feature_store.py @@ -1087,14 +1087,17 @@ def teardown(self): def get_historical_features( self, - entity_df: Union[pd.DataFrame, str], - features: Union[List[str], FeatureService], + entity_df: Optional[Union[pd.DataFrame, str]] = None, + features: Union[List[str], FeatureService] = [], full_feature_names: bool = False, + start_date: Optional[datetime] = None, + end_date: Optional[datetime] = None, ) -> RetrievalJob: """Enrich an entity dataframe with historical feature values for either training or batch scoring. This method joins historical feature data from one or more feature views to an entity dataframe by using a time - travel join. + travel join. Alternatively, features can be retrieved for a specific timestamp range without requiring an entity + dataframe. Each feature view is joined to the entity dataframe using all entities configured for the respective feature view. All configured entities must be available in the entity dataframe. Therefore, the entity dataframe must @@ -1105,16 +1108,21 @@ def get_historical_features( TTL may result in null values being returned. Args: - entity_df (Union[pd.DataFrame, str]): An entity dataframe is a collection of rows containing all entity - columns (e.g., customer_id, driver_id) on which features need to be joined, as well as a event_timestamp - column used to ensure point-in-time correctness. Either a Pandas DataFrame can be provided or a string - SQL query. The query must be of a format supported by the configured offline store (e.g., BigQuery) features: The list of features that should be retrieved from the offline store. These features can be specified either as a list of string feature references or as a feature service. String feature references must have format "feature_view:feature", e.g. "customer_fv:daily_transactions". + entity_df (Optional[Union[pd.DataFrame, str]]): An entity dataframe is a collection of rows containing all entity + columns (e.g., customer_id, driver_id) on which features need to be joined, as well as a event_timestamp + column used to ensure point-in-time correctness. Either a Pandas DataFrame can be provided or a string + SQL query. The query must be of a format supported by the configured offline store (e.g., BigQuery). + If not provided, features will be retrieved for the specified timestamp range without entity joins. full_feature_names: If True, feature names will be prefixed with the corresponding feature view name, changing them from the format "feature" to "feature_view__feature" (e.g. "daily_transactions" changes to "customer_fv__daily_transactions"). + start_date (Optional[datetime]): Start date for the timestamp range when retrieving features without entity_df. + Required when entity_df is not provided. + end_date (Optional[datetime]): End date for the timestamp range when retrieving features without entity_df. + Required when entity_df is not provided. By default, the current time is used. Returns: RetrievalJob which can be used to materialize the results. @@ -1147,6 +1155,15 @@ def get_historical_features( ... ) >>> feature_data = retrieval_job.to_df() """ + + if entity_df is not None and (start_date is not None or end_date is not None): + raise ValueError( + "Cannot specify both entity_df and start_date/end_date. Use either entity_df for entity-based retrieval or start_date/end_date for timestamp range retrieval." + ) + + if entity_df is None and end_date is None: + end_date = datetime.now() + _feature_refs = utils._get_features(self._registry, self.project, features) ( all_feature_views, @@ -1180,6 +1197,13 @@ def get_historical_features( utils._validate_feature_refs(_feature_refs, full_feature_names) provider = self._get_provider() + # Optional kwargs + kwargs = {} + if start_date is not None: + kwargs["start_date"] = start_date + if end_date is not None: + kwargs["end_date"] = end_date + job = provider.get_historical_features( self.config, feature_views, @@ -1188,6 +1212,7 @@ def get_historical_features( self._registry, self.project, full_feature_names, + **kwargs, ) return job diff --git a/sdk/python/feast/infra/offline_stores/contrib/postgres_offline_store/postgres.py b/sdk/python/feast/infra/offline_stores/contrib/postgres_offline_store/postgres.py index 1a75bb7e178..2db614f7b17 100644 --- a/sdk/python/feast/infra/offline_stores/contrib/postgres_offline_store/postgres.py +++ b/sdk/python/feast/infra/offline_stores/contrib/postgres_offline_store/postgres.py @@ -1,6 +1,6 @@ import contextlib from dataclasses import asdict -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone from enum import Enum from typing import ( Any, @@ -46,6 +46,7 @@ from feast.repo_config import RepoConfig from feast.saved_dataset import SavedDatasetStorage from feast.type_map import pg_type_code_to_arrow +from feast.utils import _utc_now, make_tzaware from .postgres_source import PostgreSQLSource @@ -119,14 +120,52 @@ def get_historical_features( config: RepoConfig, feature_views: List[FeatureView], feature_refs: List[str], - entity_df: Union[pd.DataFrame, str], + entity_df: Optional[Union[pd.DataFrame, str]], registry: BaseRegistry, project: str, full_feature_names: bool = False, + **kwargs, ) -> RetrievalJob: assert isinstance(config.offline_store, PostgreSQLOfflineStoreConfig) for fv in feature_views: assert isinstance(fv.batch_source, PostgreSQLSource) + start_date: Optional[datetime] = kwargs.get("start_date", None) + end_date: Optional[datetime] = kwargs.get("end_date", None) + + # Handle non-entity retrieval mode + if entity_df is None: + # Default to current time if end_date not provided + if end_date is None: + end_date = _utc_now() + else: + end_date = make_tzaware(end_date) + + # Calculate start_date from TTL if not provided + + if start_date is None: + # Find the maximum TTL across all feature views to ensure we capture enough data + max_ttl_seconds = 0 + for fv in feature_views: + if fv.ttl and isinstance(fv.ttl, timedelta): + ttl_seconds = int(fv.ttl.total_seconds()) + max_ttl_seconds = max(max_ttl_seconds, ttl_seconds) + + if max_ttl_seconds > 0: + # Start from (end_date - max_ttl) to ensure we capture all relevant features + start_date = end_date - timedelta(seconds=max_ttl_seconds) + else: + # If no TTL is set, default to 30 days before end_date + start_date = end_date - timedelta(days=30) + else: + start_date = make_tzaware(start_date) + + entity_df = pd.DataFrame( + { + "event_timestamp": pd.date_range( + start=start_date, end=end_date, freq="1s", tz=timezone.utc + )[:1] # Just one row + } + ) entity_schema = _get_entity_schema(entity_df, config) @@ -189,6 +228,8 @@ def query_generator() -> Iterator[str]: query_template=MULTIPLE_FEATURE_VIEW_POINT_IN_TIME_JOIN, full_feature_names=full_feature_names, use_cte=use_cte, + start_date=start_date, + end_date=end_date, ) finally: # Only cleanup if we created a table @@ -398,6 +439,8 @@ def build_point_in_time_query( query_template: str, full_feature_names: bool = False, use_cte: bool = False, + start_date: Optional[datetime] = None, + end_date: Optional[datetime] = None, ) -> str: """Build point-in-time query between each feature view table and the entity dataframe for PostgreSQL""" template = Environment(loader=BaseLoader()).from_string(source=query_template) @@ -426,6 +469,8 @@ def build_point_in_time_query( "full_feature_names": full_feature_names, "final_output_feature_names": final_output_feature_names, "use_cte": use_cte, + "start_date": start_date, + "end_date": end_date, } query = template.render(template_context) @@ -466,6 +511,113 @@ def _get_entity_schema( # https://github.com/feast-dev/feast/blob/master/sdk/python/feast/infra/offline_stores/redshift.py MULTIPLE_FEATURE_VIEW_POINT_IN_TIME_JOIN = """ +{% if start_date and end_date %} +/* + Non-entity timestamp range query - use JOINs to combine features into single rows +*/ +{% if featureviews | length == 1 %} +SELECT + "{{ featureviews[0].timestamp_field }}" AS event_timestamp, + {% if featureviews[0].created_timestamp_column %} + "{{ featureviews[0].created_timestamp_column }}" AS created_timestamp, + {% endif %} + {% for entity in featureviews[0].entities %} + "{{ entity }}", + {% endfor %} + {% for feature in featureviews[0].features %} + "{{ feature }}" AS {% if full_feature_names %}"{{ featureviews[0].name }}__{{ featureviews[0].field_mapping.get(feature, feature) }}"{% else %}"{{ featureviews[0].field_mapping.get(feature, feature) }}"{% endif %}{% if not loop.last %},{% endif %} + {% endfor %} + FROM {{ featureviews[0].table_subquery }} as fv_alias +WHERE "{{ featureviews[0].timestamp_field }}" BETWEEN '{{ start_date }}' AND '{{ end_date }}' +{% if featureviews[0].ttl != 0 and featureviews[0].min_event_timestamp %} +AND "{{ featureviews[0].timestamp_field }}" >= '{{ featureviews[0].min_event_timestamp }}' +{% endif %} +{% else %} +WITH +{% for featureview in featureviews %} +"{{ featureview.name }}__data" AS ( + SELECT + "{{ featureview.timestamp_field }}" AS event_timestamp, + {% if featureview.created_timestamp_column %} + "{{ featureview.created_timestamp_column }}" AS created_timestamp, + {% endif %} + {% for entity in featureview.entities %} + "{{ entity }}", + {% endfor %} + {% for feature in featureview.features %} + "{{ feature }}" AS {% if full_feature_names %}"{{ featureview.name }}__{{ featureview.field_mapping.get(feature, feature) }}"{% else %}"{{ featureview.field_mapping.get(feature, feature) }}"{% endif %}{% if not loop.last %},{% endif %} + {% endfor %} + FROM {{ featureview.table_subquery }} AS sub + WHERE "{{ featureview.timestamp_field }}" BETWEEN '{{ start_date }}' AND '{{ end_date }}' + {% if featureview.ttl != 0 and featureview.min_event_timestamp %} + AND "{{ featureview.timestamp_field }}" >= '{{ featureview.min_event_timestamp }}' + {% endif %} +), +{% endfor %} + +-- Create a base query with all unique entity + timestamp combinations +base_entities AS ( + {% for featureview in featureviews %} + SELECT DISTINCT + event_timestamp, + {% for entity in featureview.entities %} + "{{ entity }}"{% if not loop.last %},{% endif %} + {% endfor %} + FROM "{{ featureview.name }}__data" + {% if not loop.last %} + UNION + {% endif %} + {% endfor %} +) + +SELECT + base.event_timestamp, + {% set all_entities = [] %} + {% for featureview in featureviews %} + {% for entity in featureview.entities %} + {% if entity not in all_entities %} + {% set _ = all_entities.append(entity) %} + {% endif %} + {% endfor %} + {% endfor %} + {% for entity in all_entities %} + base."{{ entity }}", + {% endfor %} + {% set total_features = featureviews|map(attribute='features')|map('length')|sum %} + {% set feature_counter = namespace(count=0) %} + {% for featureview in featureviews %} + {% set outer_loop_index = loop.index0 %} + {% for feature in featureview.features %} + {% set feature_counter.count = feature_counter.count + 1 %} + fv_{{ outer_loop_index }}."{% if full_feature_names %}{{ featureview.name }}__{{ featureview.field_mapping.get(feature, feature) }}{% else %}{{ featureview.field_mapping.get(feature, feature) }}{% endif %}"{% if feature_counter.count < total_features %},{% endif %} + {% endfor %} + {% endfor %} +FROM base_entities base +{% for featureview in featureviews %} +{% set outer_loop_index = loop.index0 %} +LEFT JOIN LATERAL ( + SELECT DISTINCT ON ({% for entity in featureview.entities %}"{{ entity }}"{% if not loop.last %}, {% endif %}{% endfor %}) + event_timestamp, + {% for entity in featureview.entities %} + "{{ entity }}", + {% endfor %} + {% for feature in featureview.features %} + "{% if full_feature_names %}{{ featureview.name }}__{{ featureview.field_mapping.get(feature, feature) }}{% else %}{{ featureview.field_mapping.get(feature, feature) }}{% endif %}"{% if not loop.last %},{% endif %} + {% endfor %} + FROM "{{ featureview.name }}__data" fv_sub_{{ outer_loop_index }} + WHERE fv_sub_{{ outer_loop_index }}.event_timestamp <= base.event_timestamp + {% if featureview.ttl != 0 %} + AND fv_sub_{{ outer_loop_index }}.event_timestamp >= base.event_timestamp - {{ featureview.ttl }} * interval '1' second + {% endif %} + {% for entity in featureview.entities %} + AND fv_sub_{{ outer_loop_index }}."{{ entity }}" = base."{{ entity }}" + {% endfor %} + ORDER BY {% for entity in featureview.entities %}"{{ entity }}"{% if not loop.last %}, {% endif %}{% endfor %}, event_timestamp DESC +) AS fv_{{ outer_loop_index }} ON true +{% endfor %} +ORDER BY base.event_timestamp +{% endif %} +{% else %} WITH {% if use_cte %} entity_query AS ({{ left_table_query_string }}), @@ -644,4 +796,5 @@ def _get_entity_schema( FROM "{{ featureview.name }}__cleaned" ) AS "{{featureview.name}}" USING ("{{featureview.name}}__entity_row_unique_id") {% endfor %} +{% endif %} """ diff --git a/sdk/python/feast/infra/offline_stores/offline_store.py b/sdk/python/feast/infra/offline_stores/offline_store.py index 5a59e1c3234..8a15fb738d1 100644 --- a/sdk/python/feast/infra/offline_stores/offline_store.py +++ b/sdk/python/feast/infra/offline_stores/offline_store.py @@ -297,7 +297,7 @@ def get_historical_features( config: RepoConfig, feature_views: List[FeatureView], feature_refs: List[str], - entity_df: Union[pd.DataFrame, str], + entity_df: Optional[Union[pd.DataFrame, str]], registry: BaseRegistry, project: str, full_feature_names: bool = False, @@ -311,13 +311,17 @@ def get_historical_features( feature_refs: The features to be retrieved. entity_df: A collection of rows containing all entity columns on which features need to be joined, as well as the timestamp column used for point-in-time joins. Either a pandas dataframe can be - provided or a SQL query. + provided or a SQL query. If None, features will be retrieved for the specified timestamp range. registry: The registry for the current feature store. project: Feast project to which the feature views belong. full_feature_names: If True, feature names will be prefixed with the corresponding feature view name, changing them from the format "feature" to "feature_view__feature" (e.g. "daily_transactions" changes to "customer_fv__daily_transactions"). + Keyword Args: + start_date: Start date for the timestamp range when retrieving features without entity_df. + end_date: End date for the timestamp range when retrieving features without entity_df. By default, the current time is used. + Returns: A RetrievalJob that can be executed to get the features. """ diff --git a/sdk/python/feast/infra/offline_stores/offline_utils.py b/sdk/python/feast/infra/offline_stores/offline_utils.py index e951434e2a3..abd7ad4fe35 100644 --- a/sdk/python/feast/infra/offline_stores/offline_utils.py +++ b/sdk/python/feast/infra/offline_stores/offline_utils.py @@ -50,7 +50,7 @@ def assert_expected_columns_in_entity_df( entity_df_event_timestamp_col: str, ): entity_columns = set(entity_schema.keys()) - expected_columns = join_keys | {entity_df_event_timestamp_col} + expected_columns = {entity_df_event_timestamp_col} missing_keys = expected_columns - entity_columns if len(missing_keys) != 0: diff --git a/sdk/python/feast/infra/offline_stores/remote.py b/sdk/python/feast/infra/offline_stores/remote.py index 41985b9bba0..a5f50c7b45c 100644 --- a/sdk/python/feast/infra/offline_stores/remote.py +++ b/sdk/python/feast/infra/offline_stores/remote.py @@ -116,7 +116,7 @@ def __init__( client: FeastFlightClient, api: str, api_parameters: Dict[str, Any], - entity_df: Union[pd.DataFrame, str] = None, + entity_df: Optional[Union[pd.DataFrame, str]] = None, table: pa.Table = None, metadata: Optional[RetrievalMetadata] = None, ): @@ -193,7 +193,7 @@ def get_historical_features( config: RepoConfig, feature_views: List[FeatureView], feature_refs: List[str], - entity_df: Union[pd.DataFrame, str], + entity_df: Optional[Union[pd.DataFrame, str]], registry: BaseRegistry, project: str, full_feature_names: bool = False, @@ -482,8 +482,8 @@ def _get_entity_df_event_timestamp_range( def _send_retrieve_remote( api: str, api_parameters: Dict[str, Any], - entity_df: Union[pd.DataFrame, str], - table: pa.Table, + entity_df: Optional[Union[pd.DataFrame, str]], + table: Optional[pa.Table], client: FeastFlightClient, ): command_descriptor = _call_put( @@ -510,7 +510,7 @@ def _call_put( api: str, api_parameters: Dict[str, Any], client: FeastFlightClient, - entity_df: Union[pd.DataFrame, str], + entity_df: Optional[Union[pd.DataFrame, str]], table: pa.Table, ): # Generate unique command identifier @@ -535,7 +535,7 @@ def _call_put( def _put_parameters( command_descriptor: fl.FlightDescriptor, - entity_df: Union[pd.DataFrame, str], + entity_df: Optional[Union[pd.DataFrame, str]], table: pa.Table, client: FeastFlightClient, ): diff --git a/sdk/python/feast/infra/passthrough_provider.py b/sdk/python/feast/infra/passthrough_provider.py index d4b586f5c93..40b2d63f077 100644 --- a/sdk/python/feast/infra/passthrough_provider.py +++ b/sdk/python/feast/infra/passthrough_provider.py @@ -458,10 +458,11 @@ def get_historical_features( config: RepoConfig, feature_views: List[Union[FeatureView, OnDemandFeatureView]], feature_refs: List[str], - entity_df: Union[pd.DataFrame, str], + entity_df: Optional[Union[pd.DataFrame, str]], registry: BaseRegistry, project: str, full_feature_names: bool, + **kwargs, ) -> RetrievalJob: job = self.offline_store.get_historical_features( config=config, @@ -471,6 +472,7 @@ def get_historical_features( registry=registry, project=project, full_feature_names=full_feature_names, + **kwargs, ) return job diff --git a/sdk/python/feast/infra/provider.py b/sdk/python/feast/infra/provider.py index c9150c542e4..6a20b5edf03 100644 --- a/sdk/python/feast/infra/provider.py +++ b/sdk/python/feast/infra/provider.py @@ -249,10 +249,11 @@ def get_historical_features( config: RepoConfig, feature_views: List[Union[FeatureView, OnDemandFeatureView]], feature_refs: List[str], - entity_df: Union[pd.DataFrame, str], + entity_df: Optional[Union[pd.DataFrame, str]], registry: BaseRegistry, project: str, full_feature_names: bool, + **kwargs, ) -> RetrievalJob: """ Retrieves the point-in-time correct historical feature values for the specified entity rows. @@ -263,12 +264,14 @@ def get_historical_features( feature_refs: The features to be retrieved. entity_df: A collection of rows containing all entity columns on which features need to be joined, as well as the timestamp column used for point-in-time joins. Either a pandas dataframe can be - provided or a SQL query. + provided or a SQL query. If None, features will be retrieved for the specified timestamp range. registry: The registry for the current feature store. project: Feast project to which the feature views belong. full_feature_names: If True, feature names will be prefixed with the corresponding feature view name, changing them from the format "feature" to "feature_view__feature" (e.g. "daily_transactions" changes to "customer_fv__daily_transactions"). + start_date: Start date for the timestamp range when retrieving features without entity_df. + end_date: End date for the timestamp range when retrieving features without entity_df. Returns: A RetrievalJob that can be executed to get the features. diff --git a/sdk/python/feast/offline_server.py b/sdk/python/feast/offline_server.py index 9c7e04dfe31..776a0dfb96d 100644 --- a/sdk/python/feast/offline_server.py +++ b/sdk/python/feast/offline_server.py @@ -5,7 +5,7 @@ import sys import traceback from datetime import datetime -from typing import Any, Dict, List, cast +from typing import Any, Dict, List, Optional, cast import click import pyarrow as pa @@ -413,20 +413,24 @@ def list_actions(self, context): ), ] - def _validate_get_historical_features_parameters(self, command: dict, key: str): - assert key in self.flights, f"missing key={key}" + def _validate_get_historical_features_parameters( + self, command: dict, key: Optional[str] = None + ): + if key: + assert key in self.flights, f"missing key={key}" assert "feature_view_names" in command, "feature_view_names is mandatory" assert "name_aliases" in command, "name_aliases is mandatory" assert "feature_refs" in command, "feature_refs is mandatory" assert "project" in command, "project is mandatory" assert "full_feature_names" in command, "full_feature_names is mandatory" - def get_historical_features(self, command: dict, key: str): + def get_historical_features(self, command: dict, key: Optional[str] = None): self._validate_get_historical_features_parameters(command, key) - - # Extract parameters from the internal flights dictionary - entity_df_value = self.flights[key] - entity_df = pa.Table.to_pandas(entity_df_value) + entity_df = None + if key: + # Extract parameters from the internal flights dictionary + entity_df_value = self.flights[key] + entity_df = pa.Table.to_pandas(entity_df_value) feature_view_names = command["feature_view_names"] name_aliases = command["name_aliases"] diff --git a/sdk/python/tests/unit/infra/offline_stores/contrib/postgres_offline_store/test_postgres.py b/sdk/python/tests/unit/infra/offline_stores/contrib/postgres_offline_store/test_postgres.py index e220975a2b3..3a43b805607 100644 --- a/sdk/python/tests/unit/infra/offline_stores/contrib/postgres_offline_store/test_postgres.py +++ b/sdk/python/tests/unit/infra/offline_stores/contrib/postgres_offline_store/test_postgres.py @@ -1,5 +1,5 @@ import logging -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone from unittest.mock import MagicMock, patch import pandas as pd @@ -532,3 +532,421 @@ def _mock_entity(): value_type=ValueType.INT64, ) ] + + +def _mock_feature_view(name: str, ttl: timedelta = None): + """Helper to create mock feature views with configurable TTL""" + return FeatureView( + name=name, + entities=[Entity(name="driver_id", join_keys=["driver_id"])], + ttl=ttl, + source=PostgreSQLSource( + name=f"{name}_source", + table=f"{name}_table", + timestamp_field="event_timestamp", + ), + schema=[ + Field(name="feature1", dtype=Float32), + Field(name="feature2", dtype=Float32), + ], + ) + + +class TestNonEntityRetrieval: + """ + Test suite for non-entity retrieval functionality (entity_df=None) + + This test suite comprehensively covers the new non-entity retrieval mode + for PostgreSQL offline store, which enables retrieving features for specified + time ranges without requiring an entity DataFrame. + """ + + def test_non_entity_mode_with_both_dates(self): + """Test non-entity retrieval API accepts both start_date and end_date""" + test_repo_config = RepoConfig( + project="test_project", + registry="test_registry", + provider="local", + offline_store=_mock_offline_store_config(), + ) + + feature_view = _mock_feature_view("test_fv", ttl=None) + start_date = datetime(2023, 1, 1, tzinfo=timezone.utc) + end_date = datetime(2023, 1, 7, tzinfo=timezone.utc) + + # This should not raise an error - validates API signature + with patch.multiple( + "feast.infra.offline_stores.contrib.postgres_offline_store.postgres", + _get_conn=MagicMock(), + _upload_entity_df=MagicMock(), + _get_entity_schema=MagicMock(return_value={"event_timestamp": "timestamp"}), + _get_entity_df_event_timestamp_range=MagicMock( + return_value=(start_date, end_date) + ), + ): + with patch( + "feast.infra.offline_stores.contrib.postgres_offline_store.postgres.offline_utils.get_expected_join_keys", + return_value=[], + ): + with patch( + "feast.infra.offline_stores.contrib.postgres_offline_store.postgres.offline_utils.assert_expected_columns_in_entity_df" + ): + with patch( + "feast.infra.offline_stores.contrib.postgres_offline_store.postgres.offline_utils.get_feature_view_query_context", + return_value=[], + ): + try: + retrieval_job = ( + PostgreSQLOfflineStore.get_historical_features( + config=test_repo_config, + feature_views=[feature_view], + feature_refs=["test_fv:feature1"], + entity_df=None, # Non-entity mode + registry=MagicMock(), + project="test_project", + start_date=start_date, + end_date=end_date, + ) + ) + assert isinstance(retrieval_job, RetrievalJob) + except Exception as e: + # Should not fail due to API signature issues + assert "entity_df" not in str(e) + assert "start_date" not in str(e) + assert "end_date" not in str(e) + + def test_non_entity_mode_with_end_date_only(self): + """Test non-entity retrieval calculates start_date from TTL""" + test_repo_config = RepoConfig( + project="test_project", + registry="test_registry", + provider="local", + offline_store=_mock_offline_store_config(), + ) + + feature_views = [ + _mock_feature_view("user_fv", ttl=timedelta(hours=1)), + _mock_feature_view("transaction_fv", ttl=timedelta(days=1)), + ] + end_date = datetime(2023, 1, 7, tzinfo=timezone.utc) + + with patch.multiple( + "feast.infra.offline_stores.contrib.postgres_offline_store.postgres", + _get_conn=MagicMock(), + _upload_entity_df=MagicMock(), + _get_entity_schema=MagicMock(return_value={"event_timestamp": "timestamp"}), + _get_entity_df_event_timestamp_range=MagicMock( + return_value=(datetime(2023, 1, 6, tzinfo=timezone.utc), end_date) + ), + ): + with patch( + "feast.infra.offline_stores.contrib.postgres_offline_store.postgres.offline_utils.get_expected_join_keys", + return_value=[], + ): + with patch( + "feast.infra.offline_stores.contrib.postgres_offline_store.postgres.offline_utils.assert_expected_columns_in_entity_df" + ): + with patch( + "feast.infra.offline_stores.contrib.postgres_offline_store.postgres.offline_utils.get_feature_view_query_context", + return_value=[], + ): + try: + retrieval_job = ( + PostgreSQLOfflineStore.get_historical_features( + config=test_repo_config, + feature_views=feature_views, + feature_refs=[ + "user_fv:age", + "transaction_fv:amount", + ], + entity_df=None, # Non-entity mode + registry=MagicMock(), + project="test_project", + end_date=end_date, + # start_date not provided - should be calculated from max TTL + ) + ) + assert isinstance(retrieval_job, RetrievalJob) + except Exception as e: + # Should not fail due to TTL calculation issues + assert "ttl" not in str(e).lower() + + @patch("feast.utils.datetime") + def test_no_dates_provided_defaults_to_current_time(self, mock_datetime): + """Test that when no dates are provided, end_date defaults to current time""" + # Mock datetime.now() to return a fixed time + fixed_now = datetime(2023, 1, 7, 12, 0, 0, tzinfo=timezone.utc) + mock_datetime.now.return_value = fixed_now + + test_repo_config = RepoConfig( + project="test_project", + registry="test_registry", + provider="local", + offline_store=_mock_offline_store_config(), + ) + + feature_view = _mock_feature_view("test_fv", ttl=timedelta(days=1)) + + with patch.multiple( + "feast.infra.offline_stores.contrib.postgres_offline_store.postgres", + _get_conn=MagicMock(), + _upload_entity_df=MagicMock(), + _get_entity_schema=MagicMock(return_value={"event_timestamp": "timestamp"}), + _get_entity_df_event_timestamp_range=MagicMock( + return_value=( + datetime(2023, 1, 6, 12, 0, 0, tzinfo=timezone.utc), + fixed_now, + ) + ), + ): + with patch( + "feast.infra.offline_stores.contrib.postgres_offline_store.postgres.offline_utils.get_expected_join_keys", + return_value=[], + ): + with patch( + "feast.infra.offline_stores.contrib.postgres_offline_store.postgres.offline_utils.assert_expected_columns_in_entity_df" + ): + with patch( + "feast.infra.offline_stores.contrib.postgres_offline_store.postgres.offline_utils.get_feature_view_query_context", + return_value=[], + ): + try: + retrieval_job = ( + PostgreSQLOfflineStore.get_historical_features( + config=test_repo_config, + feature_views=[feature_view], + feature_refs=["test_fv:feature1"], + entity_df=None, # Non-entity mode + registry=MagicMock(), + project="test_project", + # No start_date or end_date provided + ) + ) + + # Verify that datetime.now() was called to get current time + mock_datetime.now.assert_called_with(tz=timezone.utc) + assert isinstance(retrieval_job, RetrievalJob) + except Exception as e: + # Should not fail due to datetime issues + assert "datetime" not in str(e).lower() + + def test_sql_template_ttl_filtering(self): + """Test that the SQL template includes proper TTL filtering""" + from jinja2 import BaseLoader, Environment + + # Test the template section that includes TTL filtering + template_with_ttl = """ + FROM {{ featureview.table_subquery }} AS sub + WHERE "{{ featureview.timestamp_field }}" BETWEEN '{{ start_date }}' AND '{{ end_date }}' + {% if featureview.ttl != 0 and featureview.min_event_timestamp %} + AND "{{ featureview.timestamp_field }}" >= '{{ featureview.min_event_timestamp }}' + {% endif %} + """ + + template = Environment(loader=BaseLoader()).from_string( + source=template_with_ttl + ) + + # Test case 1: Feature view with TTL + context_with_ttl = { + "featureview": { + "table_subquery": "test_table", + "timestamp_field": "event_timestamp", + "ttl": 3600, # 1 hour + "min_event_timestamp": "2023-01-06 23:00:00", + }, + "start_date": "2023-01-01", + "end_date": "2023-01-07", + } + + query_with_ttl = template.render(context_with_ttl) + # Should include the TTL timestamp value in the query + assert "2023-01-06 23:00:00" in query_with_ttl + # Should have the TTL filtering condition + assert ">=" in query_with_ttl + + # Test case 2: Feature view without TTL + context_no_ttl = { + "featureview": { + "table_subquery": "test_table", + "timestamp_field": "event_timestamp", + "ttl": 0, # No TTL + "min_event_timestamp": None, + }, + "start_date": "2023-01-01", + "end_date": "2023-01-07", + } + + query_no_ttl = template.render(context_no_ttl) + # Should not include TTL filtering when TTL is 0 or min_event_timestamp is None + assert 'AND "event_timestamp" >=' not in query_no_ttl + + def test_lateral_join_ttl_constraints(self): + """Test that LATERAL JOINs include proper TTL constraints""" + from jinja2 import BaseLoader, Environment + + lateral_template = """ + FROM "{{ featureview.name }}__data" fv_sub_{{ outer_loop_index }} + WHERE fv_sub_{{ outer_loop_index }}.event_timestamp <= base.event_timestamp + {% if featureview.ttl != 0 %} + AND fv_sub_{{ outer_loop_index }}.event_timestamp >= base.event_timestamp - {{ featureview.ttl }} * interval '1' second + {% endif %} + """ + + template = Environment(loader=BaseLoader()).from_string(source=lateral_template) + + # Test with TTL + context = { + "featureview": { + "name": "user_features", + "ttl": 86400, # 1 day + }, + "outer_loop_index": 0, + } + + query = template.render(context) + assert "86400 * interval" in query + assert "base.event_timestamp -" in query + + # Test without TTL + context_no_ttl = { + "featureview": { + "name": "user_features", + "ttl": 0, # No TTL + }, + "outer_loop_index": 0, + } + + query_no_ttl = template.render(context_no_ttl) + assert "interval" not in query_no_ttl + + def test_api_non_entity_functionality(self): + """Test that FeatureStore API accepts non-entity parameters correctly""" + from feast import FeatureStore + from feast.infra.offline_stores.offline_store import RetrievalJob + from feast.repo_config import RepoConfig + + config = RepoConfig( + project="test_project", + registry="test_registry", + provider="local", + offline_store=_mock_offline_store_config(), + ) + + # Mock the entire retrieval pipeline + with patch.object(FeatureStore, "_get_provider") as mock_provider: + mock_retrieval_job = MagicMock(spec=RetrievalJob) + mock_provider.return_value.get_historical_features.return_value = ( + mock_retrieval_job + ) + + fs = FeatureStore(config=config) + + # Mock registry and feature resolution + with ( + patch.object(fs, "_registry"), + patch("feast.utils._get_features") as mock_get_features, + patch("feast.utils._get_feature_views_to_use") as mock_get_views, + patch("feast.utils._group_feature_refs") as mock_group_refs, + ): + mock_get_features.return_value = ["test:feature"] + mock_get_views.return_value = ( + [], + [], + ) # (all_feature_views, all_on_demand_feature_views) + mock_group_refs.return_value = ([], []) # (fvs, odfvs) + + # Test non-entity API call + result = fs.get_historical_features( + features=["test:feature"], + start_date=datetime(2023, 1, 1, tzinfo=timezone.utc), + end_date=datetime(2023, 1, 7, tzinfo=timezone.utc), + ) + + # Verify the call was made correctly + assert result == mock_retrieval_job + mock_provider.return_value.get_historical_features.assert_called_once() + + # Check that the new parameters were passed (the exact call structure may vary) + # but we want to verify the API accepted the new parameters + call_args = mock_provider.return_value.get_historical_features.call_args + if call_args.kwargs: + # Called with keyword arguments + assert call_args.kwargs.get("start_date") == datetime( + 2023, 1, 1, tzinfo=timezone.utc + ) + assert call_args.kwargs.get("end_date") == datetime( + 2023, 1, 7, tzinfo=timezone.utc + ) + else: + # Called with positional arguments - just verify it was called + assert len(call_args.args) > 0 + + def test_cli_date_combinations(self): + """Test various CLI date parameter combinations""" + import tempfile + from pathlib import Path + from textwrap import dedent + + from tests.utils.cli_repo_creator import CliRunner, get_example_repo + + runner = CliRunner() + + with tempfile.TemporaryDirectory() as temp_dir: + repo_path = Path(temp_dir) + + # Setup repo + repo_config = repo_path / "feature_store.yaml" + repo_config.write_text( + dedent(f""" + project: test_cli_dates + registry: {repo_path / "registry.db"} + provider: local + offline_store: + type: file + online_store: + type: sqlite + path: {repo_path / "online_store.db"} + """) + ) + + repo_example = repo_path / "example.py" + repo_example.write_text(get_example_repo("example_feature_repo_1.py")) + + result = runner.run(["apply"], cwd=repo_path) + assert result.returncode == 0 + + # Test 1: Both dates provided - should parse correctly + result = runner.run( + [ + "get-historical-features", + "--features", + "driver_hourly_stats:conv_rate", + "--start-date", + "2023-01-01 00:00:00", + "--end-date", + "2023-01-07 00:00:00", + ], + cwd=repo_path, + ) + + # Should not fail on date parsing + stderr_output = result.stderr.decode() + assert "Error parsing" not in stderr_output + assert "time data" not in stderr_output # datetime parsing errors + + # Test 2: Only end date provided - should work (start_date calculated from TTL) + result = runner.run( + [ + "get-historical-features", + "--features", + "driver_hourly_stats:conv_rate", + "--end-date", + "2023-01-07 00:00:00", + ], + cwd=repo_path, + ) + + # Should not fail on parameter validation + stderr_output = result.stderr.decode() + assert "must be provided" not in stderr_output