Geospatial Schema Standardization & Mapping Automation

Lineage Manifest Generation

Every coordinate transformation, field cast, and validation gate in a geospatial pipeline is a claim: this feature was processed this way, at this time, with this outcome. A lineage manifest is the durable record of those claims. It is the artefact an auditor reads instead of trusting the pipeline’s word, and it is what turns a one-off transform into regulator-ready evidence. This stage sits at the foundation of Geospatial Compliance Reporting & Audit Trails: where CRS normalization and attribute ETL do the work, the lineage manifest proves the work was done, in what order, and with which result.

This page owns the record contract and the store: what a lineage event must contain, how upstream stages emit one row per event, how events are deduplicated and ordered, and how the resulting dataset is retained and queried. It stops short of two implementation details that have their own runnable guides. The physical mechanics of appending immutably to a columnar dataset — partition layout, atomic writes, schema-evolution guardrails — belong to writing append-only lineage manifests to Parquet. Rendering the stored rows into standards-compliant metadata for a catalogue is covered by generating ISO 19115 lineage statements automatically. Everything between those two — the schema, the emission contract, dedup, retention — is scoped here.

Upstream, this stage is a consumer. It ingests the lineage rows already produced by the CRS Normalization & Sync pipeline and the rejection logs emitted by Error Handling & Retry Logic, and it standardizes both into a single auditable trail. It does not re-run transformations; it records them.

Lineage Record to Partitioned Store to Audit Query Upstream CRS normalization and attribute ETL stages each emit one lineage record per event. Records pass through a deduplication and ordering gate that assigns a deterministic event_id and a monotonic sequence number. Deduplicated rows are appended as immutable part files into a Parquet dataset partitioned by run_id and target_crs. The store is never mutated in place. Downstream, an audit query engine scans partitions to answer provenance questions, and an ISO 19115 generator renders the same rows into catalogue metadata. Both readers see an append-only history. CRS normalization transform events Attribute ETL cast & validation events Dedup + order deterministic event_id monotonic sequence Append-only Parquet run_id=... / target_crs=... part-0001.parquet (immutable) part-0002.parquet (immutable) Audit query engine DuckDB / pyarrow.dataset ISO 19115 generator LI_Lineage metadata 1 row / event append scan (read-only) event flow (write path) read-only audit path — store is never mutated

The Lineage Record Contract

A manifest is only as trustworthy as the schema it enforces. Declare the lineage record as a version-controlled contract, then compile that contract into an Arrow schema so a malformed row is rejected at write time rather than discovered during an audit years later. The manifest below is the field catalogue; the Arrow schema derived from it is the enforcement mechanism.

yaml
# lineage_schema.yaml  — pyarrow >=14, pydantic >=2.0
schema_version: "1.2.0"                     # MANDATORY: bumped on any additive change
partition_keys: ["run_id", "target_crs"]    # MANDATORY: directory partition columns
fields:
  # --- MANDATORY EVENT IDENTITY ---
  - name: event_id            # deterministic hash, primary dedup key
    type: string
    required: true
  - name: run_id              # pipeline execution id; a partition key
    type: string
    required: true
  - name: sequence            # monotonic int within a run_id; total event ordering
    type: int64
    required: true
  - name: event_time_utc      # ISO 8601, UTC, when the event occurred
    type: timestamp[us, tz=UTC]
    required: true
  # --- MANDATORY PROVENANCE ---
  - name: stage               # emitting stage: crs_normalize | attr_cast | validate | ...
    type: string
    required: true
  - name: event_type          # TRANSFORM | VALIDATE | REJECT | CAST | SNAP
    type: string
    required: true
  - name: feature_id          # stable id of the feature acted on
    type: string
    required: true
  - name: source_crs          # EPSG/WKT2 of the input geometry (nullable for attr-only events)
    type: string
    required: true
  - name: target_crs          # EPSG/WKT2 of the output geometry; a partition key
    type: string
    required: true
  - name: outcome             # PASS | REJECT | WARN
    type: string
    required: true
  # --- OPTIONAL DIAGNOSTICS ---
  - name: strategy_index      # which fallback path ran (from the CRS stage)
    type: int32
    required: false
  - name: accuracy_m          # reported horizontal accuracy in metres
    type: float64
    required: false
  - name: rejection_reason    # full exception text for REJECT outcomes
    type: string
    required: false
  - name: tool_version        # "pyproj 3.6.1 / PROJ 9.3.1" — reproducibility anchor
    type: string
    required: false
  - name: audit_tag           # freeform tag propagated from upstream config
    type: string
    required: false

