[SPARK-51880][ML][PYTHON][CONNECT] Avoid eager model removal in meta algorithms when collectSubModel is true

[zhengruifeng]

What changes were proposed in this pull request?

Avoid eager model removal in meta algorithms when collectSubModel is true

No matter on classic mode or connect mode, no matter collectSubModel is true or false, __del__ of models and estimators are always invoked in _parallelFitTasks.

There seems to be an internal copy, I add log in __del__ to print the address of objects id(self)/id(self._java_obj) to be deleted and find that the ids are different from these from final model.subModels.

That is to say, internal copying and removal of model/estimator happen in _parallelFitTasks.

metric = eva.evaluate(model.transform(validation, epm[index]))

trigger

spark/python/pyspark/ml/base.py

Lines 256 to 260 in eb48553

	if isinstance(params, dict):
	if params:
	return self.copy(params)._transform(dataset)
	else:
	return self._transform(dataset)

so in this evaluate:
1, copy the model (shares the same ref_id)
2, transform and compute the metric;
3, delete the copied model

Why are the changes needed?

for feature parity

Does this PR introduce any user-facing change?

yes, bug-fix

How was this patch tested?

enabled tests

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

no

[WeichenXu123]

I suggest to use reference count in client side to make a thorough fix, pseudo code:

class RefId:
   ref_count = 1
   
   def add_ref(self):
      self.ref_count += 1
   
   def release(self):
      self.ref_count -= 1
      if self.ref_count == 0:
          # request server to delete model 
          ...

class Model:
   def __del__(self):
      self.ref_id.release()

   def copy(self):
      self.ref_id.add_ref()
      return new Model(self.ref_id)

[WeichenXu123]

This is only a fix for CrossValidator.

If user uses model.copy , it also triggers the error .

[WeichenXu123]

I filed #50707

[zhengruifeng]

close in favor of #50707