Visitar URL original
feat(server): Remote Materialization by aniketpalu · Pull Request #6649 · feast-dev/feast · GitHub
Skip to content

feat(server): Remote Materialization #6649

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ntkathole merged 13 commits into from
Aug 6, 2026
Merged

feat(server): Remote Materialization #6649

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
f61be7d
feat(server): Add async materialization via ?async=true query param
aniketpalu Jul 28, 2026
9e121fc
Merge branch 'master' into feat/remote-materialize-v2
aniketpalu Jul 28, 2026
6c3aaa7
feat(server): Add async materialization via ?async=true query param
aniketpalu Jul 28, 2026
1384c60
feat(sdk): Auto-delegate materialize to feature server when online_st…
aniketpalu Jul 28, 2026
9a97b70
fix: Conflicts resolved
aniketpalu Jul 28, 2026
b5cb173
fix(server): Let store own MATERIALIZING transitions on async path
aniketpalu Jul 28, 2026
dc6f965
Merge branch 'master' into feat/remote-materialize-v2
aniketpalu Jul 30, 2026
d8e7a99
fix: Address #6649 review — run_async, race reserve, version threading
aniketpalu Jul 31, 2026
059c2d2
Merge branch 'master' into feat/remote-materialize-v2
aniketpalu Aug 2, 2026
c602fe3
fix(server): Only reset MATERIALIZING FVs on async materialize failure
aniketpalu Aug 5, 2026
7909fc4
fix: Default run_async to False and cover version forwarding
aniketpalu Aug 5, 2026
b8aa6e7
Merge branch 'master' into feat/remote-materialize-v2
aniketpalu Aug 6, 2026
a2810c8
fix(server): Use dedicated executor for async materialize
aniketpalu Aug 6, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
feat(server): Add async materialization via ?async=true query param 
When ?async=true is passed to /materialize and /materialize-incremental,
the endpoint fires off materialization in a background thread and returns
202 Accepted immediately. Concurrent requests for an already-MATERIALIZING
FV are rejected with 409. Without ?async=true, behavior is unchanged.
Client-side additions:
- remote=True param on store.materialize() / materialize_incremental()
  to delegate to the feature server (URL/TLS from online_store config)
- wait=False support with store.poll_materialization() for status polling
- FeatureView state set to MATERIALIZING before 202, reset on failure
Server-side additions:
- ?async=true on existing /materialize and /materialize-incremental
- ?force=true to override stuck MATERIALIZING state
- Four module-level helpers for testability:
  _authorize_materialize_views, _check_already_materializing,
  _update_fv_state, _parse_materialize_timestamps
Registry fixes:
- SQL and Snowflake registries now set FV state to AVAILABLE_ONLINE
  in apply_materialization() (parity with file-based registry)
Addresses #4526

Signed-off-by: Aniket Paluskar <apaluska@redhat.com>
  • Loading branch information
@aniketpalu
aniketpalu committed Jul 28, 2026
commit 6c3aaa78e3098581ab6964a362591b64f9d2de97
216 changes: 171 additions & 45 deletions sdk/python/feast/feature_server.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@
from fastapi import (
Depends,
FastAPI,
Query,
Request,
Response,
WebSocket,
Expand All @@ -54,6 +55,7 @@
)
from feast.feast_object import FeastObject
from feast.feature_server_utils import convert_response_to_dict
from feast.feature_view import FeatureViewState
from feast.feature_view_utils import get_feature_view_from_feature_store
from feast.filter_models import ComparisonFilter, CompoundFilter
from feast.permissions.action import WRITE, AuthzedAction
Expand Down Expand Up @@ -94,12 +96,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):
Comment on lines 97 to 122

@jyejare jyejare Jul 29, 2026

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.

[Suggestion] Missing version parameter documentation and validation

The version parameter was added to both MaterializeRequest and MaterializeIncrementalRequest but there's no documentation about what values are valid or how it affects the materialization behavior.

Suggested:

