[SPARK-48634][PYTHON][CONNECT] Avoid statically initialize threadpool at ExecutePlanResponseReattachableIterator

[HyukjinKwon]

What changes were proposed in this pull request?

This PR propose to avoid having ExecutePlanResponseReattachableIterator._release_thread_pool to initialize ThreadPool.

Why are the changes needed?

This instance might be dragged in during pickle because it's statically initialized.

    _release_thread_pool: Optional[ThreadPool] = ThreadPool(os.cpu_count() if os.cpu_count() else 8)
  File "/usr/lib/python3.10/multiprocessing/pool.py", line 930, in __init__
    Pool.__init__(self, processes, initializer, initargs)
  File "/usr/lib/python3.10/multiprocessing/pool.py", line 196, in __init__
    self._change_notifier = self._ctx.SimpleQueue()
  File "/usr/lib/python3.10/multiprocessing/context.py", line 113, in SimpleQueue
    return SimpleQueue(ctx=self.get_context())
  File "/usr/lib/python3.10/multiprocessing/queues.py", line 341, in __init__
    self._rlock = ctx.Lock()
  File "/usr/lib/python3.10/multiprocessing/context.py", line 68, in Lock
    return Lock(ctx=self.get_context())
  File "/usr/lib/python3.10/multiprocessing/synchronize.py", line 162, in __init__
    SemLock.__init__(self, SEMAPHORE, 1, 1, ctx=ctx)
  File "/usr/lib/python3.10/multiprocessing/synchronize.py", line 57, in __init__
    sl = self._semlock = _multiprocessing.SemLock(
PermissionError: [Errno 13] Permission denied

which requires to change in OS level.

Does this PR introduce any user-facing change?

Yeah, potentially this could trigger some random job failures in some environment like Ubuntu

How was this patch tested?

Manually tested.

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

No.

[grundprinzip]

This code is a bit confusing to me, but it might simply because of the way Python works. My understanding was that self only gives me access to instance but not class variables, but _release_thread_pool is an actual class variable.

[HyukjinKwon]

Yeah It actually accesses to the class variable, and it is not either discouraged or encouraged in PEP 8 if I am correctly remembering this. I just used self to make it look consistent, e.g., at shutdown but I don't mind changing it back to explicit ExecutePlanResponseReattachableIterator.

[HyukjinKwon]

Merged to master.

[zhengruifeng]

@HyukjinKwon I am seeing such failure in two PRs (1, 2):

Exception ignored in: <function ExecutePlanResponseReattachableIterator.__del__ at 0x7f67e2eac180>
Traceback (most recent call last):
  File "/__w/spark/spark/python/pyspark/sql/connect/client/reattach.py", line 335, in __del__
  File "/__w/spark/spark/python/pyspark/sql/connect/client/reattach.py", line 331, in close
  File "/__w/spark/spark/python/pyspark/sql/connect/client/reattach.py", line 247, in _release_all
  File "/__w/spark/spark/python/pyspark/sql/connect/client/reattach.py", line 71, in _release_thread_pool
  File "/usr/lib/python3.11/multiprocessing/pool.py", line 930, in __init__
  File "/usr/lib/python3.11/multiprocessing/pool.py", line 196, in __init__
  File "/usr/lib/python3.11/multiprocessing/context.py", line 112, in SimpleQueue
ImportError: sys.meta_path is None, Python is likely shutting down

Had test failures in pyspark.sql.tests.connect.streaming.test_parity_listener with python3.11; see logs.

But I can reproduce it locally, would you mind taking a look?

[HyukjinKwon]

Yeah will do. I think i know the cause.