[SPARK-52337][CONNECT] Make InvalidPlanInput a user-facing error

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

This PR makes `InvalidPlanInput` a user-facing error by:
1. Refactoring `InvalidPlanInput` to implement `SparkThrowable` interface
2. Adding proper error condition (the default value is `INTERNAL_ERROR`) and message parameters support
3. Updating error creation in `InvalidInputErrors` to use the new error format
4. Adding test coverage for the new error handling

### Why are the changes needed?

The PR follows the pattern of other Spark errors by implementing the `SparkThrowable` interface, which provides a standardized way to handle and display errors to users. This makes error messages more consistent and easier to understand across the Spark ecosystem.

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

No

### How was this patch tested?

`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#51054 from heyihong/SPARK-52337.

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

Author: heyihong

sql/connect/common/src/main/scala/org/apache/spark/sql/connect/common/InvalidPlanInput.scala

@@ -16,10 +16,38 @@
  */
 package org.apache.spark.sql.connect.common
 
+import scala.jdk.CollectionConverters._
+
+import org.apache.spark.{SparkThrowable, SparkThrowableHelper}
+
 /**
  * Error thrown when a connect plan is not valid.
  */
 final case class InvalidPlanInput(
-    private val message: String = "",
-    private val cause: Throwable = None.orNull)
-    extends Exception(message, cause)
+    private val errorCondition: String,
+    private val messageParameters: Map[String, String],
+    private val causeOpt: Option[Throwable])
+    extends Exception(
+      SparkThrowableHelper.getMessage(errorCondition, messageParameters),
+      causeOpt.orNull)
+    with SparkThrowable {
+
+  override def getCondition: String = errorCondition
+
+  override def getMessageParameters: java.util.Map[String, String] = messageParameters.asJava
+}
+
+object InvalidPlanInput {
+
+  def apply(message: String): InvalidPlanInput =
+    InvalidPlanInput(
+      errorCondition = "INTERNAL_ERROR",
+      messageParameters = Map("message" -> message),
+      causeOpt = None)
+
+  def apply(errorCondition: String, messageParameters: Map[String, String]): InvalidPlanInput =
+    InvalidPlanInput(
+      errorCondition = errorCondition,
+      messageParameters = messageParameters,
+      causeOpt = None)
+}

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

@@ -19,22 +19,13 @@ package org.apache.spark.sql.connect.planner
 
 import scala.collection.mutable
 
-import org.apache.spark.SparkThrowableHelper
 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 {
 
-  // invalidPlanInput is a helper function to facilitate the migration of InvalidInputErrors
-  // to support NERF.
-  private def invalidPlanInput(
-      errorCondition: String,
-      messageParameters: Map[String, String] = Map.empty): InvalidPlanInput = {
-    InvalidPlanInput(SparkThrowableHelper.getMessage(errorCondition, messageParameters))
-  }
-
   def unknownRelationNotSupported(rel: proto.Relation): InvalidPlanInput =
     InvalidPlanInput(s"${rel.getUnknown} not supported.")
 
@@ -97,7 +88,7 @@ object InvalidInputErrors {
     InvalidPlanInput("Schema for LocalRelation is required when the input data is not provided.")
 
   def invalidSchemaStringNonStructType(schema: String, dataType: DataType): InvalidPlanInput =
-    invalidPlanInput(
+    InvalidPlanInput(
       "INVALID_SCHEMA.NON_STRUCT_TYPE",
       Map("inputSchema" -> quoteByDefault(schema), "dataType" -> toSQLType(dataType)))
 
@@ -114,7 +105,7 @@ object InvalidInputErrors {
     InvalidPlanInput(s"Does not support $what")
 
   def invalidSchemaTypeNonStruct(dataType: DataType): InvalidPlanInput =
-    invalidPlanInput("INVALID_SCHEMA_TYPE_NON_STRUCT", Map("dataType" -> toSQLType(dataType)))
+    InvalidPlanInput("INVALID_SCHEMA_TYPE_NON_STRUCT", Map("dataType" -> toSQLType(dataType)))
 
   def expressionIdNotSupported(exprId: Int): InvalidPlanInput =
     InvalidPlanInput(s"Expression with ID: $exprId is not supported")

sql/connect/server/src/test/scala/org/apache/spark/sql/connect/planner/InvalidInputErrorsSuite.scala

@@ -95,6 +95,18 @@ class InvalidInputErrorsSuite extends PlanTest with SparkConnectPlanTest {
           .build()
 
         proto.Relation.newBuilder().setRead(read).build()
+      }),
+    TestCase(
+      name = "Deduplicate needs input",
+      expectedErrorCondition = "INTERNAL_ERROR",
+      expectedParameters = Map("message" -> "Deduplicate needs a plan input"),
+      invalidInput = {
+        val deduplicate = proto.Deduplicate
+          .newBuilder()
+          .setAllColumnsAsKeys(true)
+          .build()
+
+        proto.Relation.newBuilder().setDeduplicate(deduplicate).build()
       }))
 
   // Run all test cases