fix(udf): log Python __eq__ exceptions instead of silently mapping to unequal

`PartialEq` on `PythonFunctionScalarUDF` / `PythonFunctionAggregateUDF` /
`PythonFunctionWindowUDF` used `unwrap_or(false)` when the underlying
Python `__eq__` raised. Two same-callable UDFs whose `__eq__` happened
to throw — e.g. a deliberately strict `__eq__` that runs validation —
would compare unequal silently, breaking expression dedup and cache
lookups without leaving any trace for an operator to investigate.

Rust's `PartialEq` cannot return `Result`, so `false` remains the
conservative choice (better to over-distinguish than wrongly merge),
but the exception is now logged at `log::debug` with the UDF's
registered name and the exception text. Mark each call site with a
`FIXME` pointing at the upstream `*UDFImpl` traits — when those expose
a fallible `PartialEq` we can drop the fallback and surface the error.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: timsaucer

crates/core/src/udaf.rs

@@ -243,10 +243,22 @@ impl PartialEq for PythonFunctionAggregateUDF {
             // `__eq__` only for two distinct callables.
             && (self.accumulator.as_ptr() == other.accumulator.as_ptr()
                 || Python::attach(|py| {
+                    // See `PythonFunctionScalarUDF::eq` for the
+                    // rationale on swallowing the exception as `false`
+                    // and logging at `debug`. FIXME: revisit if
+                    // upstream `AggregateUDFImpl` exposes a fallible
+                    // `PartialEq`.
                     self.accumulator
                         .bind(py)
                         .eq(other.accumulator.bind(py))
-                        .unwrap_or(false)
+                        .unwrap_or_else(|e| {
+                            log::debug!(
+                                target: "datafusion_python::udaf",
+                                "PythonFunctionAggregateUDF {:?} __eq__ raised; treating as unequal: {e}",
+                                self.name,
+                            );
+                            false
+                        })
                 }))
     }
 }

crates/core/src/udf.rs

@@ -112,7 +112,27 @@ impl PartialEq for PythonFunctionScalarUDF {
             // circuits before touching the GIL.
             && (self.func.as_ptr() == other.func.as_ptr()
                 || Python::attach(|py| {
-                    self.func.bind(py).eq(other.func.bind(py)).unwrap_or(false)
+                    // Rust's `PartialEq` cannot return `Result`, so we
+                    // have to pick a side when Python `__eq__` raises.
+                    // `false` is the conservative choice — better to
+                    // report two UDFs as distinct than to wrongly
+                    // merge them — but the silent miss can still
+                    // surface as expression-dedup or cache-lookup
+                    // anomalies. Log at `debug` so the failure is
+                    // observable without flooding production logs.
+                    // FIXME: revisit if upstream `ScalarUDFImpl`
+                    // exposes a fallible `PartialEq`.
+                    self.func
+                        .bind(py)
+                        .eq(other.func.bind(py))
+                        .unwrap_or_else(|e| {
+                            log::debug!(
+                                target: "datafusion_python::udf",
+                                "PythonFunctionScalarUDF {:?} __eq__ raised; treating as unequal: {e}",
+                                self.name,
+                            );
+                            false
+                        })
                 }))
     }
 }

crates/core/src/udwf.rs

@@ -329,10 +329,22 @@ impl PartialEq for PythonFunctionWindowUDF {
             // `__eq__` only for two distinct callables.
             && (self.evaluator.as_ptr() == other.evaluator.as_ptr()
                 || Python::attach(|py| {
+                    // See `PythonFunctionScalarUDF::eq` for the
+                    // rationale on swallowing the exception as `false`
+                    // and logging at `debug`. FIXME: revisit if
+                    // upstream `WindowUDFImpl` exposes a fallible
+                    // `PartialEq`.
                     self.evaluator
                         .bind(py)
                         .eq(other.evaluator.bind(py))
-                        .unwrap_or(false)
+                        .unwrap_or_else(|e| {
+                            log::debug!(
+                                target: "datafusion_python::udwf",
+                                "PythonFunctionWindowUDF {:?} __eq__ raised; treating as unequal: {e}",
+                                self.name,
+                            );
+                            false
+                        })
                 }))
     }
 }