[SPARK-52259][ML][CONNECT] Fix Param class binary compatibility

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

This PR fixes the following issue:

Integrating third-party ML libraries with Spark 4.0.0 may encounter compatiblity issues due to an interface change. As an example, users may encounter the following error when using xgboost4j with Spark:
```
NoSuchMethodError: 'void org.apache.spark.ml.param.Param.<init>(org.apache.spark.ml.util.Identifiable, java.lang.String, java.lang.String)'
```

### Why are the changes needed?

Fix binary compatibility

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

No.

### How was this patch tested?

Manually.

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

No.

Closes #50981 from WeichenXu123/SPARK-52259.

Lead-authored-by: Weichen Xu <weichen.xu@databricks.com>
Co-authored-by: WeichenXu <weichen.xu@databricks.com>
Signed-off-by: Weichen Xu <weichen.xu@databricks.com>

Author: WeichenXu123

mllib/src/main/scala/org/apache/spark/ml/classification/LogisticRegression.scala

@@ -191,8 +191,10 @@ private[classification] trait LogisticRegressionParams extends ProbabilisticClas
    * @group expertParam
    */
   @Since("2.2.0")
-  val lowerBoundsOnCoefficients: Param[Matrix] = new Param(this, "lowerBoundsOnCoefficients",
-    "The lower bounds on coefficients if fitting under bound constrained optimization.")
+  val lowerBoundsOnCoefficients: Param[Matrix] = new Param(this.uid, "lowerBoundsOnCoefficients",
+    "The lower bounds on coefficients if fitting under bound constrained optimization.",
+    classOf[Matrix]
+  )
 
   /** @group expertGetParam */
   @Since("2.2.0")
@@ -208,8 +210,10 @@ private[classification] trait LogisticRegressionParams extends ProbabilisticClas
    * @group expertParam
    */
   @Since("2.2.0")
-  val upperBoundsOnCoefficients: Param[Matrix] = new Param(this, "upperBoundsOnCoefficients",
-    "The upper bounds on coefficients if fitting under bound constrained optimization.")
+  val upperBoundsOnCoefficients: Param[Matrix] = new Param(this.uid, "upperBoundsOnCoefficients",
+    "The upper bounds on coefficients if fitting under bound constrained optimization.",
+    classOf[Matrix]
+  )
 
   /** @group expertGetParam */
   @Since("2.2.0")
@@ -224,8 +228,10 @@ private[classification] trait LogisticRegressionParams extends ProbabilisticClas
    * @group expertParam
    */
   @Since("2.2.0")
-  val lowerBoundsOnIntercepts: Param[Vector] = new Param(this, "lowerBoundsOnIntercepts",
-    "The lower bounds on intercepts if fitting under bound constrained optimization.")
+  val lowerBoundsOnIntercepts: Param[Vector] = new Param(this.uid, "lowerBoundsOnIntercepts",
+    "The lower bounds on intercepts if fitting under bound constrained optimization.",
+    classOf[Vector]
+  )
 
   /** @group expertGetParam */
   @Since("2.2.0")
@@ -240,8 +246,10 @@ private[classification] trait LogisticRegressionParams extends ProbabilisticClas
    * @group expertParam
    */
   @Since("2.2.0")
-  val upperBoundsOnIntercepts: Param[Vector] = new Param(this, "upperBoundsOnIntercepts",
-    "The upper bounds on intercepts if fitting under bound constrained optimization.")
+  val upperBoundsOnIntercepts: Param[Vector] = new Param(this.uid, "upperBoundsOnIntercepts",
+    "The upper bounds on intercepts if fitting under bound constrained optimization.",
+    classOf[Vector]
+  )
 
   /** @group expertGetParam */
   @Since("2.2.0")

mllib/src/main/scala/org/apache/spark/ml/classification/MultilayerPerceptronClassifier.scala

@@ -74,8 +74,8 @@ private[classification] trait MultilayerPerceptronParams extends ProbabilisticCl
    * @group expertParam
    */
   @Since("2.0.0")
-  final val initialWeights: Param[Vector] = new Param[Vector](this, "initialWeights",
-    "The initial weights of the model")
+  final val initialWeights: Param[Vector] = new Param[Vector](this.uid, "initialWeights",
+    "The initial weights of the model", classOf[Vector])
 
   /** @group expertGetParam */
   @Since("2.0.0")

mllib/src/main/scala/org/apache/spark/ml/feature/ElementwiseProduct.scala

