Geospatial Schema Standardization & Mapping Automation

Writing Append-Only Lineage Manifests to Parquet

A lineage store is only trustworthy if a prior audit record can never change. This guide is the physical write layer beneath Lineage Manifest Generation: it takes a validated Arrow table of lineage events and commits it to a pyarrow >=14 partitioned Parquet dataset so that every existing byte stays exactly as it was written. The record contract, deduplication, and retention policy are decided upstream in that stage; here we own one thing — appending immutably. Get this wrong and the whole compliance trail is repudiable; get it right and the store is a legally defensible, append-only history.

The mechanism is not “open the dataset and add rows.” Parquet files are immutable once closed, so append means write a new part file into an existing partition directory, atomically, and prove nothing else moved. Each flush becomes its own part-<uuid>.parquet; the dataset is the union of all part files across all partition directories. That is what makes the store append-only by construction rather than by convention.

Prerequisites checklist

Step 1: Configure the partitioned schema

Pin the Arrow schema and declare the partition keys explicitly. Partitioning on run_id and target_crs gives Hive-style directories (run_id=.../target_crs=.../) that let audit queries prune to a single run without scanning the whole store. Bind the schema to the dataset at configuration time so a drifting field type is caught at write, not at read.

python
# lineage/parquet_writer.py  — python 3.10+, pyarrow >=14
from __future__ import annotations
import pyarrow as pa
import pyarrow.dataset as ds

# Must match the contract compiled in the parent Lineage Manifest stage.
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()),
])

PARTITIONING = ds.partitioning(
    pa.schema([("run_id", pa.string()), ("target_crs", pa.string())]),
    flavor="hive",
)

SCHEMA_VERSION = "1.2.0"   # mirrors lineage_schema.yaml; written into every part file's metadata

Stamp the schema version into each file’s key-value metadata. When an auditor opens a part file three years from now, that string tells them exactly which contract the row was written under — the anchor for reproducibility.

Step 2: Write the partition atomically

The write must be atomic: a reader scanning the dataset mid-flush must see either the complete new part file or nothing, never a half-written file. Achieve this with a temp-then-rename move. Write to a hidden temporary path in the same partition directory, then os.replace it to its final part-<uuid>.parquet name — os.replace is atomic on POSIX filesystems, so the file appears in the directory fully formed.

python
# lineage/parquet_writer.py (continued)  — python 3.10+, pyarrow >=14
from __future__ import annotations
import logging
import os
import uuid
from pathlib import Path

import pyarrow as pa
import pyarrow.parquet as pq

logger = logging.getLogger(__name__)


def _partition_dir(root: Path, run_id: str, target_crs: str) -> Path:
    # Hive layout: run_id=<...>/target_crs=EPSG:4326/
    return root / f"run_id={run_id}" / f"target_crs={target_crs}"


def append_partition(root: Path, table: pa.Table, run_id: str, target_crs: str) -> Path:
    """Atomically append one immutable part file to a partition. Returns its path."""
    if table.schema != LINEAGE_ARROW_SCHEMA:
        raise ValueError("Table schema does not match the frozen lineage contract.")

    part_dir = _partition_dir(root, run_id, target_crs)
    part_dir.mkdir(parents=True, exist_ok=True)

    final_path = part_dir / f"part-{uuid.uuid4().hex}.parquet"
    tmp_path = part_dir / f".{final_path.name}.tmp"

    try:
        pq.write_table(
            table,
            tmp_path,
            compression="zstd",
            # Reproducibility anchor: which contract this file was written under.
            metadata_collector=None,
        )
        # Attach schema version to file-level metadata for auditors.
        _tag_schema_version(tmp_path)
        os.replace(tmp_path, final_path)   # atomic publish
    except (OSError, pa.ArrowIOError):
        logger.exception("Failed to write lineage partition to %s", final_path)
        if tmp_path.exists():
            tmp_path.unlink()              # never leave a partial temp behind
        raise

    logger.info("Committed %d rows to %s", table.num_rows, final_path)
    return final_path


def _tag_schema_version(path: Path) -> None:
    """Rewrite file metadata to embed the schema version (cheap, footer-only)."""
    existing = pq.read_metadata(path)
    meta = dict(existing.metadata or {})
    meta[b"lineage_schema_version"] = SCHEMA_VERSION.encode()
    # Re-open and persist via the table's schema metadata.
    table = pq.read_table(path)
    table = table.replace_schema_metadata(meta)
    pq.write_table(table, path, compression="zstd")

Because each flush lands in a uniquely named file, two concurrent workers writing to the same run_id/target_crs partition never collide — there is no shared file to contend for. This is why the append-only model scales horizontally where a single mutable file would serialize every writer.

Step 3: Validate no prior file was mutated

Append-only is a claim; verify it. Before the write, snapshot the checksums of every existing part file in the target partition. After the write, snapshot again. The pre-existing files must be byte-identical, and exactly one new file must have appeared.

python
# lineage/immutability.py  — python 3.10+
from __future__ import annotations
import hashlib
from pathlib import Path


def _checksums(part_dir: Path) -> dict[str, str]:
    """Map each existing part file name to its SHA-256, ignoring temp files."""
    out: dict[str, str] = {}
    for p in sorted(part_dir.glob("part-*.parquet")):
        out[p.name] = hashlib.sha256(p.read_bytes()).hexdigest()
    return out


