vxData jobs ingestion patterns

Writing ingestion jobs in mono apps/vxdata-jobs - discovery, mapping, stable identifiers, linking, and data gotchas.

When writing data ingestion jobs in mono/apps/vxdata-jobs.

Ingestion job structure

Entry tasks live in apps/vxdata-jobs/pyproject.toml under [project.scripts].
apps/vxdata-jobs/src/vxdata/jobs/ingestion.py defines the older Ingestion ABC (legacy).
Current pattern: source_2026_... and f_2026... jobs. Treat ingest_* as legacy unless required.
Each job is typically a subdirectory under src/vxdata/jobs/ with main.py, optional utils.py, and helpers.

File discovery patterns

Three common approaches:

Hierarchical DICOM trees (patient/study[/series]):
- Use Path.iterdir() or Path.rglob() for recursive traversal.
- Examples: ingest_prostatex/main.py, ingest_prus/main.py.
Filename globbing for modality/mask files:
- Match patterns to find anatomy masks, lesion masks, or specific sequences.
- Example: f_20260302_lesion_masks/prostatex.py.
Tabular source files (csv, xlsx, parquet) for clinical/pathology data:
- Parse with pandas or polars; join to existing resources via external UIDs.
- Examples: f_20260217_pathology/*.py, source_2026_01_28_ingest_files/upload.py.

Platform-side discovery is used heavily: query the vxData API for existing Patient, ImagingStudy, DICOMSeries, Volume, VoxelMap before uploading to resolve foreign keys and avoid duplicates.

Mapping and resource tree

The standard resource hierarchy:

DataSource → Patient → ImagingStudy → child resources (DICOMSeries, Volume, VoxelMap, Measurement, etc.)

Stable identifiers

Build explicit stable identifiers with small dataset-specific helper functions. Use external UIDs from the source dataset whenever possible (patient_id, study_id, series_uid, etc.).

Metadata sources

DICOM headers: utils/ingest_dicoms.py parses tags; vicom.load_volumes(...) handles DICOM/NIfTI conversion.
CSV/Excel joins: match clinical/pathology fields by patient/study identifiers.
Derived fields: some jobs compute study-level metadata from mode/frequency across files (e.g., modality, study date).

Common payload mappers

DICOM → ImagingStudy, DICOMSeries
Converted volumes → Volume / VolumeConversion
Tabular rows → Measurement, PatientStatus, PIRADSAssessment, PIRADSLesionAssessment, PathologyAssessment
Masks → VoxelMap or segmentation Volume

Linking and foreign keys

Primary tree link

Use parent_identifier to build the resource tree (e.g., ImagingStudy parent is Patient, DICOMSeries parent is ImagingStudy).

Resolution helpers

client.get_or_create_patient(...) and client.get_or_create_study(...) handle patient/study lookup or creation.
For batch jobs, rebuild maps from platform external_uid by querying all relevant resources first.

Secondary links (provenance and cross-references)

derived_from: track provenance (Volume, VoxelMap derived from other volumes).
reference_mri: link masks to their source MRI volume.
mapped_pirads_lesion_ref, index_lesion_id: connect lesion/pathology/radiology assessments.
pathology_id: associate BiopsyCoreLocation with pathology reports.

Data gotchas

Polars `group_by.agg` first() trap

pl.col(X).first() in group_by.agg is NOT “first non-null” and is NOT correlated with pl.col(Y).first().

Each first() is an independent column aggregation. When merging per-module rows where each row has NULLs in other columns (e.g., one row with pirads=5, isup=None, another with pirads=None, isup=4), plain first() will return the first value regardless of NULL, losing data.

Fix: use .drop_nulls().first() per column when aggregating sparse multi-row data:

.group_by("submission_uid", "lesion_uid")
.agg(
    pl.col("pirads").drop_nulls().first(),
    pl.col("isup").drop_nulls().first(),
)

Alternatively, restructure as separate DataFrames and join on the group key. This bug caused silent loss of ISUP scores in VBC-1250 (2026-04-20); the fix is in f_20260305_vxannotate_integration/main.py lines 191–196.

vxAnnotate placeholder scoring rows

vxAnnotate writes a placeholder scoring_annotations row when a lesion is created, with every score field blank (T2Score="", DWIScore="", level="", zones=[], or isupGrade blank/"unknown"). The real score arrives in a second row once the annotator submits.

Skip placeholder rows at parse time. Do NOT coerce empty strings to 0 via isup = isup_grade or 0 — that creates a fake score that collides with the real one and corrupts masks.

Predicate for placeholder:

PIRADS module: T2Score blank AND DWIScore blank.
Pathology module: isupGrade in (None, "", "unknown").

See f_20260305_vxannotate_integration/main.py lines 133–138 and 153–158 for the implemented skip logic.

vxAnnotate value-equal duplicate scoring rows

Independent of placeholders, vxAnnotate sometimes emits two scoring rows with identical scoring_data for the same (submission, lesion, module) — edit/resubmit artifacts.

Collapse them with .unique(maintain_order=True) on the raw score DataFrame before any strict uniqueness assertion. The assertion should only fire on genuinely-different scores (which is the real ambiguity).

High-signal reference files

apps/vxdata-jobs/src/vxdata/jobs/source_2026_01_28_ingest_files/scrape.py — tabular-driven file ingestion.
apps/vxdata-jobs/src/vxdata/jobs/f_20260305_vxannotate_integration/main.py — vxAnnotate mask export with all the above gotchas handled.
apps/vxdata-jobs/src/vxdata/jobs/f_20260217_pathology/basel.py — pathology table ingestion.
apps/vxdata-jobs/src/vxdata/jobs/f_20260302_lesion_masks/prostatex.py — lesion mask derivation.
apps/vxdata-jobs/src/vxdata/jobs/utils/store_resources.py — batch resource upload helpers.
apps/vxdata-jobs/src/vxdata/jobs/utils/ingest_dicoms.py — DICOM header parsing.

Historical note (2026-05-12): A spike evaluated a compact per-job argo.yaml contract + renderer against writing full Argo WorkflowTemplate / CronWorkflow manifests directly. Tradeoff: compact specs stay small but introduce a mini-platform that must stay correct; direct manifests are more explicit but repeat boilerplate. The spike artifacts existed in the archived data-platform repo under apps/worker/argo/ but were not carried forward into mono. Current practice is writing Argo manifests directly in virdx/infra_k8s under current/argo-workflows/.