# Copyright 2019 The Feast Authors

#

# Licensed under the Apache License, Version 2.0 (the "License");

# you may not use this file except in compliance with the License.

# You may obtain a copy of the License at

#

# https://www.apache.org/licenses/LICENSE-2.0

#

# Unless required by applicable law or agreed to in writing, software

# distributed under the License is distributed on an "AS IS" BASIS,

# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

# See the License for the specific language governing permissions and

# limitations under the License.

import copy

import itertools

import os

import warnings

from collections import Counter, defaultdict

from datetime import datetime

from pathlib import Path

from typing import (

Any,

Dict,

Iterable,

List,

Mapping,

Optional,

Sequence,

Set,

Tuple,

Union,

cast,

)

import pandas as pd

from colorama import Fore, Style

from google.protobuf.timestamp_pb2 import Timestamp

from tqdm import tqdm

from feast import feature_server, flags, flags_helper, utils

from feast.base_feature_view import BaseFeatureView

from feast.data_source import DataSource

from feast.diff.infra_diff import InfraDiff, diff_infra_protos

from feast.diff.registry_diff import RegistryDiff, apply_diff_to_registry, diff_between

from feast.entity import Entity

from feast.errors import (

EntityNotFoundException,

ExperimentalFeatureNotEnabled,

FeatureNameCollisionError,

FeatureViewNotFoundException,

RequestDataNotFoundInEntityDfException,

RequestDataNotFoundInEntityRowsException,

)

from feast.feast_object import FeastObject

from feast.feature_service import FeatureService

from feast.feature_view import (

DUMMY_ENTITY,

DUMMY_ENTITY_ID,

DUMMY_ENTITY_NAME,

DUMMY_ENTITY_VAL,

FeatureView,

)

from feast.go_server import GoServer

from feast.inference import (

update_data_sources_with_inferred_event_timestamp_col,

update_entities_with_inferred_types_from_feature_views,

update_feature_views_with_inferred_features,

)

from feast.infra.infra_object import Infra

from feast.infra.provider import Provider, RetrievalJob, get_provider

from feast.on_demand_feature_view import OnDemandFeatureView

from feast.online_response import OnlineResponse

from feast.protos.feast.core.InfraObject_pb2 import Infra as InfraProto

from feast.protos.feast.serving.ServingService_pb2 import (

FieldStatus,

GetOnlineFeaturesResponse,

)

from feast.protos.feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto

from feast.protos.feast.types.Value_pb2 import RepeatedValue, Value

from feast.registry import Registry

from feast.repo_config import RepoConfig, load_repo_config

from feast.repo_contents import RepoContents

from feast.request_feature_view import RequestFeatureView

from feast.saved_dataset import SavedDataset, SavedDatasetStorage

from feast.type_map import python_values_to_proto_values

from feast.usage import log_exceptions, log_exceptions_and_usage, set_usage_attribute

from feast.value_type import ValueType

from feast.version import get_version

warnings.simplefilter("once", DeprecationWarning)

class FeatureStore:

"""

A FeatureStore object is used to define, create, and retrieve features.

Args:

repo_path (optional): Path to a `feature_store.yaml` used to configure the

feature store.

config (optional): Configuration object used to configure the feature store.

"""

config: RepoConfig

repo_path: Path

_registry: Registry

_provider: Provider

_go_server: Optional[GoServer]

@log_exceptions

def __init__(

self,

repo_path: Optional[str] = None,

config: Optional[RepoConfig] = None,

go_server_use_thread: bool = False,

):

"""

Creates a FeatureStore object.

Raises:

ValueError: If both or neither of repo_path and config are specified.

"""

if repo_path is not None and config is not None:

raise ValueError("You cannot specify both repo_path and config.")

if config is not None:

self.repo_path = Path(os.getcwd())

self.config = config

elif repo_path is not None:

self.repo_path = Path(repo_path)

self.config = load_repo_config(Path(repo_path))

else:

raise ValueError("Please specify one of repo_path or config.")

registry_config = self.config.get_registry_config()

self._registry = Registry(registry_config, repo_path=self.repo_path)

self._registry._initialize_registry()

self._provider = get_provider(self.config, self.repo_path)

self._go_server = None

self._go_server_use_thread = go_server_use_thread

@log_exceptions

def version(self) -> str:

"""Returns the version of the current Feast SDK/CLI."""

return get_version()

@property

def registry(self) -> Registry:

