[HyukjinKwon]

What changes were proposed in this pull request?

SPARK-61166: PySpark Connect's inheritable_thread_target(session) copies the thread_local of the client, of the tags of the thread, when it decorates a function. Every thread that runs the function gets that one copy, so the threads share it: a tag that one of them adds is a tag of the others.

The client now does the same.

Rust core: the tags of a thread are a set that threads can share:

• SparkSession::copy_thread_tags copies the tags of the current thread, as of now, into a new SharedTags.
• SparkSession::share_thread_tags makes such a copy the tags of the current thread.
• The tags that a thread adds or removes are then those of all the threads given it.
• clear_tags gives the thread a new empty set of its own, as PySpark's clearTags sets the tags of the thread_local to a new set.
• A thread whose tags are empty is still forgotten, unless it shares them.

Python client: inheritable_thread_target(session) copies the tags when it decorates the function, as before, and gives each of its threads the copy (_share_thread_tags), rather than adding the tags to each thread.

InheritableThread, which copies the tags when the thread starts, is not changed. Nor are the thread-local user context extensions, which PySpark's decorator copies in the same thread_local: the client's decorator does not propagate them, before or after this change.

Why are the changes needed?

The client copied the tags into each thread instead. So a thread that the function started, from another thread of the function, lacked the tags that that other thread had added. Apache Spark's test_inheritable_tags_with_deco of test_parity_job_cancellation.py failed on this client: the nested thread lacked the tag "b" of the thread that started it.

Does this PR introduce any user-facing change?

Yes. The threads of inheritable_thread_target share their tags, as with PySpark 4.3.0. For the Rust API, SharedTags, SparkSession::copy_thread_tags and SparkSession::share_thread_tags are new.

How was this patch tested?

• A new Rust unit test checks that threads given a copy share it, that the copy is of the tags when it was made, and that a thread that clears its tags gets a set of its own.
• A new offline test runs Apache Spark's case: a decorated function in a thread, which starts a nested thread of it. It checks the tags of both threads and of the decorating thread. It passes with official PySpark 4.3.0, and fails without this change.
• The tags of the same case against a local server are those of official PySpark 4.3.0.
• Apache Spark's test_parity_job_cancellation.py passes on this client against a local server: 8 tests pass and 1 is skipped.
• The lib tests of both crates (598 passed), the build of all tests, also with wasm-udf, the offline Python suites (2222 passed, 8 skipped), and the live drop-in and Arrow suites (204 passed) and live cases (10 passed) against a local Spark 4.3.0 server. 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.