Mandatory vs optional field reference:

Field Required Arrow type Purpose
event_id Yes string Deterministic dedup key; identical across replays of the same event
run_id Yes string Pipeline execution identifier; first partition key
sequence Yes int64 Monotonic within a run; establishes total ordering for replay
event_time_utc Yes timestamp[us, tz=UTC] Wall-clock time the event occurred; never the write time
stage Yes string Which pipeline stage emitted the row
event_type Yes string Enumerated action class; enables typed audit queries
feature_id Yes string Stable feature identity across the whole pipeline
source_crs / target_crs Yes string Spatial provenance; target_crs is the second partition key
outcome Yes string PASS, REJECT, or WARN; drives conformance scoring
strategy_index No int32 Fallback tier that ran, mirrored from the CRS stage
accuracy_m No float64 Reported positional accuracy for the selected transform
rejection_reason No string Diagnostic text; populated only on REJECT
tool_version No string PROJ/pyproj build string; the reproducibility anchor an auditor needs
audit_tag No string Freeform tag carried from upstream manifests

Two decisions here are load-bearing. First, event_time_utc records when the event happened, never when the row was written — batching and retries decouple the two, and an auditor reconstructing a timeline needs the former. Second, event_id is deterministic, not random: it is a hash of the tuple that uniquely identifies an event, so a replayed batch reproduces identical ids that deduplicate cleanly. Treat lineage_schema.yaml and its schema_version as part of the audit contract; a field rename is a breaking change that must go through schema evolution, not an in-place edit.

Preprocessing Requirements

Before any stage emits a row, three preconditions must hold, or the manifest silently loses its guarantees.

  1. Every event has a stable feature_id. Lineage that cannot be joined back to a feature is noise. If source features lack a durable identifier, mint one deterministically upstream (a hash of the source path plus the original FID) and thread it through every stage. A feature_id that changes between the CRS stage and the attribute stage breaks the provenance chain for that feature.

  2. Clocks are UTC and monotonic within a run. event_time_utc must be timezone-aware UTC. The sequence field is assigned by a per-run counter, not by clock time, so two events in the same microsecond still order deterministically. Never derive ordering from event_time_utc alone.

  3. The upstream rejection log is normalized first. The Error Handling & Retry Logic stage writes rejections in its own NDJSON shape. Map those into the lineage record contract (a REJECT outcome with rejection_reason populated) before they enter the manifest, so audit queries see one uniform schema rather than two dialects.

python
# pipeline/lineage/identity.py  — python 3.10+, pyarrow >=14
from __future__ import annotations
import hashlib


def make_event_id(run_id: str, stage: str, feature_id: str, event_type: str) -> str:
    """Deterministic event identity: identical inputs always hash to the same id.

    This is the primary deduplication key. A retried batch re-emits the same
    logical events, which reproduce identical ids and collapse at audit time.
    """
    key = f"{run_id}\x1f{stage}\x1f{feature_id}\x1f{event_type}"
    return hashlib.blake2b(key.encode("utf-8"), digest_size=16).hexdigest()

Execution Engine: Emitting and Collecting Rows

The emission contract is deliberately narrow: each stage produces exactly one LineageEvent per feature-action, hands it to a collector, and never touches the store directly. Centralizing writes in a single collector is what makes the append-only and dedup guarantees enforceable — a stage cannot corrupt the manifest because it never holds a file handle to it.

python
# pipeline/lineage/collector.py  — python 3.10+, pyarrow >=14, pydantic >=2.0
from __future__ import annotations
import datetime as dt
import logging
from dataclasses import dataclass, field

import pyarrow as pa
from pyarrow import ArrowInvalid, ArrowTypeError

from pipeline.lineage.identity import make_event_id

logger = logging.getLogger(__name__)


@dataclass(slots=True)
class LineageEvent:
    run_id: str
    sequence: int
    stage: str
    event_type: str          # TRANSFORM | VALIDATE | REJECT | CAST | SNAP
    feature_id: str
    source_crs: str
    target_crs: str
    outcome: str             # PASS | REJECT | WARN
    event_time_utc: dt.datetime
    strategy_index: int | None = None
    accuracy_m: float | None = None
    rejection_reason: str | None = None
    tool_version: str | None = None
    audit_tag: str | None = None

    def event_id(self) -> str:
        return make_event_id(self.run_id, self.stage, self.feature_id, self.event_type)


