Hacktoberfest 2026: le issue che i maintainer hanno segnato per ottobre, aperte e adatte ai principianti. Sfoglia le issue Hacktoberfest

Interrupted first build leaves a partial event table which later runs use without a warning

Aperta
  • Qualcosa non va?
#1,295 1 commento 0 reazioni 0 assegnatari Vedi su GitHub

I maintainer di solito rispondono entro 3 giorni

Nessuno ha ancora preso questa issue.

Valutazione

Difficoltà
4/5
Tempo stimato
3-5 giorni
Idoneità per principianti
22/100
Tipo di issue
Bug
Chiarezza
Abbastanza chiara
Stato di attività
Attiva
Stack tecnologico
python
Ambito
data, tooling

Direzione di ricerca

The bug lives in pyhealth/datasets/base_dataset.py: global_event_df writes straight into its final folder, _event_transform cleans up only on TimeoutError and Exception, and clean_tmpdir() deletes the whole tmp/. Start by reproducing with the SIGINT script in the issue, then read those three spots. The issue's open questions (cache key bump, marker file, stale temp folders) need maintainer answers before any change, and the author already has PRs planned, so this is not a good first pick.

Scritto dal modello di indicizzazione a partire dal testo della issue.

Descrizione

Hi, PyHealth team!

If the first global_event_df build is interrupted (Ctrl-C) after Dask has written a part file, the next run on the same cache_dir takes the partial global_event_df.parquet/ as a valid cache and continues with only part of the patients, without an error or warning.

To reproduce (synthetic data, POSIX): the script below builds an event table for 3,000 patients in a child process, sends SIGINT after the first part file is written, then builds again on the same cache_dir.

dataset_cache_interrupt.py 
"""Ctrl-C during the first build of a dataset does not leave a partial cache.

The first time a dataset is used, PyHealth builds its event table and caches it
in ``cache_dir``. This script starts that build in a child process, presses
Ctrl-C (SIGINT to the child's process group) as soon as the first part of the
table is on disk, then runs the same build again on the same ``cache_dir`` and
counts the patients. The table is published only once it is complete, so the
second run builds it again and counts all 3000 patients. (Before, the second
run reused the partial table without a warning.)

Synthetic data only: three CSV tables of 60,000 rows. POSIX only, since it
sends Ctrl-C to a process group.

Usage:
    python examples/dataset_cache_interrupt.py
"""

import glob
import os
import signal
import subprocess
import sys
import tempfile
import time

import numpy as np
import pandas as pd

N_PATIENTS = 3000
ROWS_PER_TABLE = 60_000
TABLES = ["t1", "t2", "t3"]


def write_data(root: str) -> None:
    """Writes three CSV tables of random events and a config for them."""
    rng = np.random.default_rng(0)
    config = ['version: "1.0"', "tables:"]
    for table in TABLES:
        patients = rng.integers(0, N_PATIENTS, size=ROWS_PER_TABLE)
        seconds = rng.integers(0, 10**8, size=ROWS_PER_TABLE)
        times = pd.Timestamp("2020-01-01") + pd.to_timedelta(seconds, unit="s")
        pd.DataFrame(
            {
                "patient_id": [f"p{p:05d}" for p in patients],
                "time": times.strftime("%Y-%m-%d %H:%M:%S"),
                "code": "c",
            }
        ).to_csv(f"{root}/{table}.csv", index=False)
        config += [
            f"  {table}:",
            f'    file_path: "{table}.csv"',
            '    patient_id: "patient_id"',
            '    timestamp: "time"',
            '    timestamp_format: "%Y-%m-%d %H:%M:%S"',
            "    attributes:",
            '      - "code"',
        ]
    with open(f"{root}/config.yaml", "w") as f:
        f.write("\n".join(config) + "\n")


def build(root: str, cache: str) -> None:
    """Builds (or reuses) the event table and prints its patient count."""
    import polars as pl

    from pyhealth.datasets import BaseDataset

    dataset = BaseDataset(
        root=root,
        tables=TABLES,
        dataset_name="interrupt_demo",
        config_path=f"{root}/config.yaml",
        cache_dir=cache,
    )
    events = dataset.global_event_df
    n = events.select(pl.col("patient_id").n_unique()).collect().item()
    print(f"patients in the event table: {n} (expected {N_PATIENTS})", flush=True)


def first_part_written(cache: str) -> bool:
    """True once one complete parquet part of the event table is on disk."""
    for path in glob.glob(f"{cache}/*/global_event_df.parquet*/*.parquet"):
        try:
            with open(path, "rb") as f:
                f.seek(-4, os.SEEK_END)
                if f.read(4) == b"PAR1":  # the last bytes of a complete file
                    return True
        except OSError:
            pass  # empty or just removed
    return False


def main():
    with tempfile.TemporaryDirectory() as tmp:
        root, cache = f"{tmp}/data", f"{tmp}/cache"
        os.makedirs(root)
        write_data(root)
        command = [sys.executable, __file__, "build", root, cache]

        print("run 1: build, then Ctrl-C once the first part is written", flush=True)
        run = subprocess.Popen(
            command, start_new_session=True, stderr=subprocess.DEVNULL
        )
        while run.poll() is None and not first_part_written(cache):
            time.sleep(0.005)
        if run.poll() is None:
            os.killpg(run.pid, signal.SIGINT)
        run.wait()

        print("run 2: same cache_dir, no interruption", flush=True)
        subprocess.run(command, check=True)


if __name__ == "__main__":
    if sys.argv[1:2] == ["build"]:
        build(*sys.argv[2:4])
    else:
        main()

Expected: patients in the event table: 3000 (expected 3000). Actual on master a76a316 (macOS, Python 3.13.7): 1872, 1866 and 1899.

Cause: global_event_df builds straight into the final folder (base_dataset.py:757), and _event_transform removes the output only on TimeoutError and Exception (:715, :728), so Ctrl-C or a signal leaves the partial folder. Then any folder with one *.parquet file counts as complete (:749). #753 covered Exception (#740), but not interrupts.

A related bug: clean_tmpdir() removes the whole tmp/ (:545-549), so a job that finishes deletes the converted tables and Dask scratch of another job still running on the same cache_dir. I think this is part of the crashes with several jobs on one cold cache mentioned in #763.

I have 2 small fixes with tests:

  1. each dataset removes only the temp folders it created (#1296)
  2. build the event table in a private folder next to the final one and publish it with one os.replace, as download_file does since #1255, with a FileLock only to avoid duplicate builds

With both, the script prints 3000 every time.

Questions before the second PR:

  1. Old caches may already be partial and still pass the check. Should the cache key be bumped (for example with EVENT_FORMAT from #1291), or should they be kept?
  2. Is the rename enough, or do you also want a marker file?
  3. Folders left by killed jobs (tmp/<uuid>, *.part): document it, remove them by age, or handle them in clear_cache() (#765)?

Thank you!

Lingua principale
Python
Stelle
1.7k
Fork
803
Merge medio
5g 20h
PR unite (30g)
56

Preparare l'ambiente

Come iniziare

  1. Leggi tutta la issue e poi la guida ai contributi del progetto.
  2. Commenta sulla issue per dire che te ne occupi tu — evita che due persone facciano lo stesso lavoro.
  3. Fai un fork del repository e lavora su un branch.
  4. Apri una pull request che faccia riferimento al numero della issue.

Altre issue di sunlabuiuc/PyHealth

Tutte le issue di sunlabuiuc/PyHealth

Issue simili

Altre issue su Python

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.