Cross-Engine Data Reconciliation Architecture › Data Equivalence Modeling

Data Equivalence Modeling for Cross-Engine Reconciliation Pipelines

Data equivalence modeling is the discipline that decides when two heterogeneous rows count as the same logical record despite differences in storage engine, serialization format, or query semantics. It sits directly downstream of extraction and directly upstream of comparison in the Cross-Engine Data Reconciliation Architecture control plane: once both engines have been read, this stage turns raw, engine-shaped rows into canonical byte representations and deterministic digests that the diff engine can compare without ever seeing the original storage types. For data engineers, migration specialists, Python pipeline builders, and platform operations teams, this is where naive row-count checks and blind checksums are replaced with explicit definitions of identity, tolerance thresholds, type-coercion matrices, and cryptographic hashing strategies that survive engine-specific transformations.

The reason this stage exists as a distinct workload is that equivalence is not identity. A NUMERIC(38,9) value in PostgreSQL, a Decimal128 in MongoDB, and a fixed-precision string in a Parquet export can all represent the same business number while sharing not a single byte. Equivalence modeling encodes the rules that collapse those physical representations onto one canonical form, so that everything after it can rely on the simple invariant “equal bytes ⇒ equal record.”

Architectural Boundaries: What This Stage Consumes and Produces

This workload begins the moment sorted, partitioned rows arrive from the extraction layer and ends the moment a per-row (or per-column) digest is emitted for the comparison engine. It consumes iterators of typed rows keyed on a stable reconciliation key, a schema contract describing both engines, and a tolerance profile. It produces deterministic digests plus a divergence stream classified into mismatch, missing-in-source, and missing-in-target — the raw material the downstream structural mismatch detection stage turns into a discrepancy manifest.

Three concerns are isolated inside the boundary and must never leak into each other: structural mapping (which fields participate and how they align), value normalization (how a single value becomes canonical bytes), and diff computation (how two digest streams are walked). Keeping these separate is what lets the same equivalence model back both a batch snapshot job and a streaming validator. The rows themselves arrive already extracted — the mechanics of lock-light reads and streaming batches live upstream in the parallel row extraction techniques and async batching for large datasets references, and the type-translation contracts this stage depends on are defined in the cross-platform schema mapping reference.

The stage is strictly read-only and stateless with respect to row content. It holds no mutating grants on either engine and carries state only as checkpoint progress, which keeps it idempotent and safe to retry — a job re-run over the same key range must emit an identical divergence stream.

Canonicalization and row-signature folding: heterogeneous values converge on one digest Top band: a PostgreSQL NUMERIC(38,9) value and a MongoDB Decimal128 value — two physically different encodings of the same business number — both flow into canonicalize_value(), which quantizes, pins the decimal context, and prepends a type tag to produce one identical canonical byte string. That byte string is hashed by BLAKE2b into a single per-field digest. Bottom band: the per-field digests of a row are folded over sorted keys into one row signature. Because the source row and target row canonicalize to the same field digests, both fold to the same row signature, proving equal bytes imply an equal record. Canonicalization — heterogeneous physical forms collapse to identical bytes source · PostgreSQL NUMERIC(38,9) 1234.500000000 target · MongoDB Decimal128 1234.5 canonicalize_value() quantize · pin ctx · tag canonical bytes — identical d·1234.5000000000 BLAKE2b size 32 field digest a3f1…9c2 Row-signature folding — equivalent rows converge on one signature source row per-field digests (sorted keys) a3f1 · 5e08 · c19d fold key=value; row signature 7d2e…b41 target row per-field digests (sorted keys) a3f1 · 5e08 · c19d fold key=value; row signature 7d2e…b41 ≡ equal record equal bytes ⇒ equal record

Prerequisites

Before wiring an equivalence model into a reconciliation run, confirm the following are in place. Each item removes a common source of phantom divergence.

Step-by-Step Implementation

The steps below build a streaming, memory-bounded equivalence pipeline. Each step ends with an assertion or observable output so it can be verified in isolation before the next is layered on.

