# Copyright 2021 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.

from abc import ABC, abstractmethod

from datetime import datetime

from typing import Dict, List, Optional, Type, Union

from google.protobuf.json_format import MessageToJson

from google.protobuf.message import Message

from feast.data_source import DataSource

from feast.feature_view_projection import FeatureViewProjection

from feast.field import Field

from feast.protos.feast.core.FeatureView_pb2 import FeatureView as FeatureViewProto

from feast.protos.feast.core.OnDemandFeatureView_pb2 import (

OnDemandFeatureView as OnDemandFeatureViewProto,

)

from feast.protos.feast.core.StreamFeatureView_pb2 import (

StreamFeatureView as StreamFeatureViewProto,

)

class BaseFeatureView(ABC):

"""

A BaseFeatureView defines a logical group of features.

Attributes:

name: The unique name of the base feature view.

features: The list of features defined as part of this base feature view.

description: A human-readable description.

tags: A dictionary of key-value pairs to store arbitrary metadata.

owner: The owner of the base feature view, typically the email of the primary

maintainer.

projection: The feature view projection storing modifications to be applied to

this base feature view at retrieval time.

created_timestamp: The time when the base feature view was created.

last_updated_timestamp: The time when the base feature view was last

updated.

"""

name: str

features: List[Field]

description: str

tags: Dict[str, str]

owner: str

projection: FeatureViewProjection

created_timestamp: Optional[datetime]

last_updated_timestamp: Optional[datetime]

version: str

current_version_number: Optional[int]

@abstractmethod

def __init__(

self,

*,

name: str,

features: Optional[List[Field]] = None,

description: str = "",

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

owner: str = "",

source: Optional[DataSource] = None,

):

"""

Creates a BaseFeatureView object.

Args:

name: The unique name of the base feature view.

features (optional): The list of features defined as part of this base feature view.

description (optional): A human-readable description.

tags (optional): A dictionary of key-value pairs to store arbitrary metadata.

owner (optional): The owner of the base feature view, typically the email of the

primary maintainer.

source (optional): The source of data for this group of features. May be a stream source, or a batch source.

If a stream source, the source should contain a batch_source for backfills & batch materialization.

Raises:

ValueError: A field mapping conflicts with an Entity or a Feature.

"""

assert name is not None

self.name = name

self.features = features or []

self.description = description

self.tags = tags or {}

self.owner = owner

self.projection = FeatureViewProjection.from_definition(self)

self.created_timestamp = None

self.last_updated_timestamp = None

if not hasattr(self, "version"):

self.version = "latest"

if not hasattr(self, "current_version_number"):

self.current_version_number = None

self.source = source

@property

@abstractmethod

def proto_class(self) -> Type[Message]:

pass

@abstractmethod

def to_proto(

self,

) -> Union[FeatureViewProto, OnDemandFeatureViewProto, StreamFeatureViewProto]:

pass

@classmethod

@abstractmethod

def from_proto(cls, feature_view_proto):

pass

@abstractmethod

def __copy__(self):

"""Returns a deep copy of this base feature view."""

pass

def __repr__(self):

items = (f"{k} = {v}" for k, v in self.__dict__.items())

return f"<{self.__class__.__name__}({', '.join(items)})>"

def __str__(self):

return str(MessageToJson(self.to_proto()))

def __hash__(self):

return hash(self.name)

def __getitem__(self, item):

assert isinstance(item, list)

cp = self.__copy__()

if self.features:

feature_name_to_feature = {

feature.name: feature for feature in self.features

}

referenced_features = []

for feature in item:

if feature not in feature_name_to_feature:

raise ValueError(

f"Feature {feature} does not exist in this feature view."

)

referenced_features.append(feature_name_to_feature[feature])

cp.projection.features = referenced_features

else:

cp.projection.desired_features = item

return cp

def _schema_or_udf_changed(self, other: "BaseFeatureView") -> bool:

"""Check if schema or UDF-related fields have changed (version-worthy changes).

Callers always match by name first, so name comparison is omitted here.

"""

if sorted(self.features) != sorted(other.features):

return True

# Skip metadata: description, tags, owner, projection

# Skip source changes: treat as deployment/location details, not schema changes

return False

def __eq__(self, other):

if not isinstance(other, BaseFeatureView):

raise TypeError(

"Comparisons should only involve BaseFeatureView class objects."

)

if (

self.name != other.name

or sorted(self.features) != sorted(other.features)

or self.projection != other.projection

or self.description != other.description

or self.tags != other.tags

or self.owner != other.owner

):

# This is meant to ignore the File Source change to Push Source

if isinstance(type(self.source), type(other.source)):

if self.source != other.source:

return False

return False

return True

def ensure_valid(self):

"""

Validates the state of this feature view locally.

Raises:

ValueError: The feature view is invalid.

"""

if not self.name:

raise ValueError("Feature view needs a name.")

if "@" in self.name:

raise ValueError(

f"Feature view name '{self.name}' must not contain '@'. "

f"The '@' character is reserved for version-qualified references "

f"(e.g., 'fv@v2:feature')."

)

if ":" in self.name:

raise ValueError(

f"Feature view name '{self.name}' must not contain ':'. "

f"The ':' character is reserved as the separator in fully qualified "

f"feature references (e.g., 'feature_view:feature_name')."

)

def with_name(self, name: str):

"""

Returns a renamed copy of this base feature view. This renamed copy should only be

used for query operations and will not modify the underlying base feature view.

Args:

name: The name to assign to the copy.

"""

cp = self.__copy__()

cp.projection.name_alias = name

return cp

def set_projection(self, feature_view_projection: FeatureViewProjection) -> None:

"""

Sets the feature view projection of this base feature view to the given projection.

Args:

feature_view_projection: The feature view projection to be set.

Raises:

ValueError: The name or features of the projection do not match.

"""

if feature_view_projection.name != self.name:

raise ValueError(

f"The projection for the {self.name} FeatureView cannot be applied because it differs in name. "

f"The projection is named {feature_view_projection.name} and the name indicates which "

"FeatureView the projection is for."

)

for feature in feature_view_projection.features:

if feature not in self.features:

raise ValueError(

f"The projection for {self.name} cannot be applied because it contains {feature.name} which the "

"FeatureView doesn't have."

)

self.projection = feature_view_projection

def with_projection(self, feature_view_projection: FeatureViewProjection):

"""

Returns a copy of this base feature view with the feature view projection set to

the given projection.

Args:

feature_view_projection: The feature view projection to assign to the copy.

Raises:

ValueError: The name or features of the projection do not match.

"""

if feature_view_projection.name != self.name:

raise ValueError(

f"The projection for the {self.name} FeatureView cannot be applied because it differs in name. "

f"The projection is named {feature_view_projection.name} and the name indicates which "

"FeatureView the projection is for."

)

for feature in feature_view_projection.features:

if feature not in self.features:

raise ValueError(

f"The projection for {self.name} cannot be applied because it contains {feature.name} which the "

"FeatureView doesn't have."

)

cp = self.__copy__()

cp.projection = feature_view_projection

return cp