Add sources and schema parameters to tiled_transformation following on_demand_feature_view pattern

Co-authored-by: franciscojavierarceo <[email protected]>

Author: Copilot

docs/getting-started/architecture/feature-transformation.md

@@ -130,12 +130,20 @@ Examples include:
 
 ```python
 from feast.transformation import tiled_transformation
+from feast import Field
+from feast.types import Float64, Int64
 from datetime import timedelta
 
 @tiled_transformation(
-    tile_size=timedelta(hours=1),  # Process data in 1-hour tiles
-    mode="pandas",                 # Use pandas processing mode
-    overlap=timedelta(minutes=5),  # 5-minute overlap between tiles
+    sources=["transaction_source_fv"],    # Source feature views or data sources
+    schema=[                              # Output feature schema
+        Field(name="rolling_avg", dtype=Float64),
+        Field(name="cumulative_amount", dtype=Float64),
+        Field(name="transaction_count", dtype=Int64),
+    ],
+    tile_size=timedelta(hours=1),         # Process data in 1-hour tiles
+    mode="pandas",                        # Use pandas processing mode
+    overlap=timedelta(minutes=5),         # 5-minute overlap between tiles
     aggregation_functions=[
         lambda df: df.groupby('entity_id').agg({
             'transaction_amount': ['sum', 'mean', 'count']
@@ -146,7 +154,8 @@ def hourly_transaction_features(df: pd.DataFrame) -> pd.DataFrame:
     """Transform transaction data within each hour tile."""
     return df.assign(
         rolling_avg=df['transaction_amount'].rolling(window=10).mean(),
-        cumulative_amount=df['transaction_amount'].cumsum()
+        cumulative_amount=df['transaction_amount'].cumsum(),
+        transaction_count=df.groupby('entity_id').cumcount() + 1
     )
 
 # Usage in StreamFeatureView
@@ -161,6 +170,11 @@ stream_fv = StreamFeatureView(
 **Chaining Features Example:**
 ```python
 @tiled_transformation(
+    sources=["transaction_hourly_fv"],       # Source feature views
+    schema=[                                 # Output feature schema
+        Field(name="local_cumsum", dtype=Float64),
+        Field(name="global_cumsum", dtype=Float64),
+    ],
     tile_size=timedelta(hours=1),
     mode="pandas",
     chaining_functions=[
@@ -177,6 +191,8 @@ def chained_cumulative_features(df: pd.DataFrame) -> pd.DataFrame:
 ```
 
 **Configuration Options:**
+- `sources`: List of source feature views or data sources that this transformation depends on
+- `schema`: List of Field definitions specifying output feature names and data types
 - `tile_size`: Duration of each time tile (e.g., `timedelta(hours=1)`)
 - `mode`: Processing mode - currently supports `"pandas"`
 - `overlap`: Optional overlap between tiles for continuity

sdk/python/feast/transformation/pandas_tiled_transformation.py

@@ -19,6 +19,8 @@ def __init__(
         udf: Callable[[pd.DataFrame], pd.DataFrame],
         udf_string: str,
         tile_config: TileConfiguration,
+        sources: Optional[List[Any]] = None,
+        schema: Optional[List[Any]] = None,
         name: Optional[str] = None,
         tags: Optional[Dict[str, str]] = None,
         description: str = "",
@@ -30,6 +32,8 @@ def __init__(
             udf=udf,
             udf_string=udf_string,
             tile_config=tile_config,
+            sources=sources,
+            schema=schema,
             name=name,
             tags=tags,
             description=description,

sdk/python/feast/transformation/tiled_transformation.py

@@ -1,9 +1,14 @@
 from datetime import timedelta
-from typing import Any, Callable, Dict, List, Optional, Union
+from typing import Any, Callable, Dict, List, Optional, Union, TYPE_CHECKING
 
 from feast.transformation.base import Transformation
 from feast.transformation.mode import TransformationMode
 
+if TYPE_CHECKING:
+    from feast.feature_view import FeatureView
+    from feast.data_source import DataSource
+    from feast.field import Field
+
 
 class TileConfiguration:
     """
@@ -54,6 +59,8 @@ def __init__(
         udf: Callable[[Any], Any],
         udf_string: str,
         tile_config: TileConfiguration,
+        sources: Optional[List[Union[str, "FeatureView", "DataSource"]]] = None,
+        schema: Optional[List["Field"]] = None,
         name: Optional[str] = None,
         tags: Optional[Dict[str, str]] = None,
         description: str = "",
@@ -71,6 +78,8 @@ def __init__(
             owner=owner,
         )
         self.tile_config = tile_config
+        self.sources = sources or []
+        self.schema = schema or []
         self.aggregation_functions = aggregation_functions or []
         self.chaining_functions = chaining_functions or []
 
@@ -177,6 +186,8 @@ def infer_features(self, *args, **kwargs) -> Any:
 def tiled_transformation(
     tile_size: timedelta,
     mode: str = "pandas",
+    sources: Optional[List[Union[str, "FeatureView", "DataSource"]]] = None,
+    schema: Optional[List["Field"]] = None,
     overlap: Optional[timedelta] = None,
     max_tiles_in_memory: int = 10,
     enable_late_data_handling: bool = True,
@@ -193,6 +204,8 @@ def tiled_transformation(
     Args:
         tile_size: The size of each time tile (e.g., timedelta(hours=1))
         mode: The transformation mode - currently supports "pandas"
+        sources: List of source feature views or data sources that this transformation depends on
+        schema: List of Field definitions specifying output feature names and data types
         overlap: Optional overlap between tiles for continuity
         max_tiles_in_memory: Maximum number of tiles to keep in memory
         enable_late_data_handling: Whether to handle late-arriving data
@@ -207,11 +220,19 @@ def tiled_transformation(
         @tiled_transformation(
             tile_size=timedelta(hours=1),
             mode="pandas",
+            sources=["transaction_source_fv"],
+            schema=[
+                Field(name="rolling_avg", dtype=Float64),
+                Field(name="cumulative_amount", dtype=Float64),
+            ],
             overlap=timedelta(minutes=5),
             aggregation_functions=[lambda df: df.groupby('entity_id').sum()]
         )
         def my_tiled_feature(df):
-            return df.assign(derived_feature=df['value'] * 2)
+            return df.assign(
+                rolling_avg=df['value'].rolling(window=10).mean(),
+                cumulative_amount=df['value'].cumsum()
+            )
     """
     def decorator(user_function):
         import dill
@@ -238,6 +259,8 @@ def mainify(obj):
                 udf=user_function,
                 udf_string=udf_string,
                 tile_config=tile_config,
+                sources=sources,
+                schema=schema,
                 name=name or user_function.__name__,
                 tags=tags,
                 description=description,
@@ -251,6 +274,8 @@ def mainify(obj):
                 udf=user_function,
                 udf_string=udf_string,
                 tile_config=tile_config,
+                sources=sources,
+                schema=schema,
                 name=name or user_function.__name__,
                 tags=tags,
                 description=description,