Step 1 — Declare the equivalence configuration

Centralize every determinism-critical parameter in one typed configuration object. Nothing that affects a digest may live outside it.

python
import hashlib
import logging
import struct
from dataclasses import dataclass
from datetime import datetime, timezone
from decimal import Decimal, ROUND_HALF_EVEN
from typing import Iterator, Tuple, Dict, Any, Optional

logger = logging.getLogger("reconcile.equivalence")

@dataclass(frozen=True)
class EquivalenceConfig:
    hash_algorithm: str = "blake2b"
    digest_size: int = 32
    float_epsilon: float = 1e-9
    null_token: bytes = b"\x00__NULL__\x00"
    decimal_scale: int = 10
    primary_key: str = "id"

Verify the config is frozen and reproducible:

python
cfg = EquivalenceConfig()
assert cfg.digest_size == 32 and cfg.primary_key == "id"

Step 2 — Canonicalize a single value to deterministic bytes

Canonicalization is the load-bearing step. Standard json.dumps() and pickle are unsuitable: key ordering, whitespace, and platform byte representations are non-deterministic. Instead, serialize each type explicitly, quantize floats to the tolerance epsilon before packing, and pin the decimal context so rounding is host-independent.

python
def canonicalize_value(value: Any, config: EquivalenceConfig) -> bytes:
    """Deterministically serialize a single value to bytes for hashing."""
    if value is None:
        return config.null_token
    if isinstance(value, bool):
        return b"\x01" if value else b"\x02"     # keep bool distinct from int
    if isinstance(value, int):
        return b"i" + struct.pack(">q", value)
    if isinstance(value, float):
        # Quantize to the tolerance grid so micro-drift hashes identically.
        rounded = round(value / config.float_epsilon) * config.float_epsilon
        return b"f" + struct.pack(">d", rounded)
    if isinstance(value, Decimal):
        q = Decimal(10) ** -config.decimal_scale
        return b"d" + value.quantize(q, rounding=ROUND_HALF_EVEN).to_eng_string().encode()
    if isinstance(value, datetime):
        utc = value.astimezone(timezone.utc) if value.tzinfo else value.replace(tzinfo=timezone.utc)
        return b"t" + utc.isoformat(timespec="microseconds").encode()
    if isinstance(value, str):
        return b"s" + value.encode("utf-8")
    if isinstance(value, bytes):
        return b"b" + value
    raise TypeError(f"Unsupported type for canonicalization: {type(value)!r}")

The one-byte type tag (i, f, d, t, s, b) prevents cross-type collisions — the string "1" and the integer 1 must never share a digest. Verify determinism and the epsilon behaviour:

python
cfg = EquivalenceConfig()
assert canonicalize_value(1.000000000, cfg) == canonicalize_value(1.0000000004, cfg)
assert canonicalize_value("1", cfg) != canonicalize_value(1, cfg)

Step 3 — Fold field digests into a row signature

Hash each field independently and aggregate the field digests into a row-level signature. This columnar approach localizes divergence to a specific field without materializing whole rows, which is what makes targeted re-extraction possible for the column-level checksum generation workflow.

python
def compute_row_digest(row: Dict[str, Any], config: EquivalenceConfig) -> bytes:
    """Deterministic digest for a single row, order-independent across fields."""
    hasher = hashlib.new(config.hash_algorithm, digest_size=config.digest_size)
    for key in sorted(row):                        # sorted keys ⇒ stable field order
        hasher.update(key.encode("utf-8"))
        hasher.update(b"=")
        hasher.update(canonicalize_value(row[key], config))
        hasher.update(b";")                        # length-delimiter guards concatenation
    return hasher.digest()

Field ordering is normalized by sorting keys, so a document store that returns {"b":2,"a":1} and a relational row returning (a=1, b=2) produce the same signature. Verify order-independence:

python
cfg = EquivalenceConfig()
assert compute_row_digest({"a": 1, "b": 2}, cfg) == compute_row_digest({"b": 2, "a": 1}, cfg)

