merging

Signed-off-by: Francisco Javier Arceo <farceo@redhat.com>
feast-dev · franciscojavierarceo · Sep 24, 2024 · Aug 17, 2024 · Sep 24, 2024 · Aug 18, 2024
commit 16e68f096806c5e616689c0b6719f3bd58c8b497
@@ -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
@@ -278,25 +278,27 @@ def ingest_df(
         df: pd.DataFrame,
     ):
         table = pa.Table.from_pandas(df)
+        if isinstance(feature_view, OnDemandFeatureView):
+            # TODO: Update this to support On Demand Feature Views.
+            pass
+        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
+                )
 
-        # TODO: Update this to support On Demand Feature Views.
-        # 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:

diff --git a/sdk/python/feast/on_demand_feature_view.py b/sdk/python/feast/on_demand_feature_view.py
@@ -3,7 +3,7 @@
 import inspect
 import warnings
 from types import FunctionType
-from typing import Any, List, Optional, Union, get_type_hints
+from typing import Any, Optional, Union, get_type_hints, List
 
 import dill
 import pandas as pd

@@ -15,6 +15,8 @@
 from enum import Enum
 from typing import Dict, Union
 
+import pyarrow
+
 from feast.value_type import ValueType
 
 PRIMITIVE_FEAST_TYPES_TO_VALUE_TYPES = {
@@ -103,7 +105,6 @@ def __hash__(self):
 Float64 = PrimitiveFeastType.FLOAT64
 UnixTimestamp = PrimitiveFeastType.UNIX_TIMESTAMP
 
-
 SUPPORTED_BASE_TYPES = [
     Invalid,
     String,
@@ -159,7 +160,6 @@ def __str__(self):
 
 FeastType = Union[ComplexFeastType, PrimitiveFeastType]
 
-
 VALUE_TYPES_TO_FEAST_TYPES: Dict["ValueType", FeastType] = {
     ValueType.UNKNOWN: Invalid,
     ValueType.BYTES: Bytes,
@@ -180,6 +180,32 @@ 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(),
+    UnixTimestamp: pyarrow.timestamp(),
+}
+
+
+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,

@@ -43,6 +43,7 @@
 from feast.protos.feast.types.Value_pb2 import RepeatedValue as RepeatedValueProto
 from feast.protos.feast.types.Value_pb2 import Value as ValueProto
 from feast.type_map import python_values_to_proto_values
+from feast.types import from_feast_to_pyarrow_type
 from feast.value_type import ValueType
 from feast.version import get_version
 
@@ -230,6 +231,78 @@ def _convert_arrow_to_proto(
     table: Union[pyarrow.Table, pyarrow.RecordBatch],
     feature_view: "FeatureView",
     join_keys: Dict[str, ValueType],
+) -> List[Tuple[EntityKeyProto, Dict[str, ValueProto], datetime, Optional[datetime]]]:
+    if isinstance(feature_view, OnDemandFeatureView):
+        return _convert_arrow_odfv_to_proto(table, feature_view, join_keys)
+    else:
+        return _convert_arrow_fv_to_proto(table, feature_view, join_keys)
+
+
+def _convert_arrow_fv_to_proto(
+    table: Union[pyarrow.Table, pyarrow.RecordBatch],
+    feature_view: "FeatureView",
+    join_keys: Dict[str, ValueType],
+) -> List[Tuple[EntityKeyProto, Dict[str, ValueProto], datetime, Optional[datetime]]]:
+    # Avoid ChunkedArrays which guarantees `zero_copy_only` available.
+    if isinstance(table, pyarrow.Table):
+        table = table.to_batches()[0]
+
+    columns = [
+        (field.name, field.dtype.to_value_type()) for field in feature_view.features
+    ] + list(join_keys.items())
+
+    proto_values_by_column = {
+        column: python_values_to_proto_values(
+            table.column(column).to_numpy(zero_copy_only=False), value_type
+        )
+        for column, value_type in columns
+    }
+
+    entity_keys = [
+        EntityKeyProto(
+            join_keys=join_keys,
+            entity_values=[proto_values_by_column[k][idx] for k in join_keys],
+        )
+        for idx in range(table.num_rows)
+    ]
+
+    # Serialize the features per row
+    feature_dict = {
+        feature.name: proto_values_by_column[feature.name]
+        for feature in feature_view.features
+    }
+    features = [dict(zip(feature_dict, vars)) for vars in zip(*feature_dict.values())]
+
+    # Convert event_timestamps
+    event_timestamps = [
+        _coerce_datetime(val)
+        for val in pd.to_datetime(
+            table.column(feature_view.batch_source.timestamp_field).to_numpy(
+                zero_copy_only=False
+            )
+        )
+    ]
+
+    # Convert created_timestamps if they exist
+    if feature_view.batch_source.created_timestamp_column:
+        created_timestamps = [
+            _coerce_datetime(val)
+            for val in pd.to_datetime(
+                table.column(
+                    feature_view.batch_source.created_timestamp_column
+                ).to_numpy(zero_copy_only=False)
+            )
+        ]
+    else:
+        created_timestamps = [None] * table.num_rows
+
+    return list(zip(entity_keys, features, event_timestamps, created_timestamps))
+
+
+def _convert_arrow_odfv_to_proto(
+    table: Union[pyarrow.Table, pyarrow.RecordBatch],
+    feature_view: "FeatureView",
+    join_keys: Dict[str, ValueType],
 ) -> List[Tuple[EntityKeyProto, Dict[str, ValueProto], datetime, Optional[datetime]]]:
     # Avoid ChunkedArrays which guarantees `zero_copy_only` available.
     if isinstance(table, pyarrow.Table):
@@ -244,7 +317,21 @@ def _convert_arrow_to_proto(
             table.column(column).to_numpy(zero_copy_only=False), value_type
         )
         for column, value_type in columns
+        if column in table.column_names
     }
+    # Adding On Demand Features
+    for feature in feature_view.features:
+        if feature.name in [c[0] for c in columns]:
+            # initializing the column as null
+            proto_values_by_column[feature.name] = python_values_to_proto_values(
+                table.append_column(
+                    feature.name,
+                    pyarrow.array(
+                        [None] * table.shape[0],
+                        type=from_feast_to_pyarrow_type(feature.dtype),
+                    ),
+                ),
+            )
 
     entity_keys = [
         EntityKeyProto(

@@ -311,8 +311,6 @@ def test_stored_writes(self):
                 "current_datetime",
             ]
         )
-        print(online_python_response)
-        # Now this is where we need to test the stored writes, this should return the same output as the previous
 
 
 class TestOnDemandPythonTransformationAllDataTypes(unittest.TestCase):