diff --git a/sdk/python/feast/feature_server.py b/sdk/python/feast/feature_server.py index bba91130db3..8cb6dff91c6 100644 --- a/sdk/python/feast/feature_server.py +++ b/sdk/python/feast/feature_server.py @@ -90,12 +90,14 @@ class MaterializeRequest(BaseModel): feature_views: Optional[List[str]] = None disable_event_timestamp: bool = False full_feature_names: bool = False + version: Optional[str] = None class MaterializeIncrementalRequest(BaseModel): end_ts: str feature_views: Optional[List[str]] = None full_feature_names: bool = False + version: Optional[str] = None class GetOnlineFeaturesRequest(BaseModel): @@ -708,6 +710,134 @@ async def materialize_incremental(request: MaterializeIncrementalRequest) -> Non full_feature_names=request.full_feature_names, ) + def _authorize_materialize_views( + request_feature_views: Optional[List[str]], + ) -> List[str]: + """Shared authz + FV name resolution for async materialize endpoints.""" + feature_views_to_materialize = store._get_feature_views_to_materialize( + request_feature_views + ) + for fv in feature_views_to_materialize: + assert_permissions( + resource=fv, + actions=[AuthzedAction.WRITE_ONLINE], + ) + return [fv.name for fv in feature_views_to_materialize] + + def _parse_materialize_timestamps(request: MaterializeRequest): + """Parse and validate timestamps from a materialize request.""" + if request.disable_event_timestamp: + now = datetime.now() + return datetime(1970, 1, 1), now + + if not request.start_ts or not request.end_ts: + raise ValueError( + "start_ts and end_ts are required when disable_event_timestamp is False" + ) + try: + start_date = utils.make_tzaware(parser.parse(request.start_ts)) + end_date = utils.make_tzaware(parser.parse(request.end_ts)) + except (ValueError, TypeError) as e: + raise ValueError(f"Invalid timestamp format: {e}") from e + + if start_date >= end_date: + raise ValueError( + f"start_ts ({start_date}) must be before end_ts ({end_date})" + ) + return start_date, end_date + + def _check_already_materializing(fv_names: List[str]) -> Optional[JSONResponse]: + """Return 409 if any requested FV is already materializing.""" + from feast.feature_view import FeatureViewState + + for fv_name in fv_names: + try: + fv = store.registry.get_feature_view( + fv_name, store.project, allow_cache=False + ) + if getattr(fv, "state", None) == FeatureViewState.MATERIALIZING: + return JSONResponse( + status_code=409, + content={ + "error": f"Feature view '{fv_name}' is already materializing" + }, + ) + except Exception: + pass + return None + + @app.post("/materialize-async", dependencies=[Depends(inject_user_details)]) + async def materialize_async(request: MaterializeRequest) -> JSONResponse: + with feast_metrics.track_request_latency("/materialize-async"): + fv_names = _authorize_materialize_views(request.feature_views) + start_date, end_date = _parse_materialize_timestamps(request) + + conflict = _check_already_materializing(fv_names) + if conflict: + return conflict + + def _run_materialize(): + try: + store.materialize( + start_date, + end_date, + request.feature_views, + disable_event_timestamp=request.disable_event_timestamp, + full_feature_names=request.full_feature_names, + version=request.version, + _force_local=True, + ) + except Exception as e: + logger.error( + f"Async materialization failed for {fv_names}: {e}", + exc_info=True, + ) + + loop = asyncio.get_running_loop() + loop.run_in_executor(None, _run_materialize) + + return JSONResponse( + status_code=202, + content={"status": "accepted", "feature_views": fv_names}, + ) + + @app.post( + "/materialize-incremental-async", + dependencies=[Depends(inject_user_details)], + ) + async def materialize_incremental_async( + request: MaterializeIncrementalRequest, + ) -> JSONResponse: + with feast_metrics.track_request_latency("/materialize-incremental-async"): + fv_names = _authorize_materialize_views(request.feature_views) + + conflict = _check_already_materializing(fv_names) + if conflict: + return conflict + + def _run_materialize_incremental(): + try: + store.materialize_incremental( + utils.make_tzaware(parser.parse(request.end_ts)), + request.feature_views, + full_feature_names=request.full_feature_names, + version=request.version, + _force_local=True, + ) + except Exception as e: + logger.error( + f"Async incremental materialization failed for {fv_names}: {e}", + exc_info=True, + ) + + loop = asyncio.get_running_loop() + loop.run_in_executor(None, _run_materialize_incremental) + + return JSONResponse( + status_code=202, + content={"status": "accepted", "feature_views": fv_names}, + ) + @app.exception_handler(Exception) async def rest_exception_handler(request: Request, exc: Exception): # Print the original exception on the server side diff --git a/sdk/python/feast/feature_store.py b/sdk/python/feast/feature_store.py index 75f406c8e2c..633d2f67f9f 100644 --- a/sdk/python/feast/feature_store.py +++ b/sdk/python/feast/feature_store.py @@ -2192,12 +2192,201 @@ def _materialize_odfv( ) self.write_to_online_store(feature_view.name, df=transformed_df) + def _is_remote_materialize_mode(self) -> bool: + """Check if materialization should be delegated to a remote feature server.""" + fs_cfg = self.config.feature_server + if fs_cfg is None: + return False + return getattr(fs_cfg, "materialize_mode", "local") == "remote" + + def _get_feature_server_url(self) -> str: + """Get feature server URL, raising if not configured.""" + from urllib.parse import urlparse + + fs_cfg = self.config.feature_server + url = getattr(fs_cfg, "url", None) + if not url: + raise ValueError( + "feature_server.url must be set when materialize_mode='remote'. " + "Example: feature_server:\n" + " type: local\n" + " materialize_mode: remote\n" + " url: http://feast-online.feast-ns.svc.cluster.local:80" + ) + parsed = urlparse(url) + if parsed.scheme not in ("http", "https"): + raise ValueError( + f"feature_server.url must use http or https scheme, got: {url}" + ) + return url.rstrip("/") + + def _get_remote_http_session(self): + """Get an HTTP session with auth configured for the feature server.""" + import requests + + auth_config = getattr(self.config, "auth_config", None) + if auth_config and getattr(auth_config, "type", "no_auth") != "no_auth": + from feast.permissions.client.http_auth_requests_wrapper import ( + get_http_auth_requests_session, + ) + + return get_http_auth_requests_session(auth_config) + + return requests.Session() + + def _remote_materialize_common( + self, + endpoint: str, + payload: dict, + fv_names: List[str], + ) -> None: + """Shared logic for remote materialization: trigger server, poll registry. + + Args: + endpoint: Server endpoint path (e.g., "/materialize-async"). + payload: JSON payload for the POST request. + fv_names: Feature view names to track. + """ + from feast.materialization_status import ( + FVMaterializationStatus, + FVResult, + poll_materialization_status, + ) + + server_url = self._get_feature_server_url() + fs_cfg = self.config.feature_server + + _logger.info( + f"Remote materialize: triggering {len(fv_names)} feature views " + f"via {endpoint} on {server_url}" + ) + + session = self._get_remote_http_session() + try: + resp = session.post( + f"{server_url}{endpoint}", + json=payload, + timeout=getattr(fs_cfg, "http_timeout", 30), + ) + resp.raise_for_status() + except Exception as e: + raise Exception( + f"Failed to trigger remote materialization at " + f"{server_url}{endpoint}: {e}" + ) from e + finally: + session.close() + + seen_materializing: set = set() + + def registry_status_fn(fv_name: str) -> FVMaterializationStatus: + fv = self.registry.get_feature_view( + fv_name, self.project, allow_cache=False + ) + state = getattr(fv, "state", None) + if state == FeatureViewState.AVAILABLE_ONLINE: + return FVMaterializationStatus.SUCCEEDED + elif state == FeatureViewState.MATERIALIZING: + seen_materializing.add(fv_name) + return FVMaterializationStatus.RUNNING + elif state == FeatureViewState.GENERATED: + if fv_name in seen_materializing: + return FVMaterializationStatus.FAILED + return FVMaterializationStatus.PENDING + return FVMaterializationStatus.PENDING + + def on_change(result: FVResult): + symbol = { + FVMaterializationStatus.SUCCEEDED: "+", + FVMaterializationStatus.FAILED: "x", + FVMaterializationStatus.RUNNING: "~", + }.get(result.status, "?") + _logger.info( + f" [{symbol}] {result.name}: {result.status.value} " + f"({result.elapsed_seconds:.1f}s)" + ) + + timeout = getattr(fs_cfg, "materialize_timeout", 3600.0) + poll_interval = getattr(fs_cfg, "materialize_poll_interval", 5.0) + + results = poll_materialization_status( + feature_view_names=fv_names, + status_fn=registry_status_fn, + poll_interval=poll_interval, + timeout=timeout, + on_status_change=on_change, + ) + + failed = [r for r in results if r.status == FVMaterializationStatus.FAILED] + if failed: + names = ", ".join(r.name for r in failed) + errors = "; ".join(f"{r.name}: {r.error}" for r in failed if r.error) + msg = f"Remote materialization failed for: {names}" + if errors: + msg += f" ({errors})" + raise Exception(msg) + + def _remote_materialize( + self, + start_date: datetime, + end_date: datetime, + feature_views: Optional[List[str]] = None, + disable_event_timestamp: bool = False, + full_feature_names: bool = False, + version: Optional[str] = None, + ) -> None: + """Delegate materialization to the feature server, poll via registry.""" + fv_names = feature_views or [ + fv.name + for fv in self._get_feature_views_to_materialize(None) + if not isinstance(fv, OnDemandFeatureView) + ] + + payload: dict = { + "start_ts": start_date.isoformat(), + "end_ts": end_date.isoformat(), + "feature_views": fv_names, + "disable_event_timestamp": disable_event_timestamp, + "full_feature_names": full_feature_names, + } + if version is not None: + payload["version"] = version + + self._remote_materialize_common("/materialize-async", payload, fv_names) + + def _remote_materialize_incremental( + self, + end_date: datetime, + feature_views: Optional[List[str]] = None, + full_feature_names: bool = False, + version: Optional[str] = None, + ) -> None: + """Delegate incremental materialization to the feature server, poll via registry.""" + fv_names = feature_views or [ + fv.name + for fv in self._get_feature_views_to_materialize(None) + if not isinstance(fv, OnDemandFeatureView) + ] + + payload: dict = { + "end_ts": end_date.isoformat(), + "feature_views": fv_names, + "full_feature_names": full_feature_names, + } + if version is not None: + payload["version"] = version + + self._remote_materialize_common( + "/materialize-incremental-async", payload, fv_names + ) + def materialize_incremental( self, end_date: datetime, feature_views: Optional[List[str]] = None, full_feature_names: bool = False, version: Optional[str] = None, + _force_local: bool = False, ) -> None: """ Materialize incremental new data from the offline store into the online store. @@ -2231,6 +2420,14 @@ def materialize_incremental( ... """ + if not _force_local and self._is_remote_materialize_mode(): + return self._remote_materialize_incremental( + end_date=end_date, + feature_views=feature_views, + full_feature_names=full_feature_names, + version=version, + ) + parsed_version = self._validate_materialize_version(version, feature_views) feature_views_to_materialize = self._get_feature_views_to_materialize( feature_views, version=parsed_version @@ -2404,6 +2601,7 @@ def materialize( disable_event_timestamp: bool = False, full_feature_names: bool = False, version: Optional[str] = None, + _force_local: bool = False, ) -> None: """ Materialize data from the offline store into the online store. @@ -2441,6 +2639,16 @@ def materialize( f"The given start_date {start_date} is greater than the given end_date {end_date}." ) + if not _force_local and self._is_remote_materialize_mode(): + return self._remote_materialize( + start_date=start_date, + end_date=end_date, + feature_views=feature_views, + disable_event_timestamp=disable_event_timestamp, + full_feature_names=full_feature_names, + version=version, + ) + parsed_version = self._validate_materialize_version(version, feature_views) feature_views_to_materialize = self._get_feature_views_to_materialize( feature_views, version=parsed_version diff --git a/sdk/python/feast/infra/feature_servers/local_process/config.py b/sdk/python/feast/infra/feature_servers/local_process/config.py index 942927ec2e8..47435914b1f 100644 --- a/sdk/python/feast/infra/feature_servers/local_process/config.py +++ b/sdk/python/feast/infra/feature_servers/local_process/config.py @@ -1,4 +1,6 @@ -from typing import Literal +from typing import Literal, Optional + +from pydantic import model_validator from feast.infra.feature_servers.base_config import BaseFeatureServerConfig @@ -9,3 +11,25 @@ class LocalFeatureServerConfig(BaseFeatureServerConfig): # The endpoint definition for transformation_service transformation_service_endpoint: str = "localhost:6569" + + materialize_mode: Literal["local", "remote"] = "local" + """When 'remote', store.materialize() delegates to the feature server's + async endpoint instead of running the batch engine locally.""" + + url: Optional[str] = None + """Feature server URL for remote materialization (e.g. http://feast-online:80). + Required when materialize_mode is 'remote'.""" + + materialize_timeout: float = 3600.0 + """Max seconds to wait for remote materialization to complete.""" + + materialize_poll_interval: float = 5.0 + """Seconds between polling the registry for FV state updates.""" + + @model_validator(mode="after") + def _validate_remote_requires_url(self) -> "LocalFeatureServerConfig": + if self.materialize_mode == "remote" and not self.url: + raise ValueError( + "feature_server.url must be set when materialize_mode is 'remote'" + ) + return self diff --git a/sdk/python/feast/materialization_status.py b/sdk/python/feast/materialization_status.py new file mode 100644 index 00000000000..4d3d419e052 --- /dev/null +++ b/sdk/python/feast/materialization_status.py @@ -0,0 +1,93 @@ +import logging +import time +from dataclasses import dataclass +from enum import Enum +from typing import Callable, Dict, List, Optional, Set + +logger = logging.getLogger(__name__) + + +class FVMaterializationStatus(Enum): + PENDING = "pending" + RUNNING = "running" + SUCCEEDED = "succeeded" + FAILED = "failed" + + +@dataclass +class FVResult: + name: str + status: FVMaterializationStatus + elapsed_seconds: Optional[float] = None + error: Optional[str] = None + + +def poll_materialization_status( + feature_view_names: List[str], + status_fn: Callable[[str], FVMaterializationStatus], + poll_interval: float = 5.0, + timeout: float = 3600.0, + on_status_change: Optional[Callable[[FVResult], None]] = None, +) -> List[FVResult]: + """Poll materialization status for a list of feature views. + + This function is engine-agnostic. The caller provides a status_fn + that maps a feature view name to its current materialization status. + + Used by: + - Client-side remote materialize: status_fn reads FV state from remote registry + - Server-side batch engines: status_fn reads from engine job objects + + Args: + feature_view_names: FVs to track. + status_fn: Returns current status for a given FV name. + poll_interval: Seconds between polls. + timeout: Max seconds to wait before declaring timeout. + on_status_change: Callback invoked when a FV's status changes. + + Returns: + List of FVResult with final status for each FV. + """ + start = time.monotonic() + pending: Set[str] = set(feature_view_names) + results: Dict[str, FVResult] = { + name: FVResult(name=name, status=FVMaterializationStatus.PENDING) + for name in feature_view_names + } + previous_statuses: Dict[str, FVMaterializationStatus] = {} + + while pending and (time.monotonic() - start) < timeout: + for name in list(pending): + try: + current = status_fn(name) + except Exception as e: + logger.warning(f"Error polling status for {name}: {e}") + continue + + elapsed = time.monotonic() - start + + if current != previous_statuses.get(name): + result = FVResult(name=name, status=current, elapsed_seconds=elapsed) + results[name] = result + previous_statuses[name] = current + if on_status_change: + on_status_change(result) + + if current in ( + FVMaterializationStatus.SUCCEEDED, + FVMaterializationStatus.FAILED, + ): + pending.discard(name) + + if pending: + time.sleep(poll_interval) + + for name in pending: + results[name] = FVResult( + name=name, + status=FVMaterializationStatus.FAILED, + elapsed_seconds=time.monotonic() - start, + error=f"Materialization timed out after {timeout}s. Check server logs for details.", + ) + + return list(results.values()) diff --git a/sdk/python/tests/integration/materialization/test_remote_materialize.py b/sdk/python/tests/integration/materialization/test_remote_materialize.py new file mode 100644 index 00000000000..2bd17dcb4fc --- /dev/null +++ b/sdk/python/tests/integration/materialization/test_remote_materialize.py @@ -0,0 +1,229 @@ +"""Integration tests for remote materialization. + +Runs a real feature server in a background thread, configures a client +FeatureStore with materialize_mode=remote, and verifies the full flow: + client.materialize() → POST /materialize-async → server runs locally → + FV state transitions → client polls registry → returns success +""" + +import socket +import threading +import time +from datetime import datetime, timedelta, timezone + +import pandas as pd +import pytest +import uvicorn + +from feast import Entity, FeatureStore, FeatureView, Field, RepoConfig +from feast.feature_server import get_app +from feast.feature_view import FeatureViewState +from feast.infra.offline_stores.file_source import FileSource +from feast.infra.online_stores.sqlite import SqliteOnlineStoreConfig +from feast.types import Float32, Int64 + + +def _find_free_port() -> int: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + s.bind(("127.0.0.1", 0)) + return s.getsockname()[1] + + +@pytest.fixture +def e2e_repo(tmp_path): + """Set up a minimal feature repo with parquet data, registry, and online store.""" + registry_path = str(tmp_path / "registry.db") + online_store_path = str(tmp_path / "online_store.db") + data_path = str(tmp_path / "data.parquet") + + now = datetime.now(tz=timezone.utc) + df = pd.DataFrame( + { + "entity_id": [1, 2, 3, 4, 5], + "feature_a": [10.0, 20.0, 30.0, 40.0, 50.0], + "feature_b": [100, 200, 300, 400, 500], + "event_timestamp": [now - timedelta(hours=i) for i in range(5)], + "created": [now] * 5, + } + ) + df.to_parquet(data_path) + + config = RepoConfig( + project="test_remote_mat", + provider="local", + registry=registry_path, + online_store=SqliteOnlineStoreConfig(path=online_store_path), + entity_key_serialization_version=3, + ) + + store = FeatureStore(config=config) + + entity = Entity(name="entity_id", join_keys=["entity_id"]) + + source = FileSource( + path=data_path, + timestamp_field="event_timestamp", + created_timestamp_column="created", + ) + + fv = FeatureView( + name="test_feature_view", + entities=[entity], + schema=[ + Field(name="feature_a", dtype=Float32), + Field(name="feature_b", dtype=Int64), + ], + source=source, + ttl=timedelta(days=7), + ) + + store.apply([entity, fv]) + + return store, config, registry_path, online_store_path + + +@pytest.fixture +def feature_server(e2e_repo): + """Start a real feature server in a background thread on a free port.""" + store, config, registry_path, online_store_path = e2e_repo + + app = get_app(store) + server_port = _find_free_port() + + server_config = uvicorn.Config( + app, host="127.0.0.1", port=server_port, log_level="warning" + ) + server = uvicorn.Server(server_config) + + thread = threading.Thread(target=server.run, daemon=True) + thread.start() + + import httpx + + for _ in range(50): + try: + resp = httpx.get(f"http://127.0.0.1:{server_port}/health") + if resp.status_code == 200: + break + except Exception: + pass + time.sleep(0.1) + else: + pytest.fail("Feature server did not start in time") + + yield store, config, server_port + + server.should_exit = True + thread.join(timeout=5) + + +def _make_client_store(server_config, server_port): + """Create a client FeatureStore configured for remote materialization.""" + from feast.infra.feature_servers.local_process.config import ( + LocalFeatureServerConfig, + ) + + client_config = RepoConfig( + project="test_remote_mat", + provider="local", + registry=server_config.registry, + online_store=server_config.online_store, + entity_key_serialization_version=3, + feature_server=LocalFeatureServerConfig( + enabled=True, + materialize_mode="remote", + url=f"http://127.0.0.1:{server_port}", + materialize_timeout=30.0, + materialize_poll_interval=0.5, + ), + ) + return FeatureStore(config=client_config) + + +def test_remote_materialize_e2e(feature_server): + """Full E2E: client with remote mode materializes through server.""" + server_store, server_config, server_port = feature_server + client_store = _make_client_store(server_config, server_port) + + now = datetime.now(tz=timezone.utc) + start_date = now - timedelta(days=7) + end_date = now + + client_store.materialize( + start_date=start_date, + end_date=end_date, + feature_views=["test_feature_view"], + ) + + fv = client_store.registry.get_feature_view("test_feature_view", "test_remote_mat") + assert fv.state == FeatureViewState.AVAILABLE_ONLINE + + online_features = client_store.get_online_features( + features=["test_feature_view:feature_a", "test_feature_view:feature_b"], + entity_rows=[{"entity_id": 1}], + ).to_dict() + + assert online_features["feature_a"][0] == pytest.approx(10.0) + assert online_features["feature_b"][0] == 100 + + +def test_remote_materialize_incremental_e2e(feature_server): + """E2E for materialize_incremental through remote path.""" + server_store, server_config, server_port = feature_server + client_store = _make_client_store(server_config, server_port) + + now = datetime.now(tz=timezone.utc) + + client_store.materialize_incremental( + end_date=now, + feature_views=["test_feature_view"], + ) + + fv = client_store.registry.get_feature_view("test_feature_view", "test_remote_mat") + assert fv.state == FeatureViewState.AVAILABLE_ONLINE + + online_features = client_store.get_online_features( + features=["test_feature_view:feature_a"], + entity_rows=[{"entity_id": 2}], + ).to_dict() + + assert online_features["feature_a"][0] == pytest.approx(20.0) + + +def test_remote_materialize_force_local_bypasses_remote(feature_server): + """_force_local=True runs locally even with materialize_mode=remote.""" + server_store, server_config, server_port = feature_server + client_store = _make_client_store(server_config, server_port) + + now = datetime.now(tz=timezone.utc) + + # With _force_local, it should run locally (same effect, no HTTP call) + client_store.materialize( + start_date=now - timedelta(days=7), + end_date=now, + feature_views=["test_feature_view"], + _force_local=True, + ) + + fv = client_store.registry.get_feature_view("test_feature_view", "test_remote_mat") + assert fv.state == FeatureViewState.AVAILABLE_ONLINE + + +def test_remote_materialize_server_error_propagates(feature_server): + """Client gets an error when the server rejects the request. + + A non-existent feature view causes a server-side error during authz + validation, which returns HTTP 500. The client wraps this as an + Exception with a descriptive message. + """ + server_store, server_config, server_port = feature_server + client_store = _make_client_store(server_config, server_port) + + now = datetime.now(tz=timezone.utc) + + with pytest.raises(Exception, match="Failed to trigger remote materialization"): + client_store.materialize( + start_date=now - timedelta(days=1), + end_date=now, + feature_views=["nonexistent_fv"], + ) diff --git a/sdk/python/tests/unit/test_remote_materialize.py b/sdk/python/tests/unit/test_remote_materialize.py new file mode 100644 index 00000000000..a6d18639f25 --- /dev/null +++ b/sdk/python/tests/unit/test_remote_materialize.py @@ -0,0 +1,507 @@ +"""Tests for remote materialization: shared poller, client-side delegation, and async endpoints.""" + +import time +from datetime import datetime, timezone +from unittest.mock import MagicMock + +import pytest +from fastapi.testclient import TestClient + +from feast.feature_server import get_app +from feast.feature_view import FeatureViewState +from feast.materialization_status import ( + FVMaterializationStatus, + poll_materialization_status, +) + +# --------------------------------------------------------------------------- +# Tests for poll_materialization_status +# --------------------------------------------------------------------------- + + +class TestPollMaterializationStatus: + def test_all_succeed_immediately(self): + def status_fn(name): + return FVMaterializationStatus.SUCCEEDED + + results = poll_materialization_status( + feature_view_names=["fv_a", "fv_b"], + status_fn=status_fn, + poll_interval=0.01, + ) + + assert len(results) == 2 + assert all(r.status == FVMaterializationStatus.SUCCEEDED for r in results) + + def test_mixed_results(self): + statuses = { + "fv_a": FVMaterializationStatus.SUCCEEDED, + "fv_b": FVMaterializationStatus.FAILED, + } + + def status_fn(name): + return statuses[name] + + results = poll_materialization_status( + feature_view_names=["fv_a", "fv_b"], + status_fn=status_fn, + poll_interval=0.01, + ) + + result_map = {r.name: r for r in results} + assert result_map["fv_a"].status == FVMaterializationStatus.SUCCEEDED + assert result_map["fv_b"].status == FVMaterializationStatus.FAILED + + def test_timeout(self): + def status_fn(name): + return FVMaterializationStatus.RUNNING + + results = poll_materialization_status( + feature_view_names=["fv_stuck"], + status_fn=status_fn, + poll_interval=0.01, + timeout=0.05, + ) + + assert len(results) == 1 + assert results[0].status == FVMaterializationStatus.FAILED + assert "timed out" in results[0].error.lower() + + def test_transition_from_pending_to_running_to_succeeded(self): + call_count = {"n": 0} + + def status_fn(name): + call_count["n"] += 1 + if call_count["n"] <= 1: + return FVMaterializationStatus.PENDING + elif call_count["n"] <= 2: + return FVMaterializationStatus.RUNNING + return FVMaterializationStatus.SUCCEEDED + + results = poll_materialization_status( + feature_view_names=["fv_transitioning"], + status_fn=status_fn, + poll_interval=0.01, + ) + + assert results[0].status == FVMaterializationStatus.SUCCEEDED + + def test_on_status_change_callback(self): + changes = [] + call_count = {"n": 0} + + def status_fn(name): + call_count["n"] += 1 + if call_count["n"] <= 1: + return FVMaterializationStatus.RUNNING + return FVMaterializationStatus.SUCCEEDED + + poll_materialization_status( + feature_view_names=["fv_x"], + status_fn=status_fn, + poll_interval=0.01, + on_status_change=lambda r: changes.append(r), + ) + + assert len(changes) == 2 + assert changes[0].status == FVMaterializationStatus.RUNNING + assert changes[1].status == FVMaterializationStatus.SUCCEEDED + + def test_transient_exception_is_handled(self): + """Transient errors from status_fn don't crash the poller.""" + call_count = {"n": 0} + + def status_fn(name): + call_count["n"] += 1 + if call_count["n"] == 1: + raise ConnectionError("transient network error") + return FVMaterializationStatus.SUCCEEDED + + results = poll_materialization_status( + feature_view_names=["fv_err"], + status_fn=status_fn, + poll_interval=0.01, + ) + + assert results[0].status == FVMaterializationStatus.SUCCEEDED + + def test_elapsed_seconds_is_populated(self): + def status_fn(name): + return FVMaterializationStatus.SUCCEEDED + + results = poll_materialization_status( + feature_view_names=["fv_t"], + status_fn=status_fn, + poll_interval=0.01, + ) + + assert results[0].elapsed_seconds is not None + assert results[0].elapsed_seconds >= 0 + + +# --------------------------------------------------------------------------- +# Tests for feature_server async endpoints +# --------------------------------------------------------------------------- + + +class TestFeatureServerAsyncEndpoints: + def _make_client(self): + fs = MagicMock() + fs._get_provider.return_value.async_supported.online.read = False + fs.initialize = MagicMock() + fs.close = MagicMock() + fs.project = "test_project" + + mock_fv = MagicMock() + mock_fv.name = "test_fv" + mock_fv.state = FeatureViewState.GENERATED + fs._get_feature_views_to_materialize.return_value = [mock_fv] + fs.registry.get_feature_view.return_value = mock_fv + + fs.materialize = MagicMock() + fs.materialize_incremental = MagicMock() + + client = TestClient(get_app(fs)) + return client, fs + + def test_materialize_async_returns_202(self): + client, fs = self._make_client() + response = client.post( + "/materialize-async", + json={ + "start_ts": "2024-01-01T00:00:00Z", + "end_ts": "2024-01-02T00:00:00Z", + "feature_views": ["test_fv"], + }, + ) + assert response.status_code == 202 + body = response.json() + assert body["status"] == "accepted" + assert "test_fv" in body["feature_views"] + + def test_materialize_incremental_async_returns_202(self): + client, fs = self._make_client() + response = client.post( + "/materialize-incremental-async", + json={ + "end_ts": "2024-01-02T00:00:00Z", + "feature_views": ["test_fv"], + }, + ) + assert response.status_code == 202 + body = response.json() + assert body["status"] == "accepted" + assert "test_fv" in body["feature_views"] + + def test_materialize_async_validates_timestamps(self): + client, fs = self._make_client() + with pytest.raises(ValueError, match="start_ts and end_ts are required"): + client.post( + "/materialize-async", + json={ + "feature_views": ["test_fv"], + "disable_event_timestamp": False, + }, + ) + + def test_materialize_async_disable_event_timestamp(self): + client, fs = self._make_client() + response = client.post( + "/materialize-async", + json={ + "feature_views": ["test_fv"], + "disable_event_timestamp": True, + }, + ) + assert response.status_code == 202 + + def test_materialize_async_uses_force_local(self): + """Verify the server passes _force_local=True to prevent recursion.""" + client, fs = self._make_client() + response = client.post( + "/materialize-async", + json={ + "start_ts": "2024-01-01T00:00:00Z", + "end_ts": "2024-01-02T00:00:00Z", + "feature_views": ["test_fv"], + }, + ) + assert response.status_code == 202 + # Give the background task time to execute + time.sleep(0.1) + fs.materialize.assert_called_once() + call_kwargs = fs.materialize.call_args[1] + assert call_kwargs["_force_local"] is True + + def test_materialize_async_409_if_already_materializing(self): + """Return 409 if FV is already in MATERIALIZING state.""" + client, fs = self._make_client() + mock_fv = MagicMock() + mock_fv.state = FeatureViewState.MATERIALIZING + fs.registry.get_feature_view.return_value = mock_fv + + response = client.post( + "/materialize-async", + json={ + "start_ts": "2024-01-01T00:00:00Z", + "end_ts": "2024-01-02T00:00:00Z", + "feature_views": ["test_fv"], + }, + ) + assert response.status_code == 409 + + def test_materialize_async_forwards_version(self): + """Version parameter is forwarded to store.materialize.""" + client, fs = self._make_client() + response = client.post( + "/materialize-async", + json={ + "start_ts": "2024-01-01T00:00:00Z", + "end_ts": "2024-01-02T00:00:00Z", + "feature_views": ["test_fv"], + "version": "v2", + }, + ) + assert response.status_code == 202 + time.sleep(0.1) + call_kwargs = fs.materialize.call_args[1] + assert call_kwargs["version"] == "v2" + + def test_materialize_async_validates_timestamp_order(self): + """start_ts must be before end_ts.""" + client, fs = self._make_client() + with pytest.raises(ValueError, match="must be before"): + client.post( + "/materialize-async", + json={ + "start_ts": "2024-01-02T00:00:00Z", + "end_ts": "2024-01-01T00:00:00Z", + "feature_views": ["test_fv"], + }, + ) + + +# --------------------------------------------------------------------------- +# Tests for FeatureStore remote materialize gate +# --------------------------------------------------------------------------- + + +class TestFeatureStoreRemoteGate: + def test_materialize_calls_remote_when_mode_is_remote(self): + """materialize() delegates to _remote_materialize when mode is remote.""" + from feast.feature_store import FeatureStore + + fs = MagicMock() + fs._is_remote_materialize_mode = MagicMock(return_value=True) + fs._remote_materialize = MagicMock() + + # Call the actual materialize method (unbound) with our mock + FeatureStore.materialize( + fs, + start_date=datetime(2024, 1, 1, tzinfo=timezone.utc), + end_date=datetime(2024, 1, 2, tzinfo=timezone.utc), + feature_views=["fv1"], + ) + + fs._remote_materialize.assert_called_once() + + def test_materialize_skips_remote_when_force_local(self): + """materialize() with _force_local=True skips remote even if configured.""" + from feast.feature_store import FeatureStore + + fs = MagicMock() + fs._is_remote_materialize_mode = MagicMock(return_value=True) + fs._remote_materialize = MagicMock() + + # _force_local should bypass the remote gate and proceed to local logic. + # This will fail on the local path (mocked), but _remote_materialize shouldn't be called. + try: + FeatureStore.materialize( + fs, + start_date=datetime(2024, 1, 1, tzinfo=timezone.utc), + end_date=datetime(2024, 1, 2, tzinfo=timezone.utc), + feature_views=["fv1"], + _force_local=True, + ) + except Exception: + pass + + fs._remote_materialize.assert_not_called() + + def test_is_remote_materialize_mode_false_by_default(self): + from feast.feature_store import FeatureStore + + fs = MagicMock() + fs.config.feature_server = None + result = FeatureStore._is_remote_materialize_mode(fs) + assert result is False + + def test_is_remote_materialize_mode_true(self): + from feast.feature_store import FeatureStore + + fs = MagicMock() + fs.config.feature_server.materialize_mode = "remote" + result = FeatureStore._is_remote_materialize_mode(fs) + assert result is True + + def test_is_remote_materialize_mode_local(self): + from feast.feature_store import FeatureStore + + fs = MagicMock() + fs.config.feature_server.materialize_mode = "local" + result = FeatureStore._is_remote_materialize_mode(fs) + assert result is False + + def test_get_feature_server_url_raises_without_url(self): + from feast.feature_store import FeatureStore + + fs = MagicMock() + fs.config.feature_server.url = None + with pytest.raises(ValueError, match="feature_server.url must be set"): + FeatureStore._get_feature_server_url(fs) + + def test_get_feature_server_url_strips_trailing_slash(self): + from feast.feature_store import FeatureStore + + fs = MagicMock() + fs.config.feature_server.url = "http://feast-server:80/" + result = FeatureStore._get_feature_server_url(fs) + assert result == "http://feast-server:80" + + def test_get_feature_server_url_validates_scheme(self): + from feast.feature_store import FeatureStore + + fs = MagicMock() + fs.config.feature_server.url = "ftp://feast-server:80" + with pytest.raises(ValueError, match="http or https"): + FeatureStore._get_feature_server_url(fs) + + +# --------------------------------------------------------------------------- +# Tests for state mapping with seen_materializing tracking +# --------------------------------------------------------------------------- + + +class TestRegistryStatusMapping: + def test_generated_before_materializing_is_pending(self): + """GENERATED state before we've ever seen MATERIALIZING = PENDING.""" + from feast.feature_store import FeatureStore + + fs = MagicMock() + fs.config.feature_server.materialize_mode = "remote" + fs.config.feature_server.url = "http://server:80" + fs.config.feature_server.materialize_timeout = 1.0 + fs.config.feature_server.materialize_poll_interval = 0.01 + fs.config.feature_server.http_timeout = 5 + fs.config.auth_config = None + fs.project = "test" + + mock_session = MagicMock() + mock_resp = MagicMock() + mock_resp.raise_for_status = MagicMock() + mock_session.post.return_value = mock_resp + fs._get_remote_http_session = MagicMock(return_value=mock_session) + fs._get_feature_server_url = MagicMock(return_value="http://server:80") + + # Simulate: GENERATED → MATERIALIZING → AVAILABLE_ONLINE + call_count = {"n": 0} + + def mock_get_fv(name, project, allow_cache=False): + call_count["n"] += 1 + fv = MagicMock() + if call_count["n"] <= 2: + fv.state = FeatureViewState.GENERATED + elif call_count["n"] <= 4: + fv.state = FeatureViewState.MATERIALIZING + else: + fv.state = FeatureViewState.AVAILABLE_ONLINE + return fv + + fs.registry.get_feature_view = mock_get_fv + + FeatureStore._remote_materialize_common( + fs, "/materialize-async", {"feature_views": ["fv1"]}, ["fv1"] + ) + + def test_generated_after_materializing_is_failed(self): + """GENERATED state after we've seen MATERIALIZING = FAILED (rollback).""" + from feast.feature_store import FeatureStore + + fs = MagicMock() + fs.config.feature_server.materialize_mode = "remote" + fs.config.feature_server.url = "http://server:80" + fs.config.feature_server.materialize_timeout = 1.0 + fs.config.feature_server.materialize_poll_interval = 0.01 + fs.config.feature_server.http_timeout = 5 + fs.config.auth_config = None + fs.project = "test" + + mock_session = MagicMock() + mock_resp = MagicMock() + mock_resp.raise_for_status = MagicMock() + mock_session.post.return_value = mock_resp + fs._get_remote_http_session = MagicMock(return_value=mock_session) + fs._get_feature_server_url = MagicMock(return_value="http://server:80") + + # Simulate: MATERIALIZING → GENERATED (rollback = failure) + call_count = {"n": 0} + + def mock_get_fv(name, project, allow_cache=False): + call_count["n"] += 1 + fv = MagicMock() + if call_count["n"] <= 2: + fv.state = FeatureViewState.MATERIALIZING + else: + fv.state = FeatureViewState.GENERATED + return fv + + fs.registry.get_feature_view = mock_get_fv + + with pytest.raises(Exception, match="Remote materialization failed"): + FeatureStore._remote_materialize_common( + fs, "/materialize-async", {"feature_views": ["fv1"]}, ["fv1"] + ) + + +# --------------------------------------------------------------------------- +# Tests for LocalFeatureServerConfig +# --------------------------------------------------------------------------- + + +class TestLocalFeatureServerConfig: + def test_defaults(self): + from feast.infra.feature_servers.local_process.config import ( + LocalFeatureServerConfig, + ) + + cfg = LocalFeatureServerConfig() + assert cfg.materialize_mode == "local" + assert cfg.url is None + assert cfg.materialize_timeout == 3600.0 + assert cfg.materialize_poll_interval == 5.0 + + def test_remote_config(self): + from feast.infra.feature_servers.local_process.config import ( + LocalFeatureServerConfig, + ) + + cfg = LocalFeatureServerConfig( + materialize_mode="remote", + url="http://feast-server:80", + materialize_timeout=600.0, + materialize_poll_interval=2.0, + ) + assert cfg.materialize_mode == "remote" + assert cfg.url == "http://feast-server:80" + assert cfg.materialize_timeout == 600.0 + assert cfg.materialize_poll_interval == 2.0 + + def test_remote_mode_requires_url(self): + from pydantic import ValidationError + + from feast.infra.feature_servers.local_process.config import ( + LocalFeatureServerConfig, + ) + + with pytest.raises(ValidationError, match="url must be set"): + LocalFeatureServerConfig(materialize_mode="remote")