updated

Signed-off-by: Francisco Javier Arceo <farceo@redhat.com>
feast-dev · franciscojavierarceo · Nov 26, 2025 · Nov 27, 2025 · Nov 28, 2025 · Nov 28, 2025
commit 74de4670094323f0c45147805ac82d6f70860d5d
@@ -380,7 +380,6 @@ async def write_to_online_store(request: WriteToFeatureStoreRequest) -> None:
             feature_view_name=feature_view_name,
             df=df,
             allow_registry_cache=allow_registry_cache,
-            transform=request.transform_on_write,
         )
 
     @app.get("/health")

@@ -1767,7 +1767,6 @@ def push(
                     fv.name,
                     df,
                     allow_registry_cache=allow_registry_cache,
-                    transform=transform_on_write,
                 )
             if to == PushMode.OFFLINE or to == PushMode.ONLINE_AND_OFFLINE:
                 self.write_to_offline_store(
@@ -2094,10 +2093,9 @@ def write_to_online_store(
         """
 
         # Get feature view to enable schema-based transformation detection
-        registry = self._get_registry_and_project()[0]
         feature_view = cast(
             FeatureView,
-            registry.get_feature_view(
+            self._registry.get_feature_view(
                 feature_view_name, self.project, allow_cache=allow_registry_cache
             ),
         )
@@ -2156,10 +2154,9 @@ async def write_to_online_store_async(
         """
 
         # Get feature view to enable schema-based transformation detection
-        registry = self._get_registry_and_project()[0]
         feature_view = cast(
             FeatureView,
-            registry.get_feature_view(
+            self._registry.get_feature_view(
                 feature_view_name, self.project, allow_cache=allow_registry_cache
             ),
         )

@@ -93,14 +93,16 @@ def get_output_schema_columns(
         schema_columns.add(field.name)
 
     # Add entity columns (present in both input and output)
-    for entity in feature_view.entities:
-        if hasattr(entity, "join_keys"):
-            # Entity object
-            schema_columns.update(entity.join_keys)
-        elif isinstance(entity, str):
-            # Entity name string - need to get entity object from somewhere
-            # For now, we'll assume the entity name is the join key
-            schema_columns.add(entity)
+    # For OnDemandFeatureViews, we skip adding entity columns since they're not meaningful
+    if not isinstance(feature_view, OnDemandFeatureView):
+        for entity in feature_view.entities:
+            if hasattr(entity, "join_keys"):
+                # Entity object
+                schema_columns.update(entity.join_keys)
+            elif isinstance(entity, str):
+                # Entity name string - filter out dummy entities
+                if entity != "__dummy":
+                    schema_columns.add(entity)
 
     return schema_columns
 

@@ -84,6 +84,7 @@ def test_get_input_schema_columns(self):
         """Test getting input schema columns from FeatureView."""
         input_columns = get_input_schema_columns(self.feature_view)
         # Note: FileSource without explicit schema falls back to entity names + timestamp
+        # Since entities are stored as strings, we use entity name rather than join keys
         expected_columns = {"driver", "event_timestamp"}
         self.assertEqual(input_columns, expected_columns)
 
@@ -151,7 +152,7 @@ def test_should_apply_transformation_dataframe_input(self):
         # DataFrame with input schema
         input_df = pd.DataFrame(
             {
-                "driver_id": [1, 2],
+                "driver": [1, 2],
                 "value": [5, 10],
                 "event_timestamp": ["2023-01-01", "2023-01-02"],
             }
@@ -163,7 +164,7 @@ def test_should_apply_transformation_dataframe_input(self):
         )
 
         # DataFrame with output schema
-        output_df = pd.DataFrame({"driver_id": [1, 2], "doubled_value": [10, 20]})
+        output_df = pd.DataFrame({"driver": [1, 2], "doubled_value": [10, 20]})
 
         result = should_apply_transformation(self.feature_view, output_df)
         self.assertFalse(
@@ -175,7 +176,7 @@ def test_should_apply_transformation_subset_matching(self):
         """Test detection with subset schema matching (superset data)."""
         # Data is superset of input schema (extra columns are ok)
         superset_input_data = {
-            "driver_id": 1,
+            "driver": 1,
             "value": 5,
             "event_timestamp": "2023-01-01",
             "extra_field": "extra_value",
@@ -189,28 +190,29 @@ def test_should_apply_transformation_subset_matching(self):
     def test_validate_transformation_compatibility(self):
         """Test transformation compatibility validation."""
         # Valid input data
-        input_data = {"driver_id": 1, "value": 5, "event_timestamp": "2023-01-01"}
-        transformed_data = {"driver_id": 1, "doubled_value": 10}
+        input_data = {"driver": 1, "value": 5, "event_timestamp": "2023-01-01"}
+        transformed_data = {"driver": 1, "doubled_value": 10}
 
         errors = validate_transformation_compatibility(
             self.feature_view, input_data, transformed_data
         )
         self.assertEqual(len(errors), 0, "Should have no errors for valid data")
 
         # Invalid input data (missing required column)
-        invalid_input_data = {"driver_id": 1}  # Missing value and timestamp
+        invalid_input_data = {"driver": 1}  # Missing value and timestamp
 
         errors = validate_transformation_compatibility(
             self.feature_view, invalid_input_data
         )
         self.assertGreater(
             len(errors), 0, "Should have errors for missing input columns"
         )
-        self.assertIn("value", str(errors))
+        # Only event_timestamp is required from the input schema (entity + timestamp)
+        # 'value' is not part of the detected input schema for sources without explicit schema
         self.assertIn("event_timestamp", str(errors))
 
         # Invalid transformed data (missing required output column)
-        invalid_transformed_data = {"driver_id": 1}  # Missing doubled_value
+        invalid_transformed_data = {"driver": 1}  # Missing doubled_value
 
         errors = validate_transformation_compatibility(
             self.feature_view, input_data, invalid_transformed_data