Skip to content

feat: Add remote materialization support via feature server#6590

Open
aniketpalu wants to merge 10 commits into
feast-dev:masterfrom
aniketpalu:feat/remote-materialize
Open

feat: Add remote materialization support via feature server#6590
aniketpalu wants to merge 10 commits into
feast-dev:masterfrom
aniketpalu:feat/remote-materialize

Conversation

@aniketpalu

@aniketpalu aniketpalu commented Jul 8, 2026

Copy link
Copy Markdown
Contributor

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:

  • 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 Remote materialization #4526 (remote materialization).

Which issue(s) this PR fixes:

Checks

  • I've made sure the tests are passing.
  • My commits are signed off (git commit -s)
  • My PR title follows conventional commits format

Testing Strategy

  • Unit tests
  • Integration tests
  • Manual tests
  • Testing is not required for this change

Misc

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>
@aniketpalu
aniketpalu requested a review from a team as a code owner July 8, 2026 21:27
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>
Comment on lines +15 to +17
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."""

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

  1. 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.
  2. 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

  1. Backward compatibility - User can fall back without re-initializing client.
  2. Flexibility of Status polling - Full control to user to set interval of polling.
  3. No need of Operator changes.
  4. The same API can work for UI as well

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@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: 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.

  • 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.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

aniketpalu and others added 4 commits July 10, 2026 13:35
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__.
aniketpalu added a commit to aniketpalu/feast that referenced this pull request Jul 14, 2026
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>
aniketpalu added a commit to aniketpalu/feast that referenced this pull request Jul 14, 2026
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>
aniketpalu added a commit to aniketpalu/feast that referenced this pull request Jul 16, 2026
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-commenter

codecov-commenter commented Jul 18, 2026

Copy link
Copy Markdown

⚠️ Please install the 'codecov app svg image' to ensure uploads and comments are reliably processed by Codecov.

Codecov Report

❌ Patch coverage is 75.11521% with 54 lines in your changes missing coverage. Please review.
✅ Project coverage is 46.10%. Comparing base (4dc8757) to head (4b302d5).

Files with missing lines Patch % Lines
sdk/python/feast/feature_store.py 63.15% 31 Missing and 4 partials ⚠️
sdk/python/feast/feature_server.py 82.19% 12 Missing and 1 partial ⚠️
sdk/python/feast/infra/registry/snowflake.py 0.00% 3 Missing ⚠️
sdk/python/feast/infra/registry/sql.py 0.00% 3 Missing ⚠️
❗ Your organization needs to install the Codecov GitHub app to enable full functionality.
Additional details and impacted files

Impacted file tree graph

@@            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     
Flag Coverage Δ
go-feature-server 30.58% <ø> (ø)
python-unit 47.41% <75.11%> (+0.14%) ⬆️
Files with missing lines Coverage Δ
sdk/python/feast/feature_view.py 85.71% <ø> (ø)
sdk/python/feast/materialization_status.py 100.00% <100.00%> (ø)
sdk/python/feast/infra/registry/snowflake.py 39.35% <0.00%> (-0.22%) ⬇️
sdk/python/feast/infra/registry/sql.py 57.70% <0.00%> (-0.26%) ⬇️
sdk/python/feast/feature_server.py 65.43% <82.19%> (+2.59%) ⬆️
sdk/python/feast/feature_store.py 42.60% <63.15%> (+1.56%) ⬆️

Continue to review full report in Codecov by Harness.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 4dc8757...4b302d5. Read the comment docs.

🚀 New features to boost your workflow:
  • ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
  • 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

aniketpalu added a commit to aniketpalu/feast that referenced this pull request Jul 18, 2026
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 jyejare left a comment

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Comment on lines +713 to +728
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."""

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[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:

Suggested change
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]

Comment on lines +751 to +767
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

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[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:

Suggested change
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

Comment on lines +780 to +795

@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(

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[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:

Suggested change
@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)

Comment on lines 460 to +463
feature_view,
already_transitioned: list,
previous_states: dict,
_force_local: bool = False,

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[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:

Suggested change
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

Comment on lines +2424 to +2430
return FVMaterializationStatus.FAILED
return FVMaterializationStatus.PENDING
return FVMaterializationStatus.PENDING

def poll_materialization(
self,
feature_views: Optional[List[str]] = None,

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[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:

Suggested change
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

Comment on lines +2564 to +2574
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 [

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[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:

Suggested change
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()

@jyejare

jyejare commented Jul 21, 2026

Copy link
Copy Markdown
Collaborator

@aniketpalu Please fix DCO check for incorrect signoff. Also please update docs for remote materialization support.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

4 participants