fix: align Ray __eq__ with source identity; cache Spark UDF resolve

RayTransformation now compares by udf_string when present (same contract as Pandas/Python) so a future source-first from_proto will not break no-op apply. SparkTransformationNode caches resolve_udf so strip+exec runs once per node lifetime. Signed-off-by: Aniket Paluskar <apaluska@redhat.com> Co-authored-by: Cursor <cursoragent@cursor.com>
ntkathole · Aug 17, 2026 · Jul 29, 2026 · Jul 29, 2026 · Jul 30, 2026 · Jul 30, 2026
commit 39f21a670fc9bccdf142a7ed463b892beff5b6e8
@@ -613,19 +613,22 @@ def __init__(
         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.
+        yields a healthy callable. Result is cached for the lifetime of the node.
         """
-        return resolve_udf(
-            udf_string=self.udf_string,
-            fallback_udf=self.udf,
-            preferred_name=self.name,
-        )
+        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)

@@ -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):

@@ -61,6 +61,24 @@ def strip_extra_spaces(df):
     assert rows[1]["name"] == "John D."
 
 
+def test_spark_transformation_node_caches_resolved_udf():
+    def identity(df):
+        return df
+
+    input_node = MagicMock()
+    input_node.name = "source"
+    node = SparkTransformationNode(
+        "transform",
+        udf=identity,
+        inputs=[input_node],
+        udf_string="",
+    )
+    first = node._resolve_udf()
+    second = node._resolve_udf()
+    assert first is second
+    assert first is identity
+
+
 def test_spark_aggregation_node_executes_correctly(spark_session):
     # Sample input DataFrame
     input_df = spark_session.createDataFrame(

@@ -47,3 +47,13 @@ def test_python_transformation_eq_uses_udf_string_not_bytecode():
     right = PythonTransformation(udf=rehydrated, udf_string=_UDF_SOURCE)
     assert left == right
     assert left != PythonTransformation(udf=rehydrated, udf_string="other")
+
+
+def test_ray_transformation_eq_uses_udf_string_not_bytecode():
+    from feast.transformation.ray_transformation import RayTransformation
+
+    rehydrated = resolve_udf(udf_string=_UDF_SOURCE, preferred_name="pandas_udf")
+    left = RayTransformation(udf=pandas_udf, udf_string=_UDF_SOURCE)
+    right = RayTransformation(udf=rehydrated, udf_string=_UDF_SOURCE)
+    assert left == right
+    assert left != RayTransformation(udf=rehydrated, udf_string="other")