[SPARK-51014][ML][PYTHON][CONNECT] Support RFormula on connect

### What changes were proposed in this pull request?
Support RFormula on connect

### Why are the changes needed?
feature parity

### Does this PR introduce _any_ user-facing change?
yes, new algorithm supported on connect

### How was this patch tested?
added test

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

Closes apache#49703 from zhengruifeng/ml_connect_rformula.

Authored-by: Ruifeng Zheng <[email protected]>
Signed-off-by: Ruifeng Zheng <[email protected]>

Author: zhengruifeng

mllib/src/main/resources/META-INF/services/org.apache.spark.ml.Estimator

@@ -51,6 +51,7 @@ org.apache.spark.ml.recommendation.ALS
 org.apache.spark.ml.fpm.FPGrowth
 
 # feature
+org.apache.spark.ml.feature.RFormula
 org.apache.spark.ml.feature.Imputer
 org.apache.spark.ml.feature.StandardScaler
 org.apache.spark.ml.feature.MaxAbsScaler

mllib/src/main/resources/META-INF/services/org.apache.spark.ml.Transformer

@@ -73,6 +73,7 @@ org.apache.spark.ml.recommendation.ALSModel
 org.apache.spark.ml.fpm.FPGrowthModel
 
 # feature
+org.apache.spark.ml.feature.RFormulaModel
 org.apache.spark.ml.feature.ImputerModel
 org.apache.spark.ml.feature.StandardScalerModel
 org.apache.spark.ml.feature.MaxAbsScalerModel

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

@@ -349,6 +349,8 @@ class RFormulaModel private[feature](
     private[ml] val pipelineModel: PipelineModel)
   extends Model[RFormulaModel] with RFormulaBase with MLWritable {
 
+  private[ml] def this() = this(Identifiable.randomUID("rFormula"), null, null)
+
   @Since("2.0.0")
   override def transform(dataset: Dataset[_]): DataFrame = {
     checkCanTransform(dataset.schema)

python/pyspark/ml/tests/connect/test_parity_feature.py

@@ -34,14 +34,6 @@ def test_count_vectorizer_with_maxDF(self):
     def test_count_vectorizer_from_vocab(self):
         super().test_count_vectorizer_from_vocab()
 
-    @unittest.skip("Need to support.")
-    def test_rformula_force_index_label(self):
-        super().test_rformula_force_index_label()
-
-    @unittest.skip("Need to support.")
-    def test_rformula_string_indexer_order_type(self):
-        super().test_rformula_string_indexer_order_type()
-
     @unittest.skip("Need to support.")
     def test_string_indexer_handle_invalid(self):
         super().test_string_indexer_handle_invalid()

python/pyspark/ml/tests/test_feature.py

@@ -41,6 +41,7 @@
     Normalizer,
     Interaction,
     RFormula,
+    RFormulaModel,
     Tokenizer,
     SQLTransformer,
     RegexTokenizer,
@@ -1295,12 +1296,26 @@ def test_rformula_string_indexer_order_type(self):
         )
         rf = RFormula(formula="y ~ x + s", stringIndexerOrderType="alphabetDesc")
         self.assertEqual(rf.getStringIndexerOrderType(), "alphabetDesc")
-        transformedDF = rf.fit(df).transform(df)
+        model = rf.fit(df)
+        self.assertEqual(rf.uid, model.uid)
+        transformedDF = model.transform(df)
         observed = transformedDF.select("features").collect()
         expected = [[1.0, 0.0], [2.0, 1.0], [0.0, 0.0]]
         for i in range(0, len(expected)):
             self.assertTrue(all(observed[i]["features"].toArray() == expected[i]))
 
+        # save & load
+        with tempfile.TemporaryDirectory(prefix="rformula") as d:
+            rf.write().overwrite().save(d)
+            rf2 = RFormula.load(d)
+            self.assertEqual(str(rf), str(rf2))
+
+            model.write().overwrite().save(d)
+            model2 = RFormulaModel.load(d)
+            # TODO: fix str(model)
+            # self.assertEqual(str(model), str(model2))
+            self.assertEqual(model.getFormula(), model2.getFormula())
+
     def test_string_indexer_handle_invalid(self):
         df = self.spark.createDataFrame([(0, "a"), (1, "d"), (2, None)], ["id", "label"])