[SPARK-53362][ML][CONNECT] Fix IDFModel local loader bug

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

Fix IDFModel local loader bug:

https://github.com/apache/spark/blob/50a2ebe87f5ca4dbcc732cf3a543d6ebaea856ad/mllib/src/main/scala/org/apache/spark/ml/feature/IDF.scala#L234

This line code shouldn't be executed when it is loading from local disk. Otherwise file formatting error might be raised.

### Why are the changes needed?

Bugfix.

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

No.

### How was this patch tested?

updated test in mllib/src/test/scala/org/apache/spark/ml/feature/IDFSuite.scala

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

no.

Closes #52111 from WeichenXu123/SPARK-53362.

Authored-by: Weichen Xu <[email protected]>
Signed-off-by: Weichen Xu <[email protected]>

Author: WeichenXu123

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

@@ -231,7 +231,6 @@ object IDFModel extends MLReadable[IDFModel] {
     override def load(path: String): IDFModel = {
       val metadata = DefaultParamsReader.loadMetadata(path, sparkSession, className)
       val dataPath = new Path(path, "data").toString
-      val data = sparkSession.read.parquet(dataPath)
 
       val model = if (majorVersion(metadata.sparkVersion) >= 3) {
         val data = ReadWriteUtils.loadObject[Data](dataPath, sparkSession, deserializeData)
@@ -240,6 +239,7 @@ object IDFModel extends MLReadable[IDFModel] {
           new feature.IDFModel(OldVectors.fromML(data.idf), data.docFreq, data.numDocs)
         )
       } else {
+        val data = sparkSession.read.parquet(dataPath)
         val Row(idf: Vector) = MLUtils.convertVectorColumnsToML(data, "idf")
           .select("idf")
           .head()

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

@@ -120,9 +120,12 @@ class IDFSuite extends MLTest with DefaultReadWriteTest {
       new OldIDFModel(Vectors.dense(1.0, 2.0), Array(1, 2), 2))
       .setInputCol("myInputCol")
       .setOutputCol("myOutputCol")
-    val newInstance = testDefaultReadWrite(instance)
-    assert(newInstance.idf === instance.idf)
-    assert(newInstance.docFreq === instance.docFreq)
-    assert(newInstance.numDocs === instance.numDocs)
+
+    for (testSaveToLocal <- Seq(false, true)) {
+      val newInstance = testDefaultReadWrite(instance, testSaveToLocal = testSaveToLocal)
+      assert(newInstance.idf === instance.idf)
+      assert(newInstance.docFreq === instance.docFreq)
+      assert(newInstance.numDocs === instance.numDocs)
+    }
   }
 }

python/pyspark/ml/util.py

@@ -339,6 +339,9 @@ def remote_call() -> Any:
                     session.client.execute_command(create_summary_command)  # type: ignore
 
                     return remote_call()
+
+                # for other unexpected error, re-raise it.
+                raise
         else:
             return f(self, name, *args)