[SPARK-51214][ML][PYTHON][CONNECT] Don't eagerly remove the cached models for fit_transform

[zhengruifeng]

What changes were proposed in this pull request?

Don't eagerly remove the cached models for fit_transform:
1, still keep the Delete ml command protobuf, but no longer call it in __del__ in the python client side;
2, build the ml cache with guava CacheBuilder and soft references, and specify the maximum size and time out.

Why are the changes needed?

a common ml pipeline pattern is fit_transform:

def fit_transform(df):
    model = estimator.fit(df)
    return model.transform(df)

df2 = fit_transform(df)
df2.count()

existing implementation eagerly deletes the intermediate model from the ml cache, right after fit_transform, and thus causes NPE

pyspark.errors.exceptions.connect.SparkConnectGrpcException: (java.lang.NullPointerException) Cannot invoke "org.apache.spark.ml.Model.copy(org.apache.spark.ml.param.ParamMap)" because "model" is null

JVM stacktrace:
java.lang.NullPointerException
	at org.apache.spark.sql.connect.ml.ModelAttributeHelper.transform(MLHandler.scala:68)
	at org.apache.spark.sql.connect.ml.MLHandler$.transformMLRelation(MLHandler.scala:313)
	at org.apache.spark.sql.connect.planner.SparkConnectPlanner.$anonfun$transformRelation$1(SparkConnectPlanner.scala:231)
	at org.apache.spark.sql.connect.service.SessionHolder.$anonfun$usePlanCache$3(SessionHolder.scala:477)
	at scala.Option.getOrElse(Option.scala:201)
	at org.apache.spark.sql.connect.service.SessionHolder.usePlanCache(SessionHolder.scala:476)
	at org.apache.spark.sql.connect.planner.SparkConnectPlanner.transformRelation(SparkConnectPlanner.scala:147)
	at org.apache.spark.sql.connect.planner.SparkConnectPlanner.transformRelation(SparkConnectPlanner.scala:133)
	at org.apache.spark.sql.connect.planner.SparkConnectPlanner.transformRelationalGroupedAggregate(SparkConnectPlanner.scala:2318)
	at org.apache.spark.sql.connect.planner.SparkConnectPlanner.transformAggregate(SparkConnectPlanner.scala:2299)
	at org.apache.spark.sql.connect.planner.SparkConnectPlanner.$anonfun$transformRelation$1(SparkConnectPlanner.scala:165)
	at org.apache.spark.sql.connect.service.SessionHolder.$anonfun$usePlanCache$3(SessionHolder.scala:477)

Does this PR introduce any user-facing change?

yes

How was this patch tested?

added tests

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

no

[zhengruifeng]

we can add configs later (in 4.1)

[grundprinzip]

What happens if we can't find an item in the cache, do we throw a proper exception?

[zhengruifeng]

currently, it throws a NPE

let me add an error class for it