[spark] Close the commit and release the cached RDD in remove_unexisting_files

[cxzl25]

Purpose

SparkRemoveUnexistingFiles creates a TableCommit on the executor inside foreachPartition but never closes it. TableCommitImpl.close() shuts down the underlying commit and the maintainExecutor thread pool, so every invocation of the procedure leaks a thread pool on a long-lived executor. Other call sites such as PaimonPartitionManagement already wrap the commit in try/finally.

The intermediate RDD is also cached and never unpersisted. buildRDD() returned a child of the cached RDD to the caller, so the cache could only be released after the caller finished, which never happened. This change turns it into execute(), which collects the result inside the method and unpersists the cache in a finally block.

Behaviour of the procedure is unchanged; only resource lifecycle is affected.

Tests

[JingsongLi]

Resource cleanup in this procedure has practical value. The success-path cache release is improved, but the new commit close introduces an asynchronous-maintenance regression, and the cache finally block still starts too late to cover commit failures. Details are inline.

A focused JDK 8 probe using the head TableCommitImpl reproduced interruption with commit(identifier, messages) followed by close; using the one-shot commit(messages) overload completed maintenance before closing. I did not rerun the Spark procedure suite.

[JingsongLi]

[P2] Use the batch commit lifecycle before closing this committer

The existing call to commit(Long.MaxValue, messages) takes the streaming overload and does not set TableCommitImpl.batchCommitted. With snapshot.expire.execution-mode=async, maintenance runs on maintainExecutor; the newly added close immediately calls shutdownNow(), so expiration/tag maintenance can be cancelled or interrupted after the procedure reports a successful commit. This is introduced by adding close to that overload. A focused probe with the head TableCommitImpl reproduced the interruption.

Use commit(messages), which selects the one-shot batch lifecycle and performs maintenance inline, or explicitly await completion and propagate its failure before closing. The one-argument overload passed the control probe.

[JingsongLi]

[P2] Include the commit action inside the cache cleanup scope

The cached RDD is materialized by foreachPartition at lines 88–104 before this try starts. If committing a partition fails after task retries, execution never reaches this finally and the persisted blocks remain in the long-lived SparkContext. That leaves the resource leak this PR is meant to fix on the failure path. Move the try/finally to cover both foreachPartition and collect, and verify that persistent RDDs return to their initial count after an injected commit failure.