eliminate possible initialModels of the direct initialModel

Author: yinxusen

mllib/src/main/scala/org/apache/spark/ml/util/ReadWrite.scala

@@ -324,10 +324,16 @@ private[ml] object DefaultParamsWriter {
     metadataJson
   }
 
-  def saveInitialModel[T <: HasInitialModel[_ <: MLWritable]](instance: T, path: String): Unit = {
+  def saveInitialModel[T <: HasInitialModel[_ <: MLWritable with Params]](
+      instance: T, path: String): Unit = {
     if (instance.isDefined(instance.initialModel)) {
       val initialModelPath = new Path(path, "initialModel").toString
       val initialModel = instance.getOrDefault(instance.initialModel)
+      // When saving, only keep the direct initialModel by eliminating possible initialModels of the
+      // direct initialModel, to avoid unnecessary deep recursion of initialModel.
+      if (initialModel.hasParam("initialModel")) {
+        initialModel.clear(initialModel.getParam("initialModel"))
+      }
       initialModel.save(initialModelPath)
     }
   }

mllib/src/test/scala/org/apache/spark/ml/clustering/KMeansSuite.scala

@@ -186,6 +186,18 @@ class KMeansSuite extends SparkFunSuite with MLlibTestSparkContext with DefaultR
     // k is not ignored after initialModel is cleared
     assert(kmeans.setK(k - 1).getK === k - 1)
   }
+
+  test("Eliminate possible initialModels of the direct initialModel") {
+    val randomModel = KMeansSuite.generateRandomKMeansModel(dim, k)
+    val kmeans = new KMeans().setK(k).setMaxIter(1).setInitialModel(randomModel)
+    val firstLevelModel = kmeans.fit(dataset)
+    val secondLevelModel = kmeans.setInitialModel(firstLevelModel).fit(dataset)
+    assert(secondLevelModel.getInitialModel
+      .isSet(secondLevelModel.getInitialModel.getParam("initialModel")))
+    val savedThenLoadedModel = testDefaultReadWrite(secondLevelModel, testParams = false)
+    assert(!savedThenLoadedModel.getInitialModel
+      .isSet(savedThenLoadedModel.getInitialModel.getParam("initialModel")))
+  }
 }
 
 object KMeansSuite {