diff --git a/sdk/python/feast/feature_server.py b/sdk/python/feast/feature_server.py index bba91130db3..9854b1cb9da 100644 --- a/sdk/python/feast/feature_server.py +++ b/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,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.""" + 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 + + 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( + 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 diff --git a/sdk/python/feast/feature_store.py b/sdk/python/feast/feature_store.py index 563c664dab4..c6d1df0ce7e 100644 --- a/sdk/python/feast/feature_store.py +++ b/sdk/python/feast/feature_store.py @@ -437,8 +437,14 @@ def _rollback_fv_states( self, feature_views: list, previous_states: dict, + _force_local: bool = False, ) -> None: - """Restore feature views to their pre-materialization states.""" + """Restore feature views to their pre-materialization states. + + Skip when _force_local: the async handler owns failure reset (GENERATED). + """ + if _force_local: + return for fv in feature_views: prev = previous_states.get(fv.name) if ( @@ -454,13 +460,18 @@ def _transition_fv_to_materializing( feature_view, already_transitioned: list, previous_states: dict, + _force_local: bool = False, ) -> None: """ Transition a feature view to MATERIALIZING state. Rolls back all already-transitioned FVs if this one can't transition. + 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 if ( hasattr(feature_view, "state") and feature_view.state != FeatureViewState.STATE_UNSPECIFIED @@ -473,7 +484,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, @@ -482,12 +492,14 @@ def _submit_and_process_materialization_jobs( regular_fvs: list, previous_states: dict, date_range: "_MaterializationDateRange", + _force_local: bool = False, ) -> None: """ Submit all tasks to the engine in one call and process the results. For each returned job: record watermark on success, roll back state on error. If the engine itself raises, all states are rolled back. + When _force_local, skip state rollback (async handler owns GENERATED reset). """ from feast.infra.common.materialization_job import ( MaterializationJobStatus, @@ -497,11 +509,15 @@ def _submit_and_process_materialization_jobs( try: jobs = provider.batch_engine.materialize(self.registry, tasks) except Exception: - self._rollback_fv_states(regular_fvs, previous_states) + self._rollback_fv_states( + regular_fvs, previous_states, _force_local=_force_local + ) raise if len(jobs) != len(regular_fvs): - self._rollback_fv_states(regular_fvs, previous_states) + self._rollback_fv_states( + regular_fvs, previous_states, _force_local=_force_local + ) raise RuntimeError( f"Engine returned {len(jobs)} jobs for {len(regular_fvs)} tasks" ) @@ -521,7 +537,9 @@ def _submit_and_process_materialization_jobs( succeeded_fvs.append(fv) if failed_fvs: - self._rollback_fv_states(failed_fvs, previous_states) + self._rollback_fv_states( + failed_fvs, previous_states, _force_local=_force_local + ) # Engines that apply watermarks themselves (e.g. SparkApplication pod) # must not get a second apply_materialization — that duplicates intervals. @@ -552,6 +570,7 @@ def _materialize_fvs_batch( end_date: datetime, tqdm_builder, disable_event_timestamp: bool = False, + _force_local: bool = False, ) -> None: """Batch path: collect all FVs, submit to engine in one call. @@ -566,7 +585,10 @@ def _materialize_fvs_batch( for feature_view, fv_start in fv_with_dates: self._transition_fv_to_materializing( - feature_view, regular_fvs, previous_states + feature_view, + regular_fvs, + previous_states, + _force_local=_force_local, ) regular_fvs.append(feature_view) date_range.fv_start_dates[feature_view.name] = fv_start @@ -588,6 +610,7 @@ def _materialize_fvs_batch( regular_fvs, previous_states, date_range, + _force_local=_force_local, ) @property @@ -2359,12 +2382,265 @@ 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 _fv_state_to_materialization_status( + self, state, *, seen_materializing: Optional[set] = None, fv_name: str = "" + ): + """Map FeatureViewState to FVMaterializationStatus.""" + from feast.materialization_status import FVMaterializationStatus + + if state == FeatureViewState.AVAILABLE_ONLINE: + return FVMaterializationStatus.SUCCEEDED + if state == FeatureViewState.MATERIALIZING: + if seen_materializing is not None: + seen_materializing.add(fv_name) + return FVMaterializationStatus.RUNNING + if state == FeatureViewState.GENERATED: + if seen_materializing is not None and fv_name in seen_materializing: + return FVMaterializationStatus.FAILED + return FVMaterializationStatus.PENDING + return FVMaterializationStatus.PENDING + + def poll_materialization( + self, + feature_views: Optional[List[str]] = None, + ): + """One-shot check of materialization status for feature views. + + Reads current FV state from the registry (no waiting/looping). + Call repeatedly yourself if you want to track progress over time. + + Args: + feature_views: FV names to check. If None, checks all materializable FVs. + + Returns: + List of FVResult with name and current status (pending/running/succeeded). + """ + from feast.materialization_status import FVResult + + fv_names = feature_views or [ + fv.name + for fv in self._get_feature_views_to_materialize(None) + if not isinstance(fv, OnDemandFeatureView) + ] + + results = [] + for name in fv_names: + fv = self.registry.get_feature_view(name, self.project, allow_cache=False) + status = self._fv_state_to_materialization_status( + getattr(fv, "state", None) + ) + results.append(FVResult(name=name, status=status)) + return results + + def _remote_materialize_common( + self, + endpoint: str, + payload: dict, + fv_names: List[str], + timeout: float = 3600.0, + poll_interval: float = 5.0, + wait: bool = True, + ) -> None: + """Shared logic for remote materialization: trigger server, optionally wait. + + Args: + endpoint: Server endpoint path (e.g., "/materialize-async"). + payload: JSON payload for the POST request. + fv_names: Feature view names to track. + timeout: Max seconds to wait for materialization to complete. + poll_interval: Seconds between registry polls when wait=True. + wait: If True, poll until complete. If False, return after 202 Accepted. + """ + from feast.materialization_status import ( + FVMaterializationStatus, + FVResult, + poll_materialization_status, + ) + + server_url = self._get_remote_materialize_url() + + _logger.info( + f"Remote materialize: triggering {len(fv_names)} feature views " + f"via {endpoint} on {server_url}" + ) + + session = self._get_remote_http_session() + try: + 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: + session.close() + + if not wait: + _logger.info( + f"Remote materialize accepted for {fv_names} " + f"(wait=False; use poll_materialization() to check status)" + ) + return + + 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 + ) + return self._fv_state_to_materialization_status( + getattr(fv, "state", None), + seen_materializing=seen_materializing, + fv_name=fv_name, + ) + + 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)" + ) + + 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, + ) + + if hasattr(self.registry, "refresh"): + self.registry.refresh(project=self.project) + + 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, + 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 [ + 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, + timeout=timeout, + poll_interval=poll_interval, + wait=wait, + ) + + def _remote_materialize_incremental( + self, + end_date: datetime, + feature_views: Optional[List[str]] = None, + full_feature_names: bool = False, + version: Optional[str] = None, + timeout: float = 3600.0, + poll_interval: float = 5.0, + wait: bool = True, + ) -> 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, + timeout=timeout, + poll_interval=poll_interval, + wait=wait, + ) + def materialize_incremental( self, end_date: datetime, feature_views: Optional[List[str]] = None, full_feature_names: bool = False, version: Optional[str] = None, + remote: bool = False, + timeout: float = 3600.0, + poll_interval: float = 5.0, + wait: bool = True, + _force_local: bool = False, ) -> None: """ Materialize incremental new data from the offline store into the online store. @@ -2383,6 +2659,13 @@ def materialize_incremental( feature view name. version (str): Optional version to materialize (e.g., 'v2'). Requires feature_views with exactly one entry and enable_online_feature_view_versioning to be enabled. + remote (bool): If True, delegate materialization to the feature server instead of + running the batch engine locally. The server URL is taken from online_store.path. + timeout (float): Max seconds to wait when remote=True and wait=True. + poll_interval (float): Seconds between registry polls when remote=True and wait=True. + wait (bool): If True (default), block until remote materialization finishes. + If False, return after the server accepts the job; use poll_materialization() + to check status later. Raises: Exception: A feature view being materialized does not have a TTL set. @@ -2398,6 +2681,17 @@ def materialize_incremental( ... """ + if remote and not _force_local: + return self._remote_materialize_incremental( + end_date=end_date, + feature_views=feature_views, + full_feature_names=full_feature_names, + version=version, + timeout=timeout, + poll_interval=poll_interval, + wait=wait, + ) + parsed_version = self._validate_materialize_version(version, feature_views) feature_views_to_materialize = self._get_feature_views_to_materialize( feature_views, version=parsed_version @@ -2489,14 +2783,16 @@ def tqdm_builder(length): regular_fvs_with_dates, end_date_tz, tqdm_builder, + _force_local=_force_local, ) 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). + # Skip when _force_local: the async handler already set MATERIALIZING. previous_state = getattr(feature_view, "state", None) if ( - hasattr(feature_view, "state") + not _force_local + and hasattr(feature_view, "state") and feature_view.state != FeatureViewState.STATE_UNSPECIFIED ): if not feature_view.state.can_transition_to( @@ -2525,9 +2821,11 @@ def tqdm_builder(length): ) except Exception: fv_success = False - # Roll back state to previous value on failure. + # Roll back state on failure (only when not _force_local, + # since the async handler manages state via _update_fv_state). if ( - hasattr(feature_view, "state") + not _force_local + and hasattr(feature_view, "state") and previous_state is not None and previous_state != FeatureViewState.STATE_UNSPECIFIED ): @@ -2587,6 +2885,11 @@ def materialize( disable_event_timestamp: bool = False, full_feature_names: bool = False, version: Optional[str] = None, + remote: bool = False, + timeout: float = 3600.0, + poll_interval: float = 5.0, + wait: bool = True, + _force_local: bool = False, ) -> None: """ Materialize data from the offline store into the online store. @@ -2605,6 +2908,13 @@ def materialize( feature view name. version (str): Optional version to materialize (e.g., 'v2'). Requires feature_views with exactly one entry and enable_online_feature_view_versioning to be enabled. + remote (bool): If True, delegate materialization to the feature server instead of + running the batch engine locally. The server URL is taken from online_store.path. + timeout (float): Max seconds to wait when remote=True and wait=True. + poll_interval (float): Seconds between registry polls when remote=True and wait=True. + wait (bool): If True (default), block until remote materialization finishes. + If False, return after the server accepts the job; use poll_materialization() + to check status later. Examples: Materialize all features into the online store over the interval @@ -2624,6 +2934,19 @@ def materialize( f"The given start_date {start_date} is greater than the given end_date {end_date}." ) + if remote and not _force_local: + 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, + timeout=timeout, + poll_interval=poll_interval, + wait=wait, + ) + parsed_version = self._validate_materialize_version(version, feature_views) feature_views_to_materialize = self._get_feature_views_to_materialize( feature_views, version=parsed_version @@ -2681,14 +3004,16 @@ def tqdm_builder(length): end_date, tqdm_builder, disable_event_timestamp=disable_event_timestamp, + _force_local=_force_local, ) 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). + # Skip when _force_local: the async handler already set MATERIALIZING. previous_state = getattr(feature_view, "state", None) if ( - hasattr(feature_view, "state") + not _force_local + and hasattr(feature_view, "state") and feature_view.state != FeatureViewState.STATE_UNSPECIFIED ): if not feature_view.state.can_transition_to( @@ -2718,9 +3043,11 @@ def tqdm_builder(length): ) except Exception: fv_success = False - # Roll back state to previous value on failure. + # Roll back state on failure (only when not _force_local, + # since the async handler manages state via _update_fv_state). if ( - hasattr(feature_view, "state") + not _force_local + and hasattr(feature_view, "state") and previous_state is not None and previous_state != FeatureViewState.STATE_UNSPECIFIED ): diff --git a/sdk/python/feast/feature_view.py b/sdk/python/feast/feature_view.py index a5d3c8d9537..0b7e28f555b 100644 --- a/sdk/python/feast/feature_view.py +++ b/sdk/python/feast/feature_view.py @@ -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 ): return False diff --git a/sdk/python/feast/infra/registry/snowflake.py b/sdk/python/feast/infra/registry/snowflake.py index 5590e1b7574..b806c559199 100644 --- a/sdk/python/feast/infra/registry/snowflake.py +++ b/sdk/python/feast/infra/registry/snowflake.py @@ -1135,6 +1135,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, diff --git a/sdk/python/feast/infra/registry/sql.py b/sdk/python/feast/infra/registry/sql.py index edd89347be2..efaaefdac0d 100644 --- a/sdk/python/feast/infra/registry/sql.py +++ b/sdk/python/feast/infra/registry/sql.py @@ -1217,6 +1217,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" ) diff --git a/sdk/python/feast/materialization_status.py b/sdk/python/feast/materialization_status.py new file mode 100644 index 00000000000..4d3d419e052 --- /dev/null +++ b/sdk/python/feast/materialization_status.py @@ -0,0 +1,93 @@ +import logging +import time +from dataclasses import dataclass +from enum import Enum +from typing import Callable, Dict, List, Optional, Set + +logger = logging.getLogger(__name__) + + +class FVMaterializationStatus(Enum): + PENDING = "pending" + RUNNING = "running" + SUCCEEDED = "succeeded" + FAILED = "failed" + + +@dataclass +class FVResult: + name: str + status: FVMaterializationStatus + elapsed_seconds: Optional[float] = None + error: Optional[str] = None + + +def poll_materialization_status( + feature_view_names: List[str], + status_fn: Callable[[str], FVMaterializationStatus], + poll_interval: float = 5.0, + timeout: float = 3600.0, + on_status_change: Optional[Callable[[FVResult], None]] = None, +) -> List[FVResult]: + """Poll materialization status for a list of feature views. + + This function is engine-agnostic. The caller provides a status_fn + that maps a feature view name to its current materialization status. + + Used by: + - Client-side remote materialize: status_fn reads FV state from remote registry + - Server-side batch engines: status_fn reads from engine job objects + + Args: + feature_view_names: FVs to track. + status_fn: Returns current status for a given FV name. + poll_interval: Seconds between polls. + timeout: Max seconds to wait before declaring timeout. + on_status_change: Callback invoked when a FV's status changes. + + Returns: + List of FVResult with final status for each FV. + """ + start = time.monotonic() + pending: Set[str] = set(feature_view_names) + results: Dict[str, FVResult] = { + name: FVResult(name=name, status=FVMaterializationStatus.PENDING) + for name in feature_view_names + } + previous_statuses: Dict[str, FVMaterializationStatus] = {} + + while pending and (time.monotonic() - start) < timeout: + for name in list(pending): + try: + current = status_fn(name) + except Exception as e: + logger.warning(f"Error polling status for {name}: {e}") + continue + + elapsed = time.monotonic() - start + + if current != previous_statuses.get(name): + result = FVResult(name=name, status=current, elapsed_seconds=elapsed) + results[name] = result + previous_statuses[name] = current + if on_status_change: + on_status_change(result) + + if current in ( + FVMaterializationStatus.SUCCEEDED, + FVMaterializationStatus.FAILED, + ): + pending.discard(name) + + if pending: + time.sleep(poll_interval) + + for name in pending: + results[name] = FVResult( + name=name, + status=FVMaterializationStatus.FAILED, + elapsed_seconds=time.monotonic() - start, + error=f"Materialization timed out after {timeout}s. Check server logs for details.", + ) + + return list(results.values()) diff --git a/sdk/python/tests/integration/materialization/test_remote_materialize.py b/sdk/python/tests/integration/materialization/test_remote_materialize.py new file mode 100644 index 00000000000..3976a316012 --- /dev/null +++ b/sdk/python/tests/integration/materialization/test_remote_materialize.py @@ -0,0 +1,336 @@ +"""Integration tests for remote materialization. + +Runs a real feature server in a background thread, configures a client +FeatureStore with online_store pointing at the server, and verifies the +full flow using remote=True on the materialize() SDK call: + client.materialize(remote=True) → POST /materialize-async → server runs locally → + FV state transitions → client polls registry → returns success + +Uses SQL registry (SQLite) to match the production path and exercise +the state transitions that the file-based registry handles differently. +""" + +import socket +import threading +import time +from datetime import datetime, timedelta, timezone + +import pandas as pd +import pytest +import uvicorn + +from feast import Entity, FeatureStore, FeatureView, Field, RepoConfig +from feast.feature_server import get_app +from feast.feature_view import FeatureViewState +from feast.infra.offline_stores.file_source import FileSource +from feast.infra.online_stores.sqlite import SqliteOnlineStoreConfig +from feast.infra.registry.sql import SqlRegistryConfig +from feast.types import Float32, Int64 + + +def _find_free_port() -> int: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + s.bind(("127.0.0.1", 0)) + return s.getsockname()[1] + + +@pytest.fixture +def e2e_repo(tmp_path): + """Set up a minimal feature repo with SQL registry and SQLite online store.""" + registry_db = str(tmp_path / "registry.db") + online_store_path = str(tmp_path / "online_store.db") + data_path = str(tmp_path / "data.parquet") + + now = datetime.now(tz=timezone.utc) + df = pd.DataFrame( + { + "entity_id": [1, 2, 3, 4, 5], + "feature_a": [10.0, 20.0, 30.0, 40.0, 50.0], + "feature_b": [100, 200, 300, 400, 500], + "event_timestamp": [now - timedelta(hours=i) for i in range(5)], + "created": [now] * 5, + } + ) + df.to_parquet(data_path) + + config = RepoConfig( + project="test_remote_mat", + provider="local", + registry=SqlRegistryConfig( + registry_type="sql", + path=f"sqlite:///{registry_db}", + cache_ttl_seconds=1, + ), + online_store=SqliteOnlineStoreConfig(path=online_store_path), + entity_key_serialization_version=3, + ) + + store = FeatureStore(config=config) + + entity = Entity(name="entity_id", join_keys=["entity_id"]) + + source = FileSource( + path=data_path, + timestamp_field="event_timestamp", + created_timestamp_column="created", + ) + + fv = FeatureView( + name="test_feature_view", + entities=[entity], + schema=[ + Field(name="feature_a", dtype=Float32), + Field(name="feature_b", dtype=Int64), + ], + source=source, + ttl=timedelta(days=7), + ) + + store.apply([entity, fv]) + + return store, config, registry_db, online_store_path + + +@pytest.fixture +def feature_server(e2e_repo): + """Start a real feature server in a background thread on a free port.""" + store, config, registry_db, online_store_path = e2e_repo + + app = get_app(store) + server_port = _find_free_port() + + server_config = uvicorn.Config( + app, host="127.0.0.1", port=server_port, log_level="warning" + ) + server = uvicorn.Server(server_config) + + thread = threading.Thread(target=server.run, daemon=True) + thread.start() + + import httpx + + for _ in range(50): + try: + resp = httpx.get(f"http://127.0.0.1:{server_port}/health") + if resp.status_code == 200: + break + except Exception: + pass + time.sleep(0.1) + else: + pytest.fail("Feature server did not start in time") + + yield store, config, server_port + + server.should_exit = True + thread.join(timeout=5) + + +def _make_client_store(server_config, server_port): + """Create a client FeatureStore with online_store.path pointing at the server.""" + from feast.infra.online_stores.remote import RemoteOnlineStoreConfig + + client_config = RepoConfig( + project="test_remote_mat", + provider="local", + registry=server_config.registry, + online_store=RemoteOnlineStoreConfig( + type="remote", + path=f"http://127.0.0.1:{server_port}", + ), + entity_key_serialization_version=3, + ) + return FeatureStore(config=client_config) + + +def test_remote_materialize_e2e(feature_server): + """Full E2E: client with remote=True materializes through server.""" + server_store, server_config, server_port = feature_server + client_store = _make_client_store(server_config, server_port) + + now = datetime.now(tz=timezone.utc) + + client_store.materialize( + start_date=now - timedelta(days=7), + end_date=now, + feature_views=["test_feature_view"], + remote=True, + timeout=30.0, + poll_interval=0.5, + ) + + fv = client_store.registry.get_feature_view( + "test_feature_view", "test_remote_mat", allow_cache=False + ) + assert fv.state == FeatureViewState.AVAILABLE_ONLINE + + online_features = client_store.get_online_features( + features=["test_feature_view:feature_a", "test_feature_view:feature_b"], + entity_rows=[{"entity_id": 1}], + ).to_dict() + + assert online_features["feature_a"][0] == pytest.approx(10.0) + assert online_features["feature_b"][0] == 100 + + +def test_remote_materialize_incremental_e2e(feature_server): + """E2E for materialize_incremental through remote path.""" + server_store, server_config, server_port = feature_server + client_store = _make_client_store(server_config, server_port) + + now = datetime.now(tz=timezone.utc) + + client_store.materialize_incremental( + end_date=now, + feature_views=["test_feature_view"], + remote=True, + timeout=30.0, + poll_interval=0.5, + ) + + fv = client_store.registry.get_feature_view( + "test_feature_view", "test_remote_mat", allow_cache=False + ) + assert fv.state == FeatureViewState.AVAILABLE_ONLINE + + online_features = client_store.get_online_features( + features=["test_feature_view:feature_a"], + entity_rows=[{"entity_id": 2}], + ).to_dict() + + assert online_features["feature_a"][0] == pytest.approx(20.0) + + +def test_remote_materialize_re_materialize(feature_server): + """Re-materialization on already AVAILABLE_ONLINE FVs waits for completion. + + Validates the polling race fix: server sets MATERIALIZING before 202, + so the client doesn't short-circuit on stale AVAILABLE_ONLINE. + """ + server_store, server_config, server_port = feature_server + client_store = _make_client_store(server_config, server_port) + + now = datetime.now(tz=timezone.utc) + + # First materialization + client_store.materialize( + start_date=now - timedelta(days=7), + end_date=now, + feature_views=["test_feature_view"], + remote=True, + timeout=30.0, + poll_interval=0.5, + ) + + fv = client_store.registry.get_feature_view( + "test_feature_view", "test_remote_mat", allow_cache=False + ) + assert fv.state == FeatureViewState.AVAILABLE_ONLINE + + # Second materialization — must NOT return instantly + start_time = time.monotonic() + client_store.materialize( + start_date=now - timedelta(days=7), + end_date=now, + feature_views=["test_feature_view"], + remote=True, + timeout=30.0, + poll_interval=0.5, + ) + elapsed = time.monotonic() - start_time + + # Should have waited at least one poll cycle (0.5s), not returned in <0.1s + assert elapsed >= 0.3, ( + f"Re-materialization returned in {elapsed:.2f}s — polling race not fixed" + ) + + fv = client_store.registry.get_feature_view( + "test_feature_view", "test_remote_mat", allow_cache=False + ) + assert fv.state == FeatureViewState.AVAILABLE_ONLINE + + +def test_remote_materialize_force_local_bypasses_remote(feature_server): + """_force_local=True runs locally even when remote=True. + + Uses the server_store (local online store) since _force_local skips + the remote HTTP call entirely and writes to the local online store. + """ + server_store, server_config, server_port = feature_server + + now = datetime.now(tz=timezone.utc) + + server_store.materialize( + start_date=now - timedelta(days=7), + end_date=now, + feature_views=["test_feature_view"], + remote=True, + _force_local=True, + ) + + fv = server_store.registry.get_feature_view( + "test_feature_view", "test_remote_mat", allow_cache=False + ) + assert fv.state == FeatureViewState.AVAILABLE_ONLINE + + +def test_remote_materialize_server_error_propagates(feature_server): + """Client gets an error when the server rejects the request. + + A non-existent feature view causes a server-side error during authz + validation, which returns HTTP 500. The client wraps this as an + Exception with a descriptive message. + """ + server_store, server_config, server_port = feature_server + client_store = _make_client_store(server_config, server_port) + + now = datetime.now(tz=timezone.utc) + + with pytest.raises(Exception, match="Failed to trigger remote materialization"): + client_store.materialize( + start_date=now - timedelta(days=1), + end_date=now, + feature_views=["nonexistent_fv"], + remote=True, + timeout=10.0, + poll_interval=0.5, + ) + + +def test_remote_materialize_failure_resets_state(feature_server): + """When server-side materialization fails, FV state resets to GENERATED. + + The client's seen_materializing logic detects MATERIALIZING → GENERATED + as a failure and reports it promptly (within one poll cycle). + """ + server_store, server_config, server_port = feature_server + client_store = _make_client_store(server_config, server_port) + + now = datetime.now(tz=timezone.utc) + + # Break the data source so materialization fails in the background thread + fv = server_store.registry.get_feature_view( + "test_feature_view", "test_remote_mat", allow_cache=False + ) + fv.batch_source = FileSource( + path="/nonexistent/path/that/will/fail.parquet", + timestamp_field="event_timestamp", + ) + server_store.apply([fv]) + + # Client triggers remote materialization — server will fail in background + with pytest.raises(Exception, match="Remote materialization failed"): + client_store.materialize( + start_date=now - timedelta(days=7), + end_date=now, + feature_views=["test_feature_view"], + remote=True, + timeout=30.0, + poll_interval=0.5, + ) + + # FV state should be back to GENERATED (not stuck at MATERIALIZING) + fv = client_store.registry.get_feature_view( + "test_feature_view", "test_remote_mat", allow_cache=False + ) + assert fv.state == FeatureViewState.GENERATED diff --git a/sdk/python/tests/unit/test_remote_materialize.py b/sdk/python/tests/unit/test_remote_materialize.py new file mode 100644 index 00000000000..abba6feebc0 --- /dev/null +++ b/sdk/python/tests/unit/test_remote_materialize.py @@ -0,0 +1,545 @@ +"""Tests for remote materialization: shared poller, client-side delegation, and async endpoints.""" + +import time +from datetime import datetime, timezone +from unittest.mock import MagicMock + +import pytest +from fastapi.testclient import TestClient + +from feast.feature_server import get_app +from feast.feature_view import FeatureViewState +from feast.materialization_status import ( + FVMaterializationStatus, + poll_materialization_status, +) + +# --------------------------------------------------------------------------- +# Tests for poll_materialization_status +# --------------------------------------------------------------------------- + + +class TestPollMaterializationStatus: + def test_all_succeed_immediately(self): + def status_fn(name): + return FVMaterializationStatus.SUCCEEDED + + results = poll_materialization_status( + feature_view_names=["fv_a", "fv_b"], + status_fn=status_fn, + poll_interval=0.01, + ) + + assert len(results) == 2 + assert all(r.status == FVMaterializationStatus.SUCCEEDED for r in results) + + def test_mixed_results(self): + statuses = { + "fv_a": FVMaterializationStatus.SUCCEEDED, + "fv_b": FVMaterializationStatus.FAILED, + } + + def status_fn(name): + return statuses[name] + + results = poll_materialization_status( + feature_view_names=["fv_a", "fv_b"], + status_fn=status_fn, + poll_interval=0.01, + ) + + result_map = {r.name: r for r in results} + assert result_map["fv_a"].status == FVMaterializationStatus.SUCCEEDED + assert result_map["fv_b"].status == FVMaterializationStatus.FAILED + + def test_timeout(self): + def status_fn(name): + return FVMaterializationStatus.RUNNING + + results = poll_materialization_status( + feature_view_names=["fv_stuck"], + status_fn=status_fn, + poll_interval=0.01, + timeout=0.05, + ) + + assert len(results) == 1 + assert results[0].status == FVMaterializationStatus.FAILED + assert "timed out" in results[0].error.lower() + + def test_transition_from_pending_to_running_to_succeeded(self): + call_count = {"n": 0} + + def status_fn(name): + call_count["n"] += 1 + if call_count["n"] <= 1: + return FVMaterializationStatus.PENDING + elif call_count["n"] <= 2: + return FVMaterializationStatus.RUNNING + return FVMaterializationStatus.SUCCEEDED + + results = poll_materialization_status( + feature_view_names=["fv_transitioning"], + status_fn=status_fn, + poll_interval=0.01, + ) + + assert results[0].status == FVMaterializationStatus.SUCCEEDED + + def test_on_status_change_callback(self): + changes = [] + call_count = {"n": 0} + + def status_fn(name): + call_count["n"] += 1 + if call_count["n"] <= 1: + return FVMaterializationStatus.RUNNING + return FVMaterializationStatus.SUCCEEDED + + poll_materialization_status( + feature_view_names=["fv_x"], + status_fn=status_fn, + poll_interval=0.01, + on_status_change=lambda r: changes.append(r), + ) + + assert len(changes) == 2 + assert changes[0].status == FVMaterializationStatus.RUNNING + assert changes[1].status == FVMaterializationStatus.SUCCEEDED + + def test_transient_exception_is_handled(self): + """Transient errors from status_fn don't crash the poller.""" + call_count = {"n": 0} + + def status_fn(name): + call_count["n"] += 1 + if call_count["n"] == 1: + raise ConnectionError("transient network error") + return FVMaterializationStatus.SUCCEEDED + + results = poll_materialization_status( + feature_view_names=["fv_err"], + status_fn=status_fn, + poll_interval=0.01, + ) + + assert results[0].status == FVMaterializationStatus.SUCCEEDED + + def test_elapsed_seconds_is_populated(self): + def status_fn(name): + return FVMaterializationStatus.SUCCEEDED + + results = poll_materialization_status( + feature_view_names=["fv_t"], + status_fn=status_fn, + poll_interval=0.01, + ) + + assert results[0].elapsed_seconds is not None + assert results[0].elapsed_seconds >= 0 + + +# --------------------------------------------------------------------------- +# Tests for feature_server async endpoints +# --------------------------------------------------------------------------- + + +class TestFeatureServerAsyncEndpoints: + def _make_client(self): + fs = MagicMock() + fs._get_provider.return_value.async_supported.online.read = False + fs.initialize = MagicMock() + fs.close = MagicMock() + fs.project = "test_project" + + mock_fv = MagicMock() + mock_fv.name = "test_fv" + mock_fv.state = FeatureViewState.GENERATED + fs._get_feature_views_to_materialize.return_value = [mock_fv] + fs.registry.get_feature_view.return_value = mock_fv + + fs.materialize = MagicMock() + fs.materialize_incremental = MagicMock() + + client = TestClient(get_app(fs)) + return client, fs + + def test_materialize_async_returns_202(self): + client, fs = self._make_client() + response = client.post( + "/materialize-async", + json={ + "start_ts": "2024-01-01T00:00:00Z", + "end_ts": "2024-01-02T00:00:00Z", + "feature_views": ["test_fv"], + }, + ) + assert response.status_code == 202 + body = response.json() + assert body["status"] == "accepted" + assert "test_fv" in body["feature_views"] + + def test_materialize_incremental_async_returns_202(self): + client, fs = self._make_client() + response = client.post( + "/materialize-incremental-async", + json={ + "end_ts": "2024-01-02T00:00:00Z", + "feature_views": ["test_fv"], + }, + ) + assert response.status_code == 202 + body = response.json() + assert body["status"] == "accepted" + assert "test_fv" in body["feature_views"] + + def test_materialize_async_validates_timestamps(self): + client, fs = self._make_client() + with pytest.raises(ValueError, match="start_ts and end_ts are required"): + client.post( + "/materialize-async", + json={ + "feature_views": ["test_fv"], + "disable_event_timestamp": False, + }, + ) + + def test_materialize_async_disable_event_timestamp(self): + client, fs = self._make_client() + response = client.post( + "/materialize-async", + json={ + "feature_views": ["test_fv"], + "disable_event_timestamp": True, + }, + ) + assert response.status_code == 202 + + def test_materialize_async_uses_force_local(self): + """Verify the server passes _force_local=True to prevent recursion.""" + client, fs = self._make_client() + response = client.post( + "/materialize-async", + json={ + "start_ts": "2024-01-01T00:00:00Z", + "end_ts": "2024-01-02T00:00:00Z", + "feature_views": ["test_fv"], + }, + ) + assert response.status_code == 202 + # Give the background task time to execute + time.sleep(0.1) + fs.materialize.assert_called_once() + call_kwargs = fs.materialize.call_args[1] + assert call_kwargs["_force_local"] is True + + def test_materialize_async_409_if_already_materializing(self): + """Return 409 if FV is already in MATERIALIZING state.""" + client, fs = self._make_client() + mock_fv = MagicMock() + mock_fv.state = FeatureViewState.MATERIALIZING + fs.registry.get_feature_view.return_value = mock_fv + + response = client.post( + "/materialize-async", + json={ + "start_ts": "2024-01-01T00:00:00Z", + "end_ts": "2024-01-02T00:00:00Z", + "feature_views": ["test_fv"], + }, + ) + assert response.status_code == 409 + + def test_materialize_async_forwards_version(self): + """Version parameter is forwarded to store.materialize.""" + client, fs = self._make_client() + response = client.post( + "/materialize-async", + json={ + "start_ts": "2024-01-01T00:00:00Z", + "end_ts": "2024-01-02T00:00:00Z", + "feature_views": ["test_fv"], + "version": "v2", + }, + ) + assert response.status_code == 202 + time.sleep(0.1) + call_kwargs = fs.materialize.call_args[1] + assert call_kwargs["version"] == "v2" + + def test_materialize_async_validates_timestamp_order(self): + """start_ts must be before end_ts.""" + client, fs = self._make_client() + with pytest.raises(ValueError, match="must be before"): + client.post( + "/materialize-async", + json={ + "start_ts": "2024-01-02T00:00:00Z", + "end_ts": "2024-01-01T00:00:00Z", + "feature_views": ["test_fv"], + }, + ) + + +# --------------------------------------------------------------------------- +# Tests for FeatureStore remote materialize gate +# --------------------------------------------------------------------------- + + +class TestFeatureStoreRemoteGate: + def test_materialize_calls_remote_when_remote_true(self): + """materialize(remote=True) delegates to _remote_materialize.""" + from feast.feature_store import FeatureStore + + fs = MagicMock() + fs._remote_materialize = MagicMock() + + FeatureStore.materialize( + fs, + start_date=datetime(2024, 1, 1, tzinfo=timezone.utc), + end_date=datetime(2024, 1, 2, tzinfo=timezone.utc), + feature_views=["fv1"], + remote=True, + ) + + fs._remote_materialize.assert_called_once() + + def test_materialize_skips_remote_when_force_local(self): + """materialize() with _force_local=True skips remote even if remote=True.""" + from feast.feature_store import FeatureStore + + fs = MagicMock() + fs._remote_materialize = MagicMock() + + try: + FeatureStore.materialize( + fs, + start_date=datetime(2024, 1, 1, tzinfo=timezone.utc), + end_date=datetime(2024, 1, 2, tzinfo=timezone.utc), + feature_views=["fv1"], + remote=True, + _force_local=True, + ) + except Exception: + pass + + fs._remote_materialize.assert_not_called() + + def test_materialize_local_by_default(self): + """materialize() without remote=True does NOT delegate remotely.""" + from feast.feature_store import FeatureStore + + fs = MagicMock() + fs._remote_materialize = MagicMock() + + try: + FeatureStore.materialize( + fs, + start_date=datetime(2024, 1, 1, tzinfo=timezone.utc), + end_date=datetime(2024, 1, 2, tzinfo=timezone.utc), + feature_views=["fv1"], + ) + except Exception: + pass + + fs._remote_materialize.assert_not_called() + + def test_get_remote_materialize_url_uses_online_store_path(self): + from feast.feature_store import FeatureStore + + fs = MagicMock() + fs.config.online_store.path = "http://feast-server:80/" + result = FeatureStore._get_remote_materialize_url(fs) + assert result == "http://feast-server:80" + + def test_get_remote_materialize_url_raises_without_path(self): + from feast.feature_store import FeatureStore + + fs = MagicMock() + fs.config.online_store.path = None + with pytest.raises(ValueError, match="online_store.path must be set"): + FeatureStore._get_remote_materialize_url(fs) + + def test_materialize_passes_wait_false(self): + """materialize(remote=True, wait=False) forwards wait to remote helper.""" + from feast.feature_store import FeatureStore + + fs = MagicMock() + fs._remote_materialize = MagicMock() + + FeatureStore.materialize( + fs, + start_date=datetime(2024, 1, 1, tzinfo=timezone.utc), + end_date=datetime(2024, 1, 2, tzinfo=timezone.utc), + feature_views=["fv1"], + remote=True, + wait=False, + ) + + assert fs._remote_materialize.call_args.kwargs["wait"] is False + + +class TestPollMaterializationOneShot: + def test_poll_materialization_reports_current_states(self): + """One-shot poll returns status for each FV without waiting.""" + from feast.feature_store import FeatureStore + from feast.materialization_status import FVMaterializationStatus + + fs = MagicMock() + fs.project = "test" + + def mock_get_fv(name, project, allow_cache=False): + fv = MagicMock() + if name == "fv_done": + fv.state = FeatureViewState.AVAILABLE_ONLINE + elif name == "fv_running": + fv.state = FeatureViewState.MATERIALIZING + else: + fv.state = FeatureViewState.GENERATED + return fv + + fs.registry.get_feature_view = mock_get_fv + # Bind real helper methods used by poll_materialization + fs._fv_state_to_materialization_status = ( + FeatureStore._fv_state_to_materialization_status.__get__(fs, FeatureStore) + ) + + results = FeatureStore.poll_materialization( + fs, feature_views=["fv_done", "fv_running", "fv_pending"] + ) + + by_name = {r.name: r.status for r in results} + assert by_name["fv_done"] == FVMaterializationStatus.SUCCEEDED + assert by_name["fv_running"] == FVMaterializationStatus.RUNNING + assert by_name["fv_pending"] == FVMaterializationStatus.PENDING + + def test_remote_wait_false_skips_polling(self): + """wait=False returns after POST without calling the poller.""" + from feast.feature_store import FeatureStore + + fs = MagicMock() + fs.config.online_store.path = "http://server:80" + fs.config.online_store.cert = None + fs.config.auth_config = None + fs.project = "test" + fs._get_remote_materialize_url = MagicMock(return_value="http://server:80") + + mock_session = MagicMock() + mock_resp = MagicMock() + mock_resp.raise_for_status = MagicMock() + mock_session.post.return_value = mock_resp + fs._get_remote_http_session = MagicMock(return_value=mock_session) + + FeatureStore._remote_materialize_common( + fs, + "/materialize-async", + {"feature_views": ["fv1"]}, + ["fv1"], + wait=False, + ) + + mock_session.post.assert_called_once() + fs.registry.get_feature_view.assert_not_called() + + +# --------------------------------------------------------------------------- +# Tests for state mapping with seen_materializing tracking +# --------------------------------------------------------------------------- + + +class TestRegistryStatusMapping: + def _setup_fs_mock(self): + from feast.feature_store import FeatureStore + + fs = MagicMock() + fs.config.online_store.path = "http://server:80" + fs.config.online_store.cert = None + fs.config.auth_config = None + fs.project = "test" + + mock_session = MagicMock() + mock_resp = MagicMock() + mock_resp.raise_for_status = MagicMock() + mock_session.post.return_value = mock_resp + fs._get_remote_http_session = MagicMock(return_value=mock_session) + fs._get_remote_materialize_url = MagicMock(return_value="http://server:80") + fs._fv_state_to_materialization_status = ( + FeatureStore._fv_state_to_materialization_status.__get__(fs, FeatureStore) + ) + return fs + + def test_generated_before_materializing_is_pending(self): + """GENERATED state before we've ever seen MATERIALIZING = PENDING.""" + from feast.feature_store import FeatureStore + + fs = self._setup_fs_mock() + + call_count = {"n": 0} + + def mock_get_fv(name, project, allow_cache=False): + call_count["n"] += 1 + fv = MagicMock() + if call_count["n"] <= 2: + fv.state = FeatureViewState.GENERATED + elif call_count["n"] <= 4: + fv.state = FeatureViewState.MATERIALIZING + else: + fv.state = FeatureViewState.AVAILABLE_ONLINE + return fv + + fs.registry.get_feature_view = mock_get_fv + + FeatureStore._remote_materialize_common( + fs, + "/materialize-async", + {"feature_views": ["fv1"]}, + ["fv1"], + timeout=1.0, + poll_interval=0.01, + ) + + def test_generated_after_materializing_is_failed(self): + """GENERATED state after we've seen MATERIALIZING = FAILED (rollback).""" + from feast.feature_store import FeatureStore + + fs = self._setup_fs_mock() + + call_count = {"n": 0} + + def mock_get_fv(name, project, allow_cache=False): + call_count["n"] += 1 + fv = MagicMock() + if call_count["n"] <= 2: + fv.state = FeatureViewState.MATERIALIZING + else: + fv.state = FeatureViewState.GENERATED + return fv + + fs.registry.get_feature_view = mock_get_fv + + with pytest.raises(Exception, match="Remote materialization failed"): + FeatureStore._remote_materialize_common( + fs, + "/materialize-async", + {"feature_views": ["fv1"]}, + ["fv1"], + timeout=1.0, + poll_interval=0.01, + ) + + +# --------------------------------------------------------------------------- +# Tests for LocalFeatureServerConfig (no remote-specific fields) +# --------------------------------------------------------------------------- + + +class TestLocalFeatureServerConfig: + def test_defaults(self): + from feast.infra.feature_servers.local_process.config import ( + LocalFeatureServerConfig, + ) + + cfg = LocalFeatureServerConfig() + assert cfg.type == "local" + assert not hasattr(cfg, "materialize_mode") + assert not hasattr(cfg, "url")