Step 4 — Walk two sorted streams with an O(N) merge diff

With both streams sorted on the reconciliation key, a single linear merge classifies every record without loading either dataset into memory. This is the core of the comparison hand-off.

python
def streaming_diff_engine(
    source_iter: Iterator[Dict[str, Any]],
    target_iter: Iterator[Dict[str, Any]],
    config: EquivalenceConfig,
) -> Iterator[Tuple[str, Dict[str, Any]]]:
    """
    O(N) streaming diff for two key-sorted row streams.
    Yields (classification, payload). Both iterators MUST be sorted by
    config.primary_key in the same collation.
    """
    def fetch_next(it: Iterator) -> Optional[Tuple[Any, bytes, Dict[str, Any]]]:
        try:
            row = next(it)
        except StopIteration:
            return None
        except Exception:
            logger.exception("Extraction failed during diff stream")
            raise
        return (row.get(config.primary_key), compute_row_digest(row, config), row)

    src = fetch_next(source_iter)
    tgt = fetch_next(target_iter)

    while src or tgt:
        if src and tgt and src[0] == tgt[0]:
            if src[1] != tgt[1]:
                yield "mismatch", {"key": src[0], "source": src[2], "target": tgt[2]}
            src, tgt = fetch_next(source_iter), fetch_next(target_iter)
        elif src and (tgt is None or src[0] < tgt[0]):
            yield "missing_in_target", {"key": src[0], "source": src[2]}
            src = fetch_next(source_iter)
        else:
            yield "missing_in_source", {"key": tgt[0], "target": tgt[2]}
            tgt = fetch_next(target_iter)

Verify end-to-end against a known-divergent fixture:

python
cfg = EquivalenceConfig()
source = iter([{"id": 1, "amt": 10.0}, {"id": 2, "amt": 20.0}])
target = iter([{"id": 1, "amt": 10.0}, {"id": 2, "amt": 20.5}])
results = list(streaming_diff_engine(source, target, cfg))
assert results == [("mismatch", {"key": 2, "source": {"id": 2, "amt": 20.0},
                                 "target": {"id": 2, "amt": 20.5}})]

The wider set of exact/tolerance/structural fallbacks that consume this divergence stream is implemented in the fallback chain implementation reference; here the contract stops at classifying each key.

Comparison Strategy Trade-offs

Three digesting strategies dominate production equivalence models. The right choice depends on how precisely you must localize a divergence versus how much latency and storage you can spend. The compliance row matters because reconciliation frequently runs over regulated columns, and the digest that crosses a security boundary must itself be defensible.

Criterion Full-row serialization hash Field-level (columnar) digest Merkle chunk tree
Diff localization Row only — re-extract whole row Exact field that drifted Chunk (block of rows), then drill down
Latency per row Lowest (one hash) Higher (N field hashes) Low amortized; high on first divergence
Memory / cost Minimal, streaming Moderate — per-field state Tree state held per chunk
Re-extraction cost on drift Whole row Single column One chunk
Compliance / regulatory Digest may embed PII bytes; hash the masked form Field digests let regulated columns be tokenized independently Chunk roots move only 32-byte hashes across zones — smallest blast radius
Scale ceiling Excellent Good — cost grows with column count Excellent for very large, mostly-identical datasets

For most heterogeneous migrations, the field-level digest is the default: its per-column localization is what makes drift debugging tractable and lets security boundaries for reconciliation apply a per-column masking or tokenization policy before a value is ever hashed. Reserve the Merkle chunk tree for billion-row, low-divergence datasets where transferring per-row digests is itself the bottleneck. Regardless of strategy, prefer BLAKE2b or SHA-256 over MD5 — MD5’s collision weakness makes it unfit for integrity evidence in a regulated audit trail.

Scaling and Performance

Equivalence modeling is CPU-bound on hashing and memory-bound on how many rows are held simultaneously, so both dimensions must be engineered.

