[SPARK-52383][CONNECT] Improve errors in SparkConnectPlanner

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

This PR improves error handling in SparkConnectPlanner by:

1. Adding new error handling methods in `InvalidInputErrors.scala` for better error messages:
   - `invalidEnum` for handling invalid enum values
   - `invalidOneOfField` for handling invalid oneOf fields in protobuf messages
   - `cannotBeEmpty` for handling empty fields
   - `streamingQueryRunIdMismatch` and `streamingQueryNotFound` for streaming query errors

2. Replacing specific error messages with more generic ones. For examples:
   - Replacing `unknownRelationNotSupported` with `invalidOneOfField`
   - Replacing `catalogTypeNotSupported` with `invalidOneOfField`
   - Replacing `functionIdNotSupported` with `invalidOneOfField`
   - Replacing `expressionIdNotSupported` with `invalidOneOfField`
   - Replacing `dataSourceIdNotSupported` with `invalidOneOfField`

3. Improving error handling for protobuf-related issues:
   - Better handling of oneOf fields in protobuf messages
   - More descriptive error messages for invalid enum values
   - Better handling of empty fields

### Why are the changes needed?

1. Provide more specific and descriptive error messages
2. Better handle protobuf-related issues that are common in the Connect API
3. Make error messages more consistent across different parts of the codebase

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

No

### How was this patch tested?

`build/sbt "connect/testOnly *SparkConnectPlannerSuite"`
`build/sbt "connect/testOnly *InvalidInputErrorsSuite"`

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

Generated-by: Cursor 0.50.7 (Universal)

Closes apache#51062 from heyihong/SPARK-52383.

Authored-by: Yihong He <heyihong.cn@gmail.com>
Signed-off-by: Hyukjin Kwon <gurwls223@apache.org>

Author: heyihong

sql/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/InvalidInputErrors.scala

@@ -19,22 +19,20 @@ package org.apache.spark.sql.connect.planner
 
 import scala.collection.mutable
 