"""Gets the registry of this feature store."""

return self._registry

@property

def project(self) -> str:

"""Gets the project of this feature store."""

return self.config.project

def _get_provider(self) -> Provider:

# TODO: Bake self.repo_path into self.config so that we dont only have one interface to paths

return self._provider

@log_exceptions_and_usage

def refresh_registry(self):

"""Fetches and caches a copy of the feature registry in memory.

Explicitly calling this method allows for direct control of the state of the registry cache. Every time this

method is called the complete registry state will be retrieved from the remote registry store backend

(e.g., GCS, S3), and the cache timer will be reset. If refresh_registry() is run before get_online_features()

is called, then get_online_feature() will use the cached registry instead of retrieving (and caching) the

registry itself.

Additionally, the TTL for the registry cache can be set to infinity (by setting it to 0), which means that

refresh_registry() will become the only way to update the cached registry. If the TTL is set to a value

greater than 0, then once the cache becomes stale (more time than the TTL has passed), a new cache will be

downloaded synchronously, which may increase latencies if the triggering method is get_online_features()

"""

registry_config = self.config.get_registry_config()

registry = Registry(registry_config, repo_path=self.repo_path)

registry.refresh()

self._registry = registry

@log_exceptions_and_usage

def list_entities(self, allow_cache: bool = False) -> List[Entity]:

"""

Retrieves the list of entities from the registry.

Args:

allow_cache: Whether to allow returning entities from a cached registry.

Returns:

A list of entities.

"""

return self._list_entities(allow_cache)

def _list_entities(

self, allow_cache: bool = False, hide_dummy_entity: bool = True

) -> List[Entity]:

all_entities = self._registry.list_entities(

self.project, allow_cache=allow_cache

)

return [

entity

for entity in all_entities

if entity.name != DUMMY_ENTITY_NAME or not hide_dummy_entity

]

@log_exceptions_and_usage

def list_feature_services(self) -> List[FeatureService]:

"""

Retrieves the list of feature services from the registry.

Returns:

A list of feature services.

"""

return self._registry.list_feature_services(self.project)

@log_exceptions_and_usage

def list_feature_views(self, allow_cache: bool = False) -> List[FeatureView]:

"""

Retrieves the list of feature views from the registry.

Args:

allow_cache: Whether to allow returning entities from a cached registry.

Returns:

A list of feature views.

"""

return self._list_feature_views(allow_cache)

@log_exceptions_and_usage

def list_request_feature_views(

self, allow_cache: bool = False

) -> List[RequestFeatureView]:

"""

Retrieves the list of feature views from the registry.

Args:

allow_cache: Whether to allow returning entities from a cached registry.

Returns:

A list of feature views.

"""

return self._registry.list_request_feature_views(

self.project, allow_cache=allow_cache

)

def _list_feature_views(

self, allow_cache: bool = False, hide_dummy_entity: bool = True,

) -> List[FeatureView]:

feature_views = []

for fv in self._registry.list_feature_views(

self.project, allow_cache=allow_cache

):

if hide_dummy_entity and fv.entities[0] == DUMMY_ENTITY_NAME:

fv.entities = []

feature_views.append(fv)

return feature_views

@log_exceptions_and_usage

def list_on_demand_feature_views(

self, allow_cache: bool = False

) -> List[OnDemandFeatureView]:

"""

Retrieves the list of on demand feature views from the registry.

Returns:

A list of on demand feature views.

"""

return self._registry.list_on_demand_feature_views(

self.project, allow_cache=allow_cache

)

@log_exceptions_and_usage

def list_data_sources(self, allow_cache: bool = False) -> List[DataSource]:

"""

Retrieves the list of data sources from the registry.

Args:

allow_cache: Whether to allow returning data sources from a cached registry.

Returns:

A list of data sources.

"""

return self._registry.list_data_sources(self.project, allow_cache=allow_cache)

@log_exceptions_and_usage

def get_entity(self, name: str) -> Entity:

"""

Retrieves an entity.

Args:

name: Name of entity.

Returns:

The specified entity.

Raises:

EntityNotFoundException: The entity could not be found.

"""

return self._registry.get_entity(name, self.project)

@log_exceptions_and_usage

def get_feature_service(

self, name: str, allow_cache: bool = False

) -> FeatureService:

