This is an automated email from the ASF dual-hosted git repository.
Yicong-Huang pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 03473687b6e6 [SPARK-59007][PYTHON][TEST] Fix flaky Arrow grouped map
worker logging tests
03473687b6e6 is described below
commit 03473687b6e6bc37a0b2c64f9b32a667729e2133
Author: Yicong Huang <[email protected]>
AuthorDate: Thu Aug 27 23:26:49 2026 +0000
[SPARK-59007][PYTHON][TEST] Fix flaky Arrow grouped map worker logging tests
### What changes were proposed in this pull request?
Wrap the log read-and-assert step of
`ApplyInArrowTests.test_apply_in_arrow_with_logging` and
`test_apply_in_arrow_iter_with_logging` in `eventually` so it polls until the
asynchronously captured worker logs are visible. The `applyInArrow` call that
produces the logs stays outside the poll so it runs once.
### Why are the changes needed?
Observed on fork CI: the `pyspark-sql` job failed with `[DIFFERENT_ROWS]`
(100%) where `spark.tvf.python_worker_logs()` returned no rows instead of the
expected two WARNING rows, then re-ran green on the next attempt:
https://github.com/Yicong-Huang/spark/actions/runs/32828898783/job/97748219772
Root cause: worker logs are captured asynchronously. Python workers emit
log records on stdout; on the JVM side a per-worker `RedirectThread`
(`PythonWorkerLogCapture`) drains stdout and only saves the log block to the
`BlockManager` once it reads the trailing marker line. That drain runs
independently of the query result, which returns over a separate socket
channel, so `python_worker_logs()` invoked right after the query can observe
zero blocks. Polling the read side waits out the race.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Built with `build/sbt -Phive package` and ran `python/run-tests --testnames
'pyspark.sql.tests.arrow.test_arrow_grouped_map'`; both logging tests and the
full module pass.
### Was this patch authored or co-authored using generative AI tooling?
No.
Closes #58289 from Yicong-Huang/flaky-worker-logs.
Authored-by: Yicong Huang <[email protected]>
Signed-off-by: Yicong-Huang <[email protected]>
(cherry picked from commit 02cb44fe44acaa12e02c939e0cee85e347e0a114)
Signed-off-by: Yicong-Huang <[email protected]>
---
.../sql/tests/arrow/test_arrow_grouped_map.py | 73 ++++++++++++++--------
1 file changed, 46 insertions(+), 27 deletions(-)
diff --git a/python/pyspark/sql/tests/arrow/test_arrow_grouped_map.py
b/python/pyspark/sql/tests/arrow/test_arrow_grouped_map.py
index 84fdf0f0a766..8af706d8a9c5 100644
--- a/python/pyspark/sql/tests/arrow/test_arrow_grouped_map.py
+++ b/python/pyspark/sql/tests/arrow/test_arrow_grouped_map.py
@@ -27,7 +27,12 @@ from pyspark.sql import functions as sf
from pyspark.sql.functions import array, col, explode, lit, mean, stddev
from pyspark.sql.window import Window
from pyspark.testing.sqlutils import ReusedSQLTestCase
-from pyspark.testing.utils import assertDataFrameEqual, have_pyarrow,
pyarrow_requirement_message
+from pyspark.testing.utils import (
+ assertDataFrameEqual,
+ eventually,
+ have_pyarrow,
+ pyarrow_requirement_message,
+)
from pyspark.util import is_remote_only
if have_pyarrow:
@@ -472,20 +477,27 @@ class ApplyInArrowTestsMixin:
df,
)
- logs = self.spark.tvf.python_worker_logs()
+ # Worker logs are captured asynchronously from the worker's stdout
and only
+ # become visible once the trailing block is flushed, so they may
not all be
+ # present immediately after the query completes. Poll until they
show up.
+ @eventually(timeout=5, catch_assertions=True)
+ def check_logs():
+ logs = self.spark.tvf.python_worker_logs()
+
+ assertDataFrameEqual(
+ logs.select("level", "msg", "context", "logger"),
+ [
+ Row(
+ level="WARNING",
+ msg=f"arrow grouped map: {dict(id=lst, value=[v *
10 for v in lst])}",
+ context={"func_name": func_with_logging.__name__},
+ logger="test_arrow_grouped_map",
+ )
+ for lst in [[0, 2, 4, 6, 8], [1, 3, 5, 7]]
+ ],
+ )
- assertDataFrameEqual(
- logs.select("level", "msg", "context", "logger"),
- [
- Row(
- level="WARNING",
- msg=f"arrow grouped map: {dict(id=lst, value=[v * 10
for v in lst])}",
- context={"func_name": func_with_logging.__name__},
- logger="test_arrow_grouped_map",
- )
- for lst in [[0, 2, 4, 6, 8], [1, 3, 5, 7]]
- ],
- )
+ check_logs()
@unittest.skipIf(is_remote_only(), "Requires JVM access")
def test_apply_in_arrow_iter_with_logging(self):
@@ -512,20 +524,27 @@ class ApplyInArrowTestsMixin:
df,
)
- logs = self.spark.tvf.python_worker_logs()
+ # Worker logs are captured asynchronously from the worker's stdout
and only
+ # become visible once the trailing block is flushed, so they may
not all be
+ # present immediately after the query completes. Poll until they
show up.
+ @eventually(timeout=5, catch_assertions=True)
+ def check_logs():
+ logs = self.spark.tvf.python_worker_logs()
+
+ assertDataFrameEqual(
+ logs.select("level", "msg", "context", "logger"),
+ [
+ Row(
+ level="WARNING",
+ msg=f"arrow grouped map: {dict(id=lst, value=[v *
10 for v in lst])}",
+ context={"func_name": func_with_logging.__name__},
+ logger="test_arrow_grouped_map",
+ )
+ for lst in [[0, 2, 4], [6, 8], [1, 3, 5], [7]]
+ ],
+ )
- assertDataFrameEqual(
- logs.select("level", "msg", "context", "logger"),
- [
- Row(
- level="WARNING",
- msg=f"arrow grouped map: {dict(id=lst, value=[v * 10
for v in lst])}",
- context={"func_name": func_with_logging.__name__},
- logger="test_arrow_grouped_map",
- )
- for lst in [[0, 2, 4], [6, 8], [1, 3, 5], [7]]
- ],
- )
+ check_logs()
class ApplyInArrowTests(ApplyInArrowTestsMixin, ReusedSQLTestCase):
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
