Pipeline Specifications Developer Guide¶

This guide explains how to author, edit, and maintain pipeline smoke test rules, error scenarios, fixture manifests, and generated specification pages under specs/pipelines/.

1. Architecture and Directory Layout¶

Each pipeline family specification suite lives under specs/pipelines/<family>/ and is loaded by tests/pipeline_spec.py:

specs/pipelines/<family>/
├── rules.yaml                       # Rule catalogue (base rules + preset subclass rules)
├── spec/                            # Generic base pipeline smoke test cases
│   ├── 01_<case_name>/
│   │   ├── case.yaml                # Case manifest (title, rules, notes, runs)
│   │   ├── settings.toml            # Optional case/run Settings overrides (.toml, .yaml, or .json)
│   │   ├── input/                   # Staged input directory tree (files, subdirectories)
│   │   ├── expected_tables.py       # EXPECTED_TABLES: dict[str, pa.Table] (including _rejected if applicable)
│   │   └── expected.json            # Canonical JSON rows keyed by table name
│   └── 02_<error_or_edge_case>/
│       └── ...
└── presets/                         # Domain-specific preset subclass smoke test cases
    └── <preset_pipeline>/
        ├── 01_<case_name>/
        │   ├── case.yaml            # References both base rules (R-*) and preset rules
        │   ├── settings.toml
        │   ├── input/               # Domain fixtures (e.g., .xml.gz + .md5 sidecars)
        │   ├── expected_tables.py
        │   └── expected.json
        └── 02_<error_or_edge_case>/
            └── ...

The manifests in specs/pipelines/ drive two outputs:

1. End-to-end smoke and unit specification tests: tests/cdm_data_loaders/pipelines/test_pipeline_smoke_spec.py, tests/cdm_data_loaders/pipelines/xmltopyarrow/test_xmltopyarrow_spec.py, tests/cdm_data_loaders/pipelines/jsonlines/test_jsonlines_jsonschema_spec.py, and tests/cdm_data_loaders/pipelines/xsv/test_xsv_spec.py
2. Generated Markdown specification pages:
3. XML-to-PyArrow Pipeline Specification (xmltopyarrow base pipeline + pubmed preset subclass)
4. XML-to-Dict Pipeline Specification (xmltodict pipeline)
5. JSONLines Extract Pipeline Specification (jsonlines_extract pipeline)
6. JSONLines (JSON Schema) Pipeline Specification (jsonlines_jsonschema base pipeline + ncbi_datasets preset subclass)
7. JSONLines (Pydantic) Pipeline Specification (jsonlines_pydantic pipeline)
8. XSV (CSV/TSV) Schema and Pipeline Specification (xsv pipeline)

2. How Pipeline Rules Are Created and Structured¶

Rules in specs/pipelines/<pipeline>/rules.yaml define the end-to-end behavioral contract of a pipeline across three stages:

1. File Discovery and Pre-processing (for example, P1-DISCOVERY, P1-MD5-SIDECAR, N1-ENTITY-DIRS):
2. Which files input_dir and file_glob match.
3. How subdirectories map to entity tables.
4. What file pre-processors (such as validate_md5_sidecar in cdm_data_loaders.core.preprocessors) verify before reading a file.
5. Parsing, Validation, Transforms and Coercion (for example, P2-LEVEL-ROUTING, N2-JSONSCHEMA-VALIDATE, N3-FILTER-ANNOTATION):
6. How records are parsed and validated against XSD or JSON Schema.
7. How TABLE_TRANSFORMS filter, mutate, or split records across tables.
8. How records are coerced to OUTPUT_SCHEMAS or precomputed PyArrow schemas (PUBMED_SCHEMAS, SCHEMAS).
9. Error Routing and Pipeline Reporting (for example, P3-PREPROCESS-ERROR, P3-REPORT, N4-REJECTED, N4-REPORT):
10. How file pre-processing failures, JSON syntax errors, and schema validation errors populate _rejected.
11. What report.txt and _load_info record for status, ingested_files, load_errors, and pipeline_load_info_errs.

Rule Catalogue Format (rules.yaml)¶

title: PubMed XML-to-PyArrow Pipeline Specification
intro: |
  End-to-end smoke test specification for the `pubmed` pipeline
  (`cdm_data_loaders.pipelines.xmltopyarrow.pipeline:pubmed_cli`, `PubMedXmlToPyArrowSettings`).

sections:
  - id: discovery_and_preprocessing
    title: File Discovery and MD5 Pre-processing
    rules:
      - id: P1-DISCOVERY
        name: Gzip XML discovery
        rule: |
          The pipeline scans `input_dir` for files matching `file_glob` (default `*.xml.gz`).

Bidirectional Rule Invariants¶

test_pipeline_smoke_spec.py enforces two invariants on rules.yaml and case.yaml:

• Every rule is exercised (test_manifests_pass_every_rule_is_exercised): every rule ID in rules.yaml must appear in the rules: list of at least one case.yaml across spec/ or presets/ for that pipeline family.
• Every referenced rule exists (test_manifests_pass_every_case_rule_exists): every rule ID listed in any case.yaml must exist in that family's rules.yaml.
• Preset subclasses inherit base Settings and runner (test_manifests_pass_preset_subclasses_inherit_base_settings_and_runner): every preset pipeline in presets/<preset>/ must use a Settings class that subclasses the family's base Settings class and must delegate to the same run() callable.