# Compiled once from lineage_schema.yaml — the single enforcement point.
LINEAGE_ARROW_SCHEMA = pa.schema([
    ("event_id", pa.string()),
    ("run_id", pa.string()),
    ("sequence", pa.int64()),
    ("event_time_utc", pa.timestamp("us", tz="UTC")),
    ("stage", pa.string()),
    ("event_type", pa.string()),
    ("feature_id", pa.string()),
    ("source_crs", pa.string()),
    ("target_crs", pa.string()),
    ("outcome", pa.string()),
    ("strategy_index", pa.int32()),
    ("accuracy_m", pa.float64()),
    ("rejection_reason", pa.string()),
    ("tool_version", pa.string()),
    ("audit_tag", pa.string()),
])

_VALID_OUTCOMES = {"PASS", "REJECT", "WARN"}


@dataclass
class LineageCollector:
    """Accumulates events in memory and materializes a validated Arrow table.

    The collector is the ONLY component permitted to build manifest rows.
    Stages call add(); the writer (see the Parquet guide) consumes to_table().
    """
    _events: list[LineageEvent] = field(default_factory=list)

    def add(self, event: LineageEvent) -> None:
        if event.outcome not in _VALID_OUTCOMES:
            # A bad outcome value must never reach the store — fail loudly.
            raise ValueError(
                f"Invalid outcome {event.outcome!r} for feature {event.feature_id!r}; "
                f"expected one of {sorted(_VALID_OUTCOMES)}."
            )
        if event.event_time_utc.tzinfo is None:
            raise ValueError(
                f"event_time_utc for {event.feature_id!r} is naive; UTC tzinfo is required."
            )
        self._events.append(event)

    def to_table(self) -> pa.Table:
        """Materialize a schema-validated Arrow table, or raise before any write."""
        if not self._events:
            return LINEAGE_ARROW_SCHEMA.empty_table()

        columns = {name: [] for name in LINEAGE_ARROW_SCHEMA.names}
        for ev in self._events:
            columns["event_id"].append(ev.event_id())
            columns["run_id"].append(ev.run_id)
            columns["sequence"].append(ev.sequence)
            columns["event_time_utc"].append(ev.event_time_utc)
            columns["stage"].append(ev.stage)
            columns["event_type"].append(ev.event_type)
            columns["feature_id"].append(ev.feature_id)
            columns["source_crs"].append(ev.source_crs)
            columns["target_crs"].append(ev.target_crs)
            columns["outcome"].append(ev.outcome)
            columns["strategy_index"].append(ev.strategy_index)
            columns["accuracy_m"].append(ev.accuracy_m)
            columns["rejection_reason"].append(ev.rejection_reason)
            columns["tool_version"].append(ev.tool_version)
            columns["audit_tag"].append(ev.audit_tag)

        try:
            table = pa.table(columns, schema=LINEAGE_ARROW_SCHEMA)
        except (ArrowInvalid, ArrowTypeError) as exc:
            # A type mismatch means an upstream stage violated the contract.
            logger.critical("Lineage table failed schema validation: %s", exc)
            raise
        return table

The to_table() call is the choke point where the contract is enforced. If any stage emitted a value that does not fit LINEAGE_ARROW_SCHEMA — a float where an int belongs, a string where a timestamp is required — pa.table(...) raises ArrowInvalid or ArrowTypeError before a single byte reaches disk. That failure is desirable: a manifest that silently coerces bad rows is worse than one that halts.

Deduplication and Ordering

Retries, replays, and at-least-once queue delivery all reintroduce the same logical event. Deduplication is therefore not optional cleanup — it is a correctness requirement, because a lineage manifest that double-counts a REJECT will misreport a dataset’s conformance. Two mechanisms cooperate: deterministic event_id for identity and monotonic sequence for order.

python
# pipeline/lineage/dedup.py  — python 3.10+, pyarrow >=14
from __future__ import annotations
import pyarrow as pa
import pyarrow.compute as pc


