From 44d9a1152a762b161ca53f3778f5b84bd59bc79a Mon Sep 17 00:00:00 2001 From: Aniket Paluskar Date: Thu, 9 Jul 2026 02:56:36 +0530 Subject: [PATCH 1/7] feat: Add remote materialization support via feature server Allow store.materialize() and materialize_incremental() to delegate execution to the feature server instead of running the batch engine locally. This eliminates the need for engine-specific dependencies (PySpark, kubernetes client, etc.) and RBAC on the client pod. Components: - New shared poll_materialization_status() function - Client-side _remote_materialize() with registry-based FV state polling - POST /materialize-async and /materialize-incremental-async endpoints (202) - materialize_mode, url, timeout, poll_interval config fields - 19 unit tests + 2 E2E integration tests Addresses feast-dev/feast#4526 (remote materialization). Signed-off-by: Aniket Paluskar --- sdk/python/feast/feature_server.py | 95 +++++ sdk/python/feast/feature_store.py | 213 +++++++++++ .../feature_servers/local_process/config.py | 16 +- sdk/python/feast/materialization_status.py | 95 +++++ .../tests/unit/test_remote_materialize.py | 336 ++++++++++++++++++ .../tests/unit/test_remote_materialize_e2e.py | 211 +++++++++++ 6 files changed, 965 insertions(+), 1 deletion(-) create mode 100644 sdk/python/feast/materialization_status.py create mode 100644 sdk/python/tests/unit/test_remote_materialize.py create mode 100644 sdk/python/tests/unit/test_remote_materialize_e2e.py diff --git a/sdk/python/feast/feature_server.py b/sdk/python/feast/feature_server.py index bba91130db3..f6f134127e5 100644 --- a/sdk/python/feast/feature_server.py +++ b/sdk/python/feast/feature_server.py @@ -708,6 +708,101 @@ async def materialize_incremental(request: MaterializeIncrementalRequest) -> Non full_feature_names=request.full_feature_names, ) + @app.post("/materialize-async", dependencies=[Depends(inject_user_details)]) + async def materialize_async(request: MaterializeRequest) -> JSONResponse: + with feast_metrics.track_request_latency("/materialize-async"): + if request.feature_views: + for feature_view in request.feature_views: + resource = await _get_feast_object(feature_view, True) + assert_permissions( + resource=resource, + actions=[AuthzedAction.WRITE_ONLINE], + ) + else: + feature_views_to_materialize = store._get_feature_views_to_materialize( + None + ) + for fv in feature_views_to_materialize: + assert_permissions( + resource=fv, + actions=[AuthzedAction.WRITE_ONLINE], + ) + + if request.disable_event_timestamp: + now = datetime.now() + start_date = datetime(1970, 1, 1) + end_date = now + else: + 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" + ) + start_date = utils.make_tzaware(parser.parse(request.start_ts)) + end_date = utils.make_tzaware(parser.parse(request.end_ts)) + + fv_names = request.feature_views or [ + fv.name for fv in store._get_feature_views_to_materialize(None) + ] + + asyncio.get_event_loop().run_in_executor( + None, + lambda: store.materialize( + start_date, + end_date, + request.feature_views, + disable_event_timestamp=request.disable_event_timestamp, + full_feature_names=request.full_feature_names, + ), + ) + + 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"): + if request.feature_views: + for feature_view in request.feature_views: + resource = await _get_feast_object(feature_view, True) + assert_permissions( + resource=resource, + actions=[AuthzedAction.WRITE_ONLINE], + ) + else: + feature_views_to_materialize = store._get_feature_views_to_materialize( + None + ) + for fv in feature_views_to_materialize: + assert_permissions( + resource=fv, + actions=[AuthzedAction.WRITE_ONLINE], + ) + + fv_names = request.feature_views or [ + fv.name for fv in store._get_feature_views_to_materialize(None) + ] + + asyncio.get_event_loop().run_in_executor( + None, + lambda: store.materialize_incremental( + utils.make_tzaware(parser.parse(request.end_ts)), + request.feature_views, + full_feature_names=request.full_feature_names, + ), + ) + + 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..0486e2f1426 100644 --- a/sdk/python/feast/feature_store.py +++ b/sdk/python/feast/feature_store.py @@ -2192,6 +2192,203 @@ 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.""" + 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" + ) + return url.rstrip("/") + + 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, + ) -> None: + """Delegate materialization to the feature server, poll via registry.""" + import requests + + from feast.materialization_status import ( + FVMaterializationStatus, + FVResult, + poll_materialization_status, + ) + + server_url = self._get_feature_server_url() + fs_cfg = self.config.feature_server + + fv_names = feature_views or [ + fv.name + for fv in self._get_feature_views_to_materialize(None) + if not isinstance(fv, OnDemandFeatureView) + ] + + _logger.info( + f"Remote materialize: triggering {len(fv_names)} feature views " + f"on {server_url}" + ) + + resp = requests.post( + f"{server_url}/materialize-async", + json={ + "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, + }, + timeout=30, + ) + resp.raise_for_status() + + 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: + return FVMaterializationStatus.RUNNING + elif state == FeatureViewState.GENERATED: + return FVMaterializationStatus.FAILED + return FVMaterializationStatus.PENDING + + def on_change(result: FVResult): + symbol = { + FVMaterializationStatus.SUCCEEDED: "+", + FVMaterializationStatus.FAILED: "x", + FVMaterializationStatus.RUNNING: "~", + }.get(result.status, "?") + print( + 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_incremental( + self, + end_date: datetime, + feature_views: Optional[List[str]] = None, + full_feature_names: bool = False, + ) -> None: + """Delegate incremental materialization to the feature server, poll via registry.""" + import requests + + from feast.materialization_status import ( + FVMaterializationStatus, + FVResult, + poll_materialization_status, + ) + + server_url = self._get_feature_server_url() + fs_cfg = self.config.feature_server + + fv_names = feature_views or [ + fv.name + for fv in self._get_feature_views_to_materialize(None) + if not isinstance(fv, OnDemandFeatureView) + ] + + _logger.info( + f"Remote materialize_incremental: triggering {len(fv_names)} " + f"feature views on {server_url}" + ) + + resp = requests.post( + f"{server_url}/materialize-incremental-async", + json={ + "end_ts": end_date.isoformat(), + "feature_views": fv_names, + "full_feature_names": full_feature_names, + }, + timeout=30, + ) + resp.raise_for_status() + + 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: + return FVMaterializationStatus.RUNNING + elif state == FeatureViewState.GENERATED: + return FVMaterializationStatus.FAILED + return FVMaterializationStatus.PENDING + + def on_change(result: FVResult): + symbol = { + FVMaterializationStatus.SUCCEEDED: "+", + FVMaterializationStatus.FAILED: "x", + FVMaterializationStatus.RUNNING: "~", + }.get(result.status, "?") + print( + 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 incremental materialization failed for: {names}" + if errors: + msg += f" ({errors})" + raise Exception(msg) + def materialize_incremental( self, end_date: datetime, @@ -2231,6 +2428,13 @@ def materialize_incremental( ... """ + if self._is_remote_materialize_mode(): + return self._remote_materialize_incremental( + end_date=end_date, + feature_views=feature_views, + full_feature_names=full_feature_names, + ) + parsed_version = self._validate_materialize_version(version, feature_views) feature_views_to_materialize = self._get_feature_views_to_materialize( feature_views, version=parsed_version @@ -2441,6 +2645,15 @@ def materialize( f"The given start_date {start_date} is greater than the given end_date {end_date}." ) + if 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, + ) + 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..871b0e4881e 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,4 @@ -from typing import Literal +from typing import Literal, Optional from feast.infra.feature_servers.base_config import BaseFeatureServerConfig @@ -9,3 +9,17 @@ 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.""" diff --git a/sdk/python/feast/materialization_status.py b/sdk/python/feast/materialization_status.py new file mode 100644 index 00000000000..47935a48413 --- /dev/null +++ b/sdk/python/feast/materialization_status.py @@ -0,0 +1,95 @@ +import logging +import time +from dataclasses import dataclass, field +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"Timed out after {timeout}s", + ) + + return list(results.values()) 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..80354a0d5b1 --- /dev/null +++ b/sdk/python/tests/unit/test_remote_materialize.py @@ -0,0 +1,336 @@ +"""Tests for remote materialization: shared poller, client-side delegation, and async endpoints.""" + +import time +from datetime import datetime, timedelta, timezone +from unittest.mock import MagicMock, patch + +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, + FVResult, + poll_materialization_status, +) + + +# --------------------------------------------------------------------------- +# Tests for poll_materialization_status +# --------------------------------------------------------------------------- + + +class TestPollMaterializationStatus: + def test_all_succeed_immediately(self): + """All FVs report SUCCEEDED on first poll.""" + + 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): + """One FV succeeds, one fails.""" + 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): + """FV that stays RUNNING hits timeout.""" + + 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 + + def test_transition_from_pending_to_running_to_succeeded(self): + """FV transitions through states over multiple polls.""" + 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): + """Callback is invoked on state changes.""" + 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_status_fn_exception_is_handled(self): + """If status_fn raises, the poller continues gracefully.""" + call_count = {"n": 0} + + def status_fn(name): + call_count["n"] += 1 + if call_count["n"] == 1: + raise RuntimeError("transient 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): + """Results include elapsed time.""" + + 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): + """Create a test client with mocked FeatureStore.""" + fs = MagicMock() + fs._get_provider.return_value.async_supported.online.read = False + fs.initialize = MagicMock() + fs.close = MagicMock() + + mock_fv = MagicMock() + mock_fv.name = "test_fv" + fs._get_feature_views_to_materialize.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 + + +# --------------------------------------------------------------------------- +# Tests for FeatureStore remote materialize gate +# --------------------------------------------------------------------------- + + +class TestFeatureStoreRemoteGate: + @patch("feast.feature_store.FeatureStore._remote_materialize") + def test_materialize_delegates_when_remote_mode(self, mock_remote): + """materialize() calls _remote_materialize when mode is remote.""" + fs = MagicMock() + fs.config.feature_server.materialize_mode = "remote" + fs.config.feature_server.url = "http://feast-server:80" + + from feast.feature_store import FeatureStore + + # Call the unbound method with our mock self + fs._is_remote_materialize_mode = MagicMock(return_value=True) + fs._remote_materialize = MagicMock() + + # Simulate what happens: gate check returns True, calls remote + if fs._is_remote_materialize_mode(): + fs._remote_materialize( + 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_is_remote_materialize_mode_false_by_default(self): + """Without config, remote mode is False.""" + 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): + """With materialize_mode=remote, returns True.""" + 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): + """With materialize_mode=local (default), returns False.""" + 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): + """Raises ValueError when url is not set.""" + 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): + """URL is returned without trailing slash.""" + 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" + + +# --------------------------------------------------------------------------- +# 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 diff --git a/sdk/python/tests/unit/test_remote_materialize_e2e.py b/sdk/python/tests/unit/test_remote_materialize_e2e.py new file mode 100644 index 00000000000..3cfb47fed67 --- /dev/null +++ b/sdk/python/tests/unit/test_remote_materialize_e2e.py @@ -0,0 +1,211 @@ +"""End-to-end test 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 os +import tempfile +import threading +import time +from datetime import datetime, timedelta, timezone +from pathlib import Path + +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 + + +@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.""" + store, config, registry_path, online_store_path = e2e_repo + + app = get_app(store) + + server_port = 18566 + 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() + + # Wait for server to be ready + 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 test_remote_materialize_e2e(feature_server): + """Full end-to-end: client with remote mode materializes through server.""" + server_store, server_config, server_port = feature_server + + # Create a CLIENT FeatureStore that uses remote materialization. + # It shares the same registry (so it can poll FV state) but delegates + # the actual work to the server. + 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, + ), + ) + + client_store = FeatureStore(config=client_config) + + now = datetime.now(tz=timezone.utc) + start_date = now - timedelta(days=7) + end_date = now + + # Run materialize through the remote path + client_store.materialize( + start_date=start_date, + end_date=end_date, + feature_views=["test_feature_view"], + ) + + # Verify: FV should be in AVAILABLE_ONLINE state + fv = client_store.registry.get_feature_view("test_feature_view", "test_remote_mat") + assert fv.state == FeatureViewState.AVAILABLE_ONLINE + + # Verify: data actually landed in the online store + 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 + + 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, + ), + ) + + client_store = FeatureStore(config=client_config) + now = datetime.now(tz=timezone.utc) + + client_store.materialize_incremental( + end_date=now, + feature_views=["test_feature_view"], + ) + + # Verify FV state + fv = client_store.registry.get_feature_view("test_feature_view", "test_remote_mat") + assert fv.state == FeatureViewState.AVAILABLE_ONLINE + + # Verify data in online store + 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) From 82ce777cd4e082bbc1ca6945d32ea1c10e0039d5 Mon Sep 17 00:00:00 2001 From: Aniket Paluskar Date: Thu, 9 Jul 2026 03:11:08 +0530 Subject: [PATCH 2/7] Fixed lint errors & minor formatting Signed-off-by: Aniket Paluskar --- sdk/python/feast/feature_store.py | 8 ++------ sdk/python/feast/materialization_status.py | 6 ++---- sdk/python/tests/unit/test_remote_materialize.py | 13 +++++-------- .../tests/unit/test_remote_materialize_e2e.py | 3 --- 4 files changed, 9 insertions(+), 21 deletions(-) diff --git a/sdk/python/feast/feature_store.py b/sdk/python/feast/feature_store.py index 0486e2f1426..08866e95166 100644 --- a/sdk/python/feast/feature_store.py +++ b/sdk/python/feast/feature_store.py @@ -2295,9 +2295,7 @@ def on_change(result: FVResult): 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 - ) + 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})" @@ -2381,9 +2379,7 @@ def on_change(result: FVResult): 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 - ) + errors = "; ".join(f"{r.name}: {r.error}" for r in failed if r.error) msg = f"Remote incremental materialization failed for: {names}" if errors: msg += f" ({errors})" diff --git a/sdk/python/feast/materialization_status.py b/sdk/python/feast/materialization_status.py index 47935a48413..ec8e178df10 100644 --- a/sdk/python/feast/materialization_status.py +++ b/sdk/python/feast/materialization_status.py @@ -1,6 +1,6 @@ import logging import time -from dataclasses import dataclass, field +from dataclasses import dataclass from enum import Enum from typing import Callable, Dict, List, Optional, Set @@ -67,9 +67,7 @@ def poll_materialization_status( elapsed = time.monotonic() - start if current != previous_statuses.get(name): - result = FVResult( - name=name, status=current, elapsed_seconds=elapsed - ) + result = FVResult(name=name, status=current, elapsed_seconds=elapsed) results[name] = result previous_statuses[name] = current if on_status_change: diff --git a/sdk/python/tests/unit/test_remote_materialize.py b/sdk/python/tests/unit/test_remote_materialize.py index 80354a0d5b1..d3cd366dd33 100644 --- a/sdk/python/tests/unit/test_remote_materialize.py +++ b/sdk/python/tests/unit/test_remote_materialize.py @@ -1,21 +1,17 @@ """Tests for remote materialization: shared poller, client-side delegation, and async endpoints.""" -import time -from datetime import datetime, timedelta, timezone +from datetime import datetime, timezone from unittest.mock import MagicMock, patch 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, - FVResult, poll_materialization_status, ) - # --------------------------------------------------------------------------- # Tests for poll_materialization_status # --------------------------------------------------------------------------- @@ -39,7 +35,10 @@ def status_fn(name): def test_mixed_results(self): """One FV succeeds, one fails.""" - statuses = {"fv_a": FVMaterializationStatus.SUCCEEDED, "fv_b": FVMaterializationStatus.FAILED} + statuses = { + "fv_a": FVMaterializationStatus.SUCCEEDED, + "fv_b": FVMaterializationStatus.FAILED, + } def status_fn(name): return statuses[name] @@ -235,8 +234,6 @@ def test_materialize_delegates_when_remote_mode(self, mock_remote): fs.config.feature_server.materialize_mode = "remote" fs.config.feature_server.url = "http://feast-server:80" - from feast.feature_store import FeatureStore - # Call the unbound method with our mock self fs._is_remote_materialize_mode = MagicMock(return_value=True) fs._remote_materialize = MagicMock() diff --git a/sdk/python/tests/unit/test_remote_materialize_e2e.py b/sdk/python/tests/unit/test_remote_materialize_e2e.py index 3cfb47fed67..d9c74b24298 100644 --- a/sdk/python/tests/unit/test_remote_materialize_e2e.py +++ b/sdk/python/tests/unit/test_remote_materialize_e2e.py @@ -6,12 +6,9 @@ FV state transitions → client polls registry → returns success """ -import os -import tempfile import threading import time from datetime import datetime, timedelta, timezone -from pathlib import Path import pandas as pd import pytest From 3bf35a2240553a5bef1a719c0a145c8b21cae299 Mon Sep 17 00:00:00 2001 From: Aniket Paluskar Date: Thu, 9 Jul 2026 04:19:12 +0530 Subject: [PATCH 3/7] feat: Add remote materialization support via feature server Allow store.materialize() and materialize_incremental() to delegate execution to the feature server instead of running the batch engine locally. This eliminates the need for engine-specific dependencies (PySpark, kubernetes client, etc.) and RBAC on the client pod. Components: - Shared poll_materialization_status() with stateful FV state tracking - Client-side _remote_materialize_common() with auth forwarding - POST /materialize-async and /materialize-incremental-async (202) - _force_local guard prevents infinite recursion on server - Concurrency guard returns 409 if FV already materializing - version parameter forwarded through full chain - materialize_mode, url, timeout, poll_interval config fields Addresses feast-dev/feast#4526 (remote materialization). Signed-off-by: Aniket Paluskar --- sdk/python/feast/feature_server.py | 165 ++++++++----- sdk/python/feast/feature_store.py | 183 +++++++------- .../test_remote_materialize.py} | 97 +++++--- .../tests/unit/test_remote_materialize.py | 232 +++++++++++++++--- 4 files changed, 444 insertions(+), 233 deletions(-) rename sdk/python/tests/{unit/test_remote_materialize_e2e.py => integration/materialization/test_remote_materialize.py} (70%) diff --git a/sdk/python/feast/feature_server.py b/sdk/python/feast/feature_server.py index f6f134127e5..09c5f53086f 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,52 +710,89 @@ async def materialize_incremental(request: MaterializeIncrementalRequest) -> Non full_feature_names=request.full_feature_names, ) + async 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}") + + 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) + 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"): - if request.feature_views: - for feature_view in request.feature_views: - resource = await _get_feast_object(feature_view, True) - assert_permissions( - resource=resource, - actions=[AuthzedAction.WRITE_ONLINE], - ) - else: - feature_views_to_materialize = store._get_feature_views_to_materialize( - None - ) - for fv in feature_views_to_materialize: - assert_permissions( - resource=fv, - actions=[AuthzedAction.WRITE_ONLINE], - ) + fv_names = await _authorize_materialize_views(request.feature_views) + start_date, end_date = _parse_materialize_timestamps(request) - if request.disable_event_timestamp: - now = datetime.now() - start_date = datetime(1970, 1, 1) - end_date = now - else: - 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" + 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, ) - start_date = utils.make_tzaware(parser.parse(request.start_ts)) - end_date = utils.make_tzaware(parser.parse(request.end_ts)) - fv_names = request.feature_views or [ - fv.name for fv in store._get_feature_views_to_materialize(None) - ] - - asyncio.get_event_loop().run_in_executor( - None, - lambda: store.materialize( - start_date, - end_date, - request.feature_views, - disable_event_timestamp=request.disable_event_timestamp, - full_feature_names=request.full_feature_names, - ), - ) + loop = asyncio.get_running_loop() + loop.run_in_executor(None, _run_materialize) return JSONResponse( status_code=202, @@ -768,35 +807,29 @@ async def materialize_incremental_async( request: MaterializeIncrementalRequest, ) -> JSONResponse: with feast_metrics.track_request_latency("/materialize-incremental-async"): - if request.feature_views: - for feature_view in request.feature_views: - resource = await _get_feast_object(feature_view, True) - assert_permissions( - resource=resource, - actions=[AuthzedAction.WRITE_ONLINE], + fv_names = await _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, ) - else: - feature_views_to_materialize = store._get_feature_views_to_materialize( - None - ) - for fv in feature_views_to_materialize: - assert_permissions( - resource=fv, - actions=[AuthzedAction.WRITE_ONLINE], + except Exception as e: + logger.error( + f"Async incremental materialization failed for {fv_names}: {e}", + exc_info=True, ) - fv_names = request.feature_views or [ - fv.name for fv in store._get_feature_views_to_materialize(None) - ] - - asyncio.get_event_loop().run_in_executor( - None, - lambda: store.materialize_incremental( - utils.make_tzaware(parser.parse(request.end_ts)), - request.feature_views, - full_feature_names=request.full_feature_names, - ), - ) + loop = asyncio.get_running_loop() + loop.run_in_executor(None, _run_materialize_incremental) return JSONResponse( status_code=202, diff --git a/sdk/python/feast/feature_store.py b/sdk/python/feast/feature_store.py index 08866e95166..baa1e60a35c 100644 --- a/sdk/python/feast/feature_store.py +++ b/sdk/python/feast/feature_store.py @@ -2201,6 +2201,8 @@ def _is_remote_materialize_mode(self) -> bool: 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: @@ -2211,19 +2213,40 @@ def _get_feature_server_url(self) -> str: " 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 _remote_materialize( + 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, - start_date: datetime, - end_date: datetime, - feature_views: Optional[List[str]] = None, - disable_event_timestamp: bool = False, - full_feature_names: bool = False, + endpoint: str, + payload: dict, + fv_names: List[str], ) -> None: - """Delegate materialization to the feature server, poll via registry.""" - import requests + """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, @@ -2233,30 +2256,21 @@ def _remote_materialize( server_url = self._get_feature_server_url() fs_cfg = self.config.feature_server - fv_names = feature_views or [ - fv.name - for fv in self._get_feature_views_to_materialize(None) - if not isinstance(fv, OnDemandFeatureView) - ] - _logger.info( f"Remote materialize: triggering {len(fv_names)} feature views " - f"on {server_url}" + f"via {endpoint} on {server_url}" ) - resp = requests.post( - f"{server_url}/materialize-async", - json={ - "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, - }, - timeout=30, + session = self._get_remote_http_session() + resp = session.post( + f"{server_url}{endpoint}", + json=payload, + timeout=getattr(fs_cfg, "http_timeout", 30), ) resp.raise_for_status() + 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 @@ -2265,9 +2279,12 @@ def registry_status_fn(fv_name: str) -> FVMaterializationStatus: if state == FeatureViewState.AVAILABLE_ONLINE: return FVMaterializationStatus.SUCCEEDED elif state == FeatureViewState.MATERIALIZING: + seen_materializing.add(fv_name) return FVMaterializationStatus.RUNNING elif state == FeatureViewState.GENERATED: - return FVMaterializationStatus.FAILED + if fv_name in seen_materializing: + return FVMaterializationStatus.FAILED + return FVMaterializationStatus.PENDING return FVMaterializationStatus.PENDING def on_change(result: FVResult): @@ -2276,7 +2293,7 @@ def on_change(result: FVResult): FVMaterializationStatus.FAILED: "x", FVMaterializationStatus.RUNNING: "~", }.get(result.status, "?") - print( + _logger.info( f" [{symbol}] {result.name}: {result.status.value} " f"({result.elapsed_seconds:.1f}s)" ) @@ -2301,96 +2318,67 @@ def on_change(result: FVResult): msg += f" ({errors})" raise Exception(msg) - def _remote_materialize_incremental( + 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 incremental materialization to the feature server, poll via registry.""" - import requests - - from feast.materialization_status import ( - FVMaterializationStatus, - FVResult, - poll_materialization_status, - ) - - server_url = self._get_feature_server_url() - fs_cfg = self.config.feature_server - + """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) ] - _logger.info( - f"Remote materialize_incremental: triggering {len(fv_names)} " - f"feature views on {server_url}" - ) + 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 - resp = requests.post( - f"{server_url}/materialize-incremental-async", - json={ - "end_ts": end_date.isoformat(), - "feature_views": fv_names, - "full_feature_names": full_feature_names, - }, - timeout=30, - ) - resp.raise_for_status() + self._remote_materialize_common("/materialize-async", payload, fv_names) - 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: - return FVMaterializationStatus.RUNNING - elif state == FeatureViewState.GENERATED: - return FVMaterializationStatus.FAILED - return FVMaterializationStatus.PENDING - - def on_change(result: FVResult): - symbol = { - FVMaterializationStatus.SUCCEEDED: "+", - FVMaterializationStatus.FAILED: "x", - FVMaterializationStatus.RUNNING: "~", - }.get(result.status, "?") - print( - f" [{symbol}] {result.name}: {result.status.value} " - f"({result.elapsed_seconds:.1f}s)" - ) + 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) + ] - timeout = getattr(fs_cfg, "materialize_timeout", 3600.0) - poll_interval = getattr(fs_cfg, "materialize_poll_interval", 5.0) + 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 - 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, + self._remote_materialize_common( + "/materialize-incremental-async", payload, fv_names ) - 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 incremental materialization failed for: {names}" - if errors: - msg += f" ({errors})" - raise Exception(msg) - 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. @@ -2424,11 +2412,12 @@ def materialize_incremental( ... """ - if self._is_remote_materialize_mode(): + 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) @@ -2604,6 +2593,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. @@ -2641,13 +2631,14 @@ def materialize( f"The given start_date {start_date} is greater than the given end_date {end_date}." ) - if self._is_remote_materialize_mode(): + 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) diff --git a/sdk/python/tests/unit/test_remote_materialize_e2e.py b/sdk/python/tests/integration/materialization/test_remote_materialize.py similarity index 70% rename from sdk/python/tests/unit/test_remote_materialize_e2e.py rename to sdk/python/tests/integration/materialization/test_remote_materialize.py index d9c74b24298..5c998481aa8 100644 --- a/sdk/python/tests/unit/test_remote_materialize_e2e.py +++ b/sdk/python/tests/integration/materialization/test_remote_materialize.py @@ -1,4 +1,4 @@ -"""End-to-end test for remote materialization. +"""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: @@ -6,6 +6,7 @@ FV state transitions → client polls registry → returns success """ +import socket import threading import time from datetime import datetime, timedelta, timezone @@ -22,6 +23,12 @@ 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.""" @@ -77,12 +84,12 @@ def e2e_repo(tmp_path): @pytest.fixture def feature_server(e2e_repo): - """Start a real feature server in a background thread.""" + """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_port = 18566 server_config = uvicorn.Config( app, host="127.0.0.1", port=server_port, log_level="warning" ) @@ -91,7 +98,6 @@ def feature_server(e2e_repo): thread = threading.Thread(target=server.run, daemon=True) thread.start() - # Wait for server to be ready import httpx for _ in range(50): @@ -111,13 +117,8 @@ def feature_server(e2e_repo): thread.join(timeout=5) -def test_remote_materialize_e2e(feature_server): - """Full end-to-end: client with remote mode materializes through server.""" - server_store, server_config, server_port = feature_server - - # Create a CLIENT FeatureStore that uses remote materialization. - # It shares the same registry (so it can poll FV state) but delegates - # the actual work to the server. +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, ) @@ -136,25 +137,27 @@ def test_remote_materialize_e2e(feature_server): materialize_poll_interval=0.5, ), ) + return FeatureStore(config=client_config) + - client_store = 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 - # Run materialize through the remote path client_store.materialize( start_date=start_date, end_date=end_date, feature_views=["test_feature_view"], ) - # Verify: FV should be in AVAILABLE_ONLINE state fv = client_store.registry.get_feature_view("test_feature_view", "test_remote_mat") assert fv.state == FeatureViewState.AVAILABLE_ONLINE - # Verify: data actually landed in the online store online_features = client_store.get_online_features( features=["test_feature_view:feature_a", "test_feature_view:feature_b"], entity_rows=[{"entity_id": 1}], @@ -167,27 +170,8 @@ def test_remote_materialize_e2e(feature_server): 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) - 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, - ), - ) - - client_store = FeatureStore(config=client_config) now = datetime.now(tz=timezone.utc) client_store.materialize_incremental( @@ -195,14 +179,53 @@ def test_remote_materialize_incremental_e2e(feature_server): feature_views=["test_feature_view"], ) - # Verify FV state fv = client_store.registry.get_feature_view("test_feature_view", "test_remote_mat") assert fv.state == FeatureViewState.AVAILABLE_ONLINE - # Verify data in online store 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's raise_for_status() + propagates this as an exception. + """ + import requests as req_lib + + 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(req_lib.exceptions.HTTPError): + 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 index d3cd366dd33..56d0a3d05c1 100644 --- a/sdk/python/tests/unit/test_remote_materialize.py +++ b/sdk/python/tests/unit/test_remote_materialize.py @@ -1,12 +1,14 @@ """Tests for remote materialization: shared poller, client-side delegation, and async endpoints.""" +import time from datetime import datetime, timezone -from unittest.mock import MagicMock, patch +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, @@ -19,8 +21,6 @@ class TestPollMaterializationStatus: def test_all_succeed_immediately(self): - """All FVs report SUCCEEDED on first poll.""" - def status_fn(name): return FVMaterializationStatus.SUCCEEDED @@ -34,7 +34,6 @@ def status_fn(name): assert all(r.status == FVMaterializationStatus.SUCCEEDED for r in results) def test_mixed_results(self): - """One FV succeeds, one fails.""" statuses = { "fv_a": FVMaterializationStatus.SUCCEEDED, "fv_b": FVMaterializationStatus.FAILED, @@ -54,8 +53,6 @@ def status_fn(name): assert result_map["fv_b"].status == FVMaterializationStatus.FAILED def test_timeout(self): - """FV that stays RUNNING hits timeout.""" - def status_fn(name): return FVMaterializationStatus.RUNNING @@ -71,7 +68,6 @@ def status_fn(name): assert "Timed out" in results[0].error def test_transition_from_pending_to_running_to_succeeded(self): - """FV transitions through states over multiple polls.""" call_count = {"n": 0} def status_fn(name): @@ -91,7 +87,6 @@ def status_fn(name): assert results[0].status == FVMaterializationStatus.SUCCEEDED def test_on_status_change_callback(self): - """Callback is invoked on state changes.""" changes = [] call_count = {"n": 0} @@ -112,14 +107,14 @@ def status_fn(name): assert changes[0].status == FVMaterializationStatus.RUNNING assert changes[1].status == FVMaterializationStatus.SUCCEEDED - def test_status_fn_exception_is_handled(self): - """If status_fn raises, the poller continues gracefully.""" + 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 RuntimeError("transient error") + raise ConnectionError("transient network error") return FVMaterializationStatus.SUCCEEDED results = poll_materialization_status( @@ -131,8 +126,6 @@ def status_fn(name): assert results[0].status == FVMaterializationStatus.SUCCEEDED def test_elapsed_seconds_is_populated(self): - """Results include elapsed time.""" - def status_fn(name): return FVMaterializationStatus.SUCCEEDED @@ -153,15 +146,17 @@ def status_fn(name): class TestFeatureServerAsyncEndpoints: def _make_client(self): - """Create a test client with mocked FeatureStore.""" 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() @@ -220,6 +215,71 @@ def test_materialize_async_disable_event_timestamp(self): ) 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 @@ -227,77 +287,181 @@ def test_materialize_async_disable_event_timestamp(self): class TestFeatureStoreRemoteGate: - @patch("feast.feature_store.FeatureStore._remote_materialize") - def test_materialize_delegates_when_remote_mode(self, mock_remote): - """materialize() calls _remote_materialize when mode is remote.""" + 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.config.feature_server.materialize_mode = "remote" - fs.config.feature_server.url = "http://feast-server:80" + 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"], + ) - # Call the unbound method with our mock self + 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() - # Simulate what happens: gate check returns True, calls remote - if fs._is_remote_materialize_mode(): - fs._remote_materialize( + # _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_called_once() + fs._remote_materialize.assert_not_called() def test_is_remote_materialize_mode_false_by_default(self): - """Without config, remote mode is False.""" 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): - """With materialize_mode=remote, returns True.""" 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): - """With materialize_mode=local (default), returns False.""" 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): - """Raises ValueError when url is not set.""" 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): - """URL is returned without trailing slash.""" 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 From b6fe54df1551b8adb0fd430d5f322c7271fc278e Mon Sep 17 00:00:00 2001 From: Aniket Paluskar Date: Thu, 9 Jul 2026 04:57:39 +0530 Subject: [PATCH 4/7] - Remove unnecessary async from _authorize_materialize_views - Use allow_cache=False in concurrency guard to avoid stale registry reads - Close HTTP session in finally block to prevent resource leaks - Wrap remote HTTP call with descriptive error message and exception chaining - Add from-e chaining on timestamp parse errors - Improve timeout error message with actionable guidance - Add @model_validator to reject remote mode without url at config time - Add unit test for config validation Signed-off-by: Aniket Paluskar --- sdk/python/feast/feature_server.py | 12 ++++++----- sdk/python/feast/feature_store.py | 20 +++++++++++++------ .../feature_servers/local_process/config.py | 10 ++++++++++ sdk/python/feast/materialization_status.py | 2 +- .../test_remote_materialize.py | 8 +++----- .../tests/unit/test_remote_materialize.py | 12 ++++++++++- 6 files changed, 46 insertions(+), 18 deletions(-) diff --git a/sdk/python/feast/feature_server.py b/sdk/python/feast/feature_server.py index 09c5f53086f..8cb6dff91c6 100644 --- a/sdk/python/feast/feature_server.py +++ b/sdk/python/feast/feature_server.py @@ -710,7 +710,7 @@ async def materialize_incremental(request: MaterializeIncrementalRequest) -> Non full_feature_names=request.full_feature_names, ) - async def _authorize_materialize_views( + def _authorize_materialize_views( request_feature_views: Optional[List[str]], ) -> List[str]: """Shared authz + FV name resolution for async materialize endpoints.""" @@ -738,7 +738,7 @@ def _parse_materialize_timestamps(request: MaterializeRequest): 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}") + raise ValueError(f"Invalid timestamp format: {e}") from e if start_date >= end_date: raise ValueError( @@ -752,7 +752,9 @@ def _check_already_materializing(fv_names: List[str]) -> Optional[JSONResponse]: for fv_name in fv_names: try: - fv = store.registry.get_feature_view(fv_name, store.project) + 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, @@ -767,7 +769,7 @@ def _check_already_materializing(fv_names: List[str]) -> Optional[JSONResponse]: @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 = await _authorize_materialize_views(request.feature_views) + fv_names = _authorize_materialize_views(request.feature_views) start_date, end_date = _parse_materialize_timestamps(request) conflict = _check_already_materializing(fv_names) @@ -807,7 +809,7 @@ async def materialize_incremental_async( request: MaterializeIncrementalRequest, ) -> JSONResponse: with feast_metrics.track_request_latency("/materialize-incremental-async"): - fv_names = await _authorize_materialize_views(request.feature_views) + fv_names = _authorize_materialize_views(request.feature_views) conflict = _check_already_materializing(fv_names) if conflict: diff --git a/sdk/python/feast/feature_store.py b/sdk/python/feast/feature_store.py index baa1e60a35c..633d2f67f9f 100644 --- a/sdk/python/feast/feature_store.py +++ b/sdk/python/feast/feature_store.py @@ -2262,12 +2262,20 @@ def _remote_materialize_common( ) session = self._get_remote_http_session() - resp = session.post( - f"{server_url}{endpoint}", - json=payload, - timeout=getattr(fs_cfg, "http_timeout", 30), - ) - resp.raise_for_status() + 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() 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 871b0e4881e..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,5 +1,7 @@ from typing import Literal, Optional +from pydantic import model_validator + from feast.infra.feature_servers.base_config import BaseFeatureServerConfig @@ -23,3 +25,11 @@ class LocalFeatureServerConfig(BaseFeatureServerConfig): 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 index ec8e178df10..4d3d419e052 100644 --- a/sdk/python/feast/materialization_status.py +++ b/sdk/python/feast/materialization_status.py @@ -87,7 +87,7 @@ def poll_materialization_status( name=name, status=FVMaterializationStatus.FAILED, elapsed_seconds=time.monotonic() - start, - error=f"Timed out after {timeout}s", + 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 index 5c998481aa8..2bd17dcb4fc 100644 --- a/sdk/python/tests/integration/materialization/test_remote_materialize.py +++ b/sdk/python/tests/integration/materialization/test_remote_materialize.py @@ -213,17 +213,15 @@ 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's raise_for_status() - propagates this as an exception. + validation, which returns HTTP 500. The client wraps this as an + Exception with a descriptive message. """ - import requests as req_lib - 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(req_lib.exceptions.HTTPError): + with pytest.raises(Exception, match="Failed to trigger remote materialization"): client_store.materialize( start_date=now - timedelta(days=1), end_date=now, diff --git a/sdk/python/tests/unit/test_remote_materialize.py b/sdk/python/tests/unit/test_remote_materialize.py index 56d0a3d05c1..a6d18639f25 100644 --- a/sdk/python/tests/unit/test_remote_materialize.py +++ b/sdk/python/tests/unit/test_remote_materialize.py @@ -65,7 +65,7 @@ def status_fn(name): assert len(results) == 1 assert results[0].status == FVMaterializationStatus.FAILED - assert "Timed out" in results[0].error + assert "timed out" in results[0].error.lower() def test_transition_from_pending_to_running_to_succeeded(self): call_count = {"n": 0} @@ -495,3 +495,13 @@ def test_remote_config(self): 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") From 6f073fbb3018c4fcff0af9d67d38d07a907dda38 Mon Sep 17 00:00:00 2001 From: Aniket Paluskar Date: Fri, 10 Jul 2026 13:35:18 +0530 Subject: [PATCH 5/7] fix: Set FV state to AVAILABLE_ONLINE in SQL/Snowflake registries apply_materialization() in SQL and Snowflake registries did not transition FeatureViewState like the file-based registry does. This caused remote materialization polling (and any state-based monitoring) to never see completion when using these registry backends. Signed-off-by: Aniket Paluskar --- sdk/python/feast/infra/registry/snowflake.py | 4 ++++ sdk/python/feast/infra/registry/sql.py | 4 ++++ 2 files changed, 8 insertions(+) diff --git a/sdk/python/feast/infra/registry/snowflake.py b/sdk/python/feast/infra/registry/snowflake.py index 5590e1b7574..b806c559199 100644 --- a/sdk/python/feast/infra/registry/snowflake.py +++ b/sdk/python/feast/infra/registry/snowflake.py @@ -1135,6 +1135,10 @@ def apply_materialization( FeatureViewNotFoundException, ) fv.materialization_intervals.append((start_date, end_date)) + if hasattr(fv, "state"): + from feast.feature_view import FeatureViewState + + fv.state = FeatureViewState.AVAILABLE_ONLINE self._apply_object( fv_table_str, project, diff --git a/sdk/python/feast/infra/registry/sql.py b/sdk/python/feast/infra/registry/sql.py index edd89347be2..efaaefdac0d 100644 --- a/sdk/python/feast/infra/registry/sql.py +++ b/sdk/python/feast/infra/registry/sql.py @@ -1217,6 +1217,10 @@ def apply_materialization( FeatureViewNotFoundException, ) fv.materialization_intervals.append((start_date, end_date)) + if hasattr(fv, "state"): + from feast.feature_view import FeatureViewState + + fv.state = FeatureViewState.AVAILABLE_ONLINE self._apply_object( table, project, "feature_view_name", fv, "feature_view_proto" ) From cf87b03d4f01d887789eb92bb811cc8753bb9df2 Mon Sep 17 00:00:00 2001 From: Aniket Paluskar Date: Fri, 10 Jul 2026 13:46:37 +0530 Subject: [PATCH 6/7] fix: TLS verification and registry state parity for remote materialize - Pass online_store.cert as verify= on the materialize-async HTTP call (same server hosts both online store and materialize endpoints) - Set FV state to AVAILABLE_ONLINE in SQL/Snowflake registry apply_materialization() to match file-based registry behavior Signed-off-by: Aniket Paluskar --- sdk/python/feast/feature_store.py | 14 +++++++++----- 1 file changed, 9 insertions(+), 5 deletions(-) diff --git a/sdk/python/feast/feature_store.py b/sdk/python/feast/feature_store.py index 633d2f67f9f..1da5edce5bd 100644 --- a/sdk/python/feast/feature_store.py +++ b/sdk/python/feast/feature_store.py @@ -2263,11 +2263,15 @@ def _remote_materialize_common( session = self._get_remote_http_session() try: - resp = session.post( - f"{server_url}{endpoint}", - json=payload, - timeout=getattr(fs_cfg, "http_timeout", 30), - ) + post_kwargs: dict = { + "json": payload, + "timeout": getattr(fs_cfg, "http_timeout", 30), + } + online_cfg = getattr(self.config, "online_store", None) + cert = getattr(online_cfg, "cert", "") if online_cfg else "" + if cert: + post_kwargs["verify"] = cert + resp = session.post(f"{server_url}{endpoint}", **post_kwargs) resp.raise_for_status() except Exception as e: raise Exception( From ef9ff5eeb44201685c6f9d40a9d5b943b157ca74 Mon Sep 17 00:00:00 2001 From: Aniket Paluskar Date: Tue, 14 Jul 2026 12:38:58 +0530 Subject: [PATCH 7/7] refactor: Remote materialize via SDK params, not feature_server YAML Opt in with materialize(..., remote=True); URL/TLS from online_store. Add wait=False and poll_materialization(); set MATERIALIZING before 202 and reset to GENERATED on async failure; include state in FeatureView.__eq__. --- sdk/python/feast/feature_server.py | 22 +- sdk/python/feast/feature_store.py | 198 +++++++++++----- sdk/python/feast/feature_view.py | 1 + .../feature_servers/local_process/config.py | 26 +-- .../test_remote_materialize.py | 167 +++++++++++--- .../tests/unit/test_remote_materialize.py | 218 ++++++++++-------- 6 files changed, 430 insertions(+), 202 deletions(-) diff --git a/sdk/python/feast/feature_server.py b/sdk/python/feast/feature_server.py index 8cb6dff91c6..9854b1cb9da 100644 --- a/sdk/python/feast/feature_server.py +++ b/sdk/python/feast/feature_server.py @@ -746,10 +746,10 @@ def _parse_materialize_timestamps(request: MaterializeRequest): ) return start_date, end_date + from feast.feature_view import FeatureViewState + 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( @@ -766,6 +766,18 @@ def _check_already_materializing(fv_names: List[str]) -> Optional[JSONResponse]: pass return None + def _update_fv_state(fv_names: List[str], state) -> None: + """Update FV state in the registry.""" + for fv_name in fv_names: + try: + fv = store.registry.get_feature_view( + fv_name, store.project, allow_cache=False + ) + fv.state = state + store.registry.apply_feature_view(fv, store.project) + except Exception: + logger.warning(f"Failed to set state={state} for {fv_name}") + @app.post("/materialize-async", dependencies=[Depends(inject_user_details)]) async def materialize_async(request: MaterializeRequest) -> JSONResponse: with feast_metrics.track_request_latency("/materialize-async"): @@ -776,6 +788,8 @@ async def materialize_async(request: MaterializeRequest) -> JSONResponse: if conflict: return conflict + _update_fv_state(fv_names, FeatureViewState.MATERIALIZING) + def _run_materialize(): try: store.materialize( @@ -792,6 +806,7 @@ def _run_materialize(): f"Async materialization failed for {fv_names}: {e}", exc_info=True, ) + _update_fv_state(fv_names, FeatureViewState.GENERATED) loop = asyncio.get_running_loop() loop.run_in_executor(None, _run_materialize) @@ -815,6 +830,8 @@ async def materialize_incremental_async( if conflict: return conflict + _update_fv_state(fv_names, FeatureViewState.MATERIALIZING) + def _run_materialize_incremental(): try: store.materialize_incremental( @@ -829,6 +846,7 @@ def _run_materialize_incremental(): f"Async incremental materialization failed for {fv_names}: {e}", exc_info=True, ) + _update_fv_state(fv_names, FeatureViewState.GENERATED) loop = asyncio.get_running_loop() loop.run_in_executor(None, _run_materialize_incremental) diff --git a/sdk/python/feast/feature_store.py b/sdk/python/feast/feature_store.py index 1da5edce5bd..fa184ca5523 100644 --- a/sdk/python/feast/feature_store.py +++ b/sdk/python/feast/feature_store.py @@ -2192,31 +2192,14 @@ 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) + def _get_remote_materialize_url(self) -> str: + """Get the feature server URL from online_store.path for remote materialization.""" + online_cfg = self.config.online_store + url = getattr(online_cfg, "path", 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}" + "online_store.path must be set to use remote materialization. " + "Configure online_store with type: remote and a valid path." ) return url.rstrip("/") @@ -2234,18 +2217,74 @@ def _get_remote_http_session(self): return requests.Session() + def _fv_state_to_materialization_status( + self, state, *, seen_materializing: Optional[set] = None, fv_name: str = "" + ): + """Map FeatureViewState to FVMaterializationStatus.""" + from feast.materialization_status import FVMaterializationStatus + + if state == FeatureViewState.AVAILABLE_ONLINE: + return FVMaterializationStatus.SUCCEEDED + if state == FeatureViewState.MATERIALIZING: + if seen_materializing is not None: + seen_materializing.add(fv_name) + return FVMaterializationStatus.RUNNING + if state == FeatureViewState.GENERATED: + if seen_materializing is not None and fv_name in seen_materializing: + return FVMaterializationStatus.FAILED + return FVMaterializationStatus.PENDING + return FVMaterializationStatus.PENDING + + def poll_materialization( + self, + feature_views: Optional[List[str]] = None, + ): + """One-shot check of materialization status for feature views. + + Reads current FV state from the registry (no waiting/looping). + Call repeatedly yourself if you want to track progress over time. + + Args: + feature_views: FV names to check. If None, checks all materializable FVs. + + Returns: + List of FVResult with name and current status (pending/running/succeeded). + """ + from feast.materialization_status import FVResult + + fv_names = feature_views or [ + fv.name + for fv in self._get_feature_views_to_materialize(None) + if not isinstance(fv, OnDemandFeatureView) + ] + + results = [] + for name in fv_names: + fv = self.registry.get_feature_view(name, self.project, allow_cache=False) + status = self._fv_state_to_materialization_status( + getattr(fv, "state", None) + ) + results.append(FVResult(name=name, status=status)) + return results + def _remote_materialize_common( self, endpoint: str, payload: dict, fv_names: List[str], + timeout: float = 3600.0, + poll_interval: float = 5.0, + wait: bool = True, ) -> None: - """Shared logic for remote materialization: trigger server, poll registry. + """Shared logic for remote materialization: trigger server, optionally wait. Args: endpoint: Server endpoint path (e.g., "/materialize-async"). payload: JSON payload for the POST request. fv_names: Feature view names to track. + timeout: Max seconds to wait for materialization to complete. + poll_interval: Seconds between registry polls when wait=True. + wait: If True, poll until complete. If False, return after 202 Accepted. """ from feast.materialization_status import ( FVMaterializationStatus, @@ -2253,8 +2292,7 @@ def _remote_materialize_common( poll_materialization_status, ) - server_url = self._get_feature_server_url() - fs_cfg = self.config.feature_server + server_url = self._get_remote_materialize_url() _logger.info( f"Remote materialize: triggering {len(fv_names)} feature views " @@ -2265,10 +2303,9 @@ def _remote_materialize_common( try: post_kwargs: dict = { "json": payload, - "timeout": getattr(fs_cfg, "http_timeout", 30), + "timeout": 30, } - online_cfg = getattr(self.config, "online_store", None) - cert = getattr(online_cfg, "cert", "") if online_cfg else "" + cert = getattr(self.config.online_store, "cert", None) if cert: post_kwargs["verify"] = cert resp = session.post(f"{server_url}{endpoint}", **post_kwargs) @@ -2281,23 +2318,24 @@ def _remote_materialize_common( finally: session.close() + if not wait: + _logger.info( + f"Remote materialize accepted for {fv_names} " + f"(wait=False; use poll_materialization() to check status)" + ) + return + 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 + return self._fv_state_to_materialization_status( + getattr(fv, "state", None), + seen_materializing=seen_materializing, + fv_name=fv_name, + ) def on_change(result: FVResult): symbol = { @@ -2310,9 +2348,6 @@ def on_change(result: FVResult): 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, @@ -2321,6 +2356,9 @@ def on_change(result: FVResult): on_status_change=on_change, ) + if hasattr(self.registry, "refresh"): + self.registry.refresh(project=self.project) + failed = [r for r in results if r.status == FVMaterializationStatus.FAILED] if failed: names = ", ".join(r.name for r in failed) @@ -2338,6 +2376,9 @@ def _remote_materialize( disable_event_timestamp: bool = False, full_feature_names: bool = False, version: Optional[str] = None, + timeout: float = 3600.0, + poll_interval: float = 5.0, + wait: bool = True, ) -> None: """Delegate materialization to the feature server, poll via registry.""" fv_names = feature_views or [ @@ -2356,7 +2397,14 @@ def _remote_materialize( if version is not None: payload["version"] = version - self._remote_materialize_common("/materialize-async", payload, fv_names) + self._remote_materialize_common( + "/materialize-async", + payload, + fv_names, + timeout=timeout, + poll_interval=poll_interval, + wait=wait, + ) def _remote_materialize_incremental( self, @@ -2364,6 +2412,9 @@ def _remote_materialize_incremental( feature_views: Optional[List[str]] = None, full_feature_names: bool = False, version: Optional[str] = None, + timeout: float = 3600.0, + poll_interval: float = 5.0, + wait: bool = True, ) -> None: """Delegate incremental materialization to the feature server, poll via registry.""" fv_names = feature_views or [ @@ -2381,7 +2432,12 @@ def _remote_materialize_incremental( payload["version"] = version self._remote_materialize_common( - "/materialize-incremental-async", payload, fv_names + "/materialize-incremental-async", + payload, + fv_names, + timeout=timeout, + poll_interval=poll_interval, + wait=wait, ) def materialize_incremental( @@ -2390,6 +2446,10 @@ def materialize_incremental( feature_views: Optional[List[str]] = None, full_feature_names: bool = False, version: Optional[str] = None, + remote: bool = False, + timeout: float = 3600.0, + poll_interval: float = 5.0, + wait: bool = True, _force_local: bool = False, ) -> None: """ @@ -2409,6 +2469,13 @@ def materialize_incremental( feature view name. version (str): Optional version to materialize (e.g., 'v2'). Requires feature_views with exactly one entry and enable_online_feature_view_versioning to be enabled. + remote (bool): If True, delegate materialization to the feature server instead of + running the batch engine locally. The server URL is taken from online_store.path. + timeout (float): Max seconds to wait when remote=True and wait=True. + poll_interval (float): Seconds between registry polls when remote=True and wait=True. + wait (bool): If True (default), block until remote materialization finishes. + If False, return after the server accepts the job; use poll_materialization() + to check status later. Raises: Exception: A feature view being materialized does not have a TTL set. @@ -2424,12 +2491,15 @@ def materialize_incremental( ... """ - if not _force_local and self._is_remote_materialize_mode(): + if remote and not _force_local: return self._remote_materialize_incremental( end_date=end_date, feature_views=feature_views, full_feature_names=full_feature_names, version=version, + timeout=timeout, + poll_interval=poll_interval, + wait=wait, ) parsed_version = self._validate_materialize_version(version, feature_views) @@ -2510,10 +2580,11 @@ def tqdm_builder(length): end_date = utils.make_tzaware(end_date) or _utc_now() # Transition state to MATERIALIZING before starting. - # Only enforce when the state machine is active (not STATE_UNSPECIFIED). + # Skip when _force_local: the async handler already set MATERIALIZING. previous_state = getattr(feature_view, "state", None) if ( - hasattr(feature_view, "state") + not _force_local + and hasattr(feature_view, "state") and feature_view.state != FeatureViewState.STATE_UNSPECIFIED ): if not feature_view.state.can_transition_to( @@ -2542,9 +2613,11 @@ def tqdm_builder(length): ) except Exception: fv_success = False - # Roll back state to previous value on failure. + # Roll back state on failure (only when not _force_local, + # since the async handler manages state via _update_fv_state). if ( - hasattr(feature_view, "state") + not _force_local + and hasattr(feature_view, "state") and previous_state is not None and previous_state != FeatureViewState.STATE_UNSPECIFIED ): @@ -2605,6 +2678,10 @@ def materialize( disable_event_timestamp: bool = False, full_feature_names: bool = False, version: Optional[str] = None, + remote: bool = False, + timeout: float = 3600.0, + poll_interval: float = 5.0, + wait: bool = True, _force_local: bool = False, ) -> None: """ @@ -2624,6 +2701,13 @@ def materialize( feature view name. version (str): Optional version to materialize (e.g., 'v2'). Requires feature_views with exactly one entry and enable_online_feature_view_versioning to be enabled. + remote (bool): If True, delegate materialization to the feature server instead of + running the batch engine locally. The server URL is taken from online_store.path. + timeout (float): Max seconds to wait when remote=True and wait=True. + poll_interval (float): Seconds between registry polls when remote=True and wait=True. + wait (bool): If True (default), block until remote materialization finishes. + If False, return after the server accepts the job; use poll_materialization() + to check status later. Examples: Materialize all features into the online store over the interval @@ -2643,7 +2727,7 @@ 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(): + if remote and not _force_local: return self._remote_materialize( start_date=start_date, end_date=end_date, @@ -2651,6 +2735,9 @@ def materialize( disable_event_timestamp=disable_event_timestamp, full_feature_names=full_feature_names, version=version, + timeout=timeout, + poll_interval=poll_interval, + wait=wait, ) parsed_version = self._validate_materialize_version(version, feature_views) @@ -2697,10 +2784,11 @@ def tqdm_builder(length): end_date = utils.make_tzaware(end_date) # Transition state to MATERIALIZING before starting. - # Only enforce when the state machine is active (not STATE_UNSPECIFIED). + # Skip when _force_local: the async handler already set MATERIALIZING. previous_state = getattr(feature_view, "state", None) if ( - hasattr(feature_view, "state") + not _force_local + and hasattr(feature_view, "state") and feature_view.state != FeatureViewState.STATE_UNSPECIFIED ): if not feature_view.state.can_transition_to( @@ -2730,9 +2818,9 @@ def tqdm_builder(length): ) except Exception: fv_success = False - # Roll back state to previous value on failure. if ( - hasattr(feature_view, "state") + not _force_local + and hasattr(feature_view, "state") and previous_state is not None and previous_state != FeatureViewState.STATE_UNSPECIFIED ): diff --git a/sdk/python/feast/feature_view.py b/sdk/python/feast/feature_view.py index a5d3c8d9537..0b7e28f555b 100644 --- a/sdk/python/feast/feature_view.py +++ b/sdk/python/feast/feature_view.py @@ -416,6 +416,7 @@ def __eq__(self, other): or normalize_version_string(self.version) != normalize_version_string(other.version) or self.org != other.org + or self.state != other.state ): return False 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 47435914b1f..942927ec2e8 100644 --- a/sdk/python/feast/infra/feature_servers/local_process/config.py +++ b/sdk/python/feast/infra/feature_servers/local_process/config.py @@ -1,6 +1,4 @@ -from typing import Literal, Optional - -from pydantic import model_validator +from typing import Literal from feast.infra.feature_servers.base_config import BaseFeatureServerConfig @@ -11,25 +9,3 @@ 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/tests/integration/materialization/test_remote_materialize.py b/sdk/python/tests/integration/materialization/test_remote_materialize.py index 2bd17dcb4fc..3976a316012 100644 --- a/sdk/python/tests/integration/materialization/test_remote_materialize.py +++ b/sdk/python/tests/integration/materialization/test_remote_materialize.py @@ -1,9 +1,13 @@ """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 → +FeatureStore with online_store pointing at the server, and verifies the +full flow using remote=True on the materialize() SDK call: + client.materialize(remote=True) → POST /materialize-async → server runs locally → FV state transitions → client polls registry → returns success + +Uses SQL registry (SQLite) to match the production path and exercise +the state transitions that the file-based registry handles differently. """ import socket @@ -20,6 +24,7 @@ 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.infra.registry.sql import SqlRegistryConfig from feast.types import Float32, Int64 @@ -31,8 +36,8 @@ def _find_free_port() -> int: @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") + """Set up a minimal feature repo with SQL registry and SQLite online store.""" + registry_db = str(tmp_path / "registry.db") online_store_path = str(tmp_path / "online_store.db") data_path = str(tmp_path / "data.parquet") @@ -51,7 +56,11 @@ def e2e_repo(tmp_path): config = RepoConfig( project="test_remote_mat", provider="local", - registry=registry_path, + registry=SqlRegistryConfig( + registry_type="sql", + path=f"sqlite:///{registry_db}", + cache_ttl_seconds=1, + ), online_store=SqliteOnlineStoreConfig(path=online_store_path), entity_key_serialization_version=3, ) @@ -79,13 +88,13 @@ def e2e_repo(tmp_path): store.apply([entity, fv]) - return store, config, registry_path, online_store_path + return store, config, registry_db, 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 + store, config, registry_db, online_store_path = e2e_repo app = get_app(store) server_port = _find_free_port() @@ -118,44 +127,41 @@ def feature_server(e2e_repo): 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, - ) + """Create a client FeatureStore with online_store.path pointing at the server.""" + from feast.infra.online_stores.remote import RemoteOnlineStoreConfig 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, + online_store=RemoteOnlineStoreConfig( + type="remote", + path=f"http://127.0.0.1:{server_port}", ), + entity_key_serialization_version=3, ) return FeatureStore(config=client_config) def test_remote_materialize_e2e(feature_server): - """Full E2E: client with remote mode materializes through server.""" + """Full E2E: client with remote=True 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, + start_date=now - timedelta(days=7), + end_date=now, feature_views=["test_feature_view"], + remote=True, + timeout=30.0, + poll_interval=0.5, ) - fv = client_store.registry.get_feature_view("test_feature_view", "test_remote_mat") + fv = client_store.registry.get_feature_view( + "test_feature_view", "test_remote_mat", allow_cache=False + ) assert fv.state == FeatureViewState.AVAILABLE_ONLINE online_features = client_store.get_online_features( @@ -177,9 +183,14 @@ def test_remote_materialize_incremental_e2e(feature_server): client_store.materialize_incremental( end_date=now, feature_views=["test_feature_view"], + remote=True, + timeout=30.0, + poll_interval=0.5, ) - fv = client_store.registry.get_feature_view("test_feature_view", "test_remote_mat") + fv = client_store.registry.get_feature_view( + "test_feature_view", "test_remote_mat", allow_cache=False + ) assert fv.state == FeatureViewState.AVAILABLE_ONLINE online_features = client_store.get_online_features( @@ -190,22 +201,76 @@ def test_remote_materialize_incremental_e2e(feature_server): 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.""" +def test_remote_materialize_re_materialize(feature_server): + """Re-materialization on already AVAILABLE_ONLINE FVs waits for completion. + + Validates the polling race fix: server sets MATERIALIZING before 202, + so the client doesn't short-circuit on stale AVAILABLE_ONLINE. + """ 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) + # First materialization client_store.materialize( start_date=now - timedelta(days=7), end_date=now, feature_views=["test_feature_view"], + remote=True, + timeout=30.0, + poll_interval=0.5, + ) + + fv = client_store.registry.get_feature_view( + "test_feature_view", "test_remote_mat", allow_cache=False + ) + assert fv.state == FeatureViewState.AVAILABLE_ONLINE + + # Second materialization — must NOT return instantly + start_time = time.monotonic() + client_store.materialize( + start_date=now - timedelta(days=7), + end_date=now, + feature_views=["test_feature_view"], + remote=True, + timeout=30.0, + poll_interval=0.5, + ) + elapsed = time.monotonic() - start_time + + # Should have waited at least one poll cycle (0.5s), not returned in <0.1s + assert elapsed >= 0.3, ( + f"Re-materialization returned in {elapsed:.2f}s — polling race not fixed" + ) + + fv = client_store.registry.get_feature_view( + "test_feature_view", "test_remote_mat", allow_cache=False + ) + assert fv.state == FeatureViewState.AVAILABLE_ONLINE + + +def test_remote_materialize_force_local_bypasses_remote(feature_server): + """_force_local=True runs locally even when remote=True. + + Uses the server_store (local online store) since _force_local skips + the remote HTTP call entirely and writes to the local online store. + """ + server_store, server_config, server_port = feature_server + + now = datetime.now(tz=timezone.utc) + + server_store.materialize( + start_date=now - timedelta(days=7), + end_date=now, + feature_views=["test_feature_view"], + remote=True, _force_local=True, ) - fv = client_store.registry.get_feature_view("test_feature_view", "test_remote_mat") + fv = server_store.registry.get_feature_view( + "test_feature_view", "test_remote_mat", allow_cache=False + ) assert fv.state == FeatureViewState.AVAILABLE_ONLINE @@ -226,4 +291,46 @@ def test_remote_materialize_server_error_propagates(feature_server): start_date=now - timedelta(days=1), end_date=now, feature_views=["nonexistent_fv"], + remote=True, + timeout=10.0, + poll_interval=0.5, ) + + +def test_remote_materialize_failure_resets_state(feature_server): + """When server-side materialization fails, FV state resets to GENERATED. + + The client's seen_materializing logic detects MATERIALIZING → GENERATED + as a failure and reports it promptly (within one poll cycle). + """ + server_store, server_config, server_port = feature_server + client_store = _make_client_store(server_config, server_port) + + now = datetime.now(tz=timezone.utc) + + # Break the data source so materialization fails in the background thread + fv = server_store.registry.get_feature_view( + "test_feature_view", "test_remote_mat", allow_cache=False + ) + fv.batch_source = FileSource( + path="/nonexistent/path/that/will/fail.parquet", + timestamp_field="event_timestamp", + ) + server_store.apply([fv]) + + # Client triggers remote materialization — server will fail in background + with pytest.raises(Exception, match="Remote materialization failed"): + client_store.materialize( + start_date=now - timedelta(days=7), + end_date=now, + feature_views=["test_feature_view"], + remote=True, + timeout=30.0, + poll_interval=0.5, + ) + + # FV state should be back to GENERATED (not stuck at MATERIALIZING) + fv = client_store.registry.get_feature_view( + "test_feature_view", "test_remote_mat", allow_cache=False + ) + assert fv.state == FeatureViewState.GENERATED diff --git a/sdk/python/tests/unit/test_remote_materialize.py b/sdk/python/tests/unit/test_remote_materialize.py index a6d18639f25..abba6feebc0 100644 --- a/sdk/python/tests/unit/test_remote_materialize.py +++ b/sdk/python/tests/unit/test_remote_materialize.py @@ -287,40 +287,37 @@ def test_materialize_async_validates_timestamp_order(self): class TestFeatureStoreRemoteGate: - def test_materialize_calls_remote_when_mode_is_remote(self): - """materialize() delegates to _remote_materialize when mode is remote.""" + def test_materialize_calls_remote_when_remote_true(self): + """materialize(remote=True) delegates to _remote_materialize.""" 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"], + remote=True, ) 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.""" + """materialize() with _force_local=True skips remote even if remote=True.""" 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"], + remote=True, _force_local=True, ) except Exception: @@ -328,53 +325,121 @@ def test_materialize_skips_remote_when_force_local(self): fs._remote_materialize.assert_not_called() - def test_is_remote_materialize_mode_false_by_default(self): + def test_materialize_local_by_default(self): + """materialize() without remote=True does NOT delegate remotely.""" from feast.feature_store import FeatureStore fs = MagicMock() - fs.config.feature_server = None - result = FeatureStore._is_remote_materialize_mode(fs) - assert result is False + fs._remote_materialize = MagicMock() + + 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"], + ) + except Exception: + pass - def test_is_remote_materialize_mode_true(self): + fs._remote_materialize.assert_not_called() + + def test_get_remote_materialize_url_uses_online_store_path(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 + fs.config.online_store.path = "http://feast-server:80/" + result = FeatureStore._get_remote_materialize_url(fs) + assert result == "http://feast-server:80" - def test_is_remote_materialize_mode_local(self): + def test_get_remote_materialize_url_raises_without_path(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 + fs.config.online_store.path = None + with pytest.raises(ValueError, match="online_store.path must be set"): + FeatureStore._get_remote_materialize_url(fs) - def test_get_feature_server_url_raises_without_url(self): + def test_materialize_passes_wait_false(self): + """materialize(remote=True, wait=False) forwards wait to remote helper.""" 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) + fs._remote_materialize = MagicMock() + + FeatureStore.materialize( + fs, + start_date=datetime(2024, 1, 1, tzinfo=timezone.utc), + end_date=datetime(2024, 1, 2, tzinfo=timezone.utc), + feature_views=["fv1"], + remote=True, + wait=False, + ) - def test_get_feature_server_url_strips_trailing_slash(self): + assert fs._remote_materialize.call_args.kwargs["wait"] is False + + +class TestPollMaterializationOneShot: + def test_poll_materialization_reports_current_states(self): + """One-shot poll returns status for each FV without waiting.""" from feast.feature_store import FeatureStore + from feast.materialization_status import FVMaterializationStatus fs = MagicMock() - fs.config.feature_server.url = "http://feast-server:80/" - result = FeatureStore._get_feature_server_url(fs) - assert result == "http://feast-server:80" + fs.project = "test" + + def mock_get_fv(name, project, allow_cache=False): + fv = MagicMock() + if name == "fv_done": + fv.state = FeatureViewState.AVAILABLE_ONLINE + elif name == "fv_running": + fv.state = FeatureViewState.MATERIALIZING + else: + fv.state = FeatureViewState.GENERATED + return fv + + fs.registry.get_feature_view = mock_get_fv + # Bind real helper methods used by poll_materialization + fs._fv_state_to_materialization_status = ( + FeatureStore._fv_state_to_materialization_status.__get__(fs, FeatureStore) + ) + + results = FeatureStore.poll_materialization( + fs, feature_views=["fv_done", "fv_running", "fv_pending"] + ) - def test_get_feature_server_url_validates_scheme(self): + by_name = {r.name: r.status for r in results} + assert by_name["fv_done"] == FVMaterializationStatus.SUCCEEDED + assert by_name["fv_running"] == FVMaterializationStatus.RUNNING + assert by_name["fv_pending"] == FVMaterializationStatus.PENDING + + def test_remote_wait_false_skips_polling(self): + """wait=False returns after POST without calling the poller.""" 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) + fs.config.online_store.path = "http://server:80" + fs.config.online_store.cert = None + fs.config.auth_config = None + fs.project = "test" + fs._get_remote_materialize_url = MagicMock(return_value="http://server:80") + + 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) + + FeatureStore._remote_materialize_common( + fs, + "/materialize-async", + {"feature_views": ["fv1"]}, + ["fv1"], + wait=False, + ) + + mock_session.post.assert_called_once() + fs.registry.get_feature_view.assert_not_called() # --------------------------------------------------------------------------- @@ -383,16 +448,12 @@ def test_get_feature_server_url_validates_scheme(self): class TestRegistryStatusMapping: - def test_generated_before_materializing_is_pending(self): - """GENERATED state before we've ever seen MATERIALIZING = PENDING.""" + def _setup_fs_mock(self): 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.online_store.path = "http://server:80" + fs.config.online_store.cert = None fs.config.auth_config = None fs.project = "test" @@ -401,9 +462,18 @@ def test_generated_before_materializing_is_pending(self): 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") + fs._get_remote_materialize_url = MagicMock(return_value="http://server:80") + fs._fv_state_to_materialization_status = ( + FeatureStore._fv_state_to_materialization_status.__get__(fs, FeatureStore) + ) + return fs + + def test_generated_before_materializing_is_pending(self): + """GENERATED state before we've ever seen MATERIALIZING = PENDING.""" + from feast.feature_store import FeatureStore + + fs = self._setup_fs_mock() - # Simulate: GENERATED → MATERIALIZING → AVAILABLE_ONLINE call_count = {"n": 0} def mock_get_fv(name, project, allow_cache=False): @@ -420,30 +490,20 @@ def mock_get_fv(name, project, allow_cache=False): fs.registry.get_feature_view = mock_get_fv FeatureStore._remote_materialize_common( - fs, "/materialize-async", {"feature_views": ["fv1"]}, ["fv1"] + fs, + "/materialize-async", + {"feature_views": ["fv1"]}, + ["fv1"], + timeout=1.0, + poll_interval=0.01, ) 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") + fs = self._setup_fs_mock() - # Simulate: MATERIALIZING → GENERATED (rollback = failure) call_count = {"n": 0} def mock_get_fv(name, project, allow_cache=False): @@ -459,12 +519,17 @@ def mock_get_fv(name, project, allow_cache=False): with pytest.raises(Exception, match="Remote materialization failed"): FeatureStore._remote_materialize_common( - fs, "/materialize-async", {"feature_views": ["fv1"]}, ["fv1"] + fs, + "/materialize-async", + {"feature_views": ["fv1"]}, + ["fv1"], + timeout=1.0, + poll_interval=0.01, ) # --------------------------------------------------------------------------- -# Tests for LocalFeatureServerConfig +# Tests for LocalFeatureServerConfig (no remote-specific fields) # --------------------------------------------------------------------------- @@ -475,33 +540,6 @@ def test_defaults(self): ) 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") + assert cfg.type == "local" + assert not hasattr(cfg, "materialize_mode") + assert not hasattr(cfg, "url")