feast-dev · aniketpalu · Jul 20, 2026
@@ -225,7 +225,7 @@ async def _get_features(
 ):
     if request.feature_service:
         feature_service = await run_in_threadpool(
-            store.get_feature_service, request.feature_service, allow_cache=True
+            store.get_feature_service, request.feature_service, allow_cache=False
         )
         assert_permissions(
             resource=feature_service, actions=[AuthzedAction.READ_ONLINE]
@@ -237,7 +237,7 @@ async def _get_features(
             store.registry,
             store.project,
             request.features,
-            allow_cache=True,
+            allow_cache=False,
             hide_dummy_entity=False,
         )
         for feature_view in all_feature_views:

@@ -109,7 +109,10 @@ def build_dedup_node(self, view, input_node):
     def build_transformation_node(self, view, input_nodes):
         udf_name = view.feature_transformation.name
         udf = view.feature_transformation.udf
-        node = SparkTransformationNode(udf_name, udf, inputs=input_nodes)
+        udf_string = getattr(view.feature_transformation, "udf_string", "") or ""
+        node = SparkTransformationNode(
+            udf_name, udf, inputs=input_nodes, udf_string=udf_string
+        )
         self.nodes.append(node)
         return node
 

@@ -38,6 +38,7 @@
     create_offline_store_retrieval_job,
     infer_entity_timestamp_column,
 )
