[HyukjinKwon]

What changes were proposed in this pull request?

SPARK-61153: a query run inside a progress handler or a SparkSession.Hook runs, and a hook that adds reattachable = False keeps the execution reattachable, as PySpark does.

• spark_connect_core::runtime::block_on blocks in place when it is called from a thread that is already in the shared multi-thread runtime's context, which is the case for a callback that runs inside an execution. The nested query then runs, as it does in PySpark. On a current_thread runtime the call still fails as before, since such a runtime cannot block in place. No lock is held across the callback: the progress handlers and the hook are cloned out of their locks before they are called.
• ExecutePlanResponseReattachableIterator::new always appends its reattachable = True option, after the request options that the hooks added, as PySpark's ExecutePlanResponseReattachableIterator does. Before, it appended the option only when the request had none, so a hook's reattachable = False made the server treat the execution as not reattachable, and the rows were lost with INVALID_HANDLE.OPERATION_NOT_FOUND.
• Two unit tests that relied on the nested-runtime panic now use a current_thread runtime, which keeps the panic they check: the schema cache of a dataframe and the version cache of a session.

Why are the changes needed?

Both defects run Python callbacks while an execution runs:

s.registerProgressHandler(lambda **kw: s.range(4).count() if not kw["done"] else None)
s.range(2000000).selectExpr("sum(id) as t").collect()
# PanicException: Cannot start a runtime from within a runtime

class H(SparkSession.Hook):
    def on_execute_plan(self, request):
        other.range(2).count()   # the same panic
        return request

b._registerHook(lambda s: H())

b._registerHook(lambda s: AppendNotReattachable())   # appends request_options reattach_options.reattachable = False
s.range(2).count()
# INVALID_HANDLE.OPERATION_NOT_FOUND after the rows are read

Official PySpark 4.3.0 gives nested count in handler -> 4, outer -> [Row(t=1999999000000)], outer hook query -> 3, and (2, [Row(x=1)]) for the last case.

Does this PR introduce any user-facing change?

Yes. A query run in a progress handler or a session hook returns its result instead of panicking, and a hook that appends a non-reattachable option no longer fails the execution.

How was this patch tested?

• Rust test_block_on_nested_in_block_on and test_block_on_nested_on_a_worker_thread in runtime.rs: a block_on from within block_on, and from a worker thread.
• Rust test_reattach_options_set_after_a_hook_added_a_false_one in reattach.rs: the iterator's options are [false, true] after a hook added false.
• Live test_progress_handler_runs_a_query_of_its_own_as_pyspark: a handler runs a count while the execution runs. The same results with official PySpark 4.3.0.
• Live test_hook_runs_a_query_of_its_own_as_pyspark: a hook runs a query. The same results with official PySpark 4.3.0.
• Live test_hook_appending_a_non_reattachable_option_keeps_the_execution_reattachable: a count and a sql collect with a hook that appends reattachable = False. The same results with official PySpark 4.3.0.
• The tests of both crates, e2e included, against a local Spark 4.3.0 server (878 passed), the build of all tests, also with wasm-udf, the offline Python suites (2201 passed, 8 skipped), and the live drop-in, Arrow and ML suites (187 passed) and live cases (10 passed). rustfmt passes, and clippy reports no diagnostics in the changed lines.

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

Generated-by: Claude Opus 5.5

This pull request and its description were written by Isaac.