feat: propagate result nullability through JvmScalarUdf proto

Add return_nullable to JvmScalarUdf proto and plumb it through to
JvmScalarUdfExpr::nullable. Use the source RLike's own nullable on
the convert path so downstream operators can apply null-aware
optimizations instead of assuming every JVM-UDF call may yield null.

Also clean up Hash/PartialEq to delegate to the children's PhysicalExpr
impls (matching SubstringExpr) rather than stringifying via Display, and
print children using Display in the EXPLAIN output.

Author: andygrove

native/core/src/execution/planner.rs

@@ -715,6 +715,7 @@ impl PhysicalPlanner {
                     udf.class_name.clone(),
                     args,
                     return_type,
+                    udf.return_nullable,
                 )))
             }
             expr => Err(GeneralError(format!("Not implemented: {expr:?}"))),

native/proto/src/proto/expr.proto

@@ -516,4 +516,6 @@ message JvmScalarUdf {
   repeated Expr args = 2;
   // Expected return type. Used to import the result FFI_ArrowArray.
   DataType return_type = 3;
+  // Whether the result column may contain nulls.
+  bool return_nullable = 4;
 }

native/spark-expr/src/jvm_udf/mod.rs

@@ -40,48 +40,53 @@ pub struct JvmScalarUdfExpr {
     class_name: String,
     args: Vec<Arc<dyn PhysicalExpr>>,
     return_type: DataType,
+    return_nullable: bool,
 }
 
 impl JvmScalarUdfExpr {
     pub fn new(
         class_name: String,
         args: Vec<Arc<dyn PhysicalExpr>>,
         return_type: DataType,
+        return_nullable: bool,
     ) -> Self {
         Self {
             class_name,
             args,
             return_type,
+            return_nullable,
         }
     }
 }
 
 impl Display for JvmScalarUdfExpr {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        write!(f, "JvmScalarUdf({}, args={:?})", self.class_name, self.args)
+        write!(f, "JvmScalarUdf({}", self.class_name)?;
+        for a in &self.args {
+            write!(f, ", {a}")?;
+        }
+        write!(f, ")")
     }
 }
 
 impl Hash for JvmScalarUdfExpr {
     fn hash<H: Hasher>(&self, state: &mut H) {
         self.class_name.hash(state);
         for a in &self.args {
-            format!("{}", a).hash(state);
+            a.hash(state);
         }
         self.return_type.hash(state);
+        self.return_nullable.hash(state);
     }
 }
 
 impl PartialEq for JvmScalarUdfExpr {
     fn eq(&self, other: &Self) -> bool {
         self.class_name == other.class_name
             && self.return_type == other.return_type
+            && self.return_nullable == other.return_nullable
             && self.args.len() == other.args.len()
-            && self
-                .args
-                .iter()
-                .zip(other.args.iter())
-                .all(|(a, b)| format!("{}", a) == format!("{}", b))
+            && self.args.iter().zip(&other.args).all(|(a, b)| a.eq(b))
     }
 }
 
@@ -101,7 +106,7 @@ impl PhysicalExpr for JvmScalarUdfExpr {
     }
 
     fn nullable(&self, _input_schema: &Schema) -> DFResult<bool> {
-        Ok(true)
+        Ok(self.return_nullable)
     }
 
     fn evaluate(&self, batch: &RecordBatch) -> DFResult<ColumnarValue> {
@@ -229,6 +234,7 @@ impl PhysicalExpr for JvmScalarUdfExpr {
             self.class_name.clone(),
             children,
             self.return_type.clone(),
+            self.return_nullable,
         )))
     }
 }

spark/src/main/scala/org/apache/comet/serde/strings.scala

@@ -344,6 +344,7 @@ object CometRLike extends CometExpressionSerde[RLike] {
           .addArgs(subjectProto.get)
           .addArgs(patternProto.get)
           .setReturnType(returnType)
+          .setReturnNullable(expr.nullable)
         Some(
           ExprOuterClass.Expr
             .newBuilder()