[SPARK-45317][SQL][CONNECT] Handle null filename in stack traces of e…

…xceptions

### What changes were proposed in this pull request?

- Handle null filename in stack traces of exceptions
- Change the filename field in protobuf to optional

### Why are the changes needed?

- In Java exceptions, filename is the only field that can be nullable and null filename may cause NullPointerException

### Does this PR introduce _any_ user-facing change?

No

### How was this patch tested?

- `build/sbt "connect-client-jvm/testOnly *ClientE2ETestSuite"`
- `build/sbt "connect-client-jvm/testOnly *ClientStreamingQuerySuite"`

### Was this patch authored or co-authored using generative AI tooling?

Closes #43103 from heyihong/SPARK-45317.

Authored-by: Yihong He <yihong.he@databricks.com>
Signed-off-by: Hyukjin Kwon <gurwls223@apache.org>

connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/ClientE2ETestSuite.scala

@@ -45,6 +45,34 @@ import org.apache.spark.sql.types._
 
 class ClientE2ETestSuite extends RemoteSparkSession with SQLHelper with PrivateMethodTester {
 
+  test(s"throw SparkException with null filename in stack trace elements") {
+    withSQLConf("spark.sql.connect.enrichError.enabled" -> "true") {
+      val session = spark
+      import session.implicits._
+
+      val throwException =
+        udf((_: String) => {
+          val testError = new SparkException("test")
+          val stackTrace = testError.getStackTrace()
+          stackTrace(0) = new StackTraceElement(
+            stackTrace(0).getClassName,
+            stackTrace(0).getMethodName,
+            null,
+            stackTrace(0).getLineNumber)
+          testError.setStackTrace(stackTrace)
+          throw testError
+        })
+
+      val ex = intercept[SparkException] {
+        Seq("1").toDS.withColumn("udf_val", throwException($"value")).collect()
+      }
+
+      assert(ex.getCause.isInstanceOf[SparkException])
+      assert(ex.getCause.getStackTrace().length > 0)
+      assert(ex.getCause.getStackTrace()(0).getFileName == null)
+    }
+  }
+
   for (enrichErrorEnabled <- Seq(false, true)) {
     test(s"cause exception - ${enrichErrorEnabled}") {
       withSQLConf("spark.sql.connect.enrichError.enabled" -> enrichErrorEnabled.toString) {

connector/connect/common/src/main/protobuf/spark/connect/base.proto

@@ -808,7 +808,7 @@ message FetchErrorDetailsResponse {
     string method_name = 2;
 
     // The name of the file containing the execution point.
-    string file_name = 3;
+    optional string file_name = 3;
 
     // The line number of the source line containing the execution point.
     int32 line_number = 4;

connector/connect/common/src/main/scala/org/apache/spark/sql/connect/client/GrpcExceptionConverter.scala

@@ -222,7 +222,7 @@ private object GrpcExceptionConverter {
         new StackTraceElement(
           stackTraceElement.getDeclaringClass,
           stackTraceElement.getMethodName,
-          stackTraceElement.getFileName,
+          if (stackTraceElement.hasFileName) stackTraceElement.getFileName else null,
           stackTraceElement.getLineNumber)
       })
     }

connector/connect/server/src/main/scala/org/apache/spark/sql/connect/utils/ErrorUtils.scala

@@ -92,13 +92,17 @@ private[connect] object ErrorUtils extends Logging {
         builder.addAllStackTrace(
           currentError.getStackTrace
             .map { stackTraceElement =>
-              FetchErrorDetailsResponse.StackTraceElement
+              val stackTraceBuilder = FetchErrorDetailsResponse.StackTraceElement
                 .newBuilder()
                 .setDeclaringClass(stackTraceElement.getClassName)
                 .setMethodName(stackTraceElement.getMethodName)
-                .setFileName(stackTraceElement.getFileName)
                 .setLineNumber(stackTraceElement.getLineNumber)
-                .build()
+
+              if (stackTraceElement.getFileName != null) {
+                stackTraceBuilder.setFileName(stackTraceElement.getFileName)
+              }
+
+              stackTraceBuilder.build()
             }
             .toIterable
             .asJava)

connector/connect/server/src/test/scala/org/apache/spark/sql/connect/service/FetchErrorDetailsHandlerSuite.scala

@@ -163,4 +163,29 @@ class FetchErrorDetailsHandlerSuite extends SharedSparkSession with ResourceHelp
       testError = new Exception(s"test$i", testError)
     }
   }
+
+  test("null filename in stack trace elements") {
+    val testError = new Exception("test")
+    val stackTrace = testError.getStackTrace()
+    stackTrace(0) = new StackTraceElement(
+      stackTrace(0).getClassName,
+      stackTrace(0).getMethodName,
+      null,
+      stackTrace(0).getLineNumber)
+    testError.setStackTrace(stackTrace)
+
+    val errorId = UUID.randomUUID().toString()
+
+    SparkConnectService
+      .getOrCreateIsolatedSession(userId, sessionId)
+      .errorIdToError
+      .put(errorId, testError)
+
+    val response = fetchErrorDetails(userId, sessionId, errorId)
+    assert(response.hasRootErrorIdx)
+    assert(response.getRootErrorIdx == 0)
+
+    assert(response.getErrors(0).getStackTraceCount > 0)
+    assert(!response.getErrors(0).getStackTrace(0).hasFileName)
+  }
 }

python/pyspark/sql/connect/proto/base_pb2.pyi

@@ -2816,12 +2816,20 @@ class FetchErrorDetailsResponse(google.protobuf.message.Message):
             *,
             declaring_class: builtins.str = ...,
             method_name: builtins.str = ...,
-            file_name: builtins.str = ...,
+            file_name: builtins.str | None = ...,
             line_number: builtins.int = ...,
         ) -> None: ...
+        def HasField(
+            self,
+            field_name: typing_extensions.Literal[
+                "_file_name", b"_file_name", "file_name", b"file_name"
+            ],
+        ) -> builtins.bool: ...
         def ClearField(
             self,
             field_name: typing_extensions.Literal[
+                "_file_name",
+                b"_file_name",
                 "declaring_class",
                 b"declaring_class",
                 "file_name",
@@ -2832,6 +2840,9 @@ class FetchErrorDetailsResponse(google.protobuf.message.Message):
                 b"method_name",
             ],
         ) -> None: ...
+        def WhichOneof(
+            self, oneof_group: typing_extensions.Literal["_file_name", b"_file_name"]
+        ) -> typing_extensions.Literal["file_name"] | None: ...
 
     class Error(google.protobuf.message.Message):
         """Error defines the schema for the representing exception."""