min_event_timestamp: Optional[datetime]

max_event_timestamp: Optional[datetime]

# List of feature references

features: List[str]

# List of entity keys + ODFV inputs

keys: List[str]

def __init__(

self,

features: List[str],

keys: List[str],

min_event_timestamp: Optional[datetime] = None,

max_event_timestamp: Optional[datetime] = None,

):

self.features = features

self.keys = keys

self.min_event_timestamp = min_event_timestamp

"""Return ``(feature_view_names, feature_count)`` from a RetrievalJob's metadata."""

from feast.utils import _parse_feature_ref

try:

meta = job.metadata

if meta:

feature_count = len(meta.features)

feature_views = list(

{_parse_feature_ref(ref)[0] for ref in meta.features if ":" in ref}

)

return feature_views, feature_count

except (NotImplementedError, AttributeError):

pass

job: "RetrievalJob",

method: str,

status_label: str,

row_count: int,

elapsed: float,

) -> None:

"""Record offline-store request metrics and the audit log entry. Never raises."""

try:

from feast import metrics as feast_metrics

if feast_metrics._config.offline_features:

feast_metrics.offline_store_request_total.labels(

method=method, status=status_label

).inc()

feast_metrics.offline_store_request_latency_seconds.labels(

method=method

).observe(elapsed)

feast_metrics.offline_store_row_count.labels(method=method).observe(

row_count

)

if feast_metrics._config.audit_logging:

feature_views, feature_count = _extract_retrieval_metadata(job)

end_dt = datetime.now(tz=timezone.utc)

start_dt = end_dt - timedelta(seconds=elapsed)

feast_metrics.emit_offline_audit_log(

method=method,

feature_views=feature_views,

feature_count=feature_count,

row_count=row_count,

status=status_label,

start_time=start_dt.isoformat(),

end_time=end_dt.isoformat(),

duration_ms=elapsed * 1000,

)

except Exception:

logging.getLogger(__name__).debug(

"Failed to record offline store metrics", exc_info=True

"""A RetrievalJob manages the execution of a query to retrieve data from the offline store."""

def to_df(

self,

validation_reference: Optional["ValidationReference"] = None,

timeout: Optional[int] = None,

) -> pd.DataFrame:

"""

return (

self.to_arrow(validation_reference=validation_reference, timeout=timeout)

.to_pandas()

.reset_index(drop=True)

)

def to_feast_df(

self,

validation_reference: Optional["ValidationReference"] = None,

timeout: Optional[int] = None,

) -> FeastDataFrame:

"""

# Get Arrow table as before

arrow_table = self.to_arrow(

validation_reference=validation_reference, timeout=timeout

)

# Prepare metadata

metadata = {}

# Add features to metadata if available

if hasattr(self, "features"):

metadata["features"] = self.features

else:

metadata["features"] = []

# Add on-demand feature views to metadata

if hasattr(self, "on_demand_feature_views") and self.on_demand_feature_views:

metadata["on_demand_feature_views"] = [

odfv.name for odfv in self.on_demand_feature_views

]

else:

metadata["on_demand_feature_views"] = []

# Wrap in FeastDataFrame with Arrow engine and metadata

return FeastDataFrame(

data=arrow_table, engine=DataFrameEngine.ARROW, metadata=metadata

)

def to_arrow(

self,

validation_reference: Optional["ValidationReference"] = None,

timeout: Optional[int] = None,

) -> pyarrow.Table:

"""

start_wall = time.monotonic()

status_label = "success"

row_count = 0

try:

features_table = self._to_arrow_internal(timeout=timeout)

row_count = features_table.num_rows

except Exception:

status_label = "error"

raise

finally:

_emit_offline_store_request_metrics(

job=self,

method="to_arrow",

status_label=status_label,

row_count=row_count,

elapsed=time.monotonic() - start_wall,

)

if self.on_demand_feature_views:

# Build a mapping of ODFV name to requested feature names

# This ensures we only return the features that were explicitly requested

odfv_feature_refs: Dict[str, set[str]] = {}

try:

metadata = self.metadata

except NotImplementedError:

metadata = None

if metadata and metadata.features:

for feature_ref in metadata.features:

if ":" in feature_ref:

view_name, feature_name = feature_ref.split(":", 1)

# Check if this view_name matches any of the ODFVs

for odfv in self.on_demand_feature_views:

if (

odfv.name == view_name

or odfv.projection.name_to_use() == view_name

):

if view_name not in odfv_feature_refs:

odfv_feature_refs[view_name] = set()

# Store the feature name in the format that will appear in transformed_arrow

expected_col_name = (

f"{odfv.projection.name_to_use()}__{feature_name}"

if self.full_feature_names

else feature_name

)

odfv_feature_refs[view_name].add(expected_col_name)

for odfv in self.on_demand_feature_views:

transformed_arrow = odfv.transform_arrow(

features_table, self.full_feature_names

)

# Determine which columns to include from this ODFV

# If we have metadata with requested features, filter to only those

# Otherwise, include all columns (backward compatibility)

requested_features_for_odfv = (

odfv_feature_refs.get(odfv.name)

if odfv.name in odfv_feature_refs

else odfv_feature_refs.get(odfv.projection.name_to_use())

)

for col in transformed_arrow.column_names:

if col.startswith("__index"):

continue

# Only append the column if it was requested, or if we don't have feature metadata

if (

requested_features_for_odfv is None

or col in requested_features_for_odfv

):

features_table = features_table.append_column(

col, transformed_arrow[col]

)

if validation_reference:

if not flags_helper.is_test():

warnings.warn(

"Dataset validation is an experimental feature. "

"This API is unstable and it could and most probably will be changed in the future. "

"We do not guarantee that future changes will maintain backward compatibility.",

RuntimeWarning,

)

validation_result = validation_reference.profile.validate(

features_table.to_pandas()

)

if not validation_result.is_success:

raise ValidationFailed(validation_result)

return features_table

def to_arrow_reader(

self, timeout: Optional[int] = None

) -> pyarrow.RecordBatchReader:

"""

table = self.to_arrow(timeout=timeout)

return pyarrow.RecordBatchReader.from_batches(table.schema, table.to_batches())

def to_tensor(

self,

kind: str = "torch",

default_value: Any = float("nan"),

timeout: Optional[int] = None,

) -> Dict[str, Any]:

"""

if kind != "torch":

raise ValueError(

f"Unsupported tensor kind: {kind}. Only 'torch' is supported."

)

torch = get_torch()

device = "cuda" if torch.cuda.is_available() else "cpu"

df = self.to_df(timeout=timeout)

tensor_dict = {}

for column in df.columns:

values = df[column].fillna(default_value).tolist()

first_non_null = next((v for v in values if v is not None), None)

if isinstance(first_non_null, (int, float, bool)):

tensor_dict[column] = torch.tensor(values, device=device)

else:

tensor_dict[column] = values

return tensor_dict

def to_ray_dataset(self) -> Any:

"""Convert the retrieval result to a Ray Dataset.

try:

import ray.data

except ImportError as e:

raise ImportError(

"Ray is required to call to_ray_dataset(). "

"Install it with: pip install 'feast[ray]'"

) from e

return ray.data.from_arrow(self.to_arrow())

def to_sql(self) -> str:

"""

raise NotImplementedError

def _to_df_internal(self, timeout: Optional[int] = None) -> pd.DataFrame:

"""

raise NotImplementedError

def _to_arrow_internal(self, timeout: Optional[int] = None) -> pyarrow.Table:

"""

raise NotImplementedError

@property

def full_feature_names(self) -> bool:

"""Returns True if full feature names should be applied to the results of the query."""

raise NotImplementedError

@property

def on_demand_feature_views(self) -> List[OnDemandFeatureView]:

"""Returns a list containing all the on demand feature views to be handled."""

raise NotImplementedError

def persist(

self,

storage: SavedDatasetStorage,

allow_overwrite: bool = False,

timeout: Optional[int] = None,

):

"""

raise NotImplementedError

@property

def metadata(self) -> Optional[RetrievalMetadata]:

"""Returns metadata about the retrieval job."""

raise NotImplementedError

def supports_remote_storage_export(self) -> bool:

"""Returns True if the RetrievalJob supports `to_remote_storage`."""

return False

def to_remote_storage(self) -> List[str]:

"""

"""

supports_filter_by_created_timestamp: bool = False

"""Whether get_historical_features supports the filter_by_created_timestamp flag."""

def ensure_filter_by_created_timestamp_supported(self) -> None:

"""Raises NotImplementedError if this store does not support filter_by_created_timestamp."""

if not self.supports_filter_by_created_timestamp:

raise NotImplementedError(

f"filter_by_created_timestamp is not supported by {type(self).__name__}"

)

@staticmethod

def pull_latest_from_table_or_query(

config: RepoConfig,

data_source: DataSource,

join_key_columns: List[str],

feature_name_columns: List[str],

timestamp_field: str,

created_timestamp_column: Optional[str],

start_date: datetime,

end_date: datetime,

) -> RetrievalJob:

"""

raise NotImplementedError

@staticmethod

def get_historical_features(

config: RepoConfig,

feature_views: List[FeatureView],

feature_refs: List[str],

entity_df: Optional[Union[pd.DataFrame, str]],

registry: BaseRegistry,

project: str,

full_feature_names: bool = False,

) -> RetrievalJob:

"""

raise NotImplementedError

@staticmethod

def pull_all_from_table_or_query(

config: RepoConfig,

data_source: DataSource,

join_key_columns: List[str],

feature_name_columns: List[str],

timestamp_field: str,

created_timestamp_column: Optional[str] = None,

start_date: Optional[datetime] = None,

end_date: Optional[datetime] = None,

) -> RetrievalJob:

"""

raise NotImplementedError

@staticmethod

def write_logged_features(

config: RepoConfig,

data: Union[pyarrow.Table, Path],

source: LoggingSource,

logging_config: LoggingConfig,

registry: BaseRegistry,

):

"""

raise NotImplementedError

@staticmethod

def offline_write_batch(

config: RepoConfig,

feature_view: FeatureView,

table: pyarrow.Table,

progress: Optional[Callable[[int], Any]],

):

"""

raise NotImplementedError

def validate_data_source(

self,

config: RepoConfig,

data_source: DataSource,

):

"""

data_source.validate(config=config)

def get_table_column_names_and_types_from_data_source(

self,

config: RepoConfig,

data_source: DataSource,

) -> Iterable[Tuple[str, str]]:

"""

return data_source.get_table_column_names_and_types(config=config)

@staticmethod

def compute_monitoring_metrics(

config: RepoConfig,

data_source: DataSource,

feature_columns: List[Tuple[str, str]],

timestamp_field: str,

start_date: Optional[datetime] = None,

end_date: Optional[datetime] = None,

histogram_bins: int = 20,

top_n: int = 10,

) -> List[Dict[str, Any]]:

"""

raise NotImplementedError

@staticmethod

def get_monitoring_max_timestamp(

config: RepoConfig,

data_source: DataSource,

timestamp_field: str,

) -> Optional[datetime]:

"""

raise NotImplementedError

# ------------------------------------------------------------------ #

# Monitoring metrics storage (native)

# ------------------------------------------------------------------ #

MONITORING_VALID_GRANULARITIES = (

"daily",

"weekly",

"biweekly",

"monthly",

"quarterly",

)

@staticmethod

def ensure_monitoring_tables(config: RepoConfig) -> None:

"""Create the monitoring metrics tables if they do not exist.

raise NotImplementedError

@staticmethod

def save_monitoring_metrics(

config: RepoConfig,

metric_type: str,

metrics: List[Dict[str, Any]],

) -> None:

"""Persist monitoring metrics (upsert semantics).

raise NotImplementedError

@staticmethod

def query_monitoring_metrics(

config: RepoConfig,

project: str,

metric_type: str,

filters: Optional[Dict[str, Any]] = None,

start_date: Optional[date] = None,

end_date: Optional[date] = None,

) -> List[Dict[str, Any]]:

"""Read monitoring metrics with optional filtering.

raise NotImplementedError

@staticmethod

def clear_monitoring_baseline(

config: RepoConfig,

project: str,

feature_view_name: Optional[str] = None,

feature_name: Optional[str] = None,

data_source_type: Optional[str] = None,

) -> None:

"""Set is_baseline=FALSE for matching feature metric rows.