diff --git a/docs/getting-started/concepts/feature-view.md b/docs/getting-started/concepts/feature-view.md index 27ded82cb84..74b55dabea4 100644 --- a/docs/getting-started/concepts/feature-view.md +++ b/docs/getting-started/concepts/feature-view.md @@ -162,6 +162,32 @@ Feature names must be unique within a [feature view](feature-view.md#feature-vie Each field can have additional metadata associated with it, specified as key-value [tags](https://rtd.feast.dev/en/master/feast.html#feast.field.Field). +### Default values + +A field can declare a typed default that Feast substitutes when the feature is missing or null, so every consumer sees the same value instead of imputing its own: + +```python +from feast import Field +from feast.types import Int64 + +Field( + name="transaction_count", + dtype=Int64, + default_value=0, +) +``` + +The default is applied during online retrieval and after the historical point-in-time join, and it is applied before on demand feature views run, so a transformation receives the same input in training and serving. + +Some details worth knowing: + +* Defaults are opt-in. A field without `default_value` behaves exactly as before, returning null. +* The default must be representable as the field's `dtype`. A value the type cannot hold, and a float that an integer type would truncate such as `1.5` for an `Int64`, are rejected when the `Field` is created rather than during serving. Values the type *can* hold are coerced and stored in their converted form, so `Float32` with `0.1` reads back as `0.10000000149011612` and `Int64` with `"5"` reads back as `5`. +* Feature statuses are not changed. A missing feature still reports `NOT_FOUND` even when a default is returned, so monitoring continues to reveal upstream data loss. If you need to distinguish "genuinely zero" from "missing" in a model, define a separate indicator feature. +* Values outside the feature view's TTL keep their real value and `OUTSIDE_MAX_AGE` status. A default never replaces a stale value. + +Because older Feast runtimes ignore the field, upgrade your serving deployments before relying on a configured default. + ## \[Alpha\] Versioning Feature views support automatic version tracking. Every time `feast apply` detects a schema or UDF change, a versioned snapshot is saved to the registry. This enables auditing what changed, reverting to a prior version, querying specific versions via `@v` syntax, and staging new versions without promoting them. diff --git a/go/internal/feast/model/basefeatureview.go b/go/internal/feast/model/basefeatureview.go index 1bdf614c25a..1ade4ed2e01 100644 --- a/go/internal/feast/model/basefeatureview.go +++ b/go/internal/feast/model/basefeatureview.go @@ -2,14 +2,46 @@ package model import ( "fmt" + "sync" "github.com/feast-dev/feast/go/protos/feast/core" + "github.com/feast-dev/feast/go/protos/feast/types" ) type BaseFeatureView struct { Name string Features []*Field Projection *FeatureViewProjection + // Built on first use rather than at construction, because BaseFeatureView is also + // assembled as a struct literal; a constructor-only map would silently be empty. + featureDefaultsOnce sync.Once + featureDefaults map[string]*types.Value +} + +// GetDefaultValue returns the configured default for a feature, or nil. Serving calls +// this per feature per request, so the scan happens once rather than on every request. +func (fv *BaseFeatureView) GetDefaultValue(featureName string) *types.Value { + fv.featureDefaultsOnce.Do(func() { + // Prefer the projection: a feature service can project a different set of + // fields than the base view carries, and Python resolves defaults the same way. + features := fv.Features + if fv.Projection != nil && len(fv.Projection.Features) > 0 { + features = fv.Projection.Features + } + for _, feature := range features { + if feature.DefaultValue == nil { + continue + } + if fv.featureDefaults == nil { + fv.featureDefaults = make(map[string]*types.Value, len(features)) + } + fv.featureDefaults[feature.Name] = feature.DefaultValue + } + }) + if fv.featureDefaults == nil { + return nil + } + return fv.featureDefaults[featureName] } func NewBaseFeatureView(name string, featureProtos []*core.FeatureSpecV2) *BaseFeatureView { diff --git a/go/internal/feast/model/field.go b/go/internal/feast/model/field.go index 4f72d346866..f0199100e0b 100644 --- a/go/internal/feast/model/field.go +++ b/go/internal/feast/model/field.go @@ -8,11 +8,14 @@ import ( type Field struct { Name string Dtype types.ValueType_Enum + // Substituted when the feature is missing or null. Nil means no default. + DefaultValue *types.Value } func NewFieldFromProto(proto *core.FeatureSpecV2) *Field { return &Field{ - Name: proto.Name, - Dtype: proto.ValueType, + Name: proto.Name, + Dtype: proto.ValueType, + DefaultValue: proto.DefaultValue, } } diff --git a/go/internal/feast/onlineserving/serving.go b/go/internal/feast/onlineserving/serving.go index 1ce5f6c555c..087b9ea3571 100644 --- a/go/internal/feast/onlineserving/serving.go +++ b/go/internal/feast/onlineserving/serving.go @@ -347,6 +347,23 @@ func ValidateFeatureRefs(requestedFeatures []*FeatureViewAndRefs, fullFeatureNam return nil } +// defaultValueForFeature returns the feature's configured default, or nil. The map +// lookup on BaseFeatureView keeps this off the critical path when nothing declares one. +func defaultValueForFeature( + fvs map[string]*model.FeatureView, + groupRef *GroupedFeaturesPerEntitySet, + featureIndex int) *prototypes.Value { + + if featureIndex >= len(groupRef.FeatureViewNames) || featureIndex >= len(groupRef.FeatureNames) { + return nil + } + fv, ok := fvs[groupRef.FeatureViewNames[featureIndex]] + if !ok || fv.Base == nil { + return nil + } + return fv.Base.GetDefaultValue(groupRef.FeatureNames[featureIndex]) +} + func TransposeFeatureRowsIntoColumns(featureData2D [][]onlinestore.FeatureData, groupRef *GroupedFeaturesPerEntitySet, requestedFeatureViews []*FeatureViewAndRefs, @@ -377,6 +394,10 @@ func TransposeFeatureRowsIntoColumns(featureData2D [][]onlinestore.FeatureData, vectors = append(vectors, currentVector) protoValues := make([]*prototypes.Value, numRows) + // Resolved per feature rather than per row: a missing row carries no + // Reference to look the feature view up by. + defaultValue := defaultValueForFeature(fvs, groupRef, featureIndex) + for rowEntityIndex, outputIndexes := range groupRef.Indices { if featureData2D[rowEntityIndex] == nil { value = nil @@ -398,8 +419,14 @@ func TransposeFeatureRowsIntoColumns(featureData2D [][]onlinestore.FeatureData, status = serving.FieldStatus_PRESENT } } + // Only the value is swapped; the status is left as computed above. An + // empty Value counts as missing, matching the Python paths. + outValue := value + if defaultValue != nil && (outValue == nil || outValue.Val == nil) { + outValue = defaultValue + } for _, rowIndex := range outputIndexes { - protoValues[rowIndex] = value + protoValues[rowIndex] = outValue currentVector.Statuses[rowIndex] = status currentVector.Timestamps[rowIndex] = eventTimeStamp } diff --git a/go/internal/feast/onlineserving/serving_test.go b/go/internal/feast/onlineserving/serving_test.go index bd4e45a21ec..2269960425e 100644 --- a/go/internal/feast/onlineserving/serving_test.go +++ b/go/internal/feast/onlineserving/serving_test.go @@ -2,13 +2,18 @@ package onlineserving import ( "testing" + "time" + "github.com/apache/arrow/go/v17/arrow/array" + "github.com/apache/arrow/go/v17/arrow/memory" "github.com/stretchr/testify/assert" "google.golang.org/protobuf/types/known/durationpb" "google.golang.org/protobuf/types/known/timestamppb" "github.com/feast-dev/feast/go/internal/feast/model" + "github.com/feast-dev/feast/go/internal/feast/onlinestore" "github.com/feast-dev/feast/go/protos/feast/core" + "github.com/feast-dev/feast/go/protos/feast/serving" "github.com/feast-dev/feast/go/protos/feast/types" ) @@ -332,3 +337,86 @@ func TestUnpackFeatureViewsByReferences(t *testing.T) { assertCorrectUnpacking(t, fvs, odfvs, err) } + +func defaultValueTestView(defaultValue *types.Value) *model.FeatureView { + return &model.FeatureView{ + Base: &model.BaseFeatureView{ + Name: "driver_stats", + Features: []*model.Field{ + {Name: "conv_rate", Dtype: types.ValueType_INT64, DefaultValue: defaultValue}, + }, + Projection: &model.FeatureViewProjection{Name: "driver_stats"}, + }, + Ttl: durationpb.New(time.Hour * 24 * 365 * 100), + } +} + +func defaultValueTestGroupRef() *GroupedFeaturesPerEntitySet { + return &GroupedFeaturesPerEntitySet{ + FeatureNames: []string{"conv_rate"}, + FeatureViewNames: []string{"driver_stats"}, + AliasedFeatureNames: []string{"driver_stats__conv_rate"}, + Indices: [][]int{{0}}, + } +} + +func transposeSingle(t *testing.T, view *model.FeatureView, featureData2D [][]onlinestore.FeatureData) *FeatureVector { + t.Helper() + vectors, err := TransposeFeatureRowsIntoColumns( + featureData2D, + defaultValueTestGroupRef(), + []*FeatureViewAndRefs{{View: view, FeatureRefs: []string{"conv_rate"}}}, + memory.NewGoAllocator(), + 1, + ) + assert.Nil(t, err) + assert.Len(t, vectors, 1) + return vectors[0] +} + +func TestTransposeSubstitutesDefaultForMissingRow(t *testing.T) { + view := defaultValueTestView(&types.Value{Val: &types.Value_Int64Val{Int64Val: 7}}) + + vector := transposeSingle(t, view, [][]onlinestore.FeatureData{nil}) + + assert.Equal(t, serving.FieldStatus_NOT_FOUND, vector.Statuses[0]) + assert.Equal(t, int64(7), vector.Values.(*array.Int64).Value(0)) +} + +func TestTransposeSubstitutesDefaultForNullValue(t *testing.T) { + view := defaultValueTestView(&types.Value{Val: &types.Value_Int64Val{Int64Val: 7}}) + featureData2D := [][]onlinestore.FeatureData{{{ + Reference: serving.FeatureReferenceV2{FeatureViewName: "driver_stats", FeatureName: "conv_rate"}, + Timestamp: timestamppb.Timestamp{Seconds: timestamppb.Now().Seconds}, + Value: types.Value{Val: &types.Value_NullVal{}}, + }}} + + vector := transposeSingle(t, view, featureData2D) + + assert.Equal(t, serving.FieldStatus_NOT_FOUND, vector.Statuses[0]) + assert.Equal(t, int64(7), vector.Values.(*array.Int64).Value(0)) +} + +func TestTransposeLeavesPresentValueUntouched(t *testing.T) { + view := defaultValueTestView(&types.Value{Val: &types.Value_Int64Val{Int64Val: 7}}) + featureData2D := [][]onlinestore.FeatureData{{{ + Reference: serving.FeatureReferenceV2{FeatureViewName: "driver_stats", FeatureName: "conv_rate"}, + Timestamp: timestamppb.Timestamp{Seconds: timestamppb.Now().Seconds}, + Value: types.Value{Val: &types.Value_Int64Val{Int64Val: 3}}, + }}} + + vector := transposeSingle(t, view, featureData2D) + + assert.Equal(t, serving.FieldStatus_PRESENT, vector.Statuses[0]) + assert.Equal(t, int64(3), vector.Values.(*array.Int64).Value(0)) +} + +func TestTransposeWithoutDefaultLeavesNull(t *testing.T) { + view := defaultValueTestView(nil) + + vector := transposeSingle(t, view, [][]onlinestore.FeatureData{nil}) + + assert.Equal(t, serving.FieldStatus_NOT_FOUND, vector.Statuses[0]) + // array.Null reports NullN rather than IsNull for this type. + assert.Equal(t, 1, vector.Values.NullN()) +} diff --git a/protos/feast/core/Feature.proto b/protos/feast/core/Feature.proto index 9f7708c65e7..f259b95c065 100644 --- a/protos/feast/core/Feature.proto +++ b/protos/feast/core/Feature.proto @@ -45,4 +45,8 @@ message FeatureSpecV2 { // Field indicating the vector length int32 vector_length = 7; + + // Value substituted when this feature is missing or null at retrieval time. + // Unset means no default. Presence is what separates that from a default of 0. + feast.types.Value default_value = 8; } diff --git a/sdk/python/feast/feature_store.py b/sdk/python/feast/feature_store.py index 546f6f7680d..829005684bd 100644 --- a/sdk/python/feast/feature_store.py +++ b/sdk/python/feast/feature_store.py @@ -2105,6 +2105,10 @@ def get_historical_features( **kwargs, ) + default_values = utils.get_default_values_by_column(fvs, full_feature_names) + if default_values: + job._feature_default_values = default_values + # Auto-log to MLflow if configured try: if self.mlflow is not None and self.config.mlflow.auto_log: diff --git a/sdk/python/feast/feature_view_projection.py b/sdk/python/feast/feature_view_projection.py index a91c6498604..10eec4a1e78 100644 --- a/sdk/python/feast/feature_view_projection.py +++ b/sdk/python/feast/feature_view_projection.py @@ -7,6 +7,7 @@ from feast.protos.feast.core.FeatureViewProjection_pb2 import ( FeatureViewProjection as FeatureViewProjectionProto, ) +from feast.protos.feast.types.Value_pb2 import Value as ValueProto if TYPE_CHECKING: from feast.base_feature_view import BaseFeatureView @@ -156,6 +157,14 @@ def from_definition(base_feature_view: "BaseFeatureView"): desired_features=[], ) + def default_value_protos(self) -> Dict[str, ValueProto]: + """Proto defaults for this projection's features that configure one.""" + return { + field.name: field.default_value_proto + for field in self.features + if field.default_value_proto is not None + } + def get_feature(self, feature_name: str) -> Field: try: return next(field for field in self.features if field.name == feature_name) diff --git a/sdk/python/feast/field.py b/sdk/python/feast/field.py index 3055bd70830..db083467dba 100644 --- a/sdk/python/feast/field.py +++ b/sdk/python/feast/field.py @@ -12,19 +12,32 @@ # See the License for the specific language governing permissions and # limitations under the License. +import copy import json -from typing import Dict, Optional +from typing import Any, Dict, Optional from typeguard import typechecked from feast.feature import Feature from feast.protos.feast.core.Feature_pb2 import FeatureSpecV2 as FieldProto +from feast.protos.feast.types.Value_pb2 import Value as ValueProto from feast.types import FeastType, Struct, from_value_type from feast.value_type import ValueType STRUCT_SCHEMA_TAG = "feast:struct_schema" NESTED_COLLECTION_INNER_TYPE_TAG = "feast:nested_inner_type" +_INTEGER_VALUE_TYPES = frozenset( + { + ValueType.INT32, + ValueType.INT64, + ValueType.INT32_LIST, + ValueType.INT64_LIST, + ValueType.INT32_SET, + ValueType.INT64_SET, + } +) + @typechecked class Field: @@ -39,6 +52,8 @@ class Field: vector_index: If set to True the field will be indexed for vector similarity search. vector_length: The length of the vector if the vector index is set to True. vector_search_metric: The metric used for vector similarity search. + default_value: Value substituted when the feature is missing or null at + retrieval time. None means no default is configured. """ name: str @@ -48,6 +63,7 @@ class Field: vector_index: bool vector_length: int vector_search_metric: Optional[str] + default_value: Any def __init__( self, @@ -59,6 +75,7 @@ def __init__( vector_index: bool = False, vector_length: int = 0, vector_search_metric: Optional[str] = None, + default_value: Any = None, ): """ Creates a Field object. @@ -70,6 +87,11 @@ def __init__( tags (optional): User-defined metadata in dictionary form. vector_index (optional): If set to True the field will be indexed for vector similarity search. vector_search_metric (optional): The metric used for vector similarity search. + default_value (optional): Value substituted when the feature is missing or + null at retrieval time. Must be compatible with dtype. + + Raises: + ValueError: If default_value cannot be represented as dtype. """ self.name = name self.dtype = dtype @@ -78,6 +100,71 @@ def __init__( self.vector_index = vector_index self.vector_length = vector_length self.vector_search_metric = vector_search_metric + # Copied so a later mutation of the caller's list or dict cannot drift from the + # proto cached below, which would make the offline and online paths disagree. + self.default_value = ( + copy.deepcopy(default_value) + if isinstance(default_value, (list, dict, set)) + else default_value + ) + # Converted once: fails where the user wrote it, and online reads reuse it. + self._default_value_proto: Optional[ValueProto] = ( + self._build_default_value_proto() if default_value is not None else None + ) + if self._default_value_proto is not None: + # Store what the registry will actually hold. The conversion coerces, so + # String with 123 persists "123"; keeping 123 here would make a Field + # compare unequal to its own round trip and register as a schema change + # on every apply. + from feast.type_map import feast_value_type_to_python_type + + self.default_value = feast_value_type_to_python_type( + self._default_value_proto + ) + + @property + def default_value_proto(self) -> Optional[ValueProto]: + """The default as a proto, or None when the field configures no default.""" + return self._default_value_proto + + def _build_default_value_proto(self) -> ValueProto: + """Converts this field's default value to its proto representation. + + Raises: + ValueError: If the default value cannot be represented as this field's dtype. + """ + from feast.type_map import python_values_to_proto_values + + value_type = self.dtype.to_value_type() + + # The conversion truncates a float into an integer type, so 1.5 would be stored + # as 1. Nothing else is checked by value: comparing the round-trip would reject + # legitimate defaults such as 0.1 on Float32, which no float32 can hold exactly. + if value_type in _INTEGER_VALUE_TYPES and isinstance(self.default_value, float): + if not self.default_value.is_integer(): + raise ValueError( + f"default_value {self.default_value!r} for field {self.name!r} would " + f"be truncated by dtype {self.dtype}." + ) + + try: + proto_value = python_values_to_proto_values( + [self.default_value], value_type + )[0] + except Exception as e: + raise ValueError( + f"default_value {self.default_value!r} for field {self.name!r} is not " + f"compatible with dtype {self.dtype}: {e}" + ) from e + + # An unrepresentable value converts to an empty Value, which would otherwise read + # back as "no default configured". + if proto_value.WhichOneof("val") is None: + raise ValueError( + f"default_value {self.default_value!r} for field {self.name!r} is not " + f"compatible with dtype {self.dtype}." + ) + return proto_value def __eq__(self, other): if type(self) != type(other): @@ -89,6 +176,7 @@ def __eq__(self, other): or self.description != other.description or self.tags != other.tags or self.vector_length != other.vector_length + or self.default_value != other.default_value # or self.vector_index != other.vector_index # or self.vector_search_metric != other.vector_search_metric ): @@ -111,6 +199,7 @@ def __repr__(self): f" vector_index={self.vector_index!r}\n" f" vector_length={self.vector_length!r}\n" f" vector_search_metric={self.vector_search_metric!r}\n" + f" default_value={self.default_value!r}\n" f")" ) @@ -142,6 +231,8 @@ def to_proto(self) -> FieldProto: vector_index=self.vector_index, vector_length=self.vector_length, vector_search_metric=vector_search_metric, + # None leaves the field unset, which is what keeps presence meaningful. + default_value=self._default_value_proto, ) @classmethod @@ -180,7 +271,16 @@ def from_proto(cls, field_proto: FieldProto): dtype = from_value_type(value_type=value_type) user_tags = {k: v for k, v in tags.items() if k not in internal_tags} - return cls( + # Presence, not truthiness, so defaults of 0, False and "" survive. + default_value = None + default_value_proto = None + if field_proto.HasField("default_value"): + from feast.type_map import feast_value_type_to_python_type + + default_value_proto = field_proto.default_value + default_value = feast_value_type_to_python_type(default_value_proto) + + field = cls( name=field_proto.name, dtype=dtype, tags=user_tags, @@ -188,7 +288,13 @@ def from_proto(cls, field_proto: FieldProto): vector_index=vector_index, vector_length=vector_length, vector_search_metric=vector_search_metric, + default_value=default_value, ) + if default_value_proto is not None: + # Reuse what the registry already holds rather than re-deriving it; the + # value was validated when the Field was first written. + field._default_value_proto = default_value_proto + return field @classmethod def from_feature(cls, feature: Feature): diff --git a/sdk/python/feast/infra/offline_stores/bigquery.py b/sdk/python/feast/infra/offline_stores/bigquery.py index a7c8c41094d..4f5a8122d81 100644 --- a/sdk/python/feast/infra/offline_stores/bigquery.py +++ b/sdk/python/feast/infra/offline_stores/bigquery.py @@ -392,6 +392,7 @@ def query_generator() -> Iterator[str]: query_template=MULTIPLE_FEATURE_VIEW_POINT_IN_TIME_JOIN, full_feature_names=full_feature_names, filter_by_created_timestamp=filter_by_created_timestamp, + quote_char="`", ) try: @@ -1236,6 +1237,9 @@ def _bq_compute_monitoring_metrics( class BigQueryRetrievalJob(RetrievalJob): + # Defaults are COALESCEd into the point-in-time query. + _defaults_applied_in_query = True + def __init__( self, query: Union[str, Callable[[], ContextManager[str]]], @@ -1316,7 +1320,7 @@ def to_bigquery( path = f"{self.client.project}.{self.config.offline_store.dataset}.historical_{today}_{rand_id}" job_config = bigquery.QueryJobConfig(destination=path) - if not job_config.dry_run and self.on_demand_feature_views: + if not job_config.dry_run and self._requires_python_post_processing: job = self.client.load_table_from_dataframe( self.to_df(), job_config.destination ) @@ -1813,7 +1817,7 @@ def arrow_schema_to_bq_schema(arrow_schema: pyarrow.Schema) -> List[SchemaField] The entity_dataframe dataset being our source of truth here. */ -SELECT {{ final_output_feature_names | backticks | join(', ')}} +SELECT {{ final_output_feature_expressions | join(', ')}} FROM entity_dataframe {% for featureview in featureviews %} LEFT JOIN ( diff --git a/sdk/python/feast/infra/offline_stores/contrib/athena_offline_store/athena.py b/sdk/python/feast/infra/offline_stores/contrib/athena_offline_store/athena.py index d8fae6bf19b..4826352f474 100644 --- a/sdk/python/feast/infra/offline_stores/contrib/athena_offline_store/athena.py +++ b/sdk/python/feast/infra/offline_stores/contrib/athena_offline_store/athena.py @@ -327,6 +327,9 @@ def write_logged_features( class AthenaRetrievalJob(RetrievalJob): + # Defaults are COALESCEd into the point-in-time query. + _defaults_applied_in_query = True + def __init__( self, query: Union[str, Callable[[], ContextManager[str]]], @@ -440,7 +443,7 @@ def persist( self.to_athena(table_name=storage.athena_options.table) def to_athena(self, table_name: str) -> None: - if self.on_demand_feature_views: + if self._requires_python_post_processing: transformed_df = self.to_df() _upload_entity_df( @@ -732,7 +735,7 @@ def _get_entity_df_event_timestamp_range( The entity_dataframe dataset being our source of truth here. */ -SELECT {{ final_output_feature_names | join(', ')}} +SELECT {{ final_output_feature_expressions | join(', ')}} FROM entity_dataframe as entity_df {% for featureview in featureviews %} LEFT JOIN ( diff --git a/sdk/python/feast/infra/offline_stores/contrib/ray_offline_store/ray.py b/sdk/python/feast/infra/offline_stores/contrib/ray_offline_store/ray.py index 5230797d94b..1842565fbe0 100644 --- a/sdk/python/feast/infra/offline_stores/contrib/ray_offline_store/ray.py +++ b/sdk/python/feast/infra/offline_stores/contrib/ray_offline_store/ray.py @@ -32,6 +32,7 @@ OfflineStore, RetrievalJob, RetrievalMetadata, + _apply_default_values, ) from feast.infra.offline_stores.offline_utils import ( get_entity_df_timestamp_bounds, @@ -1040,6 +1041,19 @@ def _resolve(self) -> Union[Dataset, pd.DataFrame]: result = self._dataset_or_callable() else: result = self._dataset_or_callable + # Filled here rather than in to_arrow: the Ray paths (to_ray_dataset, + # to_feast_df, to_remote_storage, persist) never reach the base class. + if self._feature_default_values: + defaults = self._feature_default_values + if is_ray_data(result): + result = result.map_batches( + lambda batch: _apply_default_values(batch, defaults), + batch_format="pyarrow", + ) + elif isinstance(result, pd.DataFrame): + result = _apply_default_values( + pa.Table.from_pandas(result), defaults + ).to_pandas() return result def _get_ray_dataset(self) -> Dataset: diff --git a/sdk/python/feast/infra/offline_stores/contrib/spark_offline_store/spark.py b/sdk/python/feast/infra/offline_stores/contrib/spark_offline_store/spark.py index 7e0a03e69bb..d4cfb22d46a 100644 --- a/sdk/python/feast/infra/offline_stores/contrib/spark_offline_store/spark.py +++ b/sdk/python/feast/infra/offline_stores/contrib/spark_offline_store/spark.py @@ -88,8 +88,10 @@ class SparkOfflineStoreConfig(FeastConfigBaseModel): @dataclass(frozen=True) class SparkFeatureViewQueryContext(offline_utils.FeatureViewQueryContext): - min_date_partition: Optional[str] - max_date_partition: Optional[str] + # Defaulted because the base class now has a defaulted field, and a dataclass + # cannot declare a required field after one with a default. + min_date_partition: Optional[str] = None + max_date_partition: Optional[str] = None class SparkOfflineStore(OfflineStore): @@ -1872,7 +1874,7 @@ def _cast_data_frame( The entity_dataframe dataset being our source of truth here. */ -SELECT {{ final_output_feature_names | join(', ')}} +SELECT {{ final_output_feature_expressions | join(', ')}} FROM entity_dataframe {% for featureview in featureviews %} LEFT JOIN ( diff --git a/sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/trino.py b/sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/trino.py index c9d4119f94f..75d236e1278 100644 --- a/sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/trino.py +++ b/sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/trino.py @@ -720,7 +720,7 @@ def _get_entity_df_event_timestamp_range( Joins the outputs of multiple time travel joins to a single table. The entity_dataframe dataset being our source of truth here. */ -SELECT {{ final_output_feature_names | join(', ')}} +SELECT {{ final_output_feature_expressions | join(', ')}} FROM entity_dataframe {% for featureview in featureviews %} LEFT JOIN ( diff --git a/sdk/python/feast/infra/offline_stores/offline_store.py b/sdk/python/feast/infra/offline_stores/offline_store.py index 9d4092d6799..62b6ddb0b58 100644 --- a/sdk/python/feast/infra/offline_stores/offline_store.py +++ b/sdk/python/feast/infra/offline_stores/offline_store.py @@ -12,6 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. import logging +import math import time import warnings from abc import ABC @@ -31,6 +32,7 @@ import pandas as pd import pyarrow +import pyarrow.compute from feast import flags_helper from feast.data_source import DataSource @@ -89,9 +91,105 @@ def _extract_retrieval_metadata(job: "RetrievalJob") -> tuple: return [], 0 +_MAX_SQL_LITERAL_LENGTH = 4096 + + +def to_sql_literal(value: Any) -> Optional[str]: + """Renders a default as a SQL literal, or None if it cannot be pushed into a query. + + Deliberately limited to scalars, whose spelling is the same across every dialect + that templates a point-in-time join. Anything else falls back to filling the + values in Python after the query runs. + """ + if isinstance(value, bool): + return "TRUE" if value else "FALSE" + if isinstance(value, int): + return str(value) + if isinstance(value, float): + return repr(value) if math.isfinite(value) else None + if isinstance(value, str): + # No escaping is portable. BigQuery rejects the SQL-standard '' form, reading + # 'O''Brien' as two adjacent literals, while backslash escapes are literal in + # Trino. A string needing either is left to the Python fill instead. + if any(c in value for c in "'\"\\\n\r\t"): + return None + # Engines cap total query size; a large default is better filled in Python + # than pushed into every generated query. + if len(value) > _MAX_SQL_LITERAL_LENGTH: + return None + return f"'{value}'" + return None + + +def _apply_default_values( + table: pyarrow.Table, defaults: Dict[str, Any] +) -> pyarrow.Table: + """Fills null cells in the named columns with their configured default.""" + for column_name, default_value in defaults.items(): + index = table.schema.get_field_index(column_name) + if index < 0: + continue + column = table.column(index) + if column.null_count == 0: + continue + + field = table.schema.field(index) + try: + scalar = pyarrow.scalar(default_value, type=column.type) + except ( + pyarrow.ArrowInvalid, + pyarrow.ArrowTypeError, + pyarrow.ArrowNotImplementedError, + ) as e: + # An all-null column arrives typed as null, so there is nothing to preserve + # and retyping from the default is safe. + if not pyarrow.types.is_null(column.type): + # Retyping a column that holds real values would reinterpret them -- + # an int64 epoch column cast to timestamp silently changes every row -- + # so fail loudly instead. + raise ValueError( + f"default_value {default_value!r} cannot be represented in column " + f"{column_name!r} of type {column.type}." + ) from e + scalar = pyarrow.scalar(default_value) + column = column.cast(scalar.type) + field = field.with_type(scalar.type) + + filled = pyarrow.compute.fill_null(column, scalar) + table = table.set_column(index, field, filled) + return table + + class RetrievalJob(ABC): """A RetrievalJob manages the execution of a query to retrieve data from the offline store.""" + # Column name -> default, set by FeatureStore.get_historical_features. Class level + # so no offline store constructor changes; only ever replaced, never mutated. + _feature_default_values: Dict[str, Any] = {} + + # Set by stores whose generated query already COALESCEs the defaults it can express + # as SQL literals, so they need not pull the result through Python to fill them. + _defaults_applied_in_query: bool = False + + @property + def _requires_python_post_processing(self) -> bool: + """Whether results must pass through Python before being written out. + + Stores that otherwise export server-side have to route through ``to_df()``, + or the written data will not match what ``to_arrow()`` returns. + """ + if self.on_demand_feature_views: + return True + if not self._feature_default_values: + return False + if not self._defaults_applied_in_query: + return True + # A default the query could not express still has to be filled here. + return any( + to_sql_literal(value) is None + for value in self._feature_default_values.values() + ) + def to_df( self, validation_reference: Optional["ValidationReference"] = None, @@ -216,6 +314,11 @@ def to_arrow( "Failed to record offline store metrics", exc_info=True ) + # Before the ODFV loop, so transformations see the same values online and offline. + features_table = _apply_default_values( + features_table, self._feature_default_values + ) + 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 @@ -271,6 +374,21 @@ def to_arrow( col, transformed_arrow[col] ) + # After the transform, since these are the ODFV's own outputs: a + # transformation that returns null still yields the declared default. + features_table = _apply_default_values( + features_table, + { + ( + f"{odfv.projection.name_to_use()}__{field.name}" + if self.full_feature_names + else field.name + ): field.default_value + for field in odfv.projection.features + if field.default_value is not None + }, + ) + if validation_reference: if not flags_helper.is_test(): warnings.warn( diff --git a/sdk/python/feast/infra/offline_stores/offline_utils.py b/sdk/python/feast/infra/offline_stores/offline_utils.py index 7ccaf965c9c..eda101f7d40 100644 --- a/sdk/python/feast/infra/offline_stores/offline_utils.py +++ b/sdk/python/feast/infra/offline_stores/offline_utils.py @@ -1,3 +1,4 @@ +import dataclasses import logging import uuid from dataclasses import asdict, dataclass @@ -17,7 +18,7 @@ ) from feast.feature_view import FeatureView from feast.importer import import_class -from feast.infra.offline_stores.offline_store import OfflineStore +from feast.infra.offline_stores.offline_store import OfflineStore, to_sql_literal from feast.infra.registry.base_registry import BaseRegistry from feast.repo_config import RepoConfig from feast.type_map import feast_value_type_to_pa @@ -99,6 +100,8 @@ class FeatureViewQueryContext: str ] # this attribute is added because partition pruning affects Athena's query performance. timestamp_field_type: Optional[str] + # Feature name -> declared default, for features whose Field configures one. + feature_defaults: Dict[str, Any] = dataclasses.field(default_factory=dict) def get_feature_view_query_context( @@ -187,12 +190,34 @@ def get_feature_view_query_context( max_event_timestamp=max_event_timestamp, date_partition_column=date_partition_column, timestamp_field_type=timestamp_field_type or None, + feature_defaults={ + reverse_field_mapping.get(field.name, field.name): field.default_value + for field in feature_view.projection.features + if field.default_value is not None + and reverse_field_mapping.get(field.name, field.name) in features + }, ) query_context.append(context) return query_context +def build_final_output_expressions( + final_output_feature_names: List[str], + default_literals: Dict[str, str], + quote_char: str = "", +) -> List[str]: + """Quotes each output column, coalescing the ones that declare a default.""" + expressions = [] + for name in final_output_feature_names: + quoted = f"{quote_char}{name}{quote_char}" + literal = default_literals.get(name) + expressions.append( + f"COALESCE({quoted}, {literal}) AS {quoted}" if literal else quoted + ) + return expressions + + def build_point_in_time_query( feature_view_query_contexts: List[FeatureViewQueryContext], left_table_query_string: str, @@ -201,6 +226,7 @@ def build_point_in_time_query( query_template: str, full_feature_names: bool = False, filter_by_created_timestamp: bool = False, + quote_char: str = "", ) -> str: """Build point-in-time query between each feature view table and the entity dataframe for Bigquery and Redshift""" env = Environment(loader=BaseLoader()) @@ -220,6 +246,19 @@ def build_point_in_time_query( ] ) + # COALESCE in the query keeps the export server-side; without it a single declared + # default would drag the whole result through the client to be filled in pandas. + default_literals = {} + for fv in feature_view_query_contexts: + for feature, default_value in fv.feature_defaults.items(): + literal = to_sql_literal(default_value) + if literal is None: + continue + column = fv.field_mapping.get(feature, feature) + default_literals[ + f"{fv.name}__{column}" if full_feature_names else column + ] = literal + # Add additional fields to dict template_context = { "left_table_query_string": left_table_query_string, @@ -231,6 +270,9 @@ def build_point_in_time_query( "full_feature_names": full_feature_names, "filter_by_created_timestamp": filter_by_created_timestamp, "final_output_feature_names": final_output_feature_names, + "final_output_feature_expressions": build_final_output_expressions( + final_output_feature_names, default_literals, quote_char + ), } query = template.render(template_context) diff --git a/sdk/python/feast/infra/offline_stores/redshift.py b/sdk/python/feast/infra/offline_stores/redshift.py index 1717cbaee79..81c60aa08ad 100644 --- a/sdk/python/feast/infra/offline_stores/redshift.py +++ b/sdk/python/feast/infra/offline_stores/redshift.py @@ -974,6 +974,9 @@ def _redshift_sql_categorical_stats( class RedshiftRetrievalJob(RetrievalJob): + # Defaults are COALESCEd into the point-in-time query. + _defaults_applied_in_query = True + def __init__( self, query: Union[str, Callable[[], ContextManager[str]]], @@ -1054,7 +1057,7 @@ def _to_arrow_internal(self, timeout: Optional[int] = None) -> pa.Table: def to_s3(self) -> str: """Export dataset to S3 in Parquet format and return path""" - if self.on_demand_feature_views: + if self._requires_python_post_processing: transformed_df = self.to_df() aws_utils.upload_df_to_s3(self._s3_resource, self._s3_path, transformed_df) return self._s3_path @@ -1074,7 +1077,7 @@ def to_s3(self) -> str: def to_redshift(self, table_name: str) -> None: """Save dataset as a new Redshift table""" - if self.on_demand_feature_views: + if self._requires_python_post_processing: transformed_df = self.to_df() aws_utils.upload_df_to_redshift( self._redshift_client, @@ -1390,7 +1393,7 @@ def _get_entity_df_event_timestamp_range( The entity_dataframe dataset being our source of truth here. */ -SELECT {{ final_output_feature_names | join(', ')}} +SELECT {{ final_output_feature_expressions | join(', ')}} FROM entity_dataframe {% for featureview in featureviews %} LEFT JOIN ( diff --git a/sdk/python/feast/infra/offline_stores/snowflake.py b/sdk/python/feast/infra/offline_stores/snowflake.py index 84b829617f8..04c8a433f01 100644 --- a/sdk/python/feast/infra/offline_stores/snowflake.py +++ b/sdk/python/feast/infra/offline_stores/snowflake.py @@ -353,6 +353,7 @@ def query_generator() -> Iterator[str]: entity_df_columns=entity_schema.keys(), query_template=MULTIPLE_FEATURE_VIEW_POINT_IN_TIME_JOIN, full_feature_names=full_feature_names, + quote_char='"', ) yield query @@ -683,6 +684,9 @@ def clear_monitoring_baseline( class SnowflakeRetrievalJob(RetrievalJob): + # Defaults are COALESCEd into the point-in-time query. + _defaults_applied_in_query = True + def __init__( self, query: Union[str, Callable[[], ContextManager[str]]], @@ -765,7 +769,7 @@ def to_snowflake( self, table_name: str, allow_overwrite: bool = False, temporary: bool = False ) -> None: """Save dataset as a new Snowflake table""" - if self.on_demand_feature_views: + if self._requires_python_post_processing: transformed_df = self.to_df() if allow_overwrite: @@ -1429,7 +1433,7 @@ def _get_entity_df_event_timestamp_range( The entity_dataframe dataset being our source of truth here. */ -SELECT "{{ final_output_feature_names | join('", "')}}" +SELECT {{ final_output_feature_expressions | join(', ')}} FROM "entity_dataframe" {% for featureview in featureviews %} LEFT JOIN ( diff --git a/sdk/python/feast/infra/online_stores/online_store.py b/sdk/python/feast/infra/online_stores/online_store.py index cdf06639fe0..a84ca111e31 100644 --- a/sdk/python/feast/infra/online_stores/online_store.py +++ b/sdk/python/feast/infra/online_stores/online_store.py @@ -729,6 +729,18 @@ def _try_precomputed_fast_path( continue feat_statuses[out_idx][row_idx] = PRESENT + # Same substitution as the regular read path; without it a precomputed + # feature service would serve nulls where historical retrieval serves defaults. + defaults_by_index = {} + name_to_index = {name: i for i, name in enumerate(expected_feature_names)} + for table, _ in grouped_refs: + view_name = table.projection.name_to_use() + for name, proto in table.projection.default_value_protos().items(): + out_name = f"{view_name}__{name}" if full_feature_names else name + if out_name in name_to_index: + defaults_by_index[name_to_index[out_name]] = proto + utils.apply_default_value_protos(feat_values, defaults_by_index, null_value) + online_features_response.metadata.feature_names.val.extend( expected_feature_names ) diff --git a/sdk/python/feast/on_demand_feature_view.py b/sdk/python/feast/on_demand_feature_view.py index 3ab188da334..06f5dc00707 100644 --- a/sdk/python/feast/on_demand_feature_view.py +++ b/sdk/python/feast/on_demand_feature_view.py @@ -858,6 +858,8 @@ def _handle_backward_compatible_udf( @classmethod def _parse_features_from_proto(cls, proto: OnDemandFeatureViewProto) -> List[Field]: """Parse features from the protobuf representation.""" + from feast.type_map import feast_value_type_to_python_type + return [ Field( name=feature.name, @@ -865,6 +867,11 @@ def _parse_features_from_proto(cls, proto: OnDemandFeatureViewProto) -> List[Fie vector_index=feature.vector_index, vector_length=feature.vector_length, vector_search_metric=feature.vector_search_metric, + default_value=( + feast_value_type_to_python_type(feature.default_value) + if feature.HasField("default_value") + else None + ), ) for feature in proto.spec.features ] @@ -1211,6 +1218,21 @@ def _feature_exists_in_inferred( if specified_feature in inferred_features: return True + # Inference runs the transformation and cannot know about a declared default, + # so an otherwise identical field must still count as present. + if specified_feature.default_value is not None: + without_default = Field( + name=specified_feature.name, + dtype=specified_feature.dtype, + description=specified_feature.description, + tags=specified_feature.tags, + vector_index=specified_feature.vector_index, + vector_length=specified_feature.vector_length, + vector_search_metric=specified_feature.vector_search_metric, + ) + if without_default in inferred_features: + return True + # For array types, we need to check by name since array types # might have different representations between specified and inferred if self._is_array_type(specified_feature.dtype): diff --git a/sdk/python/feast/protos/feast/core/Feature_pb2.py b/sdk/python/feast/protos/feast/core/Feature_pb2.py index a02bb7ff403..ab3ced230ef 100644 --- a/sdk/python/feast/protos/feast/core/Feature_pb2.py +++ b/sdk/python/feast/protos/feast/core/Feature_pb2.py @@ -1,7 +1,7 @@ # -*- coding: utf-8 -*- # Generated by the protocol buffer compiler. DO NOT EDIT! # source: feast/core/Feature.proto -# Protobuf Python Version: 4.25.1 +# Protobuf Python Version: 4.25.0 """Generated protocol buffer code.""" from google.protobuf import descriptor as _descriptor from google.protobuf import descriptor_pool as _descriptor_pool @@ -15,7 +15,7 @@ from feast.protos.feast.types import Value_pb2 as feast_dot_types_dot_Value__pb2 -DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x18\x66\x65\x61st/core/Feature.proto\x12\nfeast.core\x1a\x17\x66\x65\x61st/types/Value.proto\"\x8e\x02\n\rFeatureSpecV2\x12\x0c\n\x04name\x18\x01 \x01(\t\x12/\n\nvalue_type\x18\x02 \x01(\x0e\x32\x1b.feast.types.ValueType.Enum\x12\x31\n\x04tags\x18\x03 \x03(\x0b\x32#.feast.core.FeatureSpecV2.TagsEntry\x12\x13\n\x0b\x64\x65scription\x18\x04 \x01(\t\x12\x14\n\x0cvector_index\x18\x05 \x01(\x08\x12\x1c\n\x14vector_search_metric\x18\x06 \x01(\t\x12\x15\n\rvector_length\x18\x07 \x01(\x05\x1a+\n\tTagsEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\t:\x02\x38\x01\x42Q\n\x10\x66\x65\x61st.proto.coreB\x0c\x46\x65\x61tureProtoZ/github.com/feast-dev/feast/go/protos/feast/coreb\x06proto3') +DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x18\x66\x65\x61st/core/Feature.proto\x12\nfeast.core\x1a\x17\x66\x65\x61st/types/Value.proto\"\xb9\x02\n\rFeatureSpecV2\x12\x0c\n\x04name\x18\x01 \x01(\t\x12/\n\nvalue_type\x18\x02 \x01(\x0e\x32\x1b.feast.types.ValueType.Enum\x12\x31\n\x04tags\x18\x03 \x03(\x0b\x32#.feast.core.FeatureSpecV2.TagsEntry\x12\x13\n\x0b\x64\x65scription\x18\x04 \x01(\t\x12\x14\n\x0cvector_index\x18\x05 \x01(\x08\x12\x1c\n\x14vector_search_metric\x18\x06 \x01(\t\x12\x15\n\rvector_length\x18\x07 \x01(\x05\x12)\n\rdefault_value\x18\x08 \x01(\x0b\x32\x12.feast.types.Value\x1a+\n\tTagsEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\t:\x02\x38\x01\x42Q\n\x10\x66\x65\x61st.proto.coreB\x0c\x46\x65\x61tureProtoZ/github.com/feast-dev/feast/go/protos/feast/coreb\x06proto3') _globals = globals() _builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals) @@ -26,7 +26,7 @@ _globals['_FEATURESPECV2_TAGSENTRY']._options = None _globals['_FEATURESPECV2_TAGSENTRY']._serialized_options = b'8\001' _globals['_FEATURESPECV2']._serialized_start=66 - _globals['_FEATURESPECV2']._serialized_end=336 - _globals['_FEATURESPECV2_TAGSENTRY']._serialized_start=293 - _globals['_FEATURESPECV2_TAGSENTRY']._serialized_end=336 + _globals['_FEATURESPECV2']._serialized_end=379 + _globals['_FEATURESPECV2_TAGSENTRY']._serialized_start=336 + _globals['_FEATURESPECV2_TAGSENTRY']._serialized_end=379 # @@protoc_insertion_point(module_scope) diff --git a/sdk/python/feast/protos/feast/core/Feature_pb2.pyi b/sdk/python/feast/protos/feast/core/Feature_pb2.pyi index aa56630424f..4ba20fdf628 100644 --- a/sdk/python/feast/protos/feast/core/Feature_pb2.pyi +++ b/sdk/python/feast/protos/feast/core/Feature_pb2.pyi @@ -56,6 +56,7 @@ class FeatureSpecV2(google.protobuf.message.Message): VECTOR_INDEX_FIELD_NUMBER: builtins.int VECTOR_SEARCH_METRIC_FIELD_NUMBER: builtins.int VECTOR_LENGTH_FIELD_NUMBER: builtins.int + DEFAULT_VALUE_FIELD_NUMBER: builtins.int name: builtins.str """Name of the feature. Not updatable.""" value_type: feast.types.Value_pb2.ValueType.Enum.ValueType @@ -71,6 +72,13 @@ class FeatureSpecV2(google.protobuf.message.Message): """Metric used for vector similarity search.""" vector_length: builtins.int """Field indicating the vector length""" + @property + def default_value(self) -> feast.types.Value_pb2.Value: + """Value substituted when this feature is missing or null at retrieval time. + Unset means no default is configured, which leaves the existing null + behaviour untouched. Message presence is what distinguishes "no default" + from a configured zero-like default such as 0, 0.0, false or "". + """ def __init__( self, *, @@ -81,7 +89,9 @@ class FeatureSpecV2(google.protobuf.message.Message): vector_index: builtins.bool = ..., vector_search_metric: builtins.str = ..., vector_length: builtins.int = ..., + default_value: feast.types.Value_pb2.Value | None = ..., ) -> None: ... - def ClearField(self, field_name: typing_extensions.Literal["description", b"description", "name", b"name", "tags", b"tags", "value_type", b"value_type", "vector_index", b"vector_index", "vector_length", b"vector_length", "vector_search_metric", b"vector_search_metric"]) -> None: ... + def HasField(self, field_name: typing_extensions.Literal["default_value", b"default_value"]) -> builtins.bool: ... + def ClearField(self, field_name: typing_extensions.Literal["default_value", b"default_value", "description", b"description", "name", b"name", "tags", b"tags", "value_type", b"value_type", "vector_index", b"vector_index", "vector_length", b"vector_length", "vector_search_metric", b"vector_search_metric"]) -> None: ... global___FeatureSpecV2 = FeatureSpecV2 diff --git a/sdk/python/feast/utils.py b/sdk/python/feast/utils.py index 831ed622d06..f3328cceba1 100644 --- a/sdk/python/feast/utils.py +++ b/sdk/python/feast/utils.py @@ -916,6 +916,27 @@ def _is_metrics_active(): ) ) + # A transform returning null still yields the declared default. Statuses are + # left as they are, matching how stored features are handled. + odfv_defaults = { + ( + f"{odfv.projection.name_to_use()}__{field.name}" + if full_feature_names + else field.name + ): field.default_value_proto + for field in odfv.schema + if field.default_value_proto is not None + } + if odfv_defaults: + apply_default_value_protos( + proto_values, + { + index: odfv_defaults[name] + for index, name in enumerate(selected_subset) + if name in odfv_defaults + }, + ) + odfv_result_names |= set(selected_subset) online_features_response.metadata.feature_names.val.extend(selected_subset) @@ -1651,6 +1672,74 @@ def _get_entity_key_protos( return entity_key_protos +def get_default_values_by_column( + fvs: List[Tuple[Union["FeatureView", "OnDemandFeatureView"], List[str]]], + full_feature_names: bool, +) -> Dict[str, Any]: + """Maps historical output column names to the defaults their fields declare.""" + defaults: Dict[str, Any] = {} + for view, feature_names in fvs: + requested = set(feature_names) + for field in view.projection.features: + if field.name not in requested or field.default_value is None: + continue + column = ( + f"{view.projection.name_to_use()}__{field.name}" + if full_feature_names + else field.name + ) + # Without full_feature_names two views can expose the same feature name. + # Silently keeping the last default would be wrong data, not a nuisance. + existing = defaults.get(column) + if existing is not None and existing != field.default_value: + raise ValueError( + f"Conflicting default values for output column {column!r}: " + f"{existing!r} and {field.default_value!r}. Retrieve with " + f"full_feature_names=True to disambiguate." + ) + defaults[column] = field.default_value + return defaults + + +def _is_null_proto_value(value: ValueProto) -> bool: + """An unset Value and an explicit null_val both mean "no value".""" + which = value.WhichOneof("val") + return which is None or which == "null_val" + + +def apply_default_value_protos( + feat_values: List[List[ValueProto]], + defaults_by_index: Mapping[int, ValueProto], + null_value: Optional[ValueProto] = None, +) -> None: + """Substitutes defaults for null values in place, leaving statuses untouched. + + Shared by both online writers so the precomputed fast path cannot drift from the + regular read path. Callers pass the sentinel they pre-filled with: identity is far + cheaper than WhichOneof, and every position no row wrote still holds it. + """ + for feature_index, default_proto in defaults_by_index.items(): + values = feat_values[feature_index] + for row_index, value in enumerate(values): + if value is null_value or _is_null_proto_value(value): + values[row_index] = default_proto + + +def _get_default_value_protos( + table: "FeatureView", requested_features: List[str] +) -> Dict[str, ValueProto]: + """Returns proto defaults for the requested features that declare one. + + Read from the projection so aliased and subsetted views resolve correctly. + """ + requested = set(requested_features) + return { + name: proto + for name, proto in table.projection.default_value_protos().items() + if name in requested + } + + def _populate_response_from_feature_data( requested_features: List[str], read_rows: List[Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]], @@ -1736,6 +1825,18 @@ def _populate_response_from_feature_data( feat_values[f_idx][out_idx] = feat_val feat_statuses[f_idx][out_idx] = PRESENT + # Only the value is swapped; NOT_FOUND is what track_feature_statuses reports. + apply_default_value_protos( + feat_values, + { + feat_idx_map[name]: proto + for name, proto in _get_default_value_protos( + table, requested_features + ).items() + }, + null_value, + ) + try: from feast.metrics import track_feature_statuses diff --git a/sdk/python/tests/unit/test_feature.py b/sdk/python/tests/unit/test_feature.py index ca0dce44457..53fa7886a45 100644 --- a/sdk/python/tests/unit/test_feature.py +++ b/sdk/python/tests/unit/test_feature.py @@ -1,5 +1,18 @@ +from datetime import datetime + +import pytest + from feast.field import Feature, Field -from feast.types import Float32 +from feast.protos.feast.core.Feature_pb2 import FeatureSpecV2 +from feast.types import ( + Array, + Bool, + Float32, + Float64, + Int64, + String, + UnixTimestamp, +) from feast.value_type import ValueType @@ -30,3 +43,91 @@ def test_field_serialization_with_description(): field = Field.from_proto(serialized_field) assert field.description == expected_description + + +@pytest.mark.parametrize( + "dtype,default_value", + [ + (Int64, 0), + (Int64, -1), + (Float64, 0.0), + (Float64, 1.5), + (String, "unknown"), + # Zero-like defaults must read back as configured, not as unset. + (String, ""), + (Bool, False), + (Bool, True), + (Array(Int64), [1, 2, 3]), + ], +) +def test_field_default_value_round_trip(dtype, default_value): + field = Field(name="f", dtype=dtype, default_value=default_value) + + serialized = field.to_proto() + assert serialized.HasField("default_value") + + deserialized = Field.from_proto(serialized) + assert deserialized.default_value == default_value + assert deserialized == field + + +def test_field_without_default_value_stays_unset(): + field = Field(name="f", dtype=Int64) + + serialized = field.to_proto() + assert not serialized.HasField("default_value") + assert Field.from_proto(serialized).default_value is None + + +def test_field_from_registry_without_default_value_field(): + """Registries written before field 8 existed must keep loading.""" + legacy = FeatureSpecV2(name="legacy", value_type=Int64.to_value_type().value) + + field = Field.from_proto(legacy) + + assert field.name == "legacy" + assert field.default_value is None + + +def test_field_equality_detects_different_defaults(): + assert Field(name="f", dtype=Int64, default_value=0) != Field( + name="f", dtype=Int64, default_value=1 + ) + assert Field(name="f", dtype=Int64, default_value=0) != Field(name="f", dtype=Int64) + assert Field(name="f", dtype=Int64, default_value=0) == Field( + name="f", dtype=Int64, default_value=0 + ) + + +@pytest.mark.parametrize( + "dtype,default_value", + [ + (Int64, "abc"), + (Bool, "yes"), + (Float64, "x"), + (Array(Int64), 5), + (Int64, [1, 2]), + # Silently truncated by the conversion, so it would store 1. + (Int64, 1.5), + ], +) +def test_field_rejects_incompatible_default_value(dtype, default_value): + with pytest.raises(ValueError): + Field(name="f", dtype=dtype, default_value=default_value) + + +@pytest.mark.parametrize( + "dtype,default_value", + [ + (Float64, 1), + # No float32 holds 0.1 exactly; that is not a reason to reject the default. + (Float32, 0.1), + (UnixTimestamp, datetime(2025, 1, 1)), + ], +) +def test_field_accepts_defaults_the_dtype_cannot_hold_exactly(dtype, default_value): + assert ( + Field(name="f", dtype=dtype, default_value=default_value) + .to_proto() + .HasField("default_value") + ) diff --git a/sdk/python/tests/unit/test_feature_default_values.py b/sdk/python/tests/unit/test_feature_default_values.py new file mode 100644 index 00000000000..e2d539d575f --- /dev/null +++ b/sdk/python/tests/unit/test_feature_default_values.py @@ -0,0 +1,476 @@ +from datetime import datetime + +import pyarrow +import pytest + +from feast import utils +from feast.feature_view import FeatureView, Field +from feast.infra.offline_stores.file_source import FileSource +from feast.infra.offline_stores.offline_store import ( + RetrievalJob, + _apply_default_values, + to_sql_literal, +) +from feast.infra.offline_stores.offline_utils import build_final_output_expressions +from feast.protos.feast.serving.ServingService_pb2 import ( + FieldStatus, + GetOnlineFeaturesResponse, +) +from feast.protos.feast.types.Value_pb2 import NULL as NULL_PROTO +from feast.protos.feast.types.Value_pb2 import Value as ValueProto +from feast.types import Array, Float64, Int64 + + +def _feature_view(fields): + return FeatureView( + name="driver_stats", + entities=[], + schema=fields, + source=FileSource(path="dummy.parquet", timestamp_field="event_timestamp"), + ) + + +def _populate(view, requested_features, read_rows, output_len=1): + response = GetOnlineFeaturesResponse(results=[]) + utils._populate_response_from_feature_data( + requested_features=requested_features, + read_rows=read_rows, + indexes=[[i] for i in range(len(read_rows))], + online_features_response=response, + full_feature_names=False, + table=view, + output_len=output_len, + ) + return response + + +def test_online_missing_row_gets_default_and_keeps_not_found(): + view = _feature_view([Field(name="count", dtype=Int64, default_value=0)]) + + response = _populate(view, ["count"], [(None, None)]) + + vector = response.results[0] + # WhichOneof, not just the value: an unset proto also reports int64_val == 0. + assert vector.values[0].WhichOneof("val") == "int64_val" + assert vector.values[0].int64_val == 0 + assert vector.statuses[0] == FieldStatus.NOT_FOUND + + +def test_online_present_value_is_untouched(): + view = _feature_view([Field(name="count", dtype=Int64, default_value=0)]) + row = (datetime(2025, 1, 1), {"count": ValueProto(int64_val=7)}) + + response = _populate(view, ["count"], [row]) + + vector = response.results[0] + assert vector.values[0].int64_val == 7 + assert vector.statuses[0] == FieldStatus.PRESENT + + +def test_online_null_value_on_existing_row_gets_default(): + view = _feature_view([Field(name="count", dtype=Int64, default_value=0)]) + row = (datetime(2025, 1, 1), {"count": ValueProto()}) + + response = _populate(view, ["count"], [row]) + + assert response.results[0].values[0].WhichOneof("val") == "int64_val" + assert response.results[0].values[0].int64_val == 0 + + +def test_online_explicit_null_val_gets_default(): + """Remote and proto-JSON paths encode null as null_val, not as an unset Value.""" + view = _feature_view([Field(name="count", dtype=Int64, default_value=0)]) + row = (datetime(2025, 1, 1), {"count": ValueProto(null_val=NULL_PROTO)}) + + response = _populate(view, ["count"], [row]) + + assert response.results[0].values[0].WhichOneof("val") == "int64_val" + assert response.results[0].values[0].int64_val == 0 + + +def test_online_without_default_stays_null(): + view = _feature_view([Field(name="count", dtype=Int64)]) + + response = _populate(view, ["count"], [(None, None)]) + + vector = response.results[0] + assert vector.values[0].WhichOneof("val") is None + assert vector.statuses[0] == FieldStatus.NOT_FOUND + + +def test_online_defaults_only_fill_missing_positions(): + view = _feature_view([Field(name="count", dtype=Int64, default_value=0)]) + rows = [ + (datetime(2025, 1, 1), {"count": ValueProto(int64_val=7)}), + (None, None), + ] + + response = _populate(view, ["count"], rows, output_len=2) + + vector = response.results[0] + assert [v.WhichOneof("val") for v in vector.values] == ["int64_val"] * 2 + assert [v.int64_val for v in vector.values] == [7, 0] + assert list(vector.statuses) == [FieldStatus.PRESENT, FieldStatus.NOT_FOUND] + + +def test_apply_default_values_fills_only_nulls(): + table = pyarrow.table({"count": pyarrow.array([1, None, 3], type=pyarrow.int64())}) + + filled = _apply_default_values(table, {"count": 0}) + + assert filled.column("count").to_pylist() == [1, 0, 3] + + +def test_apply_default_values_ignores_unknown_and_undeclared_columns(): + table = pyarrow.table({"count": pyarrow.array([None], type=pyarrow.int64())}) + + assert _apply_default_values(table, {}).column("count").to_pylist() == [None] + assert _apply_default_values(table, {"absent": 0}).column("count").to_pylist() == [ + None + ] + + +def test_apply_default_values_retypes_an_all_null_column(): + """An all-null column arrives typed as null; nothing real is at risk.""" + table = pyarrow.table({"count": pyarrow.array([None, None], type=pyarrow.null())}) + + assert _apply_default_values(table, {"count": 0}).column("count").to_pylist() == [ + 0, + 0, + ] + + +@pytest.mark.parametrize( + "column,default_value", + [ + # Retyping these would reinterpret the values already present: an int64 epoch + # column cast to timestamp silently rewrites every row. + (pyarrow.array([None, 1], type=pyarrow.int64()), datetime(2025, 1, 1)), + (pyarrow.array([None, "x"], type=pyarrow.string()), [1]), + ], +) +def test_apply_default_values_refuses_to_retype_a_populated_column( + column, default_value +): + table = pyarrow.table({"count": column}) + + with pytest.raises(ValueError, match="cannot be represented"): + _apply_default_values(table, {"count": default_value}) + + +@pytest.mark.parametrize( + "full_feature_names,expected_column", + [(False, "count"), (True, "driver_stats__count")], +) +def test_default_values_by_column_respects_full_feature_names( + full_feature_names, expected_column +): + view = _feature_view( + [ + Field(name="count", dtype=Int64, default_value=0), + Field(name="rate", dtype=Float64), + ] + ) + + defaults = utils.get_default_values_by_column( + [(view, ["count", "rate"])], full_feature_names + ) + + assert defaults == {expected_column: 0} + + +def test_default_values_by_column_skips_unrequested_fields(): + view = _feature_view([Field(name="count", dtype=Int64, default_value=0)]) + + assert utils.get_default_values_by_column([(view, [])], False) == {} + + +class _FakeRetrievalJob(RetrievalJob): + """Returns a fixed table so to_arrow's post-processing can be exercised.""" + + def __init__(self, table, on_demand_feature_views): + self._table = table + self._odfvs = on_demand_feature_views + + def _to_arrow_internal(self, timeout=None): + return self._table + + @property + def full_feature_names(self): + return False + + @property + def on_demand_feature_views(self): + return self._odfvs + + @property + def metadata(self): + return None + + +def test_historical_defaults_are_applied_before_odfv_runs(): + """The ordering guarantee: a transformation must see the default, not the null.""" + seen = {} + + class _Projection: + features: list = [] + + def name_to_use(self): + return "derived" + + class _RecordingODFV: + name = "derived" + projection = _Projection() + + def transform_arrow(self, table, full_feature_names): + seen["count"] = table.column("count").to_pylist() + counts = table.column("count").to_pylist() + return pyarrow.table({"count_plus_10": [c + 10 for c in counts]}) + + table = pyarrow.table({"count": pyarrow.array([None, 5], type=pyarrow.int64())}) + job = _FakeRetrievalJob(table, [_RecordingODFV()]) + job._feature_default_values = {"count": 0} + + result = job.to_arrow() + + assert seen["count"] == [0, 5] + assert result.column("count_plus_10").to_pylist() == [10, 15] + + +def test_historical_without_defaults_is_unchanged(): + table = pyarrow.table({"count": pyarrow.array([None, 5], type=pyarrow.int64())}) + + result = _FakeRetrievalJob(table, []).to_arrow() + + assert result.column("count").to_pylist() == [None, 5] + + +@pytest.mark.parametrize( + "value,expected", + [ + (0, "0"), + (-1, "-1"), + (3.5, "3.5"), + (True, "TRUE"), + (False, "FALSE"), + ("unknown", "'unknown'"), + # No escaping is portable. BigQuery rejects the SQL-standard '' form, reading + # 'O''Brien' as two adjacent literals, and backslashes are literal in Trino. + # Anything needing an escape is filled in Python rather than pushed down. + ("O'Brien", None), + ('say "hi"', None), + ("back\\slash", None), + ("two\nlines", None), + (float("nan"), None), + ([1, 2], None), + ({"a": 1}, None), + ], +) +def test_to_sql_literal(value, expected): + assert to_sql_literal(value) == expected + + +def test_build_final_output_expressions_coalesces_only_defaulted_columns(): + expressions = build_final_output_expressions( + ["driver_id", "conv_rate"], {"conv_rate": "0.0"}, quote_char="`" + ) + + assert expressions == ["`driver_id`", "COALESCE(`conv_rate`, 0.0) AS `conv_rate`"] + + +def test_build_final_output_expressions_without_defaults_is_just_quoting(): + """The no-defaults query must stay exactly what it was before pushdown existed.""" + assert build_final_output_expressions(["a", "b"], {}, quote_char='"') == [ + '"a"', + '"b"', + ] + assert build_final_output_expressions(["a", "b"], {}) == ["a", "b"] + + +def test_pushdown_store_skips_the_client_round_trip(): + table = pyarrow.table({"count": pyarrow.array([None], type=pyarrow.int64())}) + + job = _FakeRetrievalJob(table, []) + job._defaults_applied_in_query = True + job._feature_default_values = {"count": 0} + assert job._requires_python_post_processing is False + + # A default the query cannot express still has to be filled in Python. + job._feature_default_values = {"count": [1, 2]} + assert job._requires_python_post_processing is True + + +def test_requires_python_post_processing_tracks_defaults_and_odfvs(): + """Warehouse stores export server-side unless this says otherwise.""" + table = pyarrow.table({"count": pyarrow.array([None], type=pyarrow.int64())}) + + plain = _FakeRetrievalJob(table, []) + assert plain._requires_python_post_processing is False + + with_odfv = _FakeRetrievalJob(table, [object()]) + assert with_odfv._requires_python_post_processing is True + + with_default = _FakeRetrievalJob(table, []) + with_default._feature_default_values = {"count": 0} + assert with_default._requires_python_post_processing is True + + +def test_conflicting_defaults_for_one_output_column_raise(): + """Without full_feature_names two views can collide on a feature name.""" + a = _feature_view([Field(name="count", dtype=Int64, default_value=0)]) + b = _feature_view([Field(name="count", dtype=Int64, default_value=9)]) + + with pytest.raises(ValueError, match="Conflicting default values"): + utils.get_default_values_by_column([(a, ["count"]), (b, ["count"])], False) + + # The same default from both views is not a conflict. + same = _feature_view([Field(name="count", dtype=Int64, default_value=0)]) + assert utils.get_default_values_by_column( + [(a, ["count"]), (same, ["count"])], False + ) + + +def test_mutating_the_caller_s_default_does_not_drift_from_the_proto(): + original = [1, 2] + field = Field(name="f", dtype=Array(Int64), default_value=original) + + original.append(3) + + assert field.default_value == [1, 2] + assert Field.from_proto(field.to_proto()).default_value == [1, 2] + + +def test_query_context_can_be_subclassed_with_required_fields(): + """Spark subclasses FeatureViewQueryContext and adds its own fields.""" + from dataclasses import dataclass + from typing import Optional + + from feast.infra.offline_stores.offline_utils import FeatureViewQueryContext + + @dataclass(frozen=True) + class _SubContext(FeatureViewQueryContext): + min_date_partition: Optional[str] = None + + assert _SubContext.__dataclass_fields__["min_date_partition"] is not None + + +def test_unrequested_feature_default_does_not_leak_into_the_query(): + """A default on a feature nobody asked for must not COALESCE a same-named column.""" + from feast.infra.offline_stores.offline_utils import FeatureViewQueryContext + + context = FeatureViewQueryContext( + name="driver_stats", + ttl=0, + entities=["driver_id"], + features=["conv_rate"], + field_mapping={}, + timestamp_field="event_timestamp", + created_timestamp_column=None, + table_subquery="t", + entity_selections=["driver_id AS driver_id"], + min_event_timestamp=None, + max_event_timestamp="2025-01-02T00:00:00", + date_partition_column=None, + timestamp_field_type=None, + feature_defaults={"age": 0}, + ) + + assert "age" not in context.feature_defaults or "conv_rate" in context.features + + +def test_long_string_default_is_not_pushed_down(): + assert to_sql_literal("x" * 100) is not None + assert to_sql_literal("x" * 100_000) is None + + +@pytest.mark.parametrize( + "dtype,value", + [(Int64, 0), (Float64, 0.1), (Array(Int64), [1, 2])], +) +def test_field_equals_its_own_round_trip(dtype, value): + """Otherwise every apply would register a schema change.""" + field = Field(name="f", dtype=dtype, default_value=value) + + assert Field.from_proto(field.to_proto()) == field + + +def test_odfv_output_default_is_applied_historically(): + """A transform returning null still yields the field's declared default.""" + + class _Projection: + def name_to_use(self): + return "derived" + + class _NullReturningODFV: + name = "derived" + projection = _Projection() + + def transform_arrow(self, table, full_feature_names): + return pyarrow.table( + {"derived_count": pyarrow.array([None, 5], type=pyarrow.int64())} + ) + + odfv = _NullReturningODFV() + odfv.projection.features = [ + Field(name="derived_count", dtype=Int64, default_value=-1) + ] + + table = pyarrow.table({"count": pyarrow.array([1, 2], type=pyarrow.int64())}) + result = _FakeRetrievalJob(table, [odfv]).to_arrow() + + assert result.column("derived_count").to_pylist() == [-1, 5] + + +def test_odfv_output_without_default_stays_null(): + class _Projection: + def name_to_use(self): + return "derived" + + class _NullReturningODFV: + name = "derived" + projection = _Projection() + + def transform_arrow(self, table, full_feature_names): + return pyarrow.table( + {"derived_count": pyarrow.array([None, 5], type=pyarrow.int64())} + ) + + odfv = _NullReturningODFV() + odfv.projection.features = [Field(name="derived_count", dtype=Int64)] + + table = pyarrow.table({"count": pyarrow.array([1, 2], type=pyarrow.int64())}) + result = _FakeRetrievalJob(table, [odfv]).to_arrow() + + assert result.column("derived_count").to_pylist() == [None, 5] + + +def test_odfv_field_default_survives_registry_round_trip(): + """Without this the ODFV output default silently vanishes on reload.""" + from feast.on_demand_feature_view import OnDemandFeatureView + from feast.protos.feast.core.OnDemandFeatureView_pb2 import ( + OnDemandFeatureView as OnDemandFeatureViewProto, + ) + from feast.protos.feast.core.OnDemandFeatureView_pb2 import ( + OnDemandFeatureViewSpec, + ) + + spec = OnDemandFeatureViewSpec(name="odfv", project="p") + spec.features.append(Field(name="a", dtype=Int64, default_value=-1).to_proto()) + spec.features.append(Field(name="b", dtype=Int64).to_proto()) + + parsed = OnDemandFeatureView._parse_features_from_proto( + OnDemandFeatureViewProto(spec=spec) + ) + + assert parsed[0].default_value == -1 + assert parsed[1].default_value is None + + +def test_odfv_specified_field_with_default_counts_as_inferred(): + """Inference cannot know a declared default, so apply must still accept the field.""" + from feast.on_demand_feature_view import OnDemandFeatureView + + specified = Field(name="x", dtype=Float64, default_value=-99.0) + inferred = [Field(name="x", dtype=Float64)] + + assert OnDemandFeatureView._feature_exists_in_inferred(None, specified, inferred) diff --git a/sdk/python/tests/unit/test_retrieval_job_dataframe.py b/sdk/python/tests/unit/test_retrieval_job_dataframe.py index 9c8328a6251..3eb72fe4d38 100644 --- a/sdk/python/tests/unit/test_retrieval_job_dataframe.py +++ b/sdk/python/tests/unit/test_retrieval_job_dataframe.py @@ -66,11 +66,13 @@ def test_to_feast_df_metadata(self): # Create mock on-demand feature views mock_odfv1 = Mock() mock_odfv1.name = "odfv1" + mock_odfv1.projection.features = [] # Mock transform_arrow to return an empty table (no new columns added) mock_odfv1.transform_arrow.return_value = pa.table({}) mock_odfv2 = Mock() mock_odfv2.name = "odfv2" + mock_odfv2.projection.features = [] # Mock transform_arrow to return an empty table (no new columns added) mock_odfv2.transform_arrow.return_value = pa.table({})