feat: Add remote materialization support via feature server

sdk/python/feast/feature_server.py

@@ -90,12 +90,14 @@ class MaterializeRequest(BaseModel):
     feature_views: Optional[List[str]] = None
     disable_event_timestamp: bool = False
     full_feature_names: bool = False
+    version: Optional[str] = None
 
 
 class MaterializeIncrementalRequest(BaseModel):
     end_ts: str
     feature_views: Optional[List[str]] = None
     full_feature_names: bool = False
+    version: Optional[str] = None
 
 
 class GetOnlineFeaturesRequest(BaseModel):
@@ -708,6 +710,134 @@ async def materialize_incremental(request: MaterializeIncrementalRequest) -> Non
                 full_feature_names=request.full_feature_names,
             )
 
+    def _authorize_materialize_views(
+        request_feature_views: Optional[List[str]],
+    ) -> List[str]:
+        """Shared authz + FV name resolution for async materialize endpoints."""
+        feature_views_to_materialize = store._get_feature_views_to_materialize(
+            request_feature_views
+        )
+        for fv in feature_views_to_materialize:
+            assert_permissions(
+                resource=fv,
+                actions=[AuthzedAction.WRITE_ONLINE],
+            )
+        return [fv.name for fv in feature_views_to_materialize]
+
+    def _parse_materialize_timestamps(request: MaterializeRequest):
+        """Parse and validate timestamps from a materialize request."""
+        if request.disable_event_timestamp:
+            now = datetime.now()
+            return datetime(1970, 1, 1), now
+
+        if not request.start_ts or not request.end_ts:
+            raise ValueError(
+                "start_ts and end_ts are required when disable_event_timestamp is False"
+            )
+        try:
+            start_date = utils.make_tzaware(parser.parse(request.start_ts))
+            end_date = utils.make_tzaware(parser.parse(request.end_ts))
+        except (ValueError, TypeError) as e:
+            raise ValueError(f"Invalid timestamp format: {e}") from e
+
+        if start_date >= end_date:
+            raise ValueError(
+                f"start_ts ({start_date}) must be before end_ts ({end_date})"
+            )
+        return start_date, end_date
+
+    def _check_already_materializing(fv_names: List[str]) -> Optional[JSONResponse]:
+        """Return 409 if any requested FV is already materializing."""
+        from feast.feature_view import FeatureViewState
+
+        for fv_name in fv_names:
+            try:
+                fv = store.registry.get_feature_view(
+                    fv_name, store.project, allow_cache=False
+                )
+                if getattr(fv, "state", None) == FeatureViewState.MATERIALIZING:
+                    return JSONResponse(
+                        status_code=409,
+                        content={
+                            "error": f"Feature view '{fv_name}' is already materializing"
+                        },
+                    )
+            except Exception:
+                pass
+        return None
+
+    @app.post("/materialize-async", dependencies=[Depends(inject_user_details)])
+    async def materialize_async(request: MaterializeRequest) -> JSONResponse:
+        with feast_metrics.track_request_latency("/materialize-async"):
+            fv_names = _authorize_materialize_views(request.feature_views)
+            start_date, end_date = _parse_materialize_timestamps(request)
+
+            conflict = _check_already_materializing(fv_names)
+            if conflict:
+                return conflict
+
+            def _run_materialize():
+                try:
+                    store.materialize(
+                        start_date,
+                        end_date,
+                        request.feature_views,
+                        disable_event_timestamp=request.disable_event_timestamp,
+                        full_feature_names=request.full_feature_names,
+                        version=request.version,
+                        _force_local=True,
+                    )
+                except Exception as e:
+                    logger.error(
+                        f"Async materialization failed for {fv_names}: {e}",
+                        exc_info=True,
+                    )
+
+            loop = asyncio.get_running_loop()
+            loop.run_in_executor(None, _run_materialize)
+
+            return JSONResponse(
+                status_code=202,
+                content={"status": "accepted", "feature_views": fv_names},
+            )
+
+    @app.post(
+        "/materialize-incremental-async",
+        dependencies=[Depends(inject_user_details)],
+    )
+    async def materialize_incremental_async(
+        request: MaterializeIncrementalRequest,
+    ) -> JSONResponse:
+        with feast_metrics.track_request_latency("/materialize-incremental-async"):
+            fv_names = _authorize_materialize_views(request.feature_views)
+
+            conflict = _check_already_materializing(fv_names)
+            if conflict:
+                return conflict
+
+            def _run_materialize_incremental():
+                try:
+                    store.materialize_incremental(
+                        utils.make_tzaware(parser.parse(request.end_ts)),
+                        request.feature_views,
+                        full_feature_names=request.full_feature_names,
+                        version=request.version,
+                        _force_local=True,
+                    )
+                except Exception as e:
+                    logger.error(
+                        f"Async incremental materialization failed for {fv_names}: {e}",
+                        exc_info=True,
+                    )
+
+            loop = asyncio.get_running_loop()
+            loop.run_in_executor(None, _run_materialize_incremental)
+
+            return JSONResponse(
+                status_code=202,
+                content={"status": "accepted", "feature_views": fv_names},
+            )
+
     @app.exception_handler(Exception)
     async def rest_exception_handler(request: Request, exc: Exception):
         # Print the original exception on the server side