3. How Pipeline Errors Are Modeled and Tested¶

Unlike schema converters (which raise exceptions at schema-translation time), KBase CTS ingestion pipelines run on the CTS (CDM Task Service) must exit with code 0 so the container service preserves output tables, _rejected rows, _load_info, and report.txt (see Core Pipeline and KBase CTS Execution Guide).

Pipeline error scenarios are modeled as case.yaml runs that verify three error outputs:

1. _rejected Table Rows¶

Failed files and failed records are routed to the _rejected table (REJECTED_PYARROW_SCHEMA in cdm_data_loaders.utils.buffer) with five columns:

2. expected_report in case.yaml¶

Each run in case.yaml declares an expected_report block checked by assert_smoke_run_output():

runs:
  - id: mismatched_md5
    input_dir: input
    settings_file: settings.toml
    expected_tables: expected_tables.py
    expected_json: expected.json
    expected_counts:
      _rejected: 1
    expected_report:
      status: completed with errors
      ingested_files: []
      load_errors:
        - "File pre-processing failed for corrupt_sample.xml.gz: MD5 mismatch for corrupt_sample.xml.gz"
      pipeline_load_info_errs: null

• File pre-processing errors (missing, empty, malformed, or mismatched .md5 sidecar):
• Skip reading the failed file and omit it from ingested_files.
• Append "File pre-processing failed for <file>: <reason>" to report.txt load_errors.
• Set report.txt status to "completed with errors".
• Write one provenance row per failed file to _rejected.
• Record parse and schema validation errors (malformed JSON lines or JSON Schema violations):
• Keep the file in ingested_files because the file itself passed pre-processing and was read.
• Route valid records in the same file to their destination table(s) and invalid lines to _rejected.
• Keep report.txt status: completed when no file pre-processor or dlt load job failed.

4. Common Developer Workflows¶

Adding a New Rule and Smoke Test Case¶

1. Add or select rule IDs in specs/pipelines/<pipeline>/rules.yaml:
2. Add a new rule entry (id, name, rule) under the appropriate section, or reuse existing rule IDs.
3. Create specs/pipelines/<pipeline>/<NN_case_name>/:
4. input/: Place small, deterministic input files under input/.
   • When creating .gz files or .md5 sidecars for pubmed, compress with gzip.compress(data, mtime=0) so the gzip header timestamp is deterministic (0) and the MD5 digest never drifts.
5. settings.toml: Add case-level Settings overrides (for example, loader_file_format = "parquet" and buffer_size = 10).
6. expected_tables.py: Define EXPECTED_TABLES: dict[str, pa.Table] using the pipeline's schemas (PUBMED_SCHEMAS, SCHEMAS, and REJECTED_PYARROW_SCHEMA when _rejected has rows).
7. expected.json: Provide the canonical JSON representation of each expected table's rows (with null fields omitted).
8. case.yaml: Reference the rule IDs in rules: and define one or more runs: with expected_counts and expected_report.
9. Regenerate the pipeline specification page:

   uv run python -m scripts.gen_pipeline_spec
   

10. Verify fixture invariants, end-to-end execution, and documentation sync:

    uv run pytest tests/cdm_data_loaders/pipelines/test_pipeline_smoke_spec.py
    uv run pytest -m docs
    

Editing Existing Rules, Cases, or Error Expectations¶

• Updating a rule description:
• Edit specs/pipelines/<pipeline>/rules.yaml and run uv run python -m scripts.gen_pipeline_spec.
• Updating input fixtures or expected outputs:
• Update input/, expected_tables.py, expected.json, and case.yaml (expected_counts / expected_report) together.
• test_manifests_pass_expected_tables_validate checks that every pa.Table in expected_tables.py passes table.validate(full=True), matches expected_counts, and normalizes to expected.json.
• Run uv run python -m scripts.gen_pipeline_spec to refresh the embedded fixtures and expected JSON in docs/<pipeline>_specification.md.

Onboarding a New Pipeline to specs/pipelines/¶

1. Ensure the pipeline is registered in PIPELINE_TARGETS:
2. Check PIPELINE_TARGETS in tests/cdm_data_loaders/pipelines/framework.py so resolve_pipeline("<pipeline>") resolves its CLI command, module path, Settings class, and runner function.
3. Create specs/pipelines/<pipeline>/:
4. Add rules.yaml and at least one <case>/ directory with case.yaml, settings.toml, input/, expected_tables.py, and expected.json.
5. Register the pipeline in scripts/gen_pipeline_spec.py:
6. Add a PipelineSpecConfig entry to PIPELINE_SPECS pointing to specs/pipelines/<pipeline> and docs/<pipeline>_specification.md.
7. Generate and index the specification page:
8. Run uv run python -m scripts.gen_pipeline_spec.
9. Add docs/<pipeline>_specification.md to zensical.toml ([project].nav), docs/index.md, and README.md.
10. Run verification:

    uv run pytest tests/cdm_data_loaders/pipelines/test_pipeline_smoke_spec.py
    uv run pytest -m docs
    uv run --group docs zensical build --strict