daniel-leicht commented on issue #50482:
URL: https://github.com/apache/arrow/issues/50482#issuecomment-6805100901
I hit the same thing through `pyarrow.fs.PyFileSystem` (first seen with
deltalake's `DeltaTable.to_pyarrow_table()`, which reads through a
`PyFileSystem`). Adding a reproducer that needs only pyarrow (no fsspec, no
malformed file) and fails on every run here.
The script writes 20 small Parquet files to a temp folder, then starts fresh
child processes that each scan them through a `PyFileSystem` whose
`open_input_file` returns `pa.PythonFile(open(path, "rb"))`, and exit right
after `to_table()`. `--native-fs` is the control (`LocalFileSystem`).
| Python | pyarrow | PyFileSystem | `--native-fs` |
|---|---|---|---|
| 3.13.14 | 26.0.1 | 5/5 hung | 0/5 failed |
| 3.12.13 | 26.0.1 | 5/5 aborted (exit code -6, `terminate called without an
active exception`) | not run |
Debian 12, x86_64, glibc 2.36, 72 CPUs, PyPI wheels.
On 3.13.14 with 26.0.1 the stuck worker is still in `PyBuffer::~PyBuffer`
(pystack, trimmed):
```
arrow::acero::(anonymous namespace)::ExecPlanImpl::~ExecPlanImpl()
...
parquet::arrow::(anonymous namespace)::FileReaderImpl::~FileReaderImpl()
parquet::SerializedFile::~SerializedFile()
arrow::io::internal::ReadRangeCache::LazyImpl::~LazyImpl()
std::_Sp_counted_ptr<arrow::py::PyBuffer*, ...>::_M_dispose()
arrow::py::PyBuffer::~PyBuffer()
PyGILState_Ensure
take_gil
PyThread_hang_thread
pause
```
and the main thread is in `arrow::internal::ThreadPool::Shutdown(bool)` ->
`std::condition_variable::wait`, reached from libc at process exit (the libc
frames are not symbolized).
`atexit.register(time.sleep, 0.5)` in the child avoids it here. I have not
tested #54467.
<details>
<summary>repro.py</summary>
```python
"""Reproduces a hang (Python 3.13+) or abort (Python 3.11/3.12) at
interpreter exit after a dataset scan through
pyarrow.fs.PyFileSystem.
Each run is a fresh child process that reads a small Parquet dataset through
a PyFileSystem (file data comes back as
Python bytes, held by arrow::py::PyBuffer) and then exits immediately.
python repro.py # PyFileSystem: runs hang (3.13+) or crash
with SIGABRT (3.11/3.12)
python repro.py --native-fs # control, same files through
pyarrow.fs.LocalFileSystem: every run exits normally
"""
import argparse
import os
import subprocess
import sys
import tempfile
import pyarrow as pa
import pyarrow.dataset as ds
import pyarrow.fs as pafs
import pyarrow.parquet as pq
class PythonFileHandler(pafs.FileSystemHandler):
"""Local files opened as Python file objects, like any Python handler
(fsspec, deltalake's DeltaStorageHandler)."""
def __init__(self):
self._local = pafs.LocalFileSystem()
def __eq__(self, other):
return isinstance(other, PythonFileHandler)
def __ne__(self, other):
return not self == other
def get_type_name(self):
return "python-file"
def normalize_path(self, path):
return path
def get_file_info(self, paths):
return self._local.get_file_info(paths)
def get_file_info_selector(self, selector):
return self._local.get_file_info(selector)
def open_input_file(self, path):
return pa.PythonFile(open(path, "rb"), mode="r")
def open_input_stream(self, path):
return pa.PythonFile(open(path, "rb"), mode="r")
def _read_only(self, *args, **kwargs):
raise NotImplementedError("read-only")
create_dir = delete_dir = delete_dir_contents = delete_root_dir_contents
= delete_file = _read_only
move = copy_file = open_output_stream = open_append_stream = _read_only
def write_dataset(directory, files, rows):
table = pa.table({
"id": pa.array([f"doc-{i:08d}" for i in range(rows)]),
"text": pa.array([f"some text for row {i} " * 4 for i in
range(rows)]),
"vector": pa.array([[float(i % 97)] * 256 for i in range(rows)],
type=pa.list_(pa.float32())),
})
paths = []
for n in range(files):
paths.append(os.path.join(directory, f"part-{n:03d}.parquet"))
pq.write_table(table, paths[-1], row_group_size=max(1, rows // 4))
return paths
def child(paths, native_fs=False):
filesystem = pafs.LocalFileSystem() if native_fs else
pafs.PyFileSystem(PythonFileHandler())
dataset = ds.dataset(paths, filesystem=filesystem,
format=ds.ParquetFileFormat(
default_fragment_scan_options=ds.ParquetFragmentScanOptions(pre_buffer=True)))
print(f"read {dataset.to_table().num_rows} rows", flush=True)
# Return right away: the interpreter starts shutting down while Arrow
may still be tearing the scan down.
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--runs", type=int, default=10)
parser.add_argument("--files", type=int, default=20)
parser.add_argument("--rows", type=int, default=5000, help="rows per
file")
parser.add_argument("--timeout", type=float, default=20, help="seconds
before a run counts as hung")
parser.add_argument("--native-fs", action="store_true", help="control:
read through pyarrow's LocalFileSystem")
parser.add_argument("--child", nargs="*", help=argparse.SUPPRESS)
args = parser.parse_args()
if args.child is not None:
return child(args.child, args.native_fs)
print(f"Python {sys.version.split()[0]}, pyarrow {pa.__version__}, Arrow
threads: {pa.cpu_count()} CPU, "
f"{pa.io_thread_count()} IO; "
f"{'LocalFileSystem (control)' if args.native_fs else
'PyFileSystem'}")
counts = {"ok": 0, "hung": 0, "crashed": 0}
with tempfile.TemporaryDirectory() as directory:
paths = write_dataset(directory, args.files, args.rows)
for run in range(1, args.runs + 1):
command = [sys.executable, __file__, *(["--native-fs"] if
args.native_fs else []), "--child", *paths]
proc = subprocess.Popen(command, stdout=subprocess.PIPE,
stderr=subprocess.STDOUT, text=True)
try:
output, _ = proc.communicate(timeout=args.timeout)
result = "ok" if proc.returncode == 0 else "crashed"
except subprocess.TimeoutExpired:
proc.kill()
output, _ = proc.communicate()
result = "hung"
counts[result] += 1
first_lines = " | ".join(line for line in
output.strip().splitlines()[:3])
print(f"run {run:3d}: {result:7s} exit code {proc.returncode:4d}
{first_lines[:150]}", flush=True)
print(f"\n{counts['hung']} hung, {counts['crashed']} crashed,
{counts['ok']} ok out of {args.runs} runs")
if __name__ == "__main__":
main()
```
</details>
AI assistance (Claude) was used for the investigation and to draft this
comment.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