"""

Retrieves a feature service.

Args:

name: Name of feature service.

allow_cache: Whether to allow returning feature services from a cached registry.

Returns:

The specified feature service.

Raises:

FeatureServiceNotFoundException: The feature service could not be found.

"""

return self._registry.get_feature_service(name, self.project, allow_cache)

@log_exceptions_and_usage

def get_feature_view(self, name: str) -> FeatureView:

"""

Retrieves a feature view.

Args:

name: Name of feature view.

Returns:

The specified feature view.

Raises:

FeatureViewNotFoundException: The feature view could not be found.

"""

return self._get_feature_view(name)

def _get_feature_view(

self, name: str, hide_dummy_entity: bool = True

) -> FeatureView:

feature_view = self._registry.get_feature_view(name, self.project)

if hide_dummy_entity and feature_view.entities[0] == DUMMY_ENTITY_NAME:

feature_view.entities = []

return feature_view

@log_exceptions_and_usage

def get_on_demand_feature_view(self, name: str) -> OnDemandFeatureView:

"""

Retrieves a feature view.

Args:

name: Name of feature view.

Returns:

The specified feature view.

Raises:

FeatureViewNotFoundException: The feature view could not be found.

"""

return self._registry.get_on_demand_feature_view(name, self.project)

@log_exceptions_and_usage

def get_data_source(self, name: str) -> DataSource:

"""

Retrieves the list of data sources from the registry.

Args:

name: Name of the data source.

Returns:

The specified data source.

Raises:

DataSourceObjectNotFoundException: The data source could not be found.

"""

return self._registry.get_data_source(name, self.project)

@log_exceptions_and_usage

def delete_feature_view(self, name: str):

"""

Deletes a feature view.

Args:

name: Name of feature view.

Raises:

FeatureViewNotFoundException: The feature view could not be found.

"""

return self._registry.delete_feature_view(name, self.project)

@log_exceptions_and_usage

def delete_feature_service(self, name: str):

"""

Deletes a feature service.

Args:

name: Name of feature service.

Raises:

FeatureServiceNotFoundException: The feature view could not be found.

"""

return self._registry.delete_feature_service(name, self.project)

def _get_features(

self, features: Union[List[str], FeatureService], allow_cache: bool = False,

) -> List[str]:

_features = features

if not _features:

raise ValueError("No features specified for retrieval")

_feature_refs = []

if isinstance(_features, FeatureService):

feature_service_from_registry = self.get_feature_service(

_features.name, allow_cache

)

if feature_service_from_registry != _features:

warnings.warn(

"The FeatureService object that has been passed in as an argument is"

"inconsistent with the version from Registry. Potentially a newer version"

"of the FeatureService has been applied to the registry."

)

for projection in feature_service_from_registry.feature_view_projections:

_feature_refs.extend(

[

f"{projection.name_to_use()}:{f.name}"

for f in projection.features

]

)

else:

assert isinstance(_features, list)

_feature_refs = _features

return _feature_refs

def _should_use_plan(self):

"""Returns True if _plan and _apply_diffs should be used, False otherwise."""

# Currently only the local provider with sqlite online store supports _plan and _apply_diffs.

return self.config.provider == "local" and (

self.config.online_store and self.config.online_store.type == "sqlite"

)

def _validate_all_feature_views(

self,

views_to_update: List[FeatureView],

odfvs_to_update: List[OnDemandFeatureView],

request_views_to_update: List[RequestFeatureView],

):

"""Validates all feature views."""

if (

not flags_helper.enable_on_demand_feature_views(self.config)

and len(odfvs_to_update) > 0

):

raise ExperimentalFeatureNotEnabled(flags.FLAG_ON_DEMAND_TRANSFORM_NAME)

set_usage_attribute("odfv", bool(odfvs_to_update))

_validate_feature_views(

[*views_to_update, *odfvs_to_update, *request_views_to_update]

)

def _make_inferences(

self,

data_sources_to_update: List[DataSource],

entities_to_update: List[Entity],

views_to_update: List[FeatureView],

odfvs_to_update: List[OnDemandFeatureView],

):

"""Makes inferences for entities, feature views, and odfvs."""

update_entities_with_inferred_types_from_feature_views(

entities_to_update, views_to_update, self.config

)

update_data_sources_with_inferred_event_timestamp_col(

data_sources_to_update, self.config

)

update_data_sources_with_inferred_event_timestamp_col(

[view.batch_source for view in views_to_update], self.config

)

# New feature views may reference previously applied entities.