def deduplicate(table: pa.Table) -> pa.Table:
    """Collapse re-emitted events, keeping the highest sequence per event_id.

    Because event_id is deterministic, a replayed batch produces duplicate ids.
    Keeping the max sequence preserves the most recent authoritative record.
    """
    if table.num_rows == 0:
        return table

    # Stable sort by (event_id, sequence) so the last row per id is authoritative.
    order = pc.sort_indices(
        table, sort_keys=[("event_id", "ascending"), ("sequence", "ascending")]
    )
    ordered = table.take(order)

    ids = ordered.column("event_id").to_pylist()
    keep = [i for i in range(len(ids)) if i + 1 == len(ids) or ids[i] != ids[i + 1]]
    return ordered.take(pa.array(keep, type=pa.int64()))

Deduplication runs at audit-read time, not at write time. This is deliberate and it is what keeps the store genuinely append-only: the writer never deletes a superseded row, it simply appends the corrected one with a higher sequence, and the reader resolves which row is authoritative. The full physical history survives — an auditor can see that an event was retried — while queries see one row per event. The write-side mechanics that preserve that history are detailed in writing append-only lineage manifests to Parquet.

Failure Modes

A lineage manifest fails quietly if you let it. Each failure below must map to an explicit, deterministic action; none may degrade to a dropped row.

Failure mode Likely cause Deterministic action
Row rejected by to_table() Upstream stage emitted a wrong-typed value ArrowInvalid/ArrowTypeError halts the flush; fix the emitter, never coerce
Missing mandatory field Stage instrumented incompletely Reject at LineageEvent construction; the dataclass makes the field non-optional
Duplicate event_id, differing payload Non-deterministic id inputs (e.g. wall-clock in the key) Audit alert: ids must be reproducible; recompute make_event_id inputs
Naive (tz-less) event_time_utc Local-time datetime leaked from a stage ValueError at add(); require datetime.now(tz=UTC) at the source
Non-monotonic sequence within a run Parallel workers sharing a counter without coordination Assign sequence from a single ordered source per run; see partition guide
Gap in feature_id chain Feature re-keyed mid-pipeline Completeness check fails in CI: every PASS at stage N needs a prior row at N-1
Partition explosion target_crs free-text instead of canonical EPSG Normalize target_crs to EPSG:xxxx before it becomes a partition key

The partition-explosion row is the one that bites in production. If target_crs arrives as raw WKT2 strings that differ by whitespace, each variant becomes its own partition directory and the dataset fragments into thousands of tiny files. Canonicalize target_crs to an EPSG:xxxx code — the same normalization the CRS Normalization & Sync stage already performs — before it is used as a partition key.

Compliance Reporting Output

The manifest exists to be queried. Because it is a partitioned Parquet dataset, any Arrow-aware engine reads it without a database server, and partition pruning on run_id and target_crs keeps audit scans cheap even at multi-year scale.

python
# audit/report.py  — python 3.10+, pyarrow >=14
from __future__ import annotations
import pyarrow.dataset as ds
import pyarrow.compute as pc

from pipeline.lineage.dedup import deduplicate


def conformance_summary(manifest_root: str, run_id: str) -> dict[str, int]:
    """Per-outcome counts for one pipeline run, read straight from partitions."""
    dataset = ds.dataset(manifest_root, format="parquet", partitioning="hive")
    # Partition pruning: only run_id=<run_id> directories are touched.
    table = dataset.to_table(filter=ds.field("run_id") == run_id)
    table = deduplicate(table)

    outcomes = table.column("outcome")
    return {
        "total_events": table.num_rows,
        "passed": pc.sum(pc.equal(outcomes, "PASS")).as_py() or 0,
        "rejected": pc.sum(pc.equal(outcomes, "REJECT")).as_py() or 0,
        "warned": pc.sum(pc.equal(outcomes, "WARN")).as_py() or 0,
    }

That summary is the raw material for two downstream artefacts. Rendered as XML, the same rows become ISO 19115 LI_Lineage and LI_ProcessStep elements for a metadata catalogue — the subject of generating ISO 19115 lineage statements automatically. Aggregated across runs, the pass/reject ratios feed the conformance scorecards that gate merges. The lineage manifest is the single source both draw from, which is what keeps a published metadata record and a CI scorecard from ever disagreeing.

