fix: Request data api update (feast-dev#2488)

* Update request data source

Signed-off-by: Kevin Zhang <kzhang@tecton.ai>

* Fix

Signed-off-by: Kevin Zhang <kzhang@tecton.ai>

* Fix

Signed-off-by: Kevin Zhang <kzhang@tecton.ai>

* Fix

Signed-off-by: Kevin Zhang <kzhang@tecton.ai>

* Data source update

Signed-off-by: Author <kzhang@tecton.ai>
Signed-off-by: Kevin Zhang <kzhang@tecton.ai>

* Data source

Signed-off-by: Kevin Zhang <kzhang@tecton.ai>

Signed-off-by: Author <kzhang@tecton.ai>
Signed-off-by: Kevin Zhang <kzhang@tecton.ai>

* Temp fix

Signed-off-by: Kevin Zhang <kzhang@tecton.ai>

* Fix

Signed-off-by: Kevin Zhang <kzhang@tecton.ai>

* Update internal var names

Signed-off-by: Kevin Zhang <kzhang@tecton.ai>

* Add test

Signed-off-by: Kevin Zhang <kzhang@tecton.ai>

* Fix

Signed-off-by: Kevin Zhang <kzhang@tecton.ai>

* Fix

Signed-off-by: Kevin Zhang <kzhang@tecton.ai>

* Fix

Signed-off-by: Kevin Zhang <kzhang@tecton.ai>

* Fix

Signed-off-by: Kevin Zhang <kzhang@tecton.ai>

docs/getting-started/concepts/feature-view.md

@@ -130,9 +130,9 @@ Feature names must be unique within a [feature view](feature-view.md#feature-vie
 On demand feature views allows users to use existing features and request time data (features only available at request time) to transform and create new features. Users define python transformation logic which is executed in both historical retrieval and online retrieval paths:
 
 ```python
-# Define a request data source which encodes features / information only 
+# Define a request data source which encodes features / information only
 # available at request time (e.g. part of the user initiated HTTP request)
-input_request = RequestDataSource(
+input_request = RequestSource(
     name="vals_to_add",
     schema={
         "val_to_add": ValueType.INT64,

examples/java-demo/feature_repo/driver_repo.py

@@ -1,6 +1,6 @@
 import pandas as pd
 from feast import Entity, Feature, FeatureView, FileSource, ValueType
-from feast.data_source import RequestDataSource
+from feast.data_source import RequestSource
 from feast.on_demand_feature_view import on_demand_feature_view
 from feast.request_feature_view import RequestFeatureView
 from google.protobuf.duration_pb2 import Duration
@@ -28,13 +28,13 @@
 
 # Define a request data source which encodes features / information only
 # available at request time (e.g. part of the user initiated HTTP request)
-input_request = RequestDataSource(
+input_request = RequestSource(
     name="vals_to_add",
     schema={"val_to_add": ValueType.INT64, "val_to_add_2": ValueType.INT64},
 )
 
 # Define an on demand feature view which can generate new features based on
-# existing feature views and RequestDataSource features
+# existing feature views and RequestSource features
 @on_demand_feature_view(
     inputs={
         "driver_hourly_stats": driver_hourly_stats_view,
@@ -55,7 +55,7 @@ def transformed_conv_rate(inputs: pd.DataFrame) -> pd.DataFrame:
 # Define request feature view
 driver_age_request_fv = RequestFeatureView(
     name="driver_age",
-    request_data_source=RequestDataSource(
+    request_source=RequestSource(
         name="driver_age", schema={"driver_age": ValueType.INT64,}
     ),
 )

sdk/python/feast/data_source.py

@@ -12,7 +12,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-
 import enum
 import warnings
 from abc import ABC, abstractmethod
@@ -144,7 +143,7 @@ def to_proto(self) -> DataSourceProto.KinesisOptions:
     DataSourceProto.SourceType.BATCH_SNOWFLAKE: "feast.infra.offline_stores.snowflake_source.SnowflakeSource",
     DataSourceProto.SourceType.STREAM_KAFKA: "feast.data_source.KafkaSource",
     DataSourceProto.SourceType.STREAM_KINESIS: "feast.data_source.KinesisSource",
-    DataSourceProto.SourceType.REQUEST_SOURCE: "feast.data_source.RequestDataSource",
+    DataSourceProto.SourceType.REQUEST_SOURCE: "feast.data_source.RequestSource",
     DataSourceProto.SourceType.PUSH_SOURCE: "feast.data_source.PushSource",
 }
 
@@ -422,9 +421,9 @@ def get_table_query_string(self) -> str:
         raise NotImplementedError
 
 
-class RequestDataSource(DataSource):
+class RequestSource(DataSource):
     """
-    RequestDataSource that can be used to provide input features for on demand transforms
+    RequestSource that can be used to provide input features for on demand transforms
 
     Args:
         name: Name of the request data source
@@ -446,7 +445,7 @@ def __init__(
         tags: Optional[Dict[str, str]] = None,
         owner: Optional[str] = "",
     ):
-        """Creates a RequestDataSource object."""
+        """Creates a RequestSource object."""
         super().__init__(name=name, description=description, tags=tags, owner=owner)
         self.schema = schema
 
@@ -464,7 +463,7 @@ def from_proto(data_source: DataSourceProto):
         schema = {}
         for key, val in schema_pb.items():
             schema[key] = ValueType(val)
-        return RequestDataSource(
+        return RequestSource(
             name=data_source.name,
             schema=schema,
             description=data_source.description,
@@ -496,6 +495,15 @@ def source_datatype_to_feast_value_type() -> Callable[[str], ValueType]:
         raise NotImplementedError
 
 
+class RequestDataSource(RequestSource):
+    def __init__(self, *args, **kwargs):
+        warnings.warn(
+            "The 'RequestDataSource' class is deprecated and was renamed to RequestSource. Please use RequestSource instead. This class name will be removed in Feast 0.23.",
+            DeprecationWarning,
+        )
+        super().__init__(*args, **kwargs)
+
+
 class KinesisSource(DataSource):
     def validate(self, config: RepoConfig):
         pass

sdk/python/feast/feature_store.py

@@ -679,7 +679,7 @@ def apply(
             data_sources_set_to_update.add(rfv.request_data_source)
 
         for odfv in odfvs_to_update:
-            for v in odfv.source_request_data_sources.values():
+            for v in odfv.source_request_sources.values():
                 data_sources_set_to_update.add(v)
 
         data_sources_to_update = list(data_sources_set_to_update)

sdk/python/feast/inference.py

@@ -9,7 +9,7 @@
     RedshiftSource,
     SnowflakeSource,
 )
-from feast.data_source import DataSource, RequestDataSource
+from feast.data_source import DataSource, RequestSource
 from feast.errors import RegistryInferenceFailure
 from feast.feature_view import FeatureView
 from feast.repo_config import RepoConfig
@@ -78,7 +78,7 @@ def update_data_sources_with_inferred_event_timestamp_col(
     ERROR_MSG_PREFIX = "Unable to infer DataSource event_timestamp_column"
 
     for data_source in data_sources:
-        if isinstance(data_source, RequestDataSource):
+        if isinstance(data_source, RequestSource):
             continue
         if (
             data_source.event_timestamp_column is None

sdk/python/feast/on_demand_feature_view.py

@@ -8,7 +8,7 @@
 import pandas as pd
 
 from feast.base_feature_view import BaseFeatureView
-from feast.data_source import RequestDataSource
+from feast.data_source import RequestSource
 from feast.errors import RegistryInferenceFailure, SpecifiedFeaturesNotPresentError
 from feast.feature import Feature
 from feast.feature_view import FeatureView
@@ -45,8 +45,8 @@ class OnDemandFeatureView(BaseFeatureView):
         features: The list of features in the output of the on demand feature view.
         source_feature_view_projections: A map from input source names to actual input
             sources with type FeatureViewProjection.
-        source_request_data_sources: A map from input source names to the actual input
-            sources with type RequestDataSource.
+        source_request_sources: A map from input source names to the actual input
+            sources with type RequestSource.
         udf: The user defined transformation function, which must take pandas dataframes
             as inputs.
         description: A human-readable description.
@@ -59,7 +59,7 @@ class OnDemandFeatureView(BaseFeatureView):
     name: str
     features: List[Feature]
     source_feature_view_projections: Dict[str, FeatureViewProjection]
-    source_request_data_sources: Dict[str, RequestDataSource]
+    source_request_sources: Dict[str, RequestSource]
     udf: MethodType
     description: str
     tags: Dict[str, str]
@@ -71,11 +71,11 @@ def __init__(
         name: str,
         features: List[Feature],
         sources: Optional[
-            Dict[str, Union[FeatureView, FeatureViewProjection, RequestDataSource]]
+            Dict[str, Union[FeatureView, FeatureViewProjection, RequestSource]]
         ] = None,
         udf: Optional[MethodType] = None,
         inputs: Optional[
-            Dict[str, Union[FeatureView, FeatureViewProjection, RequestDataSource]]
+            Dict[str, Union[FeatureView, FeatureViewProjection, RequestSource]]
         ] = None,
         description: str = "",
         tags: Optional[Dict[str, str]] = None,
@@ -124,10 +124,10 @@ def __init__(
 
         assert sources is not None
         self.source_feature_view_projections: Dict[str, FeatureViewProjection] = {}
-        self.source_request_data_sources: Dict[str, RequestDataSource] = {}
+        self.source_request_sources: Dict[str, RequestSource] = {}
         for source_name, odfv_source in sources.items():
-            if isinstance(odfv_source, RequestDataSource):
-                self.source_request_data_sources[source_name] = odfv_source
+            if isinstance(odfv_source, RequestSource):
+                self.source_request_sources[source_name] = odfv_source
             elif isinstance(odfv_source, FeatureViewProjection):
                 self.source_feature_view_projections[source_name] = odfv_source
             else:
@@ -149,8 +149,7 @@ def __copy__(self):
             name=self.name,
             features=self.features,
             sources=dict(
-                **self.source_feature_view_projections,
-                **self.source_request_data_sources,
+                **self.source_feature_view_projections, **self.source_request_sources,
             ),
             udf=self.udf,
             description=self.description,
@@ -167,7 +166,7 @@ def __eq__(self, other):
         if (
             not self.source_feature_view_projections
             == other.source_feature_view_projections
-            or not self.source_request_data_sources == other.source_request_data_sources
+            or not self.source_request_sources == other.source_request_sources
             or not self.udf.__code__.co_code == other.udf.__code__.co_code
         ):
             return False
@@ -194,12 +193,9 @@ def to_proto(self) -> OnDemandFeatureViewProto:
             sources[source_name] = OnDemandSource(
                 feature_view_projection=fv_projection.to_proto()
             )
-        for (
-            source_name,
-            request_data_source,
-        ) in self.source_request_data_sources.items():
+        for (source_name, request_sources,) in self.source_request_sources.items():
             sources[source_name] = OnDemandSource(
-                request_data_source=request_data_source.to_proto()
+                request_data_source=request_sources.to_proto()
             )
 
         spec = OnDemandFeatureViewSpec(
@@ -241,7 +237,7 @@ def from_proto(cls, on_demand_feature_view_proto: OnDemandFeatureViewProto):
                     on_demand_source.feature_view_projection
                 )
             else:
-                sources[source_name] = RequestDataSource.from_proto(
+                sources[source_name] = RequestSource.from_proto(
                     on_demand_source.request_data_source
                 )
         on_demand_feature_view_obj = cls(
@@ -282,8 +278,8 @@ def from_proto(cls, on_demand_feature_view_proto: OnDemandFeatureViewProto):
 
     def get_request_data_schema(self) -> Dict[str, ValueType]:
         schema: Dict[str, ValueType] = {}
-        for request_data_source in self.source_request_data_sources.values():
-            schema.update(request_data_source.schema)
+        for request_source in self.source_request_sources.values():
+            schema.update(request_source.schema)
         return schema
 
     def get_transformed_features_df(
@@ -339,7 +335,7 @@ def infer_features(self):
                     dtype=dtype
                 )
                 df[f"{feature.name}"] = pd.Series(dtype=dtype)
-        for request_data in self.source_request_data_sources.values():
+        for request_data in self.source_request_sources.values():
             for feature_name, feature_type in request_data.schema.items():
                 dtype = feast_value_type_to_pandas_type(feature_type)
                 df[f"{feature_name}"] = pd.Series(dtype=dtype)
@@ -385,7 +381,7 @@ def get_requested_odfvs(feature_refs, project, registry):
 
 
 def on_demand_feature_view(
-    features: List[Feature], sources: Dict[str, Union[FeatureView, RequestDataSource]]
+    features: List[Feature], sources: Dict[str, Union[FeatureView, RequestSource]]
 ):
     """
     Declare an on-demand feature view

sdk/python/feast/request_feature_view.py

@@ -3,7 +3,7 @@
 from typing import Dict, List, Optional, Type
 
 from feast.base_feature_view import BaseFeatureView
-from feast.data_source import RequestDataSource
+from feast.data_source import RequestSource
 from feast.feature import Feature
 from feast.feature_view_projection import FeatureViewProjection
 from feast.protos.feast.core.RequestFeatureView_pb2 import (
@@ -20,7 +20,7 @@ class RequestFeatureView(BaseFeatureView):
 
     Attributes:
         name: The unique name of the request feature view.
-        request_data_source: The request data source that specifies the schema and
+        request_source: The request source that specifies the schema and
             features of the request feature view.
         features: The list of features defined as part of this request feature view.
         description: A human-readable description.
@@ -30,7 +30,7 @@ class RequestFeatureView(BaseFeatureView):
     """
 
     name: str
-    request_data_source: RequestDataSource
+    request_source: RequestSource
     features: List[Feature]
     description: str
     tags: Dict[str, str]
@@ -40,7 +40,7 @@ class RequestFeatureView(BaseFeatureView):
     def __init__(
         self,
         name: str,
-        request_data_source: RequestDataSource,
+        request_data_source: RequestSource,
         description: str = "",
         tags: Optional[Dict[str, str]] = None,
         owner: str = "",
@@ -73,7 +73,7 @@ def __init__(
             tags=tags,
             owner=owner,
         )
-        self.request_data_source = request_data_source
+        self.request_source = request_data_source
 
     @property
     def proto_class(self) -> Type[RequestFeatureViewProto]:
@@ -88,7 +88,7 @@ def to_proto(self) -> RequestFeatureViewProto:
         """
         spec = RequestFeatureViewSpec(
             name=self.name,
-            request_data_source=self.request_data_source.to_proto(),
+            request_data_source=self.request_source.to_proto(),
             description=self.description,
             tags=self.tags,
             owner=self.owner,
@@ -110,7 +110,7 @@ def from_proto(cls, request_feature_view_proto: RequestFeatureViewProto):
 
         request_feature_view_obj = cls(
             name=request_feature_view_proto.spec.name,
-            request_data_source=RequestDataSource.from_proto(
+            request_data_source=RequestSource.from_proto(
                 request_feature_view_proto.spec.request_data_source
             ),
             description=request_feature_view_proto.spec.description,
@@ -127,8 +127,6 @@ def from_proto(cls, request_feature_view_proto: RequestFeatureViewProto):
         return request_feature_view_obj
 
     def __copy__(self):
-        fv = RequestFeatureView(
-            name=self.name, request_data_source=self.request_data_source
-        )
+        fv = RequestFeatureView(name=self.name, request_data_source=self.request_source)
         fv.projection = copy.copy(self.projection)
         return fv

sdk/python/tests/integration/feature_repos/repo_configuration.py

@@ -35,7 +35,7 @@
 )
 from tests.integration.feature_repos.universal.feature_views import (
     conv_rate_plus_100_feature_view,
-    create_conv_rate_request_data_source,
+    create_conv_rate_request_source,
     create_customer_daily_profile_feature_view,
     create_driver_hourly_stats_feature_view,
     create_field_mapping_feature_view,
@@ -279,7 +279,7 @@ def construct_universal_feature_views(
         driver_odfv=conv_rate_plus_100_feature_view(
             {
                 "driver": driver_hourly_stats,
-                "input_request": create_conv_rate_request_data_source(),
+                "input_request": create_conv_rate_request_source(),
             }
         )
         if with_odfv