@@ -43,7 +43,9 @@ class ElementwiseProduct @Since("1.4.0") (@Since("1.4.0") override val uid: Stri
    * @group param
    */
   @Since("2.0.0")
-  val scalingVec: Param[Vector] = new Param(this, "scalingVec", "vector for hadamard product")
+  val scalingVec: Param[Vector] = new Param(
+    this.uid, "scalingVec", "vector for hadamard product", classOf[Vector]
+  )
 
   /** @group setParam */
   @Since("2.0.0")

mllib/src/main/scala/org/apache/spark/ml/param/params.scala

@@ -23,7 +23,6 @@ import java.util.{List => JList}
 import scala.annotation.varargs
 import scala.collection.mutable
 import scala.jdk.CollectionConverters._
-import scala.reflect.ClassTag
 
 import org.json4s._
 import org.json4s.jackson.JsonMethods._
@@ -46,20 +45,23 @@ import org.apache.spark.util.SizeEstimator
  *                See [[ParamValidators]] for factory methods for common validation functions.
  * @tparam T param value type
  */
-class Param[T: ClassTag](
-    val parent: String, val name: String, val doc: String, val isValid: T => Boolean)
-  extends Serializable {
+class Param[T](
+    val parent: String, val name: String, val doc: String, val isValid: T => Boolean,
+    val dataClass: Class[T]
+  ) extends Serializable {
 
-  // Spark Connect ML needs T type information which has been erased when compiling,
-  // Use classTag to preserve the T type.
-  val paramValueClassTag = implicitly[ClassTag[T]]
+  def this(parent: String, name: String, doc: String, isValid: T => Boolean) =
+    this(parent, name, doc, isValid, null)
 
   def this(parent: Identifiable, name: String, doc: String, isValid: T => Boolean) =
     this(parent.uid, name, doc, isValid)
 
   def this(parent: String, name: String, doc: String) =
     this(parent, name, doc, ParamValidators.alwaysTrue[T])
 
+  def this(parent: String, name: String, doc: String, dataClass: Class[T]) =
+    this(parent, name, doc, ParamValidators.alwaysTrue[T], dataClass)
+
   def this(parent: Identifiable, name: String, doc: String) = this(parent.uid, name, doc)
 
   /**
@@ -329,7 +331,7 @@ object ParamValidators {
  * Specialized version of `Param[Double]` for Java.
  */
 class DoubleParam(parent: String, name: String, doc: String, isValid: Double => Boolean)
-  extends Param[Double](parent, name, doc, isValid) {
+  extends Param[Double](parent, name, doc, isValid, classOf[Double]) {
 
   def this(parent: String, name: String, doc: String) =
     this(parent, name, doc, ParamValidators.alwaysTrue)
@@ -387,7 +389,7 @@ private[param] object DoubleParam {
  * Specialized version of `Param[Int]` for Java.
  */
 class IntParam(parent: String, name: String, doc: String, isValid: Int => Boolean)
-  extends Param[Int](parent, name, doc, isValid) {
+  extends Param[Int](parent, name, doc, isValid, classOf[Int]) {
 
   def this(parent: String, name: String, doc: String) =
     this(parent, name, doc, ParamValidators.alwaysTrue)
@@ -414,7 +416,7 @@ class IntParam(parent: String, name: String, doc: String, isValid: Int => Boolea
  * Specialized version of `Param[Float]` for Java.
  */
 class FloatParam(parent: String, name: String, doc: String, isValid: Float => Boolean)
-  extends Param[Float](parent, name, doc, isValid) {
+  extends Param[Float](parent, name, doc, isValid, classOf[Float]) {
 
   def this(parent: String, name: String, doc: String) =
     this(parent, name, doc, ParamValidators.alwaysTrue)
@@ -473,7 +475,7 @@ private object FloatParam {
  * Specialized version of `Param[Long]` for Java.
  */
 class LongParam(parent: String, name: String, doc: String, isValid: Long => Boolean)
-  extends Param[Long](parent, name, doc, isValid) {
+  extends Param[Long](parent, name, doc, isValid, classOf[Long]) {
 
   def this(parent: String, name: String, doc: String) =
     this(parent, name, doc, ParamValidators.alwaysTrue)
@@ -500,7 +502,8 @@ class LongParam(parent: String, name: String, doc: String, isValid: Long => Bool
  * Specialized version of `Param[Boolean]` for Java.
  */
 class BooleanParam(parent: String, name: String, doc: String) // No need for isValid
-  extends Param[Boolean](parent, name, doc) {
+  extends Param[Boolean](parent, name, doc, ParamValidators.alwaysTrue[Boolean], classOf[Boolean])
+{
 
   def this(parent: Identifiable, name: String, doc: String) = this(parent.uid, name, doc)
 
@@ -521,7 +524,7 @@ class BooleanParam(parent: String, name: String, doc: String) // No need for isV
  * Specialized version of `Param[Array[String]]` for Java.
  */
 class StringArrayParam(parent: Params, name: String, doc: String, isValid: Array[String] => Boolean)
-  extends Param[Array[String]](parent, name, doc, isValid) {
+  extends Param[Array[String]](parent.uid, name, doc, isValid, classOf[Array[String]]) {
 
   def this(parent: Params, name: String, doc: String) =
     this(parent, name, doc, ParamValidators.alwaysTrue)
@@ -544,7 +547,7 @@ class StringArrayParam(parent: Params, name: String, doc: String, isValid: Array
  * Specialized version of `Param[Array[Double]]` for Java.
  */
 class DoubleArrayParam(parent: Params, name: String, doc: String, isValid: Array[Double] => Boolean)
-  extends Param[Array[Double]](parent, name, doc, isValid) {
+  extends Param[Array[Double]](parent.uid, name, doc, isValid, classOf[Array[Double]]) {
 
   def this(parent: Params, name: String, doc: String) =
     this(parent, name, doc, ParamValidators.alwaysTrue)
@@ -576,7 +579,9 @@ class DoubleArrayArrayParam(
     name: String,
     doc: String,
     isValid: Array[Array[Double]] => Boolean)
-  extends Param[Array[Array[Double]]](parent, name, doc, isValid) {
+  extends Param[Array[Array[Double]]](
+    parent.uid, name, doc, isValid, classOf[Array[Array[Double]]]
+  ) {
 
   def this(parent: Params, name: String, doc: String) =
     this(parent, name, doc, ParamValidators.alwaysTrue)
@@ -610,7 +615,7 @@ class DoubleArrayArrayParam(
  * Specialized version of `Param[Array[Int]]` for Java.
  */
 class IntArrayParam(parent: Params, name: String, doc: String, isValid: Array[Int] => Boolean)
-  extends Param[Array[Int]](parent, name, doc, isValid) {
+  extends Param[Array[Int]](parent.uid, name, doc, isValid, classOf[Array[Int]]) {
 
   def this(parent: Params, name: String, doc: String) =
     this(parent, name, doc, ParamValidators.alwaysTrue)

mllib/src/test/java/org/apache/spark/ml/param/JavaTestParams.java

@@ -21,7 +21,6 @@
 import java.util.List;
 
 import org.apache.spark.ml.util.Identifiable$;
-import scala.reflect.ClassTag;
 
 /**
  * A subclass of Params for testing.
@@ -111,7 +110,7 @@ private void init() {
       ParamValidators.inRange(0.0, 1.0));
     List<String> validStrings = Arrays.asList("a", "b");
     myStringParam_ = new Param<>(this, "myStringParam", "this is a string param",
-      ParamValidators.inArray(validStrings), ClassTag.apply(String.class));
+      ParamValidators.inArray(validStrings));
     myDoubleArrayParam_ =
       new DoubleArrayParam(this, "myDoubleArrayParam", "this is a double param");
 

sql/connect/server/src/main/scala/org/apache/spark/sql/connect/ml/MLUtils.scala

@@ -157,9 +157,22 @@ private[ml] object MLUtils {
           }
 
         case _ =>
-          reconcileParam(
-            p.paramValueClassTag.runtimeClass,
-            LiteralValueProtoConverter.toCatalystValue(literal))
+          val paramValue = LiteralValueProtoConverter.toCatalystValue(literal)
+          val paramType: Class[_] = if (p.dataClass == null) {
+            if (paramValue.isInstanceOf[String]) {
+              classOf[String]
+            } else if (paramValue.isInstanceOf[Boolean]) {
+              classOf[Boolean]
+            } else {
+              throw MlUnsupportedException(
+                "Spark Connect ML requires the customized ML Param class setting 'dataClass' " +
+                  "parameter if the param value type is not String or Boolean type, " +
+                  s"but the param $name does not have the required dataClass.")
+            }
+          } else {
+            p.dataClass
+          }
+          reconcileParam(paramType, paramValue)
       }
       instance.set(p, value)
     }