Retention. Partition by run date so expiry is a directory operation, never a row-level delete against an append-only store. Set the window to the strictest obligation the data is subject to: INSPIRE-aligned datasets retain lineage for a minimum of six years; FGDC and ISO 19115 programs commonly require the life of the dataset plus a fixed archival tail. Because partitions are immutable directories, a retention job is a safe rm -rf of run_date= prefixes older than the window, with no risk of corrupting live history.

CI Integration

Two properties must be gated in continuous integration: the manifest is complete (every downstream event has its upstream predecessor) and the store is hygienic (canonical partitions, no orphaned schema versions). Run both on every pull request that touches pipeline instrumentation.

python
# tests/test_lineage_manifest.py  — pytest >=7, pyarrow >=14
import pyarrow as pa
from pipeline.lineage.collector import LineageCollector, LineageEvent, LINEAGE_ARROW_SCHEMA
from pipeline.lineage.dedup import deduplicate


def test_schema_is_frozen():
    """The Arrow schema must match the versioned contract exactly."""
    assert LINEAGE_ARROW_SCHEMA.field("event_time_utc").type == pa.timestamp("us", tz="UTC")
    assert LINEAGE_ARROW_SCHEMA.field("sequence").type == pa.int64()
    assert "rejection_reason" in LINEAGE_ARROW_SCHEMA.names


def test_replay_deduplicates(sample_event):
    """Emitting the same logical event twice must collapse to one row."""
    c = LineageCollector()
    c.add(sample_event)
    c.add(sample_event)          # simulated retry: identical event_id
    table = deduplicate(c.to_table())
    assert table.num_rows == 1


def test_reject_carries_reason():
    """A REJECT outcome without a rejection_reason is a broken audit trail."""
    c = LineageCollector()
    # ... build a REJECT event without rejection_reason ...
    # completeness rule enforced in the full suite:
    table = c.to_table()
    for outcome, reason in zip(
        table.column("outcome").to_pylist(),
        table.column("rejection_reason").to_pylist(),
    ):
        if outcome == "REJECT":
            assert reason, "REJECT rows must populate rejection_reason"
yaml
# .github/workflows/validate-lineage.yml
name: Validate Lineage Manifest
on: [push, pull_request]

jobs:
  lineage:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"
      - name: Install dependencies
        run: pip install "pyarrow>=14" "pydantic>=2.0" "pandas>=2.0" "pytest>=7"
      - name: Validate lineage schema contract
        run: python -m pipeline.lineage.validate_schema lineage_schema.yaml
      - name: Run manifest tests
        run: pytest tests/test_lineage_manifest.py -v --tb=short
      - name: Assert partition hygiene
        run: python -m audit.check_partitions --root ./_manifest --require-epsg-target

The check_partitions step is the one most teams omit and most regret: it walks the dataset directory and fails the build if any target_crs= partition is not a canonical EPSG:xxxx value, catching the fragmentation failure before it reaches production storage. Wrap transient object-store write failures in the retry patterns from Error Handling & Retry Logic — a flaky S3 PUT should back off and retry, not drop a lineage row.

Frequently Asked Questions

Why a partitioned Parquet dataset instead of a relational audit table? An append-only Parquet dataset is immutable by construction — each flush writes a new file and never rewrites an old one — which matches the legal shape of an audit trail far better than a mutable SQL table that a stray UPDATE can silently rewrite. It is also columnar, so conformance scans over billions of rows read only the two or three columns they need, and it is portable to any object store without keeping a database running for the full multi-year retention window.

How is the manifest truly append-only if deduplication removes rows? Deduplication never touches the store. The writer only ever appends; a corrected event is a new row with a higher sequence, and the superseded row stays on disk forever. Deduplication happens in the reader, at query time, resolving which row is authoritative. The physical file history is complete and immutable; only the logical view collapses duplicates.

What stops a retried batch from double-counting events? event_id is a deterministic hash of (run_id, stage, feature_id, event_type), so re-processing the same batch reproduces byte-identical ids. At audit time those identical ids collapse to one row per event. Randomly generated ids would defeat this entirely, which is why the id must be derived, never assigned.

How does this stage relate to the CRS pipeline’s own lineage records? The CRS Normalization & Sync stage emits transform lineage in its own local shape for operational debugging. This stage consumes those rows, maps them onto the canonical record contract, and standardizes them alongside attribute-ETL and rejection events into one auditable manifest. It is an aggregation and standardization layer, not a second source of truth.