entities = self._list_entities()

update_feature_views_with_inferred_features(

views_to_update, entities + entities_to_update, self.config

)

for odfv in odfvs_to_update:

odfv.infer_features()

@log_exceptions_and_usage

def _plan(

self, desired_repo_contents: RepoContents

) -> Tuple[RegistryDiff, InfraDiff, Infra]:

"""Dry-run registering objects to metadata store.

The plan method dry-runs registering one or more definitions (e.g., Entity, FeatureView), and produces

a list of all the changes the that would be introduced in the feature repo. The changes computed by the plan

command are for informational purpose, and are not actually applied to the registry.

Args:

objects: A single object, or a list of objects that are intended to be registered with the Feature Store.

objects_to_delete: A list of objects to be deleted from the registry.

partial: If True, apply will only handle the specified objects; if False, apply will also delete

all the objects in objects_to_delete.

Raises:

ValueError: The 'objects' parameter could not be parsed properly.

Examples:

Generate a plan adding an Entity and a FeatureView.

>>> from feast import FeatureStore, Entity, FeatureView, Feature, ValueType, FileSource, RepoConfig

>>> from feast.feature_store import RepoContents

>>> from datetime import timedelta

>>> fs = FeatureStore(repo_path="feature_repo")

>>> driver = Entity(name="driver_id", value_type=ValueType.INT64, description="driver id")

>>> driver_hourly_stats = FileSource(

... path="feature_repo/data/driver_stats.parquet",

... event_timestamp_column="event_timestamp",

... created_timestamp_column="created",

... )

>>> driver_hourly_stats_view = FeatureView(

... name="driver_hourly_stats",

... entities=["driver_id"],

... ttl=timedelta(seconds=86400 * 1),

... batch_source=driver_hourly_stats,

... )

>>> registry_diff, infra_diff, new_infra = fs._plan(RepoContents(

... data_sources={driver_hourly_stats},

... feature_views={driver_hourly_stats_view},

... on_demand_feature_views=set(),

... request_feature_views=set(),

... entities={driver},

... feature_services=set())) # register entity and feature view

"""

# Validate and run inference on all the objects to be registered.

self._validate_all_feature_views(

list(desired_repo_contents.feature_views),

list(desired_repo_contents.on_demand_feature_views),

list(desired_repo_contents.request_feature_views),

)

_validate_data_sources(list(desired_repo_contents.data_sources))

self._make_inferences(

list(desired_repo_contents.data_sources),

list(desired_repo_contents.entities),

list(desired_repo_contents.feature_views),

list(desired_repo_contents.on_demand_feature_views),

)

# Compute the desired difference between the current objects in the registry and

# the desired repo state.

registry_diff = diff_between(

self._registry, self.project, desired_repo_contents

)

# Compute the desired difference between the current infra, as stored in the registry,

# and the desired infra.

self._registry.refresh()

current_infra_proto = (

self._registry.cached_registry_proto.infra.__deepcopy__()

if self._registry.cached_registry_proto

else InfraProto()

)

desired_registry_proto = desired_repo_contents.to_registry_proto()

new_infra = self._provider.plan_infra(self.config, desired_registry_proto)

new_infra_proto = new_infra.to_proto()

infra_diff = diff_infra_protos(current_infra_proto, new_infra_proto)

return (registry_diff, infra_diff, new_infra)

@log_exceptions_and_usage

def _apply_diffs(

self, registry_diff: RegistryDiff, infra_diff: InfraDiff, new_infra: Infra

):

"""Applies the given diffs to the metadata store and infrastructure.

Args:

registry_diff: The diff between the current registry and the desired registry.

infra_diff: The diff between the current infra and the desired infra.

new_infra: The desired infra.

"""

infra_diff.update()

apply_diff_to_registry(

self._registry, registry_diff, self.project, commit=False

)

self._registry.update_infra(new_infra, self.project, commit=True)

@log_exceptions_and_usage

def apply(

self,

objects: Union[

DataSource,

Entity,

FeatureView,

OnDemandFeatureView,

RequestFeatureView,

FeatureService,

List[FeastObject],

],

objects_to_delete: Optional[List[FeastObject]] = None,

partial: bool = True,

):

