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 #4526 (remote materialization). Signed-off-by: Aniket Paluskar <apaluska@redhat.com>
aniketpalu · Jul 8, 2026 · Jul 8, 2026 · Jul 8, 2026 · Jul 8, 2026 · Jul 10, 2026
commit 44d9a1152a762b161ca53f3778f5b84bd59bc79a
@@ -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

@@ -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(
             <BLANKLINE>
             ...
         """
+        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
@@ -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."""