[SPARK-22799][ML] Bucketizer should throw exception if single- and multi-column params are both set

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

@@ -34,9 +34,9 @@ import org.apache.spark.sql.types.{DoubleType, StructField, StructType}
 /**
  * `Bucketizer` maps a column of continuous features to a column of feature buckets. Since 2.3.0,
  * `Bucketizer` can map multiple columns at once by setting the `inputCols` parameter. Note that
- * when both the `inputCol` and `inputCols` parameters are set, a log warning will be printed and
- * only `inputCol` will take effect, while `inputCols` will be ignored. The `splits` parameter is
- * only used for single column usage, and `splitsArray` is for multiple columns.
+ * when both the `inputCol` and `inputCols` parameters are set, an Exception will be thrown. The
+ * `splits` parameter is only used for single column usage, and `splitsArray` is for multiple
+ * columns.
  */
 @Since("1.4.0")
 final class Bucketizer @Since("1.4.0") (@Since("1.4.0") override val uid: String)
@@ -137,18 +137,10 @@ final class Bucketizer @Since("1.4.0") (@Since("1.4.0") override val uid: String
   /**
    * Determines whether this `Bucketizer` is going to map multiple columns. If and only if
    * `inputCols` is set, it will map multiple columns. Otherwise, it just maps a column specified
-   * by `inputCol`. A warning will be printed if both are set.
+   * by `inputCol`.
    */
   private[feature] def isBucketizeMultipleColumns(): Boolean = {
-    if (isSet(inputCols) && isSet(inputCol)) {
-      logWarning("Both `inputCol` and `inputCols` are set, we ignore `inputCols` and this " +
-        "`Bucketizer` only map one column specified by `inputCol`")
-      false
-    } else if (isSet(inputCols)) {
-      true
-    } else {
-      false
-    }
+    isSet(inputCols)
   }
 
   @Since("2.0.0")
@@ -201,6 +193,13 @@ final class Bucketizer @Since("1.4.0") (@Since("1.4.0") override val uid: String
 
   @Since("1.4.0")
   override def transformSchema(schema: StructType): StructType = {
+    ParamValidators.checkMultiColumnParams(this)
+    if (isSet(inputCol) && isSet(splitsArray)) {
+      ParamValidators.raiseIncompatibleParamsException("inputCol", "splitsArray")
+    }
+    if (isSet(inputCols) && isSet(splits)) {
+      ParamValidators.raiseIncompatibleParamsException("inputCols", "splits")
+    }
     if (isBucketizeMultipleColumns()) {
       var transformedSchema = schema
       $(inputCols).zip($(outputCols)).zipWithIndex.map { case ((inputCol, outputCol), idx) =>

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

@@ -31,6 +31,7 @@ import org.json4s.jackson.JsonMethods._
 import org.apache.spark.SparkException
 import org.apache.spark.annotation.{DeveloperApi, Since}
 import org.apache.spark.ml.linalg.{JsonMatrixConverter, JsonVectorConverter, Matrix, Vector}
+import org.apache.spark.ml.param.shared._
 import org.apache.spark.ml.util.Identifiable
 
 /**
@@ -249,6 +250,31 @@ object ParamValidators {
   def arrayLengthGt[T](lowerBound: Double): Array[T] => Boolean = { (value: Array[T]) =>
     value.length > lowerBound
   }
+
+  /**
+   * Checks that either inputCols and outputCols are set or inputCol and outputCol are set. If
+   * this is not true, an `IllegalArgumentException` is raised.
+   * @param model
+   */
+  private[spark] def checkMultiColumnParams(model: Params): Unit = {
+    model match {
+      case m: HasInputCols with HasInputCol if m.isSet(m.inputCols) && m.isSet(m.inputCol) =>
+        raiseIncompatibleParamsException("inputCols", "inputCol")
+      case m: HasOutputCols with HasInputCol if m.isSet(m.outputCols) && m.isSet(m.inputCol) =>
+        raiseIncompatibleParamsException("outputCols", "inputCol")
+      case m: HasInputCols with HasOutputCol if m.isSet(m.inputCols) && m.isSet(m.outputCol) =>
+        raiseIncompatibleParamsException("inputCols", "outputCol")
+      case m: HasOutputCols with HasOutputCol if m.isSet(m.outputCols) && m.isSet(m.outputCol) =>
+        raiseIncompatibleParamsException("outputCols", "outputCol")
+      case _ =>
+    }
+  }
+
+  private[spark] def raiseIncompatibleParamsException(
+      paramName1: String,
+      paramName2: String): Unit = {
+    throw new IllegalArgumentException(s"`$paramName1` and `$paramName2` cannot be both set.")
+  }
 }
 
 // specialize primitive-typed params because Java doesn't recognize scala.Double, scala.Int, ...

mllib/src/test/scala/org/apache/spark/ml/feature/BucketizerSuite.scala

@@ -401,15 +401,9 @@ class BucketizerSuite extends SparkFunSuite with MLlibTestSparkContext with Defa
     }
   }
 
-  test("Both inputCol and inputCols are set") {
-    val bucket = new Bucketizer()
-      .setInputCol("feature1")
-      .setOutputCol("result")
-      .setSplits(Array(-0.5, 0.0, 0.5))
-      .setInputCols(Array("feature1", "feature2"))
-
-    // When both are set, we ignore `inputCols` and just map the column specified by `inputCol`.
-    assert(bucket.isBucketizeMultipleColumns() == false)
+  test("assert exception is thrown is both multi-column and single-column params are set") {
+    val df = Seq((0.5, 0.3), (0.5, -0.4)).toDF("feature1", "feature2")
+    ParamsSuite.testMultiColumnParams(classOf[Bucketizer], df)
   }
 }
 

mllib/src/test/scala/org/apache/spark/ml/param/ParamsSuite.scala

@@ -20,8 +20,11 @@ package org.apache.spark.ml.param
 import java.io.{ByteArrayOutputStream, ObjectOutputStream}
 
 import org.apache.spark.SparkFunSuite
+import org.apache.spark.ml.{Estimator, Transformer}
 import org.apache.spark.ml.linalg.{Vector, Vectors}
+import org.apache.spark.ml.param.shared.{HasInputCol, HasInputCols, HasOutputCol, HasOutputCols}
 import org.apache.spark.ml.util.MyParams
+import org.apache.spark.sql.Dataset
 
 class ParamsSuite extends SparkFunSuite {
 
@@ -430,4 +433,45 @@ object ParamsSuite extends SparkFunSuite {
     require(copyReturnType === obj.getClass,
       s"${clazz.getName}.copy should return ${clazz.getName} instead of ${copyReturnType.getName}.")
   }
+
+  /**
+   * Checks that the class throws an exception in case both `inputCols` and `inputCol` are set and
+   * in case both `outputCols` and `outputCol` are set.
+   * These checks are performed only whether the class extends respectively both `HasInputCols` and
+   * `HasInputCol` and both `HasOutputCols` and `HasOutputCol`.
+   *
+   * @param paramsClass The Class to be checked
+   * @param dataset A `Dataset` to use in the tests
+   */
+  def testMultiColumnParams(paramsClass: Class[_ <: Params], dataset: Dataset[_]): Unit = {
+    val cols = dataset.columns
+
+    if (paramsClass.isAssignableFrom(classOf[HasInputCols])
+        && paramsClass.isAssignableFrom(classOf[HasInputCol])) {
+      val model = paramsClass.newInstance()
+      model.set(model.asInstanceOf[HasInputCols].inputCols, cols)
+      model.set(model.asInstanceOf[HasInputCol].inputCol, cols(0))
+      val e = intercept[IllegalArgumentException] {
+        model match {
+          case t: Transformer => t.transform(dataset)
+          case e: Estimator[_] => e.fit(dataset)
+        }
+      }
+      assert(e.getMessage.contains("cannot be both set"))
+    }
+
+    if (paramsClass.isAssignableFrom(classOf[HasOutputCols])
+      && paramsClass.isAssignableFrom(classOf[HasOutputCol])) {
+      val model = paramsClass.newInstance()
+      model.set(model.asInstanceOf[HasOutputCols].outputCols, cols)
+      model.set(model.asInstanceOf[HasOutputCol].outputCol, cols(0))
+      val e = intercept[IllegalArgumentException] {
+        model match {
+          case t: Transformer => t.transform(dataset)
+          case e: Estimator[_] => e.fit(dataset)
+        }
+      }
+      assert(e.getMessage.contains("cannot be both set"))
+    }
+  }
 }