GeezerVison is showing github.com — fix: UDF/ODFV source rehydrate (+ Postgres / online cache) by aniketpalu · Pull Request #6655 · feast-dev/feast
fix: UDF/ODFV source rehydrate (+ Postgres / online cache)
#6655
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement . We’ll occasionally send you account related emails.
Already on GitHub?
Sign in
to your account
Uh oh!
There was an error while loading. Please reload this page.
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
11 commits
Select commit
Hold shift + click to select a range
98e2582
fix: prefer UDF source over dill; Postgres SELECT *; fresh online reg…
aniketpalu Jul 29, 2026
12ae434
Lint
aniketpalu Jul 29, 2026
b36a953
fix: drop global allow_cache=False on online path
aniketpalu Jul 30, 2026
45ad64a
fix: replace UDF decorator strip regex with linear scan
aniketpalu Jul 30, 2026
708ff26
fix: cast resolve_udf result to FunctionType for mypy
aniketpalu Aug 6, 2026
f2878fb
Merge branch 'master' into fix/udf-source-rehydrate-postgres-online-c…
aniketpalu Aug 11, 2026
8485ae7
fix: compare UDF transformations by udf_string when present
aniketpalu Aug 11, 2026
9dbe4f1
Merge branch 'master' into fix/udf-source-rehydrate-postgres-online-c…
aniketpalu Aug 11, 2026
4e3aaf4
lint
aniketpalu Aug 12, 2026
39f21a6
fix: align Ray __eq__ with source identity; cache Spark UDF resolve
aniketpalu Aug 17, 2026
0c6e240
Merge branch 'master' into fix/udf-source-rehydrate-postgres-online-c…
aniketpalu Aug 17, 2026
File filter
Filter by extension
.baseline
(1)
.py
(11)
All 2 file types selected
Viewed files
Conversations
Failed to load comments.
Loading
Uh oh!
There was an error while loading. Please reload this page.
Jump to
Jump to file
Failed to load files.
Loading
Uh oh!
There was an error while loading. Please reload this page.
Diff view
Hide whitespace
Diff view
Hide whitespace
There are no files selected for viewing
Oops, something went wrong.
Uh oh!
There was an error while loading. Please reload this page.
Original file line number
Diff line number
Diff line change
Expand Up
@@ -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
Expand Down
Original file line number
Diff line number
Diff line change
Expand Up
@@ -45,6 +45,7 @@
from feast.infra.offline_stores.contrib.spark_offline_store.spark_source import (
SparkSource,
)
from feast.transformation.udf_rehydrate import resolve_udf
logger = logging.getLogger(__name__)
Expand Down
Expand Up
@@ -602,9 +603,32 @@ 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 ""
self._resolved_udf: Optional[Callable] = None
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. Result is cached for the lifetime of the node.
"""
if self._resolved_udf is None:
self._resolved_udf = resolve_udf(
udf_string=self.udf_string,
fallback_udf=self.udf,
preferred_name=self.name,
)
return self._resolved_udf
def execute(self, context: ExecutionContext) -> DAGValue:
input_values = self.get_input_values(context)
Expand All
@@ -613,7 +637,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}
Expand Down
Original file line number
Diff line number
Diff line change
Expand Up
@@ -101,14 +101,35 @@ 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 ""}
Expand Down
Expand Up
@@ -275,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,
Expand Down
Original file line number
Diff line number
Diff line change
@@ -1,7 +1,6 @@
import inspect
from typing import Any, Callable, Optional, cast, get_type_hints
import dill
import pandas as pd
import pyarrow
Expand Down
Expand Up
@@ -134,17 +133,28 @@ def __eq__(self, other):
if not isinstance(other, PandasTransformation):
return False
if (
self.udf_string != other.udf_string
or self.udf.__code__.co_code != other.udf.__code__.co_code
):
return False
# udf_string is the canonical diff identity. Source-first from_proto
# rebuilds a new callable (strip+exec) whose bytecode differs from the
# live repo function even when the source is unchanged — do not require
# co_code equality when both sides have source text.
left = self.udf_string or ""
right = other.udf_string or ""
if left and right:
return left == right
return True
return self.udf.__code__.co_code == other.udf.__code__.co_code
@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 ,
)
Original file line number
Diff line number
Diff line change
@@ -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
Expand Down
Expand Up
@@ -145,13 +144,16 @@ def __eq__(self, other):
if not isinstance(other, PythonTransformation):
return False
if (
self.udf_string != other.udf_string
or self.udf.__code__.co_code != other.udf.__code__.co_code
):
return False
# udf_string is the canonical diff identity. Source-first from_proto
# rebuilds a new callable (strip+exec) whose bytecode differs from the
# live repo function even when the source is unchanged — do not require
# co_code equality when both sides have source text.
left = self.udf_string or ""
right = other.udf_string or ""
if left and right:
return left == right
return True
return self.udf.__code__.co_code == other.udf.__code__.co_code
def __reduce__(self):
"""Support for pickle/dill serialization."""
Expand All
@@ -162,7 +164,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=cast(FunctionType, udf ),
udf_string=udf_string ,
)
Original file line number
Diff line number
Diff line change
Expand Up
@@ -272,13 +272,14 @@ def __eq__(self, other):
if not isinstance(other, RayTransformation):
return False
if (
self.udf_string != other.udf_string
or self.udf.__code__.co_code != other.udf.__code__.co_code
):
return False
return True
# Match Pandas/Python: udf_string is the canonical diff identity so a
# future source-first from_proto does not break no-op feast apply.
left = self.udf_string or ""
right = other.udf_string or ""
if left and right:
return left == right
return self.udf.__code__.co_code == other.udf.__code__.co_code
@classmethod
def from_proto(cls, user_defined_function_proto: UserDefinedFunctionProto):
Expand Down
Loading
Oops, something went wrong.
Uh oh!
There was an error while loading. Please reload this page.
Loading
Oops, something went wrong.
Uh oh!
There was an error while loading. Please reload this page.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.