sdk/python/feast/feature_store.py

@@ -2192,12 +2192,201 @@ def _materialize_odfv(
         )
         self.write_to_online_store(feature_view.name, df=transformed_df)
 
+    def _is_remote_materialize_mode(self) -> bool:
+        """Check if materialization should be delegated to a remote feature server."""
+        fs_cfg = self.config.feature_server
+        if fs_cfg is None:
+            return False
+        return getattr(fs_cfg, "materialize_mode", "local") == "remote"
+
+    def _get_feature_server_url(self) -> str:
+        """Get feature server URL, raising if not configured."""
+        from urllib.parse import urlparse
+
+        fs_cfg = self.config.feature_server
+        url = getattr(fs_cfg, "url", None)
+        if not url:
+            raise ValueError(
+                "feature_server.url must be set when materialize_mode='remote'. "
+                "Example: feature_server:\n"
+                "  type: local\n"
+                "  materialize_mode: remote\n"
+                "  url: http://feast-online.feast-ns.svc.cluster.local:80"
+            )
+        parsed = urlparse(url)
+        if parsed.scheme not in ("http", "https"):
+            raise ValueError(
+                f"feature_server.url must use http or https scheme, got: {url}"
+            )
+        return url.rstrip("/")
+
+    def _get_remote_http_session(self):
+        """Get an HTTP session with auth configured for the feature server."""
+        import requests
+
+        auth_config = getattr(self.config, "auth_config", None)
+        if auth_config and getattr(auth_config, "type", "no_auth") != "no_auth":
+            from feast.permissions.client.http_auth_requests_wrapper import (
+                get_http_auth_requests_session,
+            )
+
+            return get_http_auth_requests_session(auth_config)
+
+        return requests.Session()
+
+    def _remote_materialize_common(
+        self,
+        endpoint: str,
+        payload: dict,
+        fv_names: List[str],
+    ) -> None:
+        """Shared logic for remote materialization: trigger server, poll registry.
+
+        Args:
+            endpoint: Server endpoint path (e.g., "/materialize-async").
+            payload: JSON payload for the POST request.
+            fv_names: Feature view names to track.
+        """
+        from feast.materialization_status import (
+            FVMaterializationStatus,
+            FVResult,
+            poll_materialization_status,
+        )
+
+        server_url = self._get_feature_server_url()
+        fs_cfg = self.config.feature_server
+
+        _logger.info(
+            f"Remote materialize: triggering {len(fv_names)} feature views "
+            f"via {endpoint} on {server_url}"
+        )
+
+        session = self._get_remote_http_session()
+        try:
+            resp = session.post(
+                f"{server_url}{endpoint}",
+                json=payload,
+                timeout=getattr(fs_cfg, "http_timeout", 30),
+            )
+            resp.raise_for_status()
+        except Exception as e:
+            raise Exception(
+                f"Failed to trigger remote materialization at "
+                f"{server_url}{endpoint}: {e}"
+            ) from e
+        finally:
+            session.close()
+
+        seen_materializing: set = set()
+
+        def registry_status_fn(fv_name: str) -> FVMaterializationStatus:
+            fv = self.registry.get_feature_view(
+                fv_name, self.project, allow_cache=False
+            )
+            state = getattr(fv, "state", None)
+            if state == FeatureViewState.AVAILABLE_ONLINE:
+                return FVMaterializationStatus.SUCCEEDED
+            elif state == FeatureViewState.MATERIALIZING:
+                seen_materializing.add(fv_name)
+                return FVMaterializationStatus.RUNNING
+            elif state == FeatureViewState.GENERATED:
+                if fv_name in seen_materializing:
+                    return FVMaterializationStatus.FAILED
+                return FVMaterializationStatus.PENDING
+            return FVMaterializationStatus.PENDING
+
+        def on_change(result: FVResult):
+            symbol = {
+                FVMaterializationStatus.SUCCEEDED: "+",
+                FVMaterializationStatus.FAILED: "x",
+                FVMaterializationStatus.RUNNING: "~",
+            }.get(result.status, "?")
+            _logger.info(
+                f"  [{symbol}] {result.name}: {result.status.value} "
+                f"({result.elapsed_seconds:.1f}s)"
+            )
+
+        timeout = getattr(fs_cfg, "materialize_timeout", 3600.0)
+        poll_interval = getattr(fs_cfg, "materialize_poll_interval", 5.0)
+
+        results = poll_materialization_status(
+            feature_view_names=fv_names,
+            status_fn=registry_status_fn,
+            poll_interval=poll_interval,
+            timeout=timeout,
+            on_status_change=on_change,
+        )
+
+        failed = [r for r in results if r.status == FVMaterializationStatus.FAILED]
+        if failed:
+            names = ", ".join(r.name for r in failed)
+            errors = "; ".join(f"{r.name}: {r.error}" for r in failed if r.error)
+            msg = f"Remote materialization failed for: {names}"
+            if errors:
+                msg += f" ({errors})"
+            raise Exception(msg)
+
+    def _remote_materialize(
+        self,
+        start_date: datetime,
+        end_date: datetime,
+        feature_views: Optional[List[str]] = None,
+        disable_event_timestamp: bool = False,
+        full_feature_names: bool = False,
+        version: Optional[str] = None,
+    ) -> None:
+        """Delegate materialization to the feature server, poll via registry."""
+        fv_names = feature_views or [
+            fv.name
+            for fv in self._get_feature_views_to_materialize(None)
+            if not isinstance(fv, OnDemandFeatureView)
+        ]
+
+        payload: dict = {
+            "start_ts": start_date.isoformat(),
+            "end_ts": end_date.isoformat(),
+            "feature_views": fv_names,
+            "disable_event_timestamp": disable_event_timestamp,
+            "full_feature_names": full_feature_names,
+        }
+        if version is not None:
+            payload["version"] = version
+
+        self._remote_materialize_common("/materialize-async", payload, fv_names)
+
+    def _remote_materialize_incremental(
+        self,
+        end_date: datetime,
+        feature_views: Optional[List[str]] = None,
+        full_feature_names: bool = False,
+        version: Optional[str] = None,
+    ) -> None:
+        """Delegate incremental materialization to the feature server, poll via registry."""
+        fv_names = feature_views or [
+            fv.name
+            for fv in self._get_feature_views_to_materialize(None)
+            if not isinstance(fv, OnDemandFeatureView)
+        ]
+
+        payload: dict = {
+            "end_ts": end_date.isoformat(),
+            "feature_views": fv_names,
+            "full_feature_names": full_feature_names,
+        }
+        if version is not None:
+            payload["version"] = version
+
+        self._remote_materialize_common(
+            "/materialize-incremental-async", payload, fv_names
+        )
+
     def materialize_incremental(
         self,
         end_date: datetime,
         feature_views: Optional[List[str]] = None,
         full_feature_names: bool = False,
         version: Optional[str] = None,
+        _force_local: bool = False,
     ) -> None:
         """
         Materialize incremental new data from the offline store into the online store.
@@ -2231,6 +2420,14 @@ def materialize_incremental(
             <BLANKLINE>
             ...
         """