def assert_append_only(before: dict[str, str], after: dict[str, str]) -> None:
    """Prove the write appended exactly one file and mutated nothing prior."""
    # 1. Every file that existed before must be unchanged.
    for name, digest in before.items():
        if name not in after:
            raise AssertionError(f"Prior part file vanished: {name} (store not append-only).")
        if after[name] != digest:
            raise AssertionError(f"Prior part file mutated: {name} (append-only violated).")
    # 2. Exactly one new file must have appeared.
    new_files = set(after) - set(before)
    if len(new_files) != 1:
        raise AssertionError(f"Expected exactly one new part file, found {len(new_files)}: {new_files}")

Run assert_append_only in the write path during development and in CI against a fixture partition. In production, sample it — checksumming every prior file on every flush is O(partition size) and will dominate runtime on large partitions, so verify on a schedule rather than per-write once the invariant is trusted.

Step 4: Log the manifest pointer

A part file that no index knows about is invisible to audits. After each atomic commit, append a pointer record — the file path, row count, schema version, and a content digest — to a lightweight pointer log. This log is the manifest’s table of contents and the fastest way to reconstruct exactly which files a given run produced.

python
# lineage/pointer_log.py  — python 3.10+
from __future__ import annotations
import datetime as dt
import hashlib
import json
from pathlib import Path


def log_manifest_pointer(
    pointer_log: Path, part_path: Path, run_id: str, row_count: int, schema_version: str
) -> None:
    """Append one NDJSON pointer describing a committed part file."""
    digest = hashlib.sha256(part_path.read_bytes()).hexdigest()
    record = {
        "run_id": run_id,
        "part_path": str(part_path),
        "row_count": row_count,
        "schema_version": schema_version,
        "sha256": digest,
        "committed_utc": dt.datetime.now(tz=dt.timezone.utc).isoformat(),
    }
    with pointer_log.open("a") as fh:
        fh.write(json.dumps(record) + "\n")

The sha256 in the pointer log doubles as tamper evidence: re-checksumming a part file and comparing against its logged digest proves the stored history has not been altered since it was committed — precisely the assurance an ISO 19115 or INSPIRE audit asks for.

Verification

Confirm the full write path end to end — do not trust a clean exit alone. Assert the partition layout, the atomic commit, and the immutability invariant together:

python
# python 3.10+, pyarrow >=14
from pathlib import Path
import pyarrow.dataset as ds
from lineage.parquet_writer import append_partition, LINEAGE_ARROW_SCHEMA, PARTITIONING
from lineage.immutability import _checksums, assert_append_only

root = Path("/tmp/_manifest")
part_dir = root / "run_id=run-2026-07-13" / "target_crs=EPSG:4326"
part_dir.mkdir(parents=True, exist_ok=True)

before = _checksums(part_dir)
path = append_partition(root, sample_table, "run-2026-07-13", "EPSG:4326")
after = _checksums(part_dir)

# 1. The write appended exactly one immutable file, mutating nothing prior.
assert_append_only(before, after)

# 2. The new file is discoverable through the partitioned dataset.
dataset = ds.dataset(root, format="parquet", partitioning=PARTITIONING)
tbl = dataset.to_table(filter=ds.field("run_id") == "run-2026-07-13")
assert tbl.num_rows == sample_table.num_rows, "row count mismatch after append"

# 3. The schema round-trips unchanged.
assert dataset.schema.field("event_time_utc").type == LINEAGE_ARROW_SCHEMA.field("event_time_utc").type

A healthy run logs one Committed N rows line per flush, assert_append_only raises nothing, and a second identical write produces a second file rather than overwriting the first. On the CLI, find _manifest -name 'part-*.parquet' | wc -l should increase by exactly one per commit, and no .tmp files should remain.

Troubleshooting

Symptom Likely cause Fix
ValueError: Table schema does not match the frozen lineage contract Upstream stage drifted a field type, or a nullable column arrived as a different Arrow type Re-materialize the table against LINEAGE_ARROW_SCHEMA; do not cast silently in the writer
Readers occasionally see a truncated file Writing directly to the final path instead of temp-then-rename Route every write through os.replace; never pq.write_table straight to part-*.parquet
assert_append_only reports a prior file mutated A retention or compaction job rewrote part files in place Make compaction write new files and delete old ones as whole units; never edit a part file
Thousands of tiny part files, slow scans One flush per event instead of per batch, or an un-canonicalized target_crs exploding partitions Batch events per flush; canonicalize target_crs to EPSG:xxxx before Step 2
pa.ArrowInvalid: Schema at index N was different on read Two part files in one dataset carry incompatible schemas Gate schema evolution: only additive, nullable columns; bump SCHEMA_VERSION and backfill readers
.tmp files accumulating in partitions Write crashed after temp write, before os.replace The except branch unlinks temps; add a startup sweep for stale .*.tmp files

Schema evolution is the subtle one. Parquet tolerates reading files with different-but-compatible schemas only when new columns are nullable additions. Never drop, rename, or retype a column in place — that splits the dataset into unreadable halves. Additive changes bump SCHEMA_VERSION, and the version stamped in each file’s metadata lets a reader apply the right interpretation per file.