Merging changes continued

Signed-off-by: Francisco Javier Arceo <farceo@redhat.com>
feast-dev · franciscojavierarceo · Oct 10, 2024 · Sep 24, 2024 · Aug 17, 2024 · Sep 24, 2024
commit 82f3f8b48eaf64f07768710e2396549e73a76a9d
@@ -290,11 +290,6 @@ def ingest_df(
             self.online_write_batch(
                 self.repo_config, feature_view, rows_to_write, progress=None
             )
-            # We get these two sets of feature view projections, which are similar to the batch source for regular feature_views
-            # We need to extract these dependencies to get the data types and field mappings to run the _convert_arrow_to_proto function
-            #print(feature_view.source_feature_view_projections)
-            #print(feature_view.source_request_sources)
-            #print('something should happen here')
         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

@@ -322,18 +322,24 @@ def _convert_arrow_odfv_to_proto(
     }
     # Adding On Demand Features
     for feature in feature_view.features:
-        if feature.name in [c[0] for c in columns] and feature.name not in proto_values_by_column:
+        if (
+            feature.name in [c[0] for c in columns]
+            and feature.name not in proto_values_by_column
+        ):
             # initializing the column as null
             null_column = pyarrow.array(
-                        [None] * table.num_rows,
-                        type=from_feast_to_pyarrow_type(feature.dtype),
-                    )
+                [None] * table.num_rows,
+                type=from_feast_to_pyarrow_type(feature.dtype),
+            )
             updated_table = pyarrow.RecordBatch.from_arrays(
                 table.columns + [null_column],
-                schema=table.schema.append(pyarrow.field(feature.name, null_column.type))
+                schema=table.schema.append(
+                    pyarrow.field(feature.name, null_column.type)
+                ),
             )
             proto_values_by_column[feature.name] = python_values_to_proto_values(
-                updated_table.column(feature.name).to_numpy(zero_copy_only=False), feature.dtype.to_value_type(),
+                updated_table.column(feature.name).to_numpy(zero_copy_only=False),
+                feature.dtype.to_value_type(),
             )
 
     entity_keys = [
@@ -351,28 +357,15 @@ def _convert_arrow_odfv_to_proto(
     }
     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
-            )
-        )
-    ]
+    # We need to artificially add event_timestamps and created_timestamps
+    event_timestamps = []
+    timestamp_values = pd.to_datetime([_utc_now() for i in range(table.num_rows)])
 
-    # 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
+    for val in timestamp_values:
+        event_timestamps.append(_coerce_datetime(val))
+
+    # setting them equivalent
+    created_timestamps = event_timestamps
 
     return list(zip(entity_keys, features, event_timestamps, created_timestamps))
 

@@ -68,13 +68,6 @@ def setUp(self):
                 timestamp_field="event_timestamp",
                 created_timestamp_column="created",
             )
-            input_request_source = RequestSource(
-                name="counter_source",
-                schema=[
-                    Field(name="counter", dtype=Int64),
-                    Field(name="input_datetime", dtype=UnixTimestamp),
-                ],
-            )
 
             driver_stats_fv = FeatureView(
                 name="driver_hourly_stats",
@@ -273,6 +266,7 @@ def test_python_docs_demo(self):
             == online_python_response["conv_rate_plus_val2_python"][0]
         )
 
+
 class TestOnDemandPythonTransformationAllDataTypes(unittest.TestCase):
     def setUp(self):
         with tempfile.TemporaryDirectory() as data_dir:
@@ -511,7 +505,6 @@ def python_view(inputs: dict[str, Any]) -> dict[str, Any]:
             store.apply([request_source, python_view])
 
 
-
 class TestOnDemandTransformationsWithWrites(unittest.TestCase):
     def setUp(self):
         with tempfile.TemporaryDirectory() as data_dir:
@@ -616,10 +609,16 @@ def python_stored_writes_feature_view(
             assert len(self.store.list_feature_views()) == 1
             assert len(self.store.list_on_demand_feature_views()) == 1
             assert len(self.store.list_stream_feature_views()) == 0
-            assert driver_stats_fv.entity_columns == \
-                   self.store.get_feature_view('driver_hourly_stats').entity_columns
-            assert self.store.get_on_demand_feature_view('python_stored_writes_feature_view').entity_columns == \
-                     self.store.get_feature_view('driver_hourly_stats').entity_columns
+            assert (
+                driver_stats_fv.entity_columns
+                == self.store.get_feature_view("driver_hourly_stats").entity_columns
+            )
+            assert (
+                self.store.get_on_demand_feature_view(
+                    "python_stored_writes_feature_view"
+                ).entity_columns
+                == self.store.get_feature_view("driver_hourly_stats").entity_columns
+            )
 
     def test_stored_writes(self):
         current_datetime = _utc_now()
@@ -666,4 +665,3 @@ def test_stored_writes(self):
                 "input_datetime",
             ]
         )
-