+        if not _force_local and self._is_remote_materialize_mode():
+            return self._remote_materialize_incremental(
+                end_date=end_date,
+                feature_views=feature_views,
+                full_feature_names=full_feature_names,
+                version=version,
+            )
+
         parsed_version = self._validate_materialize_version(version, feature_views)
         feature_views_to_materialize = self._get_feature_views_to_materialize(
             feature_views, version=parsed_version
@@ -2404,6 +2601,7 @@ def materialize(
         disable_event_timestamp: bool = False,
         full_feature_names: bool = False,
         version: Optional[str] = None,
+        _force_local: bool = False,
     ) -> None:
         """
         Materialize data from the offline store into the online store.
@@ -2441,6 +2639,16 @@ def materialize(
                 f"The given start_date {start_date} is greater than the given end_date {end_date}."
             )
 
+        if not _force_local and self._is_remote_materialize_mode():
+            return self._remote_materialize(
+                start_date=start_date,
+                end_date=end_date,
+                feature_views=feature_views,
+                disable_event_timestamp=disable_event_timestamp,
+                full_feature_names=full_feature_names,
+                version=version,
+            )
+
         parsed_version = self._validate_materialize_version(version, feature_views)
         feature_views_to_materialize = self._get_feature_views_to_materialize(
             feature_views, version=parsed_version