Partitioning strategy. Shard the reconciliation space on contiguous ranges of the reconciliation key — [min_key, max_key) slices — so each worker owns a disjoint, independently checkpointable partition. Keyset (seek) pagination, not OFFSET, keeps extraction O(batch) and avoids page drift as the partition is walked.

Batch sizing. Size batches to keep a worker’s resident set within a fixed budget: batch_rows ≈ memory_budget / (avg_row_bytes × safety_factor). Because streaming_diff_engine holds at most one row per side, the dominant memory cost is the extraction buffer, not the diff — tune the extractor batch, not the comparator.

Memory bounding. Never materialize a full table. The generator-based merge above bounds resident rows to two, so a partition of arbitrary size costs constant comparison memory; only the sorted extraction feeding it needs care.

GIL and parallelism. Hashing releases the GIL inside hashlib, so a ProcessPoolExecutor sharded by key range gives near-linear throughput on multi-core hosts; threads help far less because the Python-level canonicalization loop is GIL-bound. Prefer process-per-partition and aggregate divergence streams centrally.

Failure Modes and Diagnostic Runbook

Each named failure mode below lists its cause, the signal that detects it, and the remediation.

  • Phantom divergence from serialization skew. Cause: the same value serialized differently across hosts (locale, PYTHONHASHSEED, unpinned decimal context). Signal: divergence disappears when both sides are re-hashed on one host. Remediation: pin LC_ALL=C.UTF-8, PYTHONHASHSEED=0, and the decimal context; assert byte-equality of canonicalize_value in CI.
  • Float micro-drift false positives. Cause: IEEE-754 rounding differences below business significance. Signal: mismatches cluster on float columns with deltas near machine epsilon. Remediation: quantize to float_epsilon before packing (Step 2) and tune the profile per the threshold tuning for tolerance reference.
  • Key-collation mismatch. Cause: source sorted by database collation, target by byte order, so the merge misaligns. Signal: long runs of alternating missing-in-source / missing-in-target. Remediation: force one explicit collation on both extraction queries; assert monotonic non-decreasing keys at the diff boundary.
  • Null vs empty-string divergence. Cause: relational NULL mapped against NoSQL "" or a missing key. Signal: mismatches concentrate on nullable text columns. Remediation: route every empty representation through the single null_token; declare the tri-state rule in the coercion matrix.
  • Schema drift mid-run. Cause: an upstream migration adds or renames a column during a long run. Signal: an abrupt, total divergence spike from a known offset onward. Remediation: pin the schema version into the job identifier and fail closed; re-run through schema validation pre-checks before resuming.
  • OOM on wide rows. Cause: extraction batch sized by row count on a table with large blob columns. Signal: worker RSS grows linearly until the OOM killer fires. Remediation: size batches by byte budget, exclude non-participating blob columns from the digest set.

In This Reference

This equivalence model is developed further in a dedicated companion reference:

Frequently Asked Questions

Why not just compare row counts or a whole-table checksum?

A matching row count says nothing about whether the values agree, and a whole-table checksum tells you only that something diverged, never what or where. Equivalence modeling produces per-record — and optionally per-field — digests so a divergence is attributable to a specific key and column, which is what makes remediation targeted instead of a full re-migration.

Why quantize floats before hashing instead of comparing with an epsilon later?

Hashing collapses a value to a fixed digest, so two floats that differ below the tolerance must map to the same bytes before hashing — otherwise their digests differ and the tolerance never gets a chance to apply. Quantizing to the epsilon grid in canonicalize_value guarantees equivalent floats produce an identical row signature.

BLAKE2b or SHA-256 for the digest?

Both are collision-resistant and suitable for integrity evidence. BLAKE2b is typically faster in pure-Python pipelines and supports a configurable digest size; SHA-256 is often preferred where a NIST-standardized algorithm is mandated by an audit regime. Avoid MD5 entirely — its collision weakness disqualifies it from a defensible audit trail.

Does this stage ever write to either engine?

No. Equivalence modeling is strictly read-only. It emits a classified divergence stream that a separately authorized backfill job may act on; the reconciliation identity holds SELECT/SCAN grants only and never mutates source or target data.