[SPARK-54441][INFRA] Enable coverage data for workers

### What changes were proposed in this pull request?

By patching the worker in coverage runs, we save coverage data for code executed in workers. This code will only be triggered in coverage runs, not normal runs.

Also we uses the zip lib for many tests, this PR will also combine the coverage data with the `[paths]` config in `.coveragerc`.

### Why are the changes needed?

To have a better idea of what test coverage we are missing in our tests suite.

### Does this PR introduce _any_ user-facing change?

No

### How was this patch tested?

Manually test that coverage data shows up for the worker file.

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

No.

Closes #53145 from gaogaotiantian/enable-worker-coverage.

Authored-by: Tian Gao <[email protected]>
Signed-off-by: Hyukjin Kwon <[email protected]>

Author: gaogaotiantian

python/.coveragerc

@@ -35,3 +35,8 @@ omit =
     */pyspark/streaming/tests/*
     */pyspark/tests/*
     */pyspark/testing/tests/*
+
+[paths]
+source =
+    pyspark/
+    lib/pyspark.zip/pyspark

python/test_coverage/sitecustomize.py

@@ -21,6 +21,32 @@
 # variable is set or not. If set, it starts to run the coverage.
 try:
     import coverage
-    coverage.process_startup()
+    cov = coverage.process_startup()
+    if cov:
+        import os
+
+        def patch_worker():
+            # If it's a worker forked from the daemon, we need to patch it to save
+            # the coverage data. Otherwise the worker will be killed by a signal and
+            # the coverage data will not be saved.
+            import sys
+            frame = sys._getframe(1)
+            if (
+                frame.f_code.co_name == "manager" and
+                "daemon.py" in frame.f_code.co_filename and
+                "worker" in frame.f_globals
+            ):
+
+                def save_when_exit(func):
+                    def wrapper(*args, **kwargs):
+                        result = func(*args, **kwargs)
+                        cov.save()
+                        return result
+                    return wrapper
+
+                frame.f_globals["worker"] = save_when_exit(frame.f_globals["worker"])
+
+        os.register_at_fork(after_in_child=patch_worker)
+
 except ImportError:
     pass