Skip to content

feat: Add remote materialization support via feature server #6590

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

Open
aniketpalu wants to merge 10 commits into
base: master
Choose a base branch
from
Open

feat: Add remote materialization support via feature server #6590

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
44d9a11
feat: Add remote materialization support via feature server
aniketpalu Jul 8, 2026
82ce777
Fixed lint errors & minor formatting
aniketpalu Jul 8, 2026
3bf35a2
feat: Add remote materialization support via feature server
aniketpalu Jul 8, 2026
b6fe54d
- Remove unnecessary async from _authorize_materialize_views
aniketpalu Jul 8, 2026
6f073fb
fix: Set FV state to AVAILABLE_ONLINE in SQL/Snowflake registries
aniketpalu Jul 10, 2026
cf87b03
fix: TLS verification and registry state parity for remote materialize
aniketpalu Jul 10, 2026
ef9ff5e
refactor: Remote materialize via SDK params, not feature_server YAML
aniketpalu Jul 14, 2026
a1a316e
Merge branch 'master' into feat/remote-materialize
aniketpalu Jul 14, 2026
c1f53c9
Merge branch 'master' into feat/remote-materialize
aniketpalu Jul 18, 2026
4b302d5
Merge branch 'master' into feat/remote-materialize
aniketpalu Jul 20, 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
148 changes: 148 additions & 0 deletions sdk/python/feast/feature_server.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand Down Expand Up @@ -708,6 +710,152 @@ 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."""
Comment on lines +713 to +728

@jyejare jyejare Jul 21, 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] 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]

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

from feast.feature_view import FeatureViewState

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

@jyejare jyejare Jul 21, 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] 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


def _update_fv_state(fv_names: List[str], state) -> None:
"""Update FV state in the registry."""
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}")

@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(
Comment on lines +780 to +795

@jyejare jyejare Jul 21, 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] 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)

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,
)
_update_fv_state(fv_names, FeatureViewState.GENERATED)

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

_update_fv_state(fv_names, FeatureViewState.MATERIALIZING)

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,
)
_update_fv_state(fv_names, FeatureViewState.GENERATED)

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
Expand Down
Loading
Loading