[pyspark] Validate the validation indicator column type.

Author: trivialfis

python-package/xgboost/spark/core.py

@@ -51,6 +51,7 @@
 from pyspark.sql.functions import col, countDistinct, pandas_udf, rand, struct
 from pyspark.sql.types import (
     ArrayType,
+    BooleanType,
     DoubleType,
     FloatType,
     IntegerType,
@@ -844,6 +845,11 @@ def _prepare_input(self, dataset: DataFrame) -> Tuple[DataFrame, FeatureProp]:
         sc = _get_spark_session().sparkContext
         max_concurrent_tasks = _get_max_num_concurrent_tasks(sc)
 
+        if feature_prop.has_validation_col:
+            dtype = dataset.schema[alias.valid].dataType
+            if not isinstance(dtype, BooleanType):
+                raise TypeError("The validation indicator must be boolean type.")
+
         if num_workers > max_concurrent_tasks:
             get_logger(self.__class__.__name__).warning(
                 "The num_workers %s set for xgboost distributed "

tests/test_distributed/test_with_spark/test_spark_local.py

@@ -1308,6 +1308,27 @@ def test_classifier_xgb_summary_with_validation(
             atol=1e-3,
         )
 
+    def test_valid_type(self, spark: SparkSession) -> None:
+        # integer as indicator is invalid
+        df_train = spark.createDataFrame(
+            [
+                (Vectors.dense(1.0, 2.0, 3.0), 0, 0),
+                (Vectors.sparse(3, {1: 1.0, 2: 5.5}), 1, 0),
+                (Vectors.dense(4.0, 5.0, 6.0), 0, 1),
+                (Vectors.sparse(3, {1: 6.0, 2: 7.5}), 1, 1),
+            ],
+            ["features", "label", "isVal"],
+        )
+        reg = SparkXGBRegressor(
+            features_col="features",
+            label_col="label",
+            validation_indicator_col="isVal",
+        )
+        # Use Exception here since spark might return a py4j error instead of specific
+        # exception.
+        with pytest.raises(Exception, match="The validation indicator must be boolean"):
+            reg.fit(df_train)
+
 
 class XgboostLocalTest(SparkTestCase):
     def setUp(self):