Merge branch 'master' into oidc-rbac-ssl-logging

Author: aniketpalu

docs/reference/beta-on-demand-feature-view.md

@@ -69,6 +69,42 @@ def driver_aggregated_stats(inputs):
 
 Aggregated columns are automatically named using the pattern `{function}_{column}` (e.g., `sum_trips`, `mean_rating`).
 
+### Using `input_schema` with Aggregations
+
+When the input data is not already stored as a feature view, use `input_schema` instead of `sources` to describe the fields that will be passed at request time. Feast will create an internal `RequestSource` automatically.
+
+```python
+from datetime import timedelta
+from feast import Field, on_demand_feature_view
+from feast.aggregation import Aggregation
+from feast.types import Float64, Int64
+
+@on_demand_feature_view(
+    input_schema=[
+        Field(name="txn_amount", dtype=Float64),
+    ],
+    schema=[
+        Field(name="txn_count", dtype=Int64),
+        Field(name="total_txn_amount", dtype=Float64),
+        Field(name="avg_txn_amount", dtype=Float64),
+    ],
+    aggregations=[
+        Aggregation(column="txn_amount", function="count", name="txn_count",
+                    time_window=timedelta(days=30)),
+        Aggregation(column="txn_amount", function="sum", name="total_txn_amount",
+                    time_window=timedelta(days=30)),
+        Aggregation(column="txn_amount", function="mean", name="avg_txn_amount",
+                    time_window=timedelta(days=30)),
+    ],
+    entities=[user],
+)
+def user_transaction_stats(inputs):
+    # Aggregations replace the transformation function — no body needed.
+    pass
+```
+
+`input_schema` also accepts fields that are not aggregation columns — for example, thresholds, currency codes, or other contextual values passed at request time that your UDF needs but that are not stored as features.
+
 ## Example
 See [https://github.com/feast-dev/on-demand-feature-views-demo](https://github.com/feast-dev/on-demand-feature-views-demo) for an example on how to use on demand feature views.
 

sdk/python/feast/entity.py

@@ -203,7 +203,7 @@ def to_proto(self) -> EntityProto:
 
         spec = EntitySpecProto(
             name=self.name,
-            value_type=self.value_type.value,
+            value_type=self.value_type.value,  # type: ignore[arg-type]
             join_key=self.join_key,
             description=self.description,
             tags=self.tags,

sdk/python/feast/field.py

@@ -136,7 +136,7 @@ def to_proto(self) -> FieldProto:
             tags[NESTED_COLLECTION_INNER_TYPE_TAG] = _feast_type_to_str(self.dtype)
         return FieldProto(
             name=self.name,
-            value_type=value_type.value,
+            value_type=value_type.value,  # type: ignore[arg-type]
             description=self.description,
             tags=tags,
             vector_index=self.vector_index,

sdk/python/feast/infra/compute_engines/feature_builder.py

@@ -158,12 +158,14 @@ def get_column_info(
         # we need to read ALL source columns, not just the output feature columns.
         # This is specifically for transformations that create new columns or need raw data.
         mode = getattr(getattr(view, "feature_transformation", None), "mode", None)
-        if mode in ("ray", "pandas") or getattr(mode, "value", None) in (
+        if mode in ("ray", "pandas", "python") or getattr(mode, "value", None) in (
             "ray",
             "pandas",
+            "python",
         ):
-            # Signal to read all columns by passing empty list for feature_cols
-            # The transformation will produce the output columns defined in the schema
+            # Signal to read all columns by passing empty list for feature_cols.
+            # "python" (BatchFeatureView) transformations need all raw source columns — the
+            # UDF computes output features from raw input, not from pre-existing feature cols.
             feature_cols = []
 
         return ColumnInfo(

sdk/python/feast/infra/offline_stores/contrib/spark_offline_store/spark.py

@@ -387,12 +387,18 @@ def pull_all_from_table_or_query(
         timestamp_fields = [timestamp_field]
         if created_timestamp_column:
             timestamp_fields.append(created_timestamp_column)
-        (fields_with_aliases, aliases) = _get_fields_with_aliases(
-            fields=join_key_columns + feature_name_columns + timestamp_fields,
-            field_mappings=data_source.field_mapping,
-        )
 
-        fields_with_alias_string = ", ".join(fields_with_aliases)
+        if feature_name_columns:
+            (fields_with_aliases, _) = _get_fields_with_aliases(
+                fields=join_key_columns + feature_name_columns + timestamp_fields,
+                field_mappings=data_source.field_mapping,
+            )
+            fields_with_alias_string = ", ".join(fields_with_aliases)
+        else:
+            # Empty feature_name_columns signals "read all source columns".
+            # Used by BatchFeatureView with TransformationMode.PYTHON/ray/pandas where
+            # the UDF computes output features from raw input — don't project upfront.
+            fields_with_alias_string = "*"
 
         from_expression = data_source.get_table_query_string()
         timestamp_filter = get_timestamp_filter_sql(

sdk/python/feast/infra/online_stores/remote.py

@@ -501,7 +501,7 @@ def _construct_online_read_api_json_request(
         for row in entity_keys:
             entity_key = row.join_keys[0]
             entity_values.append(
-                getattr(row.entity_values[0], row.entity_values[0].WhichOneof("val"))
+                getattr(row.entity_values[0], row.entity_values[0].WhichOneof("val"))  # type: ignore[arg-type]
             )
 
         return {

sdk/python/feast/on_demand_feature_view.py

@@ -134,6 +134,7 @@ class OnDemandFeatureView(BaseFeatureView):
     """
 
     _TRACK_METRICS_TAG = "feast:track_metrics"
+    _INPUT_SCHEMA_SOURCE_PREFIX = "__input_schema__"
 
     name: str
     entities: Optional[List[str]]
@@ -158,7 +159,8 @@ def __init__(  # noqa: C901
         name: str,
         entities: Optional[List[Entity]] = None,
         schema: Optional[List[Field]] = None,
-        sources: List[OnDemandSourceType],
+        sources: Optional[List[OnDemandSourceType]] = None,
+        input_schema: Optional[List[Field]] = None,
         udf: Optional[FunctionType] = None,
         udf_string: Optional[str] = "",
         feature_transformation: Optional[Transformation] = None,
@@ -183,6 +185,11 @@ def __init__(  # noqa: C901
             sources: A map from input source names to the actual input sources, which may be
                 feature views, or request data sources. These sources serve as inputs to the udf,
                 which will refer to them by name.
+            input_schema (optional): A list of Fields describing data that is accepted as input
+                but not stored directly as features — e.g. aggregation columns, normalization
+                parameters, thresholds, or other contextual values passed at request time.
+                When provided, sources is not required — an internal RequestSource will be
+                created automatically.
             udf: The user defined transformation function, which must take pandas
                 dataframes as inputs.
             udf_string: The source code version of the udf (for diffing and displaying in Web UI)
@@ -214,15 +221,44 @@ def __init__(  # noqa: C901
         self.version = version
         schema = schema or []
         self.entities = [e.name for e in entities] if entities else [DUMMY_ENTITY_NAME]
-        self.sources = sources
+        self.input_schema = input_schema
         self.mode = mode.lower()
         self.udf = udf
         self.udf_string = udf_string
         self.source_feature_view_projections: dict[str, FeatureViewProjection] = {}
         self.source_request_sources: dict[str, RequestSource] = {}
+        self._input_schema_sentinel: Optional[RequestSource] = None
+
+        # Strip any existing sentinel from sources (handles __copy__ round-trip)
+        effective_sources: List[OnDemandSourceType] = [
+            s
+            for s in (sources or [])
+            if not (
+                isinstance(s, RequestSource)
+                and s.name.startswith(self._INPUT_SCHEMA_SOURCE_PREFIX)
+            )
+        ]
+
+        if input_schema is not None:
+            # Automatically create an internal RequestSource from input_schema.
+            # Stored privately so it does not appear in source_request_sources for
+            # external consumers (e.g. the feature server, apply(), utils.py).
+            self._input_schema_sentinel = RequestSource(
+                name=f"{self._INPUT_SCHEMA_SOURCE_PREFIX}{name}",
+                schema=input_schema,
+            )
+            self.source_request_sources[self._input_schema_sentinel.name] = (
+                self._input_schema_sentinel
+            )
+        elif not effective_sources:
+            raise ValueError(
+                "Either 'sources' or 'input_schema' must be provided for OnDemandFeatureView."
+            )
+
+        self.sources = effective_sources
 
         # Process each source with explicit type handling
-        for odfv_source in sources:
+        for odfv_source in effective_sources:
             self._add_source_to_collections(odfv_source)
 
         features: List[Field] = []
@@ -274,6 +310,20 @@ def __init__(  # noqa: C901
         self.track_metrics = track_metrics
         self.aggregations = aggregations or []
 
+        if input_schema is not None and self.aggregations:
+            input_field_names = {f.name for f in input_schema}
+            unknown = [
+                agg.column
+                for agg in self.aggregations
+                if agg.column and agg.column not in input_field_names
+            ]
+            if unknown:
+                raise ValueError(
+                    f"Aggregation column(s) {unknown} not found in input_schema "
+                    f"for OnDemandFeatureView '{name}'. "
+                    f"Available fields: {sorted(input_field_names)}"
+                )
+
     def _add_source_to_collections(self, odfv_source: OnDemandSourceType) -> None:
         """
         Add a source to the appropriate collection with explicit type checking.
@@ -328,6 +378,7 @@ def __copy__(self):
             schema=self.features,
             sources=list(self.source_feature_view_projections.values())
             + list(self.source_request_sources.values()),
+            input_schema=self.input_schema,
             feature_transformation=self.feature_transformation,
             mode=self.mode,
             description=self.description,
@@ -337,6 +388,7 @@ def __copy__(self):
             singleton=self.singleton,
             version=self.version,
             track_metrics=self.track_metrics,
+            aggregations=self.aggregations,
         )
         fv.entities = self.entities
         fv.features = self.features
@@ -475,6 +527,10 @@ def _validate_sources_config(self) -> None:
 
     def _validate_transformation_config(self) -> None:
         """Validate transformation configuration."""
+        # Aggregations provide their own transformation; no udf/feature_transformation required.
+        if self.aggregations:
+            return
+
         if not self.feature_transformation:
             raise ValueError(ODFVErrorMessages.no_transformation_provided())
 
@@ -536,6 +592,14 @@ def to_proto(self) -> OnDemandFeatureViewProto:
                 request_data_source=request_sources.to_proto()
             )
 
+        # Serialize the input_schema sentinel so that from_proto() can reconstruct
+        # input_schema correctly; it is excluded from source_request_sources so that
+        # external consumers never see it as a real data source.
+        if self._input_schema_sentinel is not None:
+            sources[self._input_schema_sentinel.name] = OnDemandSource(
+                request_data_source=self._input_schema_sentinel.to_proto()
+            )
+
         feature_transformation = transformation_to_proto(self.feature_transformation)
 
         tags = dict(self.tags) if self.tags else {}
@@ -559,7 +623,7 @@ def to_proto(self) -> OnDemandFeatureViewProto:
             owner=self.owner,
             write_to_online_store=self.write_to_online_store,
             singleton=self.singleton or False,
-            aggregations=self.aggregations,
+            aggregations=[agg.to_proto() for agg in self.aggregations],
             version=self.version,
         )
         return OnDemandFeatureViewProto(spec=spec, meta=meta)
@@ -585,6 +649,18 @@ def from_proto(
             on_demand_feature_view_proto, skip_udf=skip_udf
         )
 
+        # Detect and strip input_schema sentinel from sources
+        input_schema: Optional[List[Field]] = None
+        sources_without_sentinel: List[OnDemandSourceType] = []
+        for source in sources:
+            if isinstance(source, RequestSource) and source.name.startswith(
+                cls._INPUT_SCHEMA_SOURCE_PREFIX
+            ):
+                input_schema = source.schema
+            else:
+                sources_without_sentinel.append(source)
+        sources = sources_without_sentinel
+
         # Parse transformation from proto (skip UDF deserialization if requested)
         transformation = cls._parse_transformation_from_proto(
             on_demand_feature_view_proto, skip_udf=skip_udf
@@ -607,6 +683,7 @@ def from_proto(
             name=on_demand_feature_view_proto.spec.name,
             schema=cls._parse_features_from_proto(on_demand_feature_view_proto),
             sources=cast(List[OnDemandSourceType], sources),
+            input_schema=input_schema,
             feature_transformation=transformation,
             mode=on_demand_feature_view_proto.spec.mode or "pandas",
             description=on_demand_feature_view_proto.spec.description,
@@ -710,6 +787,8 @@ def _parse_transformation_from_proto(
                 feature_transformation.substrait_transformation
             )
         elif transformation_type is None:
+            if proto.spec.aggregations:
+                return None
             # Handle backward compatibility case where feature_transformation is cleared
             return cls._handle_backward_compatible_udf(proto)
         else:
@@ -817,6 +896,10 @@ def get_request_data_schema(self) -> dict[str, ValueType]:
                 raise TypeError(
                     f"Request source schema is not correct type: ${str(type(request_source.schema))}"
                 )
+        # Include fields from the input_schema sentinel (stored privately)
+        if self._input_schema_sentinel is not None:
+            for field in self._input_schema_sentinel.schema:
+                schema[field.name] = field.dtype.to_value_type()
         return schema
 
     def _get_projected_feature_name(self, feature: str) -> str:
@@ -1036,6 +1119,13 @@ def _preprocess_feature_dict(
         return preprocessed_dict, columns_to_cleanup
 
     def infer_features(self) -> None:
+        if self.aggregations and not self.feature_transformation:
+            if not self.features:
+                raise RegistryInferenceFailure(
+                    "OnDemandFeatureView",
+                    f"Could not infer Features for the feature view '{self.name}'.",
+                )
+            return
         assert self.feature_transformation is not None
         random_input = self._construct_random_input(singleton=self.singleton)
         inferred_features = self.feature_transformation.infer_features(
@@ -1092,7 +1182,7 @@ def _is_array_type(self, dtype) -> bool:
         """Check if the dtype represents an array type."""
         # Use proper type checking instead of string comparison
         dtype_str = str(dtype)
-        return "Array" in dtype_str or "List" in dtype_str
+        return "Array" in dtype_str or "List" in dtype_str or "Set" in dtype_str
 
     def _construct_random_input(
         self, singleton: bool = False
@@ -1137,6 +1227,13 @@ def _construct_random_input(
                 sample_value = sample_values.get(value_type, default_value)
                 feature_dict[field.name] = sample_value
 
+        # Add input_schema fields (stored privately outside source_request_sources)
+        if self._input_schema_sentinel is not None:
+            for field in self._input_schema_sentinel.schema:
+                value_type = field.dtype.to_value_type()
+                sample_value = sample_values.get(value_type, default_value)
+                feature_dict[field.name] = sample_value
+
         return feature_dict
 
     def _get_sample_values_by_type(self) -> dict[ValueType, list[Any]]:
@@ -1224,13 +1321,17 @@ def on_demand_feature_view(
     name: Optional[str] = None,
     entities: Optional[List[Entity]] = None,
     schema: list[Field],
-    sources: list[
-        Union[
-            FeatureView,
-            RequestSource,
-            FeatureViewProjection,
+    sources: Optional[
+        list[
+            Union[
+                FeatureView,
+                RequestSource,
+                FeatureViewProjection,
+            ]
         ]
-    ],
+    ] = None,
+    input_schema: Optional[list[Field]] = None,
+    aggregations: Optional[List[Aggregation]] = None,
     mode: str = "pandas",
     description: str = "",
     tags: Optional[dict[str, str]] = None,
@@ -1252,6 +1353,10 @@ def on_demand_feature_view(
         sources: A map from input source names to the actual input sources, which may be
             feature views, or request data sources. These sources serve as inputs to the udf,
             which will refer to them by name.
+        input_schema (optional): A list of Fields describing data that is accepted as input
+            but not stored directly as features — e.g. aggregation columns, normalization
+            parameters, thresholds, or other contextual values passed at request time.
+            When provided, sources is not required.
         mode: The mode of execution (e.g,. Pandas or Python Native)
         description (optional): A human-readable description.
         tags (optional): A dictionary of key-value pairs to store arbitrary metadata.
@@ -1279,6 +1384,7 @@ def decorator(user_function):
         on_demand_feature_view_obj = OnDemandFeatureView(
             name=name if name is not None else user_function.__name__,
             sources=sources,
+            input_schema=input_schema,
             schema=schema,
             mode=mode,
             description=description,
@@ -1288,6 +1394,7 @@ def decorator(user_function):
             entities=entities,
             singleton=singleton,
             track_metrics=track_metrics,
+            aggregations=aggregations,
             udf=user_function,
             udf_string=udf_string,
             version=version,