"""Register objects to metadata store and update related infrastructure.

The apply method registers one or more definitions (e.g., Entity, FeatureView) and registers or updates these

objects in the Feast registry. Once the apply method has updated the infrastructure (e.g., create tables in

an online store), it will commit the updated registry. All operations are idempotent, meaning they can safely

be rerun.

Args:

objects: A single object, or a list of objects that should be registered with the Feature Store.

objects_to_delete: A list of objects to be deleted from the registry and removed from the

provider's infrastructure. This deletion will only be performed if partial is set to False.

partial: If True, apply will only handle the specified objects; if False, apply will also delete

all the objects in objects_to_delete, and tear down any associated cloud resources.

Raises:

ValueError: The 'objects' parameter could not be parsed properly.

Examples:

Register an Entity and a FeatureView.

>>> from feast import FeatureStore, Entity, FeatureView, Feature, ValueType, FileSource, RepoConfig

>>> from datetime import timedelta

>>> fs = FeatureStore(repo_path="feature_repo")

>>> driver = Entity(name="driver_id", value_type=ValueType.INT64, description="driver id")

>>> driver_hourly_stats = FileSource(

... path="feature_repo/data/driver_stats.parquet",

... event_timestamp_column="event_timestamp",

... created_timestamp_column="created",

... )

>>> driver_hourly_stats_view = FeatureView(

... name="driver_hourly_stats",

... entities=["driver_id"],

... ttl=timedelta(seconds=86400 * 1),

... batch_source=driver_hourly_stats,

... )

>>> fs.apply([driver_hourly_stats_view, driver]) # register entity and feature view

"""

# TODO: Add locking

if not isinstance(objects, Iterable):

objects = [objects]

assert isinstance(objects, list)

if not objects_to_delete:

objects_to_delete = []

# Separate all objects into entities, feature services, and different feature view types.

entities_to_update = [ob for ob in objects if isinstance(ob, Entity)]

views_to_update = [ob for ob in objects if isinstance(ob, FeatureView)]

request_views_to_update = [

ob for ob in objects if isinstance(ob, RequestFeatureView)

]

odfvs_to_update = [ob for ob in objects if isinstance(ob, OnDemandFeatureView)]

services_to_update = [ob for ob in objects if isinstance(ob, FeatureService)]

data_sources_to_update = [ob for ob in objects if isinstance(ob, DataSource)]

if len(entities_to_update) + len(views_to_update) + len(

request_views_to_update

) + len(odfvs_to_update) + len(services_to_update) + len(

data_sources_to_update

) != len(

objects

):

raise ValueError("Unknown object type provided as part of apply() call")

# Validate all feature views and make inferences.

self._validate_all_feature_views(

views_to_update, odfvs_to_update, request_views_to_update

)

self._make_inferences(

data_sources_to_update, entities_to_update, views_to_update, odfvs_to_update

)

# Handle all entityless feature views by using DUMMY_ENTITY as a placeholder entity.

entities_to_update.append(DUMMY_ENTITY)

# Add all objects to the registry and update the provider's infrastructure.

for ds in data_sources_to_update:

self._registry.apply_data_source(ds, project=self.project, commit=False)

for view in itertools.chain(

views_to_update, odfvs_to_update, request_views_to_update

):

self._registry.apply_feature_view(view, project=self.project, commit=False)

for ent in entities_to_update:

self._registry.apply_entity(ent, project=self.project, commit=False)

for feature_service in services_to_update:

self._registry.apply_feature_service(

feature_service, project=self.project, commit=False

)

if not partial:

# Delete all registry objects that should not exist.

entities_to_delete = [

ob for ob in objects_to_delete if isinstance(ob, Entity)

]

views_to_delete = [

ob for ob in objects_to_delete if isinstance(ob, FeatureView)

]

request_views_to_delete = [

ob for ob in objects_to_delete if isinstance(ob, RequestFeatureView)

]

odfvs_to_delete = [

ob for ob in objects_to_delete if isinstance(ob, OnDemandFeatureView)

]

services_to_delete = [

ob for ob in objects_to_delete if isinstance(ob, FeatureService)

]

data_sources_to_delete = [

ob for ob in objects_to_delete if isinstance(ob, DataSource)

]

for data_source in data_sources_to_delete:

self._registry.delete_data_source(

data_source.name, project=self.project, commit=False

)

for entity in entities_to_delete:

self._registry.delete_entity(

entity.name, project=self.project, commit=False

)

for view in views_to_delete:

self._registry.delete_feature_view(

view.name, project=self.project, commit=False

)

for request_view in request_views_to_delete:

self._registry.delete_feature_view(

request_view.name, project=self.project, commit=False

)

for odfv in odfvs_to_delete:

self._registry.delete_feature_view(

odfv.name, project=self.project, commit=False

)

for service in services_to_delete:

self._registry.delete_feature_service(

service.name, project=self.project, commit=False

)

self._get_provider().update_infra(

project=self.project,

tables_to_delete=views_to_delete if not partial else [],

tables_to_keep=views_to_update,

entities_to_delete=entities_to_delete if not partial else [],

entities_to_keep=entities_to_update,

partial=partial,

)

self._registry.commit()

@log_exceptions_and_usage

def teardown(self):

"""Tears down all local and cloud resources for the feature store."""

tables: List[FeatureView] = []

feature_views = self.list_feature_views()

tables.extend(feature_views)

entities = self.list_entities()

self._get_provider().teardown_infra(self.project, tables, entities)

self._registry.teardown()

@log_exceptions_and_usage

def get_historical_features(

self,

entity_df: Union[pd.DataFrame, str],

features: Union[List[str], FeatureService],

full_feature_names: bool = False,

) -> RetrievalJob:

"""Enrich an entity dataframe with historical feature values for either training or batch scoring.

This method joins historical feature data from one or more feature views to an entity dataframe by using a time

travel join.

Each feature view is joined to the entity dataframe using all entities configured for the respective feature

view. All configured entities must be available in the entity dataframe. Therefore, the entity dataframe must

contain all entities found in all feature views, but the individual feature views can have different entities.

Time travel is based on the configured TTL for each feature view. A shorter TTL will limit the

amount of scanning that will be done in order to find feature data for a specific entity key. Setting a short

TTL may result in null values being returned.

Args:

entity_df (Union[pd.DataFrame, str]): An entity dataframe is a collection of rows containing all entity

columns (e.g., customer_id, driver_id) on which features need to be joined, as well as a event_timestamp

column used to ensure point-in-time correctness. Either a Pandas DataFrame can be provided or a string

SQL query. The query must be of a format supported by the configured offline store (e.g., BigQuery)

features: A list of features, that should be retrieved from the offline store.

Either a list of string feature references can be provided or a FeatureService object.

Feature references are of the format "feature_view:feature", e.g., "customer_fv:daily_transactions".

full_feature_names: A boolean that provides the option to add the feature view prefixes to the feature names,

changing them from the format "feature" to "feature_view__feature" (e.g., "daily_transactions" changes to

"customer_fv__daily_transactions"). By default, this value is set to False.

Returns:

RetrievalJob which can be used to materialize the results.

Raises:

ValueError: Both or neither of features and feature_refs are specified.

Examples:

Retrieve historical features from a local offline store.

>>> from feast import FeatureStore, RepoConfig

>>> import pandas as pd

>>> fs = FeatureStore(repo_path="feature_repo")

>>> entity_df = pd.DataFrame.from_dict(

... {

... "driver_id": [1001, 1002],

... "event_timestamp": [

... datetime(2021, 4, 12, 10, 59, 42),

... datetime(2021, 4, 12, 8, 12, 10),

... ],

... }

... )

>>> retrieval_job = fs.get_historical_features(

... entity_df=entity_df,

... features=[

... "driver_hourly_stats:conv_rate",

... "driver_hourly_stats:acc_rate",

... "driver_hourly_stats:avg_daily_trips",

... ],

... )

>>> feature_data = retrieval_job.to_df()

"""

_feature_refs = self._get_features(features)

(

all_feature_views,

all_request_feature_views,

all_on_demand_feature_views,

) = self._get_feature_views_to_use(features)

# TODO(achal): _group_feature_refs returns the on demand feature views, but it's no passed into the provider.

# This is a weird interface quirk - we should revisit the `get_historical_features` to

# pass in the on demand feature views as well.

fvs, odfvs, request_fvs, request_fv_refs = _group_feature_refs(

_feature_refs,

all_feature_views,

all_request_feature_views,

all_on_demand_feature_views,

)

feature_views = list(view for view, _ in fvs)

on_demand_feature_views = list(view for view, _ in odfvs)

request_feature_views = list(view for view, _ in request_fvs)

set_usage_attribute("odfv", bool(on_demand_feature_views))

set_usage_attribute("request_fv", bool(request_feature_views))

# Check that the right request data is present in the entity_df

if type(entity_df) == pd.DataFrame:

entity_pd_df = cast(pd.DataFrame, entity_df)

for fv in request_feature_views:

for feature in fv.features:

if feature.name not in entity_pd_df.columns:

raise RequestDataNotFoundInEntityDfException(

feature_name=feature.name, feature_view_name=fv.name

)

for odfv in on_demand_feature_views:

odfv_request_data_schema = odfv.get_request_data_schema()

for feature_name in odfv_request_data_schema.keys():

if feature_name not in entity_pd_df.columns:

raise RequestDataNotFoundInEntityDfException(

feature_name=feature_name, feature_view_name=odfv.name,

)

_validate_feature_refs(_feature_refs, full_feature_names)

# Drop refs that refer to RequestFeatureViews since they don't need to be fetched and

# already exist in the entity_df

_feature_refs = [ref for ref in _feature_refs if ref not in request_fv_refs]

provider = self._get_provider()

job = provider.get_historical_features(

self.config,

feature_views,

_feature_refs,

entity_df,

self._registry,

self.project,

full_feature_names,

)

return job

@log_exceptions_and_usage

def create_saved_dataset(

self,

from_: RetrievalJob,

name: str,

storage: SavedDatasetStorage,

tags: Optional[Dict[str, str]] = None,

feature_service: Optional[FeatureService] = None,

) -> SavedDataset:

"""

Execute provided retrieval job and persist its outcome in given storage.

Storage type (eg, BigQuery or Redshift) must be the same as globally configured offline store.

After data successfully persisted saved dataset object with dataset metadata is committed to the registry.

Name for the saved dataset should be unique within project, since it's possible to overwrite previously stored dataset

with the same name.

Returns:

SavedDataset object with attached RetrievalJob

Raises:

ValueError if given retrieval job doesn't have metadata

"""

warnings.warn(

"Saving dataset 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,

)

if not from_.metadata:

raise ValueError(

"RetrievalJob must contains metadata. "

"Use RetrievalJob produced by get_historical_features"

)

dataset = SavedDataset(

name=name,

features=from_.metadata.features,

join_keys=from_.metadata.keys,

full_feature_names=from_.full_feature_names,

storage=storage,

tags=tags,

feature_service_name=feature_service.name if feature_service else None,

)

dataset.min_event_timestamp = from_.metadata.min_event_timestamp

dataset.max_event_timestamp = from_.metadata.max_event_timestamp

from_.persist(storage)

dataset = dataset.with_retrieval_job(

self._get_provider().retrieve_saved_dataset(

config=self.config, dataset=dataset

)

)

self._registry.apply_saved_dataset(dataset, self.project, commit=True)

return dataset

@log_exceptions_and_usage

def get_saved_dataset(self, name: str) -> SavedDataset:

"""

Find a saved dataset in the registry by provided name and

create a retrieval job to pull whole dataset from storage (offline store).

If dataset couldn't be found by provided name SavedDatasetNotFound exception will be raised.

Data will be retrieved from globally configured offline store.

Returns:

SavedDataset with RetrievalJob attached

Raises:

SavedDatasetNotFound

"""

warnings.warn(

"Retrieving datasets 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,

)

dataset = self._registry.get_saved_dataset(name, self.project)

provider = self._get_provider()

retrieval_job = provider.retrieve_saved_dataset(

config=self.config, dataset=dataset

)

return dataset.with_retrieval_job(retrieval_job)

@log_exceptions_and_usage

def materialize_incremental(

self, end_date: datetime, feature_views: Optional[List[str]] = None,

) -> None:

"""

Materialize incremental new data from the offline store into the online store.

This method loads incremental new feature data up to the specified end time from either

the specified feature views, or all feature views if none are specified,

into the online store where it is available for online serving. The start time of

the interval materialized is either the most recent end time of a prior materialization or

(now - ttl) if no such prior materialization exists.

Args:

end_date (datetime): End date for time range of data to materialize into the online store

feature_views (List[str]): Optional list of feature view names. If selected, will only run

materialization for the specified feature views.

Raises:

Exception: A feature view being materialized does not have a TTL set.

Examples:

Materialize all features into the online store up to 5 minutes ago.

>>> from feast import FeatureStore, RepoConfig

>>> from datetime import datetime, timedelta

>>> fs = FeatureStore(repo_path="feature_repo")

>>> fs.materialize_incremental(end_date=datetime.utcnow() - timedelta(minutes=5))

Materializing...