PR check fixes

Author: Shekharrajak

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

@@ -22,13 +22,13 @@ package org.apache.comet.serde
 import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression}
 
 import org.apache.comet.serde.ExprOuterClass.Expr
-import org.apache.comet.serde.QueryPlanSerde.{exprToProtoInternal, optExprWithInfo, scalarFunctionExprToProtoWithReturnType}
+import org.apache.comet.serde.QueryPlanSerde.{exprToProtoInternal, optExprWithInfo, scalarFunctionExprToProto}
 
 /** Serde for scalar function. */
 case class CometScalarFunction[T <: Expression](name: String) extends CometExpressionSerde[T] {
   override def convert(expr: T, inputs: Seq[Attribute], binding: Boolean): Option[Expr] = {
     val childExpr = expr.children.map(exprToProtoInternal(_, inputs, binding))
-    val optExpr = scalarFunctionExprToProtoWithReturnType(name, expr.dataType, false, childExpr: _*)
+    val optExpr = scalarFunctionExprToProto(name, childExpr: _*)
     optExprWithInfo(optExpr, expr, expr.children: _*)
   }
 }

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

@@ -166,7 +166,7 @@ object QueryPlanSerde extends Logging with CometExprShim {
     classOf[StringRPad] -> CometStringRPad,
     classOf[StringLPad] -> CometStringLPad,
     classOf[StringSpace] -> CometScalarFunction("string_space"),
-    classOf[StringSplit] -> CometScalarFunction("split"),
+    classOf[StringSplit] -> CometStringSplit,
     classOf[StringTranslate] -> CometScalarFunction("translate"),
     classOf[StringTrim] -> CometScalarFunction("trim"),
     classOf[StringTrimBoth] -> CometScalarFunction("btrim"),

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

@@ -21,14 +21,14 @@ package org.apache.comet.serde
 
 import java.util.Locale
 
-import org.apache.spark.sql.catalyst.expressions.{Attribute, Cast, Concat, Expression, InitCap, Left, Length, Like, Literal, Lower, RegExpReplace, RLike, StringLPad, StringRepeat, StringRPad, Substring, Upper}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, Cast, Concat, Expression, InitCap, Left, Length, Like, Literal, Lower, RegExpReplace, RLike, StringLPad, StringRepeat, StringRPad, StringSplit, Substring, Upper}
 import org.apache.spark.sql.types.{BinaryType, DataTypes, LongType, StringType}
 
 import org.apache.comet.CometConf
 import org.apache.comet.CometSparkSessionExtensions.withInfo
 import org.apache.comet.expressions.{CometCast, CometEvalMode, RegExp}
 import org.apache.comet.serde.ExprOuterClass.Expr
-import org.apache.comet.serde.QueryPlanSerde.{createBinaryExpr, exprToProtoInternal, optExprWithInfo, scalarFunctionExprToProto}
+import org.apache.comet.serde.QueryPlanSerde.{createBinaryExpr, exprToProtoInternal, optExprWithInfo, scalarFunctionExprToProto, scalarFunctionExprToProtoWithReturnType}
 
 object CometStringRepeat extends CometExpressionSerde[StringRepeat] {
 
@@ -289,6 +289,32 @@ object CometRegExpReplace extends CometExpressionSerde[RegExpReplace] {
   }
 }
 
+/**
+ * Serde for StringSplit expression.
+ * This is a custom Comet function (not a built-in DataFusion function),
+ * so we need to include the return type in the protobuf to avoid
+ * DataFusion registry lookup failures.
+ */
+object CometStringSplit extends CometExpressionSerde[StringSplit] {
+
+  override def convert(
+      expr: StringSplit,
+      inputs: Seq[Attribute],
+      binding: Boolean): Option[Expr] = {
+    val strExpr = exprToProtoInternal(expr.str, inputs, binding)
+    val regexExpr = exprToProtoInternal(expr.regex, inputs, binding)
+    val limitExpr = exprToProtoInternal(expr.limit, inputs, binding)
+    val optExpr = scalarFunctionExprToProtoWithReturnType(
+      "split",
+      expr.dataType,
+      false,
+      strExpr,
+      regexExpr,
+      limitExpr)
+    optExprWithInfo(optExpr, expr, expr.str, expr.regex, expr.limit)
+  }
+}
+
 trait CommonStringExprs {
 
   def stringDecode(