feat: Adding write capabability to online store to on demand feature …

sdk/python/feast/feature_store.py

@@ -608,7 +608,12 @@ def _make_inferences(
         update_feature_views_with_inferred_features_and_entities(
             sfvs_to_update, entities + entities_to_update, self.config
         )
-        # TODO(kevjumba): Update schema inferrence
+        # We need to attach the time stamp fields to the underlying data sources
+        # and cascade the dependencies
+        update_feature_views_with_inferred_features_and_entities(
+            odfvs_to_update, entities + entities_to_update, self.config
+        )
+        # TODO(kevjumba): Update schema inference
         for sfv in sfvs_to_update:
             if not sfv.schema:
                 raise ValueError(
@@ -618,8 +623,13 @@ def _make_inferences(
         for odfv in odfvs_to_update:
             odfv.infer_features()
 
+        odfvs_to_write = [
+            odfv for odfv in odfvs_to_update if odfv.write_to_online_store
+        ]
+        # Update to include ODFVs with write to online store
         fvs_to_update_map = {
-            view.name: view for view in [*views_to_update, *sfvs_to_update]
+            view.name: view
+            for view in [*views_to_update, *sfvs_to_update, *odfvs_to_write]
         }
         for feature_service in feature_services_to_update:
             feature_service.infer_features(fvs_to_update=fvs_to_update_map)
@@ -847,6 +857,11 @@ def apply(
         ]
         sfvs_to_update = [ob for ob in objects if isinstance(ob, StreamFeatureView)]
         odfvs_to_update = [ob for ob in objects if isinstance(ob, OnDemandFeatureView)]
+        odfvs_with_writes_to_update = [
+            ob
+            for ob in objects
+            if isinstance(ob, OnDemandFeatureView) and ob.write_to_online_store
+        ]
         services_to_update = [ob for ob in objects if isinstance(ob, FeatureService)]
         data_sources_set_to_update = {
             ob for ob in objects if isinstance(ob, DataSource)
@@ -868,10 +883,22 @@ def apply(
         for batch_source in batch_sources_to_add:
             data_sources_set_to_update.add(batch_source)
 
-        for fv in itertools.chain(views_to_update, sfvs_to_update):
-            data_sources_set_to_update.add(fv.batch_source)
-            if fv.stream_source:
-                data_sources_set_to_update.add(fv.stream_source)
+        for fv in itertools.chain(
+            views_to_update, sfvs_to_update, odfvs_with_writes_to_update
+        ):
+            if isinstance(fv, FeatureView):
+                data_sources_set_to_update.add(fv.batch_source)
+            if isinstance(fv, StreamFeatureView):
+                if fv.stream_source:
+                    data_sources_set_to_update.add(fv.stream_source)
+            if isinstance(fv, OnDemandFeatureView):
+                for source_fvp in fv.source_feature_view_projections:
+                    if fv.source_feature_view_projections[source_fvp].batch_source:
+                        data_sources_set_to_update.add(
+                            fv.source_feature_view_projections[source_fvp].batch_source
+                        )
+            else:
+                pass
 
         for odfv in odfvs_to_update:
             for v in odfv.source_request_sources.values():
@@ -884,7 +911,9 @@ def apply(
 
         # Validate all feature views and make inferences.
         self._validate_all_feature_views(
-            views_to_update, odfvs_to_update, sfvs_to_update
+            views_to_update,
+            odfvs_to_update,
+            sfvs_to_update,
         )
         self._make_inferences(
             data_sources_to_update,
@@ -989,7 +1018,9 @@ def apply(
         tables_to_delete: List[FeatureView] = (
             views_to_delete + sfvs_to_delete if not partial else []  # type: ignore
         )
-        tables_to_keep: List[FeatureView] = views_to_update + sfvs_to_update  # type: ignore
+        tables_to_keep: List[FeatureView] = (
+            views_to_update + sfvs_to_update + odfvs_with_writes_to_update
+        )  # type: ignore
 
         self._get_provider().update_infra(
             project=self.project,
@@ -1444,19 +1475,18 @@ def write_to_online_store(
             inputs: Optional the dictionary object to be written
             allow_registry_cache (optional): Whether to allow retrieving feature views from a cached registry.
         """
-        # TODO: restrict this to work with online StreamFeatureViews and validate the FeatureView type
+        feature_view_dict = {
+            fv_proto.name: fv_proto
+            for fv_proto in self.list_all_feature_views(allow_registry_cache)
+        }
         try:
-            feature_view: FeatureView = self.get_stream_feature_view(
-                feature_view_name, allow_registry_cache=allow_registry_cache
-            )
+            feature_view = feature_view_dict[feature_view_name]
         except FeatureViewNotFoundException:
-            feature_view = self.get_feature_view(
-                feature_view_name, allow_registry_cache=allow_registry_cache
-            )
+            raise FeatureViewNotFoundException(feature_view_name, self.project)
         if df is not None and inputs is not None:
             raise ValueError("Both df and inputs cannot be provided at the same time.")
         if df is None and inputs is not None:
-            if isinstance(inputs, dict):
+            if isinstance(inputs, dict) or isinstance(inputs, List):
                 try:
                     df = pd.DataFrame(inputs)
                 except Exception as _:
@@ -1465,8 +1495,20 @@ def write_to_online_store(
                 pass
             else:
                 raise ValueError("inputs must be a dictionary or a pandas DataFrame.")
+        if df is not None and inputs is None:
+            if isinstance(df, dict) or isinstance(df, List):
+                try:
+                    df = pd.DataFrame(df)
+                except Exception as _:
+                    raise DataFrameSerializationError
+
         provider = self._get_provider()
-        provider.ingest_df(feature_view, df)
+        if isinstance(feature_view, OnDemandFeatureView):
+            # TODO: add projection mapping
+            projection_mapping = {}
+            provider.ingest_df(feature_view, df, projection_mapping)
+        else:
+            provider.ingest_df(feature_view, df)
 
     def write_to_offline_store(
         self,

sdk/python/feast/inference.py

@@ -209,6 +209,7 @@ def _infer_features_and_entities(
             fv, join_keys, run_inference_for_features, config
         )
 
+    entity_columns = []
     columns_to_exclude = {
         fv.batch_source.timestamp_field,
         fv.batch_source.created_timestamp_column,
@@ -218,6 +219,7 @@ def _infer_features_and_entities(
             columns_to_exclude.remove(mapped_col)
             columns_to_exclude.add(original_col)
 
+    # this is what gets the right stuff
     table_column_names_and_types = fv.batch_source.get_table_column_names_and_types(
         config
     )
@@ -233,9 +235,9 @@ def _infer_features_and_entities(
                 ),
             )
             if field.name not in [
-                entity_column.name for entity_column in fv.entity_columns
+                entity_column.name for entity_column in entity_columns
             ]:
-                fv.entity_columns.append(field)
+                entity_columns.append(field)
         elif not re.match(
             "^__|__$", col_name
         ):  # double underscores often signal an internal-use column
@@ -256,6 +258,8 @@ def _infer_features_and_entities(
                 if field.name not in [feature.name for feature in fv.features]:
                     fv.features.append(field)
 
+    fv.entity_columns = entity_columns
+
 
 def _infer_on_demand_features_and_entities(
     fv: OnDemandFeatureView,
@@ -282,26 +286,29 @@ def _infer_on_demand_features_and_entities(
 
         batch_source = getattr(source_feature_view, "batch_source")
         batch_field_mapping = getattr(batch_source or None, "field_mapping")
-        if batch_field_mapping:
-            for (
-                original_col,
-                mapped_col,
-            ) in batch_field_mapping.items():
-                if mapped_col in columns_to_exclude:
-                    columns_to_exclude.remove(mapped_col)
-                    columns_to_exclude.add(original_col)
+        for (
+            original_col,
+            mapped_col,
+        ) in batch_field_mapping.items():
+            if mapped_col in columns_to_exclude:
+                columns_to_exclude.remove(mapped_col)
+                columns_to_exclude.add(original_col)
+
+        table_column_names_and_types = batch_source.get_table_column_names_and_types(
+            config
+        )
+        batch_field_mapping = getattr(batch_source, "field_mapping", {})
 
-            table_column_names_and_types = (
-                batch_source.get_table_column_names_and_types(config)
-            )
         for col_name, col_datatype in table_column_names_and_types:
             if col_name in columns_to_exclude:
                 continue
             elif col_name in join_keys:
                 field = Field(
                     name=col_name,
                     dtype=from_value_type(
-                        batch_source.source_datatype_to_feast_value_type()(col_datatype)
+                        batch_source.source_datatype_to_feast_value_type()(
+                            col_datatype
+                        )
                     ),
                 )
                 if field.name not in [

sdk/python/feast/infra/passthrough_provider.py

@@ -5,7 +5,7 @@
 import pyarrow as pa
 from tqdm import tqdm
 
-from feast import importer
+from feast import OnDemandFeatureView, importer
 from feast.batch_feature_view import BatchFeatureView
 from feast.data_source import DataSource
 from feast.entity import Entity
@@ -276,23 +276,38 @@ def ingest_df(
         self,
         feature_view: FeatureView,
         df: pd.DataFrame,
+        field_mapping: Optional[Dict] = None,
     ):
         table = pa.Table.from_pandas(df)
-
-        if feature_view.batch_source.field_mapping is not None:
-            table = _run_pyarrow_field_mapping(
-                table, feature_view.batch_source.field_mapping
+        if isinstance(feature_view, OnDemandFeatureView):
+            table = _run_pyarrow_field_mapping(table, field_mapping)
+            join_keys = {
+                entity.name: entity.dtype.to_value_type()
+                for entity in feature_view.entity_columns
+            }
+            rows_to_write = _convert_arrow_to_proto(table, feature_view, join_keys)
+
+            self.online_write_batch(
+                self.repo_config, feature_view, rows_to_write, progress=None
             )
+        else:
+            # Note: A dictionary mapping of column names in this data
+            #   source to feature names in a feature table or view. Only used for feature
+            #   columns, not entity or timestamp columns.
+            if feature_view.batch_source.field_mapping is not None:
+                table = _run_pyarrow_field_mapping(
+                    table, feature_view.batch_source.field_mapping
+                )
 
-        join_keys = {
-            entity.name: entity.dtype.to_value_type()
-            for entity in feature_view.entity_columns
-        }
-        rows_to_write = _convert_arrow_to_proto(table, feature_view, join_keys)
+            join_keys = {
+                entity.name: entity.dtype.to_value_type()
+                for entity in feature_view.entity_columns
+            }
+            rows_to_write = _convert_arrow_to_proto(table, feature_view, join_keys)
 
-        self.online_write_batch(
-            self.repo_config, feature_view, rows_to_write, progress=None
-        )
+            self.online_write_batch(
+                self.repo_config, feature_view, rows_to_write, progress=None
+            )
 
     def ingest_df_to_offline_store(self, feature_view: FeatureView, table: pa.Table):
         if feature_view.batch_source.field_mapping is not None:

sdk/python/feast/types.py

@@ -12,9 +12,12 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 from abc import ABC, abstractmethod
+from datetime import datetime, timezone
 from enum import Enum
 from typing import Dict, Union
 
+import pyarrow
+
 from feast.value_type import ValueType
 
 PRIMITIVE_FEAST_TYPES_TO_VALUE_TYPES = {
@@ -30,6 +33,10 @@
 }
 
 
+def _utc_now() -> datetime:
+    return datetime.now(tz=timezone.utc)
+
+
 class ComplexFeastType(ABC):
     """
     A ComplexFeastType represents a structured type that is recognized by Feast.
@@ -103,7 +110,6 @@ def __hash__(self):
 Float64 = PrimitiveFeastType.FLOAT64
 UnixTimestamp = PrimitiveFeastType.UNIX_TIMESTAMP
 
-
 SUPPORTED_BASE_TYPES = [
     Invalid,
     String,
@@ -159,7 +165,6 @@ def __str__(self):
 
 FeastType = Union[ComplexFeastType, PrimitiveFeastType]
 
-
 VALUE_TYPES_TO_FEAST_TYPES: Dict["ValueType", FeastType] = {
     ValueType.UNKNOWN: Invalid,
     ValueType.BYTES: Bytes,
@@ -180,6 +185,33 @@ def __str__(self):
     ValueType.UNIX_TIMESTAMP_LIST: Array(UnixTimestamp),
 }
 
+FEAST_TYPES_TO_PYARROW_TYPES = {
+    String: pyarrow.string(),
+    Bool: pyarrow.bool_(),
+    Int32: pyarrow.int32(),
+    Int64: pyarrow.int64(),
+    Float32: pyarrow.float32(),
+    Float64: pyarrow.float64(),
+    # Note: datetime only supports microseconds https://github.com/python/cpython/blob/3.8/Lib/datetime.py#L1559
+    UnixTimestamp: pyarrow.timestamp("us", tz=_utc_now().tzname()),
+}
+
+
+def from_feast_to_pyarrow_type(feast_type: FeastType) -> pyarrow.DataType:
+    """
+    Converts a Feast type to a PyArrow type.
+
+    Args:
+        feast_type: The Feast type to be converted.
+
+    Raises:
+        ValueError: The conversion could not be performed.
+    """
+    if feast_type in FEAST_TYPES_TO_PYARROW_TYPES:
+        return FEAST_TYPES_TO_PYARROW_TYPES[feast_type]
+
+    raise ValueError(f"Could not convert Feast type {feast_type} to PyArrow type.")
+
 
 def from_value_type(
     value_type: ValueType,