diff --git a/.secrets.baseline b/.secrets.baseline index d2838a1317f..2f403d38eeb 100644 --- a/.secrets.baseline +++ b/.secrets.baseline @@ -1485,14 +1485,14 @@ "filename": "sdk/python/tests/universal/feature_repos/repo_configuration.py", "hashed_secret": "d90e76ef629fb00c95f4e84fec29fbda111e2392", "is_verified": false, - "line_number": 466 + "line_number": 475 }, { "type": "Secret Keyword", "filename": "sdk/python/tests/universal/feature_repos/repo_configuration.py", "hashed_secret": "5baa61e4c9b93f3f0682250b6cf8331b7ee68fd8", "is_verified": false, - "line_number": 468 + "line_number": 477 } ], "sdk/python/tests/universal/feature_repos/universal/data_sources/file.py": [ @@ -1564,5 +1564,5 @@ } ] }, - "generated_at": "2026-07-30T09:40:48Z" + "generated_at": "2026-07-30T13:06:49Z" } diff --git a/sdk/python/feast/infra/offline_stores/duckdb.py b/sdk/python/feast/infra/offline_stores/duckdb.py index 27940feeb2f..51526b275f0 100644 --- a/sdk/python/feast/infra/offline_stores/duckdb.py +++ b/sdk/python/feast/infra/offline_stores/duckdb.py @@ -46,6 +46,7 @@ opt_float, ) from feast.repo_config import FeastConfigBaseModel, RepoConfig +from feast.utils import compute_non_entity_date_range def _read_data_source(data_source: DataSource, repo_path: str) -> Table: @@ -495,6 +496,59 @@ class DuckDBOfflineStoreConfig(FeastConfigBaseModel): staging_location_endpoint_override: Optional[str] = None +def _build_entity_df_from_sources( + config: RepoConfig, + feature_views: List[FeatureView], + start_date: datetime, + end_date: datetime, + data_source_reader: Callable[[DataSource, str], Table], +) -> pd.DataFrame: + """Build a synthetic entity DataFrame for non-entity mode. + + Scans each feature view's source within the date range to discover all unique + entity key combinations, then creates a single row per entity with + event_timestamp = end_date so the PIT join returns the latest feature values. + """ + entity_dfs: List[pd.DataFrame] = [] + repo_path = str(config.repo_path) if config.repo_path else "" + + for fv in feature_views: + source = fv.batch_source + assert source is not None, f"Feature view '{fv.name}' has no batch_source" + table = data_source_reader(source, repo_path) + ts_col = source.timestamp_field + + if fv.projection.join_key_map: + join_keys = list(fv.projection.join_key_map.values()) + elif fv.entity_columns: + join_keys = [e.name for e in fv.entity_columns] + else: + join_keys = list(fv.entities) if fv.entities else [] + + filtered = table.filter( + (table[ts_col] >= ibis.literal(start_date)) + & (table[ts_col] <= ibis.literal(end_date)) + ) + + if join_keys: + sub_df = filtered.select(join_keys).execute().drop_duplicates() + else: + sub_df = pd.DataFrame(index=[0]) + + entity_dfs.append(sub_df) + + if not entity_dfs: + return pd.DataFrame(columns=["event_timestamp"]) + + combined = pd.concat(entity_dfs, ignore_index=True).drop_duplicates() + combined["event_timestamp"] = ( + pd.Timestamp(end_date).tz_convert("UTC") + if pd.Timestamp(end_date).tzinfo + else pd.Timestamp(end_date, tz="UTC") + ) + return combined + + class DuckDBOfflineStore(OfflineStore): @staticmethod def pull_latest_from_table_or_query( @@ -527,11 +581,30 @@ 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, + start_date: Optional[datetime] = None, + end_date: Optional[datetime] = None, ) -> RetrievalJob: + non_entity_mode = entity_df is None + + if non_entity_mode: + start_date, end_date = compute_non_entity_date_range( + feature_views, + start_date=start_date, + end_date=end_date, + ) + + entity_df = _build_entity_df_from_sources( + config=config, + feature_views=feature_views, + start_date=start_date, + end_date=end_date, + data_source_reader=_read_data_source, + ) + return get_historical_features_ibis( config=config, feature_views=feature_views, diff --git a/sdk/python/tests/integration/offline_store/test_non_entity_mode.py b/sdk/python/tests/integration/offline_store/test_non_entity_mode.py index f17352fb356..9879f331a09 100644 --- a/sdk/python/tests/integration/offline_store/test_non_entity_mode.py +++ b/sdk/python/tests/integration/offline_store/test_non_entity_mode.py @@ -11,8 +11,7 @@ @pytest.mark.integration -@pytest.mark.universal_offline_stores -@pytest.mark.ray_offline_stores_only +@pytest.mark.universal_offline_stores(only=["file", "duckdb"]) def test_non_entity_mode_basic(environment, universal_data_sources): """Test historical features retrieval without entity_df (non-entity mode). diff --git a/sdk/python/tests/integration/offline_store/test_universal_historical_retrieval.py b/sdk/python/tests/integration/offline_store/test_universal_historical_retrieval.py index af0de479f3e..92d89a74f3a 100644 --- a/sdk/python/tests/integration/offline_store/test_universal_historical_retrieval.py +++ b/sdk/python/tests/integration/offline_store/test_universal_historical_retrieval.py @@ -731,7 +731,7 @@ def test_historical_features_field_mapping( @pytest.mark.integration -@pytest.mark.universal_offline_stores(only=["file"]) +@pytest.mark.universal_offline_stores(only=["file", "duckdb"]) def test_historical_features_non_entity_retrieval(environment): """Test get_historical_features with entity_df=None using start_date/end_date. diff --git a/sdk/python/tests/unit/infra/offline_stores/test_duckdb_non_entity.py b/sdk/python/tests/unit/infra/offline_stores/test_duckdb_non_entity.py new file mode 100644 index 00000000000..68652a0e61e --- /dev/null +++ b/sdk/python/tests/unit/infra/offline_stores/test_duckdb_non_entity.py @@ -0,0 +1,554 @@ +from datetime import datetime, timedelta, timezone +from unittest.mock import MagicMock, patch + +import ibis +import pandas as pd + +from feast.entity import Entity +from feast.feature_view import FeatureView, Field +from feast.infra.offline_stores import duckdb as duckdb_mod +from feast.infra.offline_stores.duckdb import ( + DuckDBOfflineStore, + DuckDBOfflineStoreConfig, +) +from feast.infra.offline_stores.file_source import FileSource +from feast.infra.offline_stores.offline_store import RetrievalJob +from feast.repo_config import RepoConfig +from feast.types import Float32, Int64, ValueType + + +def _mock_duckdb_offline_store_config(): + return DuckDBOfflineStoreConfig(type="duckdb") + + +def _mock_entity(): + return [ + Entity( + name="driver_id", + join_keys=["driver_id"], + description="Driver ID", + value_type=ValueType.INT64, + ) + ] + + +def _mock_feature_view(name: str = "driver_stats", ttl: timedelta = None): + return FeatureView( + name=name, + entities=_mock_entity(), + schema=[ + Field(name="conv_rate", dtype=Float32), + ], + source=FileSource( + path="dummy.parquet", + timestamp_field="event_timestamp", + ), + ttl=ttl, + ) + + +def _mock_data_source_reader(src_df): + """Return a data_source_reader that wraps a pandas DataFrame as an ibis memtable.""" + + def reader(data_source, repo_path): + return ibis.memtable(src_df) + + return reader + + +def test_duckdb_non_entity_historical_retrieval_accepts_dates(monkeypatch): + src = pd.DataFrame( + { + "driver_id": [1], + "event_timestamp": pd.to_datetime(["2023-01-01T12:00:00Z"]), + "conv_rate": [0.5], + } + ) + monkeypatch.setattr(duckdb_mod, "_read_data_source", _mock_data_source_reader(src)) + + repo_config = RepoConfig( + project="test_project", + registry="test_registry", + provider="local", + offline_store=_mock_duckdb_offline_store_config(), + ) + + fv = _mock_feature_view() + + retrieval_job = DuckDBOfflineStore.get_historical_features( + config=repo_config, + feature_views=[fv], + feature_refs=["driver_stats:conv_rate"], + entity_df=None, + registry=MagicMock(), + project="test_project", + full_feature_names=False, + start_date=datetime(2023, 1, 1, tzinfo=timezone.utc), + end_date=datetime(2023, 1, 2, tzinfo=timezone.utc), + ) + + assert isinstance(retrieval_job, RetrievalJob) + + +class TestNonEntityRetrieval: + """Test suite for non-entity retrieval functionality (entity_df=None) in DuckDB offline store.""" + + def test_duckdb_non_entity_snapshot_ttl_and_dedup(self, monkeypatch): + src = pd.DataFrame( + { + "driver_id": [1, 1, 1, 2, 2], + "event_timestamp": pd.to_datetime( + [ + "2025-01-01T10:00:00Z", + "2025-01-01T10:00:00Z", + "2024-12-30T10:00:00Z", + "2025-01-01T12:00:00Z", + "2025-01-02T11:00:00Z", + ] + ), + "created_ts": pd.to_datetime( + [ + "2025-01-01T10:00:01Z", + "2025-01-01T10:00:02Z", + "2024-12-30T10:00:00Z", + "2025-01-01T12:00:00Z", + "2025-01-02T11:00:00Z", + ] + ), + "conv_rate": [0.1, 0.2, 0.05, 0.3, 0.4], + } + ) + monkeypatch.setattr( + duckdb_mod, "_read_data_source", _mock_data_source_reader(src) + ) + + fv = FeatureView( + name="driver_stats", + entities=_mock_entity(), + schema=[ + Field(name="driver_id", dtype=Int64), + Field(name="conv_rate", dtype=Float32), + ], + source=FileSource( + path="unused", + timestamp_field="event_timestamp", + created_timestamp_column="created_ts", + ), + ttl=timedelta(days=1), + ) + + repo_config = RepoConfig( + project="proj", + registry="unused", + provider="local", + offline_store=_mock_duckdb_offline_store_config(), + ) + + end = datetime(2025, 1, 2, 10, 0, tzinfo=timezone.utc) + start = end - timedelta(days=7) + + job = DuckDBOfflineStore.get_historical_features( + config=repo_config, + feature_views=[fv], + feature_refs=["driver_stats:conv_rate"], + entity_df=None, + registry=MagicMock(), + project="proj", + full_feature_names=False, + start_date=start, + end_date=end, + ) + + df = job.to_df() + + assert set(df["driver_id"]) == {1, 2} + out = df.set_index("driver_id")["conv_rate"].to_dict() + assert out[1] == 0.2 + assert out[2] == 0.3 + + def test_non_entity_mode_with_both_dates_retrieves_data(self, monkeypatch): + src = pd.DataFrame( + { + "driver_id": [1, 1, 1, 2], + "event_timestamp": pd.to_datetime( + [ + "2023-01-01T08:00:00Z", + "2023-01-03T10:00:00Z", + "2023-01-05T12:00:00Z", + "2023-01-08T14:00:00Z", + ] + ), + "conv_rate": [0.1, 0.2, 0.3, 0.4], + } + ) + monkeypatch.setattr( + duckdb_mod, "_read_data_source", _mock_data_source_reader(src) + ) + + repo_config = RepoConfig( + project="test_project", + registry="test_registry", + provider="local", + offline_store=_mock_duckdb_offline_store_config(), + ) + + fv = FeatureView( + name="test_fv", + entities=_mock_entity(), + schema=[ + Field(name="driver_id", dtype=Int64), + Field(name="conv_rate", dtype=Float32), + ], + source=FileSource(path="unused", timestamp_field="event_timestamp"), + ttl=None, + ) + + start_date = datetime(2023, 1, 2, tzinfo=timezone.utc) + end_date = datetime(2023, 1, 7, tzinfo=timezone.utc) + + job = DuckDBOfflineStore.get_historical_features( + config=repo_config, + feature_views=[fv], + feature_refs=["test_fv:conv_rate"], + entity_df=None, + registry=MagicMock(), + project="test_project", + start_date=start_date, + end_date=end_date, + ) + + df = job.to_df() + + assert job.metadata.min_event_timestamp == end_date + assert job.metadata.max_event_timestamp == end_date + + assert len(df) >= 1 + assert set(df["driver_id"]) == {1} + driver1_data = df[df["driver_id"] == 1] + assert 0.3 in driver1_data["conv_rate"].values + + def test_non_entity_mode_with_end_date_only_calculates_start_from_ttl( + self, monkeypatch + ): + src = pd.DataFrame( + { + "driver_id": [1, 1, 1], + "event_timestamp": pd.to_datetime( + [ + "2023-01-05T08:00:00Z", + "2023-01-06T10:00:00Z", + "2023-01-07T12:00:00Z", + ] + ), + "conv_rate": [0.1, 0.2, 0.3], + } + ) + monkeypatch.setattr( + duckdb_mod, "_read_data_source", _mock_data_source_reader(src) + ) + + repo_config = RepoConfig( + project="test_project", + registry="test_registry", + provider="local", + offline_store=_mock_duckdb_offline_store_config(), + ) + + fv = FeatureView( + name="test_fv", + entities=_mock_entity(), + schema=[ + Field(name="driver_id", dtype=Int64), + Field(name="conv_rate", dtype=Float32), + ], + source=FileSource(path="unused", timestamp_field="event_timestamp"), + ttl=timedelta(days=1), + ) + + end_date = datetime(2023, 1, 7, 12, 0, tzinfo=timezone.utc) + + job = DuckDBOfflineStore.get_historical_features( + config=repo_config, + feature_views=[fv], + feature_refs=["test_fv:conv_rate"], + entity_df=None, + registry=MagicMock(), + project="test_project", + end_date=end_date, + ) + + df = job.to_df() + + driver1_data = df[df["driver_id"] == 1] + assert len(driver1_data) >= 1 + assert 0.3 in driver1_data["conv_rate"].values + + @patch("feast.utils.datetime") + def test_no_dates_provided_defaults_to_current_time_and_filters_data( + self, mock_datetime, monkeypatch + ): + fixed_now = datetime(2023, 1, 7, 12, 0, 0, tzinfo=timezone.utc) + mock_datetime.now.return_value = fixed_now + mock_datetime.side_effect = lambda *args, **kw: datetime(*args, **kw) + + src = pd.DataFrame( + { + "driver_id": [1, 1, 1], + "event_timestamp": pd.to_datetime( + [ + "2023-01-05T12:00:00Z", + "2023-01-06T18:00:00Z", + "2023-01-07T11:00:00Z", + ] + ), + "conv_rate": [0.1, 0.2, 0.3], + } + ) + monkeypatch.setattr( + duckdb_mod, "_read_data_source", _mock_data_source_reader(src) + ) + + repo_config = RepoConfig( + project="test_project", + registry="test_registry", + provider="local", + offline_store=_mock_duckdb_offline_store_config(), + ) + + fv = FeatureView( + name="test_fv", + entities=_mock_entity(), + schema=[ + Field(name="driver_id", dtype=Int64), + Field(name="conv_rate", dtype=Float32), + ], + source=FileSource(path="unused", timestamp_field="event_timestamp"), + ttl=timedelta(days=1), + ) + + job = DuckDBOfflineStore.get_historical_features( + config=repo_config, + feature_views=[fv], + feature_refs=["test_fv:conv_rate"], + entity_df=None, + registry=MagicMock(), + project="test_project", + ) + + df = job.to_df() + + driver1_data = df[df["driver_id"] == 1] + assert len(driver1_data) >= 1 + assert 0.3 in driver1_data["conv_rate"].values + + def test_ttl_filtering_excludes_old_rows(self, monkeypatch): + src = pd.DataFrame( + { + "driver_id": [1, 1, 1], + "event_timestamp": pd.to_datetime( + [ + "2025-01-01T08:00:00Z", + "2025-01-01T09:30:00Z", + "2025-01-01T10:00:00Z", + ] + ), + "conv_rate": [0.1, 0.2, 0.3], + } + ) + monkeypatch.setattr( + duckdb_mod, "_read_data_source", _mock_data_source_reader(src) + ) + + fv = FeatureView( + name="driver_stats", + entities=_mock_entity(), + schema=[ + Field(name="driver_id", dtype=Int64), + Field(name="conv_rate", dtype=Float32), + ], + source=FileSource(path="unused", timestamp_field="event_timestamp"), + ttl=timedelta(hours=1), + ) + + repo_config = RepoConfig( + project="proj", + registry="unused", + provider="local", + offline_store=_mock_duckdb_offline_store_config(), + ) + + end = datetime(2025, 1, 1, 10, 0, tzinfo=timezone.utc) + start = end - timedelta(days=1) + + job = DuckDBOfflineStore.get_historical_features( + config=repo_config, + feature_views=[fv], + feature_refs=["driver_stats:conv_rate"], + entity_df=None, + registry=MagicMock(), + project="proj", + full_feature_names=False, + start_date=start, + end_date=end, + ) + + df = job.to_df() + + driver1_data = df[df["driver_id"] == 1] + assert len(driver1_data) >= 1 + assert ( + 0.2 in driver1_data["conv_rate"].values + or 0.3 in driver1_data["conv_rate"].values + ) + assert 0.1 not in driver1_data["conv_rate"].values + + def test_multiple_feature_views_with_different_ttls(self, monkeypatch): + src1 = pd.DataFrame( + { + "driver_id": [1, 1], + "event_timestamp": pd.to_datetime( + [ + "2025-01-01T08:00:00Z", + "2025-01-01T09:30:00Z", + ] + ), + "age": [25, 26], + } + ) + + src2 = pd.DataFrame( + { + "driver_id": [1, 1], + "event_timestamp": pd.to_datetime( + [ + "2025-01-01T08:00:00Z", + "2025-01-01T09:30:00Z", + ] + ), + "total_trips": [100, 101], + } + ) + + def mock_read_data_source(data_source, repo_path): + if data_source.path == "unused1": + return ibis.memtable(src1) + return ibis.memtable(src2) + + monkeypatch.setattr(duckdb_mod, "_read_data_source", mock_read_data_source) + + fv1 = FeatureView( + name="user_fv", + entities=_mock_entity(), + schema=[ + Field(name="driver_id", dtype=Int64), + Field(name="age", dtype=Float32), + ], + source=FileSource(path="unused1", timestamp_field="event_timestamp"), + ttl=timedelta(hours=1), + ) + + fv2 = FeatureView( + name="transaction_fv", + entities=_mock_entity(), + schema=[ + Field(name="driver_id", dtype=Int64), + Field(name="total_trips", dtype=Float32), + ], + source=FileSource(path="unused2", timestamp_field="event_timestamp"), + ttl=timedelta(days=1), + ) + + repo_config = RepoConfig( + project="proj", + registry="unused", + provider="local", + offline_store=_mock_duckdb_offline_store_config(), + ) + + end = datetime(2025, 1, 1, 10, 0, tzinfo=timezone.utc) + start = end - timedelta(days=1) + + job = DuckDBOfflineStore.get_historical_features( + config=repo_config, + feature_views=[fv1, fv2], + feature_refs=["user_fv:age", "transaction_fv:total_trips"], + entity_df=None, + registry=MagicMock(), + project="proj", + full_feature_names=False, + start_date=start, + end_date=end, + ) + + df = job.to_df() + + assert "age" in df.columns or "user_fv__age" in df.columns + assert ( + "total_trips" in df.columns or "transaction_fv__total_trips" in df.columns + ) + + if "age" in df.columns: + age_values = df["age"].values + assert 26 in age_values + assert 25 not in age_values + + if "total_trips" in df.columns: + trips_values = df["total_trips"].values + assert 101 in trips_values + + def test_entity_df_still_works(self, monkeypatch): + """Verify standard entity_df path is not broken by the changes.""" + src = pd.DataFrame( + { + "driver_id": [1, 2], + "event_timestamp": pd.to_datetime( + ["2025-01-01T10:00:00Z", "2025-01-01T11:00:00Z"] + ), + "conv_rate": [0.5, 0.6], + } + ) + monkeypatch.setattr( + duckdb_mod, "_read_data_source", _mock_data_source_reader(src) + ) + + repo_config = RepoConfig( + project="proj", + registry="unused", + provider="local", + offline_store=_mock_duckdb_offline_store_config(), + ) + + fv = FeatureView( + name="driver_stats", + entities=_mock_entity(), + schema=[ + Field(name="driver_id", dtype=Int64), + Field(name="conv_rate", dtype=Float32), + ], + source=FileSource(path="unused", timestamp_field="event_timestamp"), + ttl=timedelta(days=1), + ) + + entity_df = pd.DataFrame( + { + "driver_id": [1, 2], + "event_timestamp": pd.to_datetime( + ["2025-01-01T12:00:00Z", "2025-01-01T12:00:00Z"] + ), + } + ) + + job = DuckDBOfflineStore.get_historical_features( + config=repo_config, + feature_views=[fv], + feature_refs=["driver_stats:conv_rate"], + entity_df=entity_df, + registry=MagicMock(), + project="proj", + full_feature_names=False, + ) + + df = job.to_df() + + assert set(df["driver_id"]) == {1, 2} + assert len(df) == 2 diff --git a/sdk/python/tests/universal/feature_repos/repo_configuration.py b/sdk/python/tests/universal/feature_repos/repo_configuration.py index 7da8d54f433..f79445e24f6 100644 --- a/sdk/python/tests/universal/feature_repos/repo_configuration.py +++ b/sdk/python/tests/universal/feature_repos/repo_configuration.py @@ -93,6 +93,15 @@ "file": ("local", FileDataSourceCreator), } +try: + from tests.universal.feature_repos.duckdb_repo_configuration import ( + DuckDBDataSourceCreator, + ) + + OFFLINE_STORE_TO_PROVIDER_CONFIG["duckdb"] = ("local", DuckDBDataSourceCreator) +except ImportError: + pass + AVAILABLE_OFFLINE_STORES: List[Tuple[str, Type[DataSourceCreator]]] = [ ("local", FileDataSourceCreator), ("local", RemoteOfflineStoreDataSourceCreator),