[SPARK-52216][SQL][CONNECT] Make InvalidCommandInput a user-facing error

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

- Specify a standardized SQLSTATE, an error condition, and message parameters for each InvalidCommandInput error

### Why are the changes needed?

- Support new error framework

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

No

### How was this patch tested?

`build/sbt "connect/testOnly *SparkConnectProtoSuite"`

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

No

Closes #50934 from heyihong/SPARK-52216.

Authored-by: Yihong He <[email protected]>
Signed-off-by: Hyukjin Kwon <[email protected]>

Author: heyihong

common/utils/src/main/resources/error/error-conditions.json

@@ -2386,6 +2386,12 @@
     ],
     "sqlState" : "42601"
   },
+  "INVALID_BUCKET_COUNT" : {
+    "message" : [
+      "BucketBy must specify a bucket count > 0, received <numBuckets> instead."
+    ],
+    "sqlState" : "22003"
+  },
   "INVALID_BUCKET_FILE" : {
     "message" : [
       "Invalid bucket file: <path>."

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

@@ -16,10 +16,21 @@
  */
 package org.apache.spark.sql.connect.common
 
+import scala.jdk.CollectionConverters._
+
+import org.apache.spark.{SparkThrowable, SparkThrowableHelper}
+
 /**
  * Error thrown when a connect command is not valid.
  */
 final case class InvalidCommandInput(
-    private val message: String = "",
+    private val errorCondition: String,
+    private val messageParameters: Map[String, String] = Map.empty,
     private val cause: Throwable = null)
-    extends Exception(message, cause)
+    extends Exception(SparkThrowableHelper.getMessage(errorCondition, messageParameters), cause)
+    with SparkThrowable {
+
+  override def getCondition: String = errorCondition
+
+  override def getMessageParameters: java.util.Map[String, String] = messageParameters.asJava
+}

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

@@ -211,7 +211,7 @@ object InvalidInputErrors {
     InvalidPlanInput("UnionByName `allowMissingCol` can be true only if `byName` is true.")
 
   def invalidBucketCount(numBuckets: Int): InvalidCommandInput =
-    InvalidCommandInput("BucketBy must specify a bucket count > 0, received $numBuckets instead.")
+    InvalidCommandInput("INVALID_BUCKET_COUNT", Map("numBuckets" -> numBuckets.toString))
 
   def invalidPythonUdtfReturnType(actualType: String): InvalidPlanInput =
     InvalidPlanInput(

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

@@ -658,9 +658,12 @@ class SparkConnectProtoSuite extends PlanTest with SparkConnectPlanTest {
 
   test("Write with invalid bucketBy configuration") {
     val cmd = localRelation.write(bucketByCols = Seq("id"), numBuckets = Some(0))
-    assertThrows[InvalidCommandInput] {
-      transform(cmd)
-    }
+    checkError(
+      exception = intercept[InvalidCommandInput] {
+        transform(cmd)
+      },
+      condition = "INVALID_BUCKET_COUNT",
+      parameters = Map("numBuckets" -> "0"))
   }
 
   test("Write to Path") {