Hacktoberfest 2026: die Issues, die Maintainer für den Oktober markiert haben – offen und einsteigerfreundlich. Hacktoberfest-Issues durchsuchen

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

Offen
  • Stimmt etwas nicht?
#1,295 1 Kommentar 0 Reaktionen 0 zugewiesene Personen Auf GitHub ansehen

Maintainer antworten meist innerhalb von 3 Tagen

Dieses Issue hat noch niemand übernommen.

Bewertung

Schwierigkeit
4/5
Geschätzter Aufwand
3-5 Tage
Anfängerfreundlichkeit
22/100
Issue-Typ
Bug
Klarheit
Größtenteils klar
Aktivitätsstatus
Aktiv
Tech-Stack
python
Bereich
data, tooling

Rechercherichtung

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.

Vom Indexierungsmodell aus dem Issue-Text verfasst.

Beschreibung

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!

Vorherrschende Sprache
Python
Sterne
1.7k
Forks
803
Ø Merge
5 T. 7 Std.
Gemergte PRs (30 T.)
58

Entwicklungsumgebung

Erste Schritte

  1. Lesen Sie das ganze Issue und danach den Beitragsleitfaden des Projekts.
  2. Schreiben Sie ins Issue, dass Sie es übernehmen — das erspart doppelte Arbeit.
  3. Forken Sie das Repository und arbeiten Sie in einem Branch.
  4. Öffnen Sie einen Pull Request, der die Issue-Nummer nennt.

Mehr aus sunlabuiuc/PyHealth

Alle Issues in sunlabuiuc/PyHealth

Ähnliche Issues

Weitere Issues zu Python

Neue Issues direkt in Ihr Postfach

Eine kurze Übersicht über anfängerfreundliche GitHub-Issues.