+from feast.transformation.udf_rehydrate import resolve_udf
 from feast.infra.offline_stores.contrib.spark_offline_store.spark import (
     SparkRetrievalJob,
     _get_entity_schema,
@@ -602,9 +603,29 @@ def execute(self, context: ExecutionContext) -> DAGValue:
 
 
 class SparkTransformationNode(DAGNode):
-    def __init__(self, name: str, udf: Callable, inputs: List[DAGNode]):
+    def __init__(
+        self,
+        name: str,
+        udf: Callable,
+        inputs: List[DAGNode],
+        udf_string: str = "",
+    ):
         super().__init__(name, inputs)
         self.udf = udf
+        self.udf_string = udf_string or ""
+
+    def _resolve_udf(self) -> Callable:
+        """Prefer source reconstruction over dill callables.
+
+        Dill-deserialized functions that call DataFrame.withColumn / __getitem__
+        can segfault (exit 139) on Spark 4.0.1. Re-executing ``udf_string``
+        yields a healthy callable.
+        """
+        return resolve_udf(
+            udf_string=self.udf_string,
+            fallback_udf=self.udf,
+            preferred_name=self.name,
+        )
 
     def execute(self, context: ExecutionContext) -> DAGValue:
         input_values = self.get_input_values(context)
@@ -613,7 +634,7 @@ def execute(self, context: ExecutionContext) -> DAGValue:
 
         input_dfs: List[DataFrame] = [val.data for val in input_values]
 
-        transformed_df = self.udf(*input_dfs)
+        transformed_df = self._resolve_udf()(*input_dfs)
 
         return DAGValue(
             data=transformed_df, format=DAGFormat.SPARK, metadata={"transformed": True}

@@ -99,14 +99,39 @@ def pull_latest_from_table_or_query(
         if created_timestamp_column:
             timestamps.append(created_timestamp_column)
         timestamp_desc_string = " DESC, ".join(_append_alias(timestamps, "a")) + " DESC"
-        a_field_string = ", ".join(
-            _append_alias(join_key_columns + feature_name_columns + timestamps, "a")
-        )
-        b_field_string = ", ".join(
-            _append_alias(join_key_columns + feature_name_columns + timestamps, "b")
-        )
-
-        query = f"""
+        # Empty feature_name_columns means "all source columns". BatchFeatureView
+        # python/pandas/ray transforms signal this via get_column_info. Selecting
+        # only join keys + timestamps would starve the UDF of input features.
+        if not feature_name_columns:
+            distinct_on = ", ".join(f'a."{c}"' for c in join_key_columns) or (
+                f'a."{timestamp_field}"'
+            )
+            order_by_parts = [f'a."{c}"' for c in join_key_columns] + [
+                f'a."{timestamp_field}" DESC'
+            ]
+            if created_timestamp_column:
+                order_by_parts.append(f'a."{created_timestamp_column}" DESC')
+            order_by = ", ".join(order_by_parts)
+            query = f"""
+            SELECT DISTINCT ON ({distinct_on})
+                a.*
+                {f", {repr(DUMMY_ENTITY_VAL)} AS {DUMMY_ENTITY_ID}" if not join_key_columns else ""}
+            FROM {from_expression} a
+            WHERE a."{timestamp_field}" BETWEEN '{start_date}'::timestamptz AND '{end_date}'::timestamptz
+            ORDER BY {order_by}
+            """
+        else:
+            a_field_string = ", ".join(
+                _append_alias(
+                    join_key_columns + feature_name_columns + timestamps, "a"
+                )
+            )
+            b_field_string = ", ".join(
+                _append_alias(
+                    join_key_columns + feature_name_columns + timestamps, "b"
+                )
+            )
+            query = f"""
             SELECT
                 {b_field_string}
                 {f", {repr(DUMMY_ENTITY_VAL)} AS {DUMMY_ENTITY_ID}" if not join_key_columns else ""}
@@ -271,12 +296,17 @@ def pull_all_from_table_or_query(
         timestamp_fields = [timestamp_field]
         if created_timestamp_column:
             timestamp_fields.append(created_timestamp_column)
-        field_string = ", ".join(
-            _append_alias(
-                join_key_columns + feature_name_columns + timestamp_fields,
-                "paftoq_alias",
+        # Empty feature_name_columns => SELECT * (BatchFeatureView python mode).
+        # Default materialization uses pull_all (pull_latest_features=False).
+        if not feature_name_columns:
+            field_string = "paftoq_alias.*"
+        else:
+            field_string = ", ".join(
+                _append_alias(
+                    join_key_columns + feature_name_columns + timestamp_fields,
+                    "paftoq_alias",
+                )
             )
-        )
 
         timestamp_filter = get_timestamp_filter_sql(
             start_date,

@@ -1,7 +1,6 @@
 import inspect
 from typing import Any, Callable, Optional, cast, get_type_hints
 
-import dill
 import pandas as pd
 import pyarrow
 
@@ -146,7 +145,15 @@ def __eq__(self, other):
 
     @classmethod
     def from_proto(cls, user_defined_function_proto: UserDefinedFunctionProto):
+        from feast.transformation.udf_rehydrate import resolve_udf
+
+        udf_string = user_defined_function_proto.body_text or ""
+        udf = resolve_udf(
+            udf_string=udf_string,
+            body=user_defined_function_proto.body or None,
+            preferred_name=user_defined_function_proto.name or None,
+        )
         return PandasTransformation(
-            udf=dill.loads(user_defined_function_proto.body),
-            udf_string=user_defined_function_proto.body_text,
+            udf=udf,
+            udf_string=udf_string,
         )
@@ -1,7 +1,6 @@
 from types import FunctionType
 from typing import Any, Dict, Optional, cast
 
-import dill
 import pyarrow
 
 from feast.field import Field, from_value_type
@@ -164,7 +163,15 @@ def __reduce__(self):
 
     @classmethod
     def from_proto(cls, user_defined_function_proto: UserDefinedFunctionProto):
+        from feast.transformation.udf_rehydrate import resolve_udf
+
+        udf_string = user_defined_function_proto.body_text or ""
+        udf = resolve_udf(
+            udf_string=udf_string,
+            body=user_defined_function_proto.body or None,
+            preferred_name=user_defined_function_proto.name or None,
+        )
         return PythonTransformation(
-            udf=dill.loads(user_defined_function_proto.body),
-            udf_string=user_defined_function_proto.body_text,
+            udf=udf,
+            udf_string=udf_string,
         )
@@ -1,8 +1,6 @@
 import inspect
 from typing import Any, Callable, Optional, cast, get_type_hints
 
-import dill
-
 from feast.field import Field, from_value_type
 from feast.protos.feast.core.Transformation_pb2 import (
     UserDefinedFunctionV2 as UserDefinedFunctionProto,
@@ -284,9 +282,17 @@ def __eq__(self, other):
 
     @classmethod
     def from_proto(cls, user_defined_function_proto: UserDefinedFunctionProto):
+        from feast.transformation.udf_rehydrate import resolve_udf
+
+        udf_string = user_defined_function_proto.body_text or ""
+        udf = resolve_udf(
+            udf_string=udf_string,
+            body=user_defined_function_proto.body or None,
+            preferred_name=user_defined_function_proto.name or None,
+        )
         return RayTransformation(
-            udf=dill.loads(user_defined_function_proto.body),
-            udf_string=user_defined_function_proto.body_text,
+            udf=udf,
+            udf_string=udf_string,
             name=user_defined_function_proto.name
             if user_defined_function_proto.name
             else None,

@@ -0,0 +1,115 @@
+"""Rehydrate trusted UDF callables from registry source text.
+
+Feast persists both dill bytes (``body``) and source (``body_text`` / ``udf_string``).
+Preferring source avoids:
+
+* Spark driver segfaults (exit 139) when dill-restored callables touch DataFrames
+* Cross-Python-version serve failures when apply and feature-server differ
+"""
+
+from __future__ import annotations
+
+import logging
+import re
+from typing import Callable, Optional
+
+import dill
+
+logger = logging.getLogger(__name__)
+
+# Leading decorators (e.g. @on_demand_feature_view(...)) are often included in
+# dill.source.getsource() output. They are not needed to recover the callable and
+# usually reference names unavailable at rehydrate time.
+_DECORATOR_BLOCK = re.compile(
+    r"^(?:\s*@[\s\S]*?\n)+(?=\s*(?:async\s+)?def\s+)",
+    re.MULTILINE,
+)
+
+
+def _strip_leading_decorators(udf_string: str) -> str:
+    stripped = _DECORATOR_BLOCK.sub("", udf_string.lstrip(), count=1)
+    return stripped if stripped.strip() else udf_string
+
+
+def _exec_namespace() -> dict:
+    """Minimal globals so typical UDF source can exec without full feast apply ctx."""
+    ns: dict = {"__name__": "feast_udf_rehydrate"}
+    try:
+        import pandas as pd
+
+        ns["pd"] = pd
+        ns["pandas"] = pd
+    except ImportError:
+        pass
+    try:
+        import numpy as np
+
+        ns["np"] = np
+        ns["numpy"] = np
+    except ImportError:
+        pass
+    return ns
+
+
+def rehydrate_udf_from_source(
+    udf_string: str,
+    *,
+    preferred_name: Optional[str] = None,
+) -> Optional[Callable]:
+    """Exec ``udf_string`` and return the resulting callable, or None.
+
+    ``udf_string`` is treated as trusted registry content written by ``feast apply``.
+    Returns None on failure so callers can fall back to dill.
+    """
+    if not (udf_string or "").strip():
+        return None
+
+    source = _strip_leading_decorators(udf_string)
+    ns = _exec_namespace()
+    try:
+        exec(source, ns)  # noqa: S102 — trusted registry source
+    except Exception as e:
+        logger.debug("udf source rehydrate failed: %s", e)
+        return None
+
+    if preferred_name and preferred_name in ns and callable(ns[preferred_name]):
+        return ns[preferred_name]
+
+    for value in ns.values():
+        if not callable(value):
+            continue
+        name = getattr(value, "__name__", None)
+        if name in (None, "<lambda>", "__build_class__"):
+            continue
+        # Skip imported modules / classes we seeded
+        if name in ("DataFrame",):
+            continue
+        return value
+
+    return None
+
+
+def resolve_udf(
+    *,
+    udf_string: str = "",
+    body: Optional[bytes] = None,
+    fallback_udf: Optional[Callable] = None,
+    preferred_name: Optional[str] = None,
+) -> Callable:
+    """Resolve a UDF: source first, then ``fallback_udf``, then dill ``body``."""
+    rehydrated = rehydrate_udf_from_source(
+        udf_string, preferred_name=preferred_name
+    )
+    if rehydrated is not None:
+        return rehydrated
+
+    if fallback_udf is not None:
+        return fallback_udf
+
+    if body:
+        return dill.loads(body)
+
+    raise ValueError(
+        "Cannot resolve UDF: empty udf_string/body_text and no dill body or "
+        "fallback callable. Re-run feast apply so body_text is persisted."
+    )
@@ -1315,7 +1315,10 @@ def _get_online_request_context(
 ):
     from feast.feature_view import DUMMY_ENTITY_NAME
 
-    _feature_refs = _get_features(registry, project, features, allow_cache=True)
+    # Online serve must see fresh FeatureViewState (AVAILABLE_ONLINE vs
+    # MATERIALIZING/GENERATED). Cached registry reads cause brief or stuck 500s
+    # after remote materialize completes.
+    _feature_refs = _get_features(registry, project, features, allow_cache=False)
 
     (
         requested_feature_views,
@@ -1324,7 +1327,7 @@ def _get_online_request_context(
         registry=registry,
         project=project,
         features=features,
-        allow_cache=True,
+        allow_cache=False,
         hide_dummy_entity=False,
     )