+import com.google.protobuf.Descriptors.Descriptor
+import com.google.protobuf.Internal.EnumLite
+import com.google.protobuf.ProtocolMessageEnum
+
 import org.apache.spark.connect.proto
 import org.apache.spark.sql.connect.common.InvalidPlanInput
 import org.apache.spark.sql.errors.DataTypeErrors.{quoteByDefault, toSQLType}
 import org.apache.spark.sql.types.DataType
 
 object InvalidInputErrors {
 
-  def unknownRelationNotSupported(rel: proto.Relation): InvalidPlanInput =
-    InvalidPlanInput(s"${rel.getUnknown} not supported.")
-
   def noHandlerFoundForExtension(): InvalidPlanInput =
     InvalidPlanInput("No handler found for extension")
 
-  def catalogTypeNotSupported(catType: proto.Catalog.CatTypeCase): InvalidPlanInput =
-    InvalidPlanInput(s"$catType not supported.")
-
   def invalidSQLWithReferences(query: proto.WithRelations): InvalidPlanInput =
     InvalidPlanInput(s"$query is not a valid relation for SQL with references")
 
@@ -61,9 +59,6 @@ object InvalidInputErrors {
   def functionEvalTypeNotSupported(evalType: Int): InvalidPlanInput =
     InvalidPlanInput(s"Function with EvalType: $evalType is not supported")
 
-  def functionIdNotSupported(functionId: Int): InvalidPlanInput =
-    InvalidPlanInput(s"Function with ID: $functionId is not supported")
-
   def groupingExpressionAbsentForKeyValueGroupedDataset(): InvalidPlanInput =
     InvalidPlanInput("The grouping expression cannot be absent for KeyValueGroupedDataset")
 
@@ -101,15 +96,31 @@ object InvalidInputErrors {
   def multiplePathsNotSupportedForStreamingSource(): InvalidPlanInput =
     InvalidPlanInput("Multiple paths are not supported for streaming source")
 
-  def doesNotSupport(what: String): InvalidPlanInput =
-    InvalidPlanInput(s"Does not support $what")
+  def invalidEnum(protoEnum: Enum[_] with ProtocolMessageEnum): InvalidPlanInput =
+    InvalidPlanInput(
+      s"This enum value of ${protoEnum.getDescriptorForType.getFullName}" +
+        s" is invalid: ${protoEnum.name()}(${protoEnum.getNumber})")
+
+  def invalidOneOfField(
+      enumCase: Enum[_] with EnumLite,
+      descriptor: Descriptor): InvalidPlanInput = {
+    // If the oneOf field is not set, the enum number will be 0.
+    if (enumCase.getNumber == 0) {
+      InvalidPlanInput(
+        s"This oneOf field in ${descriptor.getFullName} is not set: ${enumCase.name()}")
+    } else {
+      InvalidPlanInput(
+        s"This oneOf field message in ${descriptor.getFullName} is not supported: " +
+          s"${enumCase.name()}(${enumCase.getNumber})")
+    }
+  }
+
+  def cannotBeEmpty(fieldName: String, descriptor: Descriptor): InvalidPlanInput =
+    InvalidPlanInput(s"$fieldName in ${descriptor.getFullName} cannot be empty")
 
   def invalidSchemaTypeNonStruct(dataType: DataType): InvalidPlanInput =
     InvalidPlanInput("INVALID_SCHEMA_TYPE_NON_STRUCT", Map("dataType" -> toSQLType(dataType)))
 
-  def expressionIdNotSupported(exprId: Int): InvalidPlanInput =
-    InvalidPlanInput(s"Expression with ID: $exprId is not supported")
-
   def lambdaFunctionArgumentCountInvalid(got: Int): InvalidPlanInput =
     InvalidPlanInput(s"LambdaFunction requires 1 ~ 3 arguments, but got $got ones!")
 
@@ -127,18 +138,9 @@ object InvalidInputErrors {
   def windowFunctionRequired(): InvalidPlanInput =
     InvalidPlanInput("WindowFunction is required in WindowExpression")
 
-  def unknownFrameType(
-      frameType: proto.Expression.Window.WindowFrame.FrameType): InvalidPlanInput =
-    InvalidPlanInput(s"Unknown FrameType $frameType")
-
   def lowerBoundRequiredInWindowFrame(): InvalidPlanInput =
     InvalidPlanInput("LowerBound is required in WindowFrame")
 
-  def unknownFrameBoundary(
-      boundary: proto.Expression.Window.WindowFrame.FrameBoundary.BoundaryCase)
-      : InvalidPlanInput =
-    InvalidPlanInput(s"Unknown FrameBoundary $boundary")
-
   def upperBoundRequiredInWindowFrame(): InvalidPlanInput =
     InvalidPlanInput("UpperBound is required in WindowFrame")
 
@@ -151,36 +153,18 @@ object InvalidInputErrors {
   def intersectDoesNotSupportUnionByName(): InvalidPlanInput =
     InvalidPlanInput("Intersect does not support union_by_name")
 
-  def unsupportedSetOperation(op: Int): InvalidPlanInput =
-    InvalidPlanInput(s"Unsupported set operation $op")
-
-  def joinTypeNotSupported(t: proto.Join.JoinType): InvalidPlanInput =
-    InvalidPlanInput(s"Join type $t is not supported")
-
   def aggregateNeedsPlanInput(): InvalidPlanInput =
     InvalidPlanInput("Aggregate needs a plan input")
 
   def aggregateWithPivotRequiresPivot(): InvalidPlanInput =
     InvalidPlanInput("Aggregate with GROUP_TYPE_PIVOT requires a Pivot")
 
-  def runnerCannotBeEmptyInExecuteExternalCommand(): InvalidPlanInput =
-    InvalidPlanInput("runner cannot be empty in executeExternalCommand")
-
-  def commandCannotBeEmptyInExecuteExternalCommand(): InvalidPlanInput =
-    InvalidPlanInput("command cannot be empty in executeExternalCommand")
-
-  def unexpectedForeachBatchFunction(): InvalidPlanInput =
-    InvalidPlanInput("Unexpected foreachBatch function")
-
   def invalidWithRelationReference(): InvalidPlanInput =
     InvalidPlanInput("Invalid WithRelation reference")
 
   def assertionFailure(message: String): InvalidPlanInput =
     InvalidPlanInput(message)
 
-  def unsupportedMergeActionType(actionType: proto.MergeAction.ActionType): InvalidPlanInput =
-    InvalidPlanInput(s"Unsupported merge action type $actionType")
-
   def unresolvedNamedLambdaVariableRequiresNamePart(): InvalidPlanInput =
     InvalidPlanInput("UnresolvedNamedLambdaVariable requires at least one name part!")
 
@@ -190,15 +174,6 @@ object InvalidInputErrors {
   def sqlCommandExpectsSqlOrWithRelations(other: proto.Relation.RelTypeCase): InvalidPlanInput =
     InvalidPlanInput(s"SQL command expects either a SQL or a WithRelations, but got $other")
 
-  def unknownGroupType(groupType: proto.Aggregate.GroupType): InvalidPlanInput =
-    InvalidPlanInput(s"Unknown group type $groupType")
-
-  def dataSourceIdNotSupported(dataSourceId: Int): InvalidPlanInput =
-    InvalidPlanInput(s"Data source id $dataSourceId is not supported")
-
-  def unknownSubqueryType(subqueryType: proto.SubqueryExpression.SubqueryType): InvalidPlanInput =
-    InvalidPlanInput(s"Unknown subquery type $subqueryType")
-
   def reduceShouldCarryScalarScalaUdf(got: mutable.Buffer[proto.Expression]): InvalidPlanInput =
     InvalidPlanInput(s"reduce should carry a scalar scala udf, but got $got")
 
@@ -207,4 +182,15 @@ object InvalidInputErrors {
 
   def unsupportedUserDefinedFunctionImplementation(clazz: Class[_]): InvalidPlanInput =
     InvalidPlanInput(s"Unsupported UserDefinedFunction implementation: ${clazz}")
+
+  def streamingQueryRunIdMismatch(
+      id: String,
+      runId: String,
+      serverRunId: String): InvalidPlanInput =
+    InvalidPlanInput(
+      s"Run id mismatch for query id $id. Run id in the request $runId " +
+        s"does not match one on the server $serverRunId. The query might have restarted.")
+
+  def streamingQueryNotFound(id: String): InvalidPlanInput =
+    InvalidPlanInput(s"Streaming query $id is not found")
 }