Suggested change
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):
version: Optional[str] = Field(
None,
description="Optional version to materialize (e.g., 'v2'). Requires feature_views with exactly one entry."
)

@aniketpalu aniketpalu Jul 31, 2026

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.

Done

Expand Down Expand Up @@ -333,6 +337,96 @@ async def load_static_artifacts(app: FastAPI, store):
logger.warning(f"Failed to load static artifacts: {e}")


def _authorize_materialize_views(
store: "feast.FeatureStore",
feature_view_names: Optional[List[str]],
) -> List[str]:
"""Resolve + authorize feature views for materialization.

Returns the resolved list of FV names (all eligible FVs when
feature_view_names is None).
"""
feature_views_to_materialize = store._get_feature_views_to_materialize(
feature_view_names
)
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(
store: "feast.FeatureStore",
fv_names: List[str],
) -> Optional[JSONResponse]:
"""Return a 409 JSONResponse if any requested FV is already MATERIALIZING."""
conflicting: List[str] = []
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:
conflicting.append(fv_name)
except Exception:
pass
if conflicting:
return JSONResponse(
status_code=409,
content={
"error": (
f"Cannot start async materialization — the following feature "
f"views are already in MATERIALIZING state: {conflicting}. "
f"Use ?force=true to override."
),
"feature_views": conflicting,
},
)
return None


def _update_fv_state(
Comment on lines +397 to +409

@jyejare jyejare Jul 29, 2026

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] Silent exception handling could hide real errors

The _reset_stuck_materializing_to_generated and _check_already_materializing functions silently ignore all exceptions when accessing feature views. This could hide legitimate errors like registry corruption or network issues, making debugging difficult.

Suggested:

Suggested change
content={
"error": (
f"Cannot start async materialization — the following feature "
f"views are already in MATERIALIZING state: {conflicting}. "
f"Use ?force=true to override."
),
"feature_views": conflicting,
},
)
return None
def _update_fv_state(
except (FeatureViewNotFoundException, KeyError):
# Expected when FV doesn't exist
pass
except Exception as e:
logger.warning(f"Unexpected error checking state for {fv_name}: {e}")
pass

@aniketpalu aniketpalu Jul 31, 2026

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.

Accepted

store: "feast.FeatureStore",
fv_names: List[str],
state: FeatureViewState,
) -> None:
"""Set FV state in the registry for each named feature view."""
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}")


def _parse_materialize_timestamps(
request: "MaterializeRequest",
) -> tuple:
"""Parse and validate start/end timestamps from a MaterializeRequest."""
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 get_app(
store: "feast.FeatureStore",
registry_ttl_sec: int = DEFAULT_FEATURE_SERVER_REGISTRY_TTL,
Expand Down Expand Up @@ -798,36 +892,47 @@ async def chat_ui():
return Response(content=content, media_type="text/html")

@app.post("/materialize", dependencies=[Depends(inject_user_details)])
async def materialize(request: MaterializeRequest) -> None:
async def materialize(
request: MaterializeRequest,
async_mode: bool = Query(False, alias="async"),
force: bool = Query(False),
):
with feast_metrics.track_request_latency("/materialize"):
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 = _authorize_materialize_views(store, request.feature_views)
start_date, end_date = _parse_materialize_timestamps(request)

if async_mode:
if not force:
conflict = _check_already_materializing(store, fv_names)
if conflict:
return conflict

_update_fv_state(store, fv_names, FeatureViewState.MATERIALIZING)

def _run_materialize():
try:
store.materialize(
start_date,
end_date,
fv_names,
disable_event_timestamp=request.disable_event_timestamp,
full_feature_names=request.full_feature_names,
version=request.version,
)
except Exception as e:
logger.error(
f"Async materialization failed for {fv_names}: {e}",
exc_info=True,
)
_update_fv_state(store, fv_names, FeatureViewState.GENERATED)

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))
loop = asyncio.get_running_loop()
loop.run_in_executor(None, _run_materialize)

