feat: Add remote materialization support via feature server#6590
feat: Add remote materialization support via feature server#6590aniketpalu wants to merge 10 commits into
Conversation
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#4526 (remote materialization). Signed-off-by: Aniket Paluskar <apaluska@redhat.com>
Signed-off-by: Aniket Paluskar <apaluska@redhat.com>
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#4526 (remote materialization). Signed-off-by: Aniket Paluskar <apaluska@redhat.com>
- 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 <apaluska@redhat.com>
| 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.""" |
There was a problem hiding this comment.
I think we dont need to provide the materialize_mode config. We should strictly feature server take care of materialization delegation to compute engine. Leaving no choice to do materialization on client side.
There was a problem hiding this comment.
A cleaner approach might be: a RemoteComputeEngine (like RemoteOnlineStore/RemoteOfflineStore) that POSTs to the existing /materialize endpoint, returning a job handle with a dedicated /materialize-status/{job_id} endpoint - keeping the materialization delegation at the engine layer rather than the FeatureStore layer.
There was a problem hiding this comment.
if you mean something like
batch_engine:
type: remote.engine
path: https://feast-online:443 # same host as online
cert: /var/run/secrets/.../service-ca.crt
I thought about this before choosing current approach. I see two points which made me not take this decision.
- This adds another config section which has to indicate only one thing, i.e. trigger materialization on feature server. For this, it add 1 config section & user has to configure 2 redundant fields.
- Once client is initialized, user wont have option to try normal materialization flow if remote materialization fails for any reason.
My proposed approach is
store.materialize(remote=True, wait=False) --> pre-process check & return 202 Accepted response
store.poll_materialization() --> returns status
This approach has
- Backward compatibility - User can fall back without re-initializing client.
- Flexibility of Status polling - Full control to user to set interval of polling.
- No need of Operator changes.
- The same API can work for UI as well
There was a problem hiding this comment.
@aniketpalu Ok, so, I think this PR is mixing three orthogonal concerns: async execution, remote delegation, and job tracking.
-
Remote delegation already works today - POST /materialize runs materialization server-side. A client with
online_store: remotecan call this endpoint directly. As @jyejare mentioned, the feature server should strictly handle materialization - no per-call remote=True flag needed; the deployment topology already decides this. -
Async execution - The real gap is that long-running engines (SparkApplication) block the HTTP caller. The existing /materialize endpoint just needs an
asyncmode (?async=true → 202 with job_id) rather than entirely new/materialize-asyncendpoints. This serves both UI on-demand requests and SDK wait=False use cases, while CronJob continues using the existing sync path unchanged. -
Job tracking - Polling FeatureViewState from the registry is fragile (race conditions, no job_id, ambiguous failure, breaks FeatureView.eq). If we need job tracking, it should be a dedicated MaterializationJob entity in the registry with proper RPCs - not piggybacking on FV metadata. For SparkApplication, the SparkApp CR status on K8s is already the source of truth.
May be we can scope this PR to just remote delegation and async execution ? ?async=true on the existing /materialize endpoint (fire-and-forget, returns 202. Completion is already observable via existing mechanisms like materialization_intervals on the FV, SparkApp CR status) + SDK support for async with remote delegation.
Job tracking with dedicated status endpoints / registry entities can come as a separate effort.
There was a problem hiding this comment.
Remote delegation already works today - POST /materialize runs materialization server-side. A client with online_store: remote can call this endpoint directly. As @jyejare mentioned, the feature server should strictly handle materialization - no per-call remote=True flag needed; the deployment topology already decides this.
Async execution - The real gap is that long-running engines (SparkApplication) block the HTTP caller. The existing /materialize endpoint just needs an async mode (?async=true → 202 with job_id) rather than entirely new /materialize-async endpoints. This serves both UI on-demand requests and SDK wait=False use cases, while CronJob continues using the existing sync path unchanged.
agree on both point
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 <apaluska@redhat.com>
- 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 <apaluska@redhat.com>
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__.
Keep feast-dev#6550 batching in materialize loops; skip FV state transition when _force_local so async /materialize-async (already MATERIALIZING) works. Co-authored-by: Cursor <cursoragent@cursor.com>
Keep feast-dev#6550 batching; skip FV state transition when _force_local so /materialize-async (already MATERIALIZING) works with SparkApplication. Co-authored-by: Cursor <cursoragent@cursor.com>
Keep feast-dev#6550 batching; skip FV state transition when _force_local so /materialize-async (already MATERIALIZING) works with SparkApplication. Co-authored-by: Cursor <cursoragent@cursor.com>
Bring in feast-dev#6550 SparkApplication (supports_batch / _materialize_fvs_batch). Resolve feature_store.py by keeping master batch structure and threading _force_local through transition/rollback/batch helpers so async remote-mat does not double-set MATERIALIZING or wrongly roll back state. Co-authored-by: Cursor <cursoragent@cursor.com>
|
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #6590 +/- ##
==========================================
+ Coverage 45.96% 46.10% +0.14%
==========================================
Files 412 413 +1
Lines 48814 49027 +213
Branches 6908 6945 +37
==========================================
+ Hits 22437 22605 +168
- Misses 24826 24869 +43
- Partials 1551 1553 +2
Continue to review full report in Codecov by Harness.
🚀 New features to boost your workflow:
|
Refresh with current feast-dev#6590 tip (includes master merge with feast-dev#6550). Keep spark_application Dockerfile postgres extra for E2E. Co-authored-by: Cursor <cursoragent@cursor.com>
jyejare
left a comment
There was a problem hiding this comment.
This PR adds remote materialization support via feature server, allowing clients to delegate materialization to a remote server instead of running it locally. The implementation includes async endpoints in the feature server, client-side remote delegation logic, status polling mechanisms, and comprehensive test coverage. The feature is well-architected with proper error handling and state management.
| 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.""" |
There was a problem hiding this comment.
[Critical] Function defined inside endpoint handler scope
The function _authorize_materialize_views is defined inside the endpoint handler scope but references store from the outer scope. This creates a closure that captures the store variable, which could lead to unexpected behavior if the store changes or if this function is called in different contexts. This pattern violates the principle of explicit dependencies and makes testing difficult.
Suggested:
| 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.""" | |
| def _authorize_materialize_views( | |
| store: FeatureStore, 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 _check_already_materializing(fv_names: List[str]) -> Optional[JSONResponse]: | ||
| """Return 409 if any requested FV is already materializing.""" | ||
| 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 |
There was a problem hiding this comment.
[Critical] Function defined inside endpoint handler with closure dependencies
Similar to the previous issue, _check_already_materializing and other helper functions are defined inside the endpoint handler and capture variables from outer scope. This makes the code harder to test and reason about. These functions should be extracted as proper module-level functions with explicit parameters.
Suggested:
| def _check_already_materializing(fv_names: List[str]) -> Optional[JSONResponse]: | |
| """Return 409 if any requested FV is already materializing.""" | |
| 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 | |
| def _check_already_materializing( | |
| store: FeatureStore, fv_names: List[str] | |
| ) -> Optional[JSONResponse]: | |
| """Return 409 if any requested FV is already materializing.""" | |
| 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 | ||
|
|
||
| _update_fv_state(fv_names, FeatureViewState.MATERIALIZING) | ||
|
|
||
| def _run_materialize(): | ||
| try: | ||
| store.materialize( |
There was a problem hiding this comment.
[Warning] Materialize async endpoint lacks proper error handling
The /materialize-async endpoint catches all exceptions in the background thread but only logs them. If the materialization fails during the initial setup (before the background thread), the state won't be properly reset. The endpoint should have a try-catch around the initial setup and reset states appropriately.
Suggested:
| @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 | |
| _update_fv_state(fv_names, FeatureViewState.MATERIALIZING) | |
| def _run_materialize(): | |
| try: | |
| store.materialize( | |
| 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, | |
| ) | |
| # Set state to AVAILABLE_ONLINE on success | |
| _update_fv_state(fv_names, FeatureViewState.AVAILABLE_ONLINE) | |
| except Exception as e: | |
| logger.error( | |
| f"Async materialization failed for {fv_names}: {e}", | |
| exc_info=True, | |
| ) | |
| _update_fv_state(fv_names, FeatureViewState.GENERATED) |
| feature_view, | ||
| already_transitioned: list, | ||
| previous_states: dict, | ||
| _force_local: bool = False, |
There was a problem hiding this comment.
[Warning] Empty return in state transition function breaks existing logic
Adding an early return when _force_local=True means the previous_states[feature_view.name] = previous_state line is never reached, which could break rollback logic elsewhere. The function should still record the previous state even when skipping the transition.
Suggested:
| feature_view, | |
| already_transitioned: list, | |
| previous_states: dict, | |
| _force_local: bool = False, | |
| Skip when _force_local: the async handler already set MATERIALIZING. | |
| """ | |
| previous_state = getattr(feature_view, "state", None) | |
| previous_states[feature_view.name] = previous_state | |
| if _force_local: | |
| return # Skip state transition but preserve previous_state recording |
| return FVMaterializationStatus.FAILED | ||
| return FVMaterializationStatus.PENDING | ||
| return FVMaterializationStatus.PENDING | ||
|
|
||
| def poll_materialization( | ||
| self, | ||
| feature_views: Optional[List[str]] = None, |
There was a problem hiding this comment.
[Warning] State mapping logic could be more explicit
The _fv_state_to_materialization_status function uses a mutable seen_materializing set that's passed by reference. This side-effect pattern makes the function harder to reason about and test. Consider returning a tuple or using a more functional approach.
Suggested:
| return FVMaterializationStatus.FAILED | |
| return FVMaterializationStatus.PENDING | |
| return FVMaterializationStatus.PENDING | |
| def poll_materialization( | |
| self, | |
| feature_views: Optional[List[str]] = None, | |
| def _fv_state_to_materialization_status( | |
| self, state, *, seen_materializing: Optional[set] = None, fv_name: str = "" | |
| ) -> Tuple[FVMaterializationStatus, bool]: | |
| """Map FeatureViewState to FVMaterializationStatus. | |
| Returns: | |
| Tuple of (status, was_materializing) where was_materializing indicates | |
| if this FV was seen in MATERIALIZING state. | |
| """ | |
| from feast.materialization_status import FVMaterializationStatus | |
| if state == FeatureViewState.AVAILABLE_ONLINE: | |
| return FVMaterializationStatus.SUCCEEDED, False | |
| if state == FeatureViewState.MATERIALIZING: | |
| if seen_materializing is not None: | |
| seen_materializing.add(fv_name) | |
| return FVMaterializationStatus.RUNNING, True | |
| if state == FeatureViewState.GENERATED: | |
| was_materializing = seen_materializing is not None and fv_name in seen_materializing | |
| status = FVMaterializationStatus.FAILED if was_materializing else FVMaterializationStatus.PENDING | |
| return status, was_materializing | |
| return FVMaterializationStatus.PENDING, False |
| end_date: datetime, | ||
| feature_views: Optional[List[str]] = None, | ||
| 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 [ |
There was a problem hiding this comment.
[Warning] HTTP session not properly closed in all paths
The HTTP session is closed in a finally block, but if an exception occurs during session creation or configuration, the session might never be assigned, leading to a NameError. The session creation and closure should be more robust.
Suggested:
| end_date: datetime, | |
| feature_views: Optional[List[str]] = None, | |
| 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 [ | |
| session = None | |
| try: | |
| session = self._get_remote_http_session() | |
| post_kwargs: dict = { | |
| "json": payload, | |
| "timeout": 30, | |
| } | |
| cert = getattr(self.config.online_store, "cert", None) | |
| 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( | |
| f"Failed to trigger remote materialization at " | |
| f"{server_url}{endpoint}: {e}" | |
| ) from e | |
| finally: | |
| if session is not None: | |
| session.close() |
|
@aniketpalu Please fix DCO check for incorrect signoff. Also please update docs for remote materialization support. |
What this PR does / why we need it:
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:
Which issue(s) this PR fixes:
Checks
git commit -s)Testing Strategy
Misc