return JSONResponse(
status_code=202,
content={"status": "accepted", "feature_views": fv_names},
)

await run_in_threadpool(
store.materialize,
Expand All @@ -839,27 +944,48 @@ async def materialize(request: MaterializeRequest) -> None:
)

Comment on lines 998 to 1016

@jyejare jyejare Jul 29, 2026

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.

[Nitpick] Inconsistent parameter passing in sync mode

In the synchronous code path for materialize, the version parameter is passed but the feature_views parameter uses the original request.feature_views instead of the resolved fv_names list.

Suggested:

Suggested change
await run_in_threadpool(
store.materialize,
start_date,
end_date,
fv_names, # Use resolved names for consistency
disable_event_timestamp=request.disable_event_timestamp,
full_feature_names=request.full_feature_names,
version=request.version, # Don't forget version parameter
)

@aniketpalu aniketpalu Jul 31, 2026

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.

sync path now passes resolved fv_names and version=request.version (aligned with async).

@app.post("/materialize-incremental", dependencies=[Depends(inject_user_details)])
async def materialize_incremental(request: MaterializeIncrementalRequest) -> None:
async def materialize_incremental(
request: MaterializeIncrementalRequest,
async_mode: bool = Query(False, alias="async"),
force: bool = Query(False),
):
with feast_metrics.track_request_latency("/materialize-incremental"):
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
fv_names = _authorize_materialize_views(store, request.feature_views)
end_date = utils.make_tzaware(parser.parse(request.end_ts))

if async_mode:
if not force:
conflict = _check_already_materializing(store, fv_names)
if conflict:
return conflict

_update_fv_state(store, fv_names, FeatureViewState.MATERIALIZING)

def _run_materialize_incremental():
try:
store.materialize_incremental(
end_date,
fv_names,
full_feature_names=request.full_feature_names,
)
except Exception as e:
logger.error(
f"Async materialize-incremental failed for {fv_names}: {e}",
exc_info=True,
)
_update_fv_state(store, fv_names, FeatureViewState.GENERATED)

loop = asyncio.get_running_loop()
loop.run_in_executor(None, _run_materialize_incremental)

@ntkathole ntkathole Jul 31, 2026

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.

non-blocking but I think it's better if we used dedicated executor, instead of default shared pool, for materialization so that it won't block other operations if multiple executors in progress

@aniketpalu aniketpalu Aug 6, 2026

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.

Agreed, added separate executor


return JSONResponse(
status_code=202,
content={"status": "accepted", "feature_views": fv_names},
)
for fv in feature_views_to_materialize:
assert_permissions(
resource=fv,
actions=[AuthzedAction.WRITE_ONLINE],
)

await run_in_threadpool(
store.materialize_incremental,
utils.make_tzaware(parser.parse(request.end_ts)),
end_date,
request.feature_views,
full_feature_names=request.full_feature_names,
)
Expand Down
34 changes: 26 additions & 8 deletions sdk/python/feast/feature_store.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -510,7 +510,7 @@ def _transition_fv_to_materializing(

Rolls back all already-transitioned FVs if this one can't transition.
"""
previous_state = getattr(feature_view, "state", None)
previous_states[feature_view.name] = getattr(feature_view, "state", None)

@jyejare jyejare Jul 29, 2026

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] Race condition in state tracking initialization

The line previous_states[feature_view.name] = getattr(feature_view, "state", None) was moved before the state transition logic, but this creates a critical bug. If the transition fails and we need to rollback other FVs, the previous_states dict won't have entries for FVs that failed before this one was reached.

Suggested:

Suggested change
Rolls back all already-transitioned FVs if this one can't transition.
"""
previous_state = getattr(feature_view, "state", None)
previous_states[feature_view.name] = getattr(feature_view, "state", None)
previous_state = getattr(feature_view, "state", None)
if (
hasattr(feature_view, "state")
and feature_view.state != FeatureViewState.STATE_UNSPECIFIED
):
# ... validation logic ...
feature_view.state = FeatureViewState.MATERIALIZING
self.registry.apply_feature_view(feature_view, self.project, commit=True)
previous_states[feature_view.name] = previous_state

@aniketpalu aniketpalu Jul 31, 2026 •
edited
Loading

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.

Thanks for the careful look — walked through both placements with the caller loop in mind; they have the same observable outcome (including failure cases).
Caller (simplified):

for feature_view, fv_start in fv_with_dates:
    self._transition_fv_to_materializing(feature_view, regular_fvs, previous_states)
    regular_fvs.append(feature_view)  # only after a successful return

On a failed transition, the failing FV is not in already_transitioned / regular_fvs. Rollback only restores FVs that already succeeded.

if (
hasattr(feature_view, "state")
and feature_view.state != FeatureViewState.STATE_UNSPECIFIED
Expand All @@ -523,7 +523,6 @@ def _transition_fv_to_materializing(
)
feature_view.state = FeatureViewState.MATERIALIZING
self.registry.apply_feature_view(feature_view, self.project, commit=True)
previous_states[feature_view.name] = previous_state

def _submit_and_process_materialization_jobs(
self,
Expand Down Expand Up @@ -2427,6 +2426,31 @@ def _materialize_odfv(
)
self.write_to_online_store(feature_view.name, df=transformed_df)

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(
"online_store.path must be set to use remote materialization. "
"Configure online_store with type: remote and a valid path."
)
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 materialize_incremental(
self,
end_date: datetime,
Expand Down Expand Up @@ -2560,8 +2584,6 @@ def tqdm_builder(length):
)
else:
for feature_view, start_date in regular_fvs_with_dates:
# Transition state to MATERIALIZING before starting.
# Only enforce when the state machine is active (not STATE_UNSPECIFIED).
previous_state = getattr(feature_view, "state", None)
if (
hasattr(feature_view, "state")
Expand Down Expand Up @@ -2593,7 +2615,6 @@ def tqdm_builder(length):
)
except Exception:
fv_success = False
# Roll back state to previous value on failure.
if (
hasattr(feature_view, "state")
and previous_state is not None
Expand Down Expand Up @@ -2752,8 +2773,6 @@ def tqdm_builder(length):
)
else:
for feature_view, fv_start in regular_fvs_with_dates:
# Transition state to MATERIALIZING before starting.
# Only enforce when the state machine is active (not STATE_UNSPECIFIED).
previous_state = getattr(feature_view, "state", None)
if (
hasattr(feature_view, "state")
Expand Down Expand Up @@ -2786,7 +2805,6 @@ def tqdm_builder(length):
)
except Exception:
fv_success = False
# Roll back state to previous value on failure.
if (
hasattr(feature_view, "state")
and previous_state is not None
Expand Down
1 change: 1 addition & 0 deletions sdk/python/feast/feature_view.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Comment thread
aniketpalu marked this conversation as resolved.
Outdated

@jyejare jyejare Jul 29, 2026

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.

[Suggestion] State comparison added to eq affects hash consistency

Adding state to the equality comparison means two otherwise identical FeatureViews will be considered different if they're in different states (e.g., GENERATED vs MATERIALIZING). This could break set operations, dict lookups, and caching logic that expects state changes to not affect object identity.

Suggested:

Suggested change
or self.state != other.state
# Consider if state should be included in equality. If FeatureViews should be
# considered equal regardless of materialization state, remove this line.
# If state is semantically important for equality, ensure __hash__ is also updated
# or make the class unhashable.

@aniketpalu aniketpalu Jul 31, 2026

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.

Agreed

):
return False

Expand Down
4 changes: 4 additions & 0 deletions sdk/python/feast/infra/registry/snowflake.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1147,6 +1147,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,
Expand Down
4 changes: 4 additions & 0 deletions sdk/python/feast/infra/registry/sql.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1228,6 +1228,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"
)
Expand Down