Input Dataset + Transform Version + Run Context -> Output Dataset

Can we prove that Output Dataset is the expected consequence of Input Dataset?

case_id must not be null
amount must be >= 0
status must be one of OPEN, CLOSED, ESCALATED
assigned_officer_id must exist in officer reference table

All source cases updated between T1 and T2 are represented in the canonical case table.
The total penalty amount in the gold report equals the sum of validated silver penalty facts.
Every Kafka input offset up to N has either produced a sink effect or a quarantine record.
Every source file in the manifest has exactly one import ledger entry.

Data quality asks: Is this data valid?
Reconciliation asks: Is this the complete and correct consequence of the prior state?

public record ReconciliationResult(
    String reconciliationId,
    String pipelineId,
    String runId,
    String assetName,
    ReconciliationScope scope,
    ReconciliationMethod method,
    ReconciliationStatus status,
    Map<String, MetricValue> expected,
    Map<String, MetricValue> actual,
    List<ReconciliationDifference> differences,
    String transformVersion,
    Instant evaluatedAt
) {}

public enum ReconciliationStatus {
    PASSED,
    PASSED_WITH_TOLERANCE,
    FAILED,
    INCONCLUSIVE,
    SKIPPED
}

public record ReconciliationScope(
    String sourceDataset,
    String targetDataset,
    Instant sourceFromInclusive,
    Instant sourceToExclusive,
    Map<String, String> partitions,
    Map<String, String> sourcePositions
) {}

public enum ReconciliationMethod {
    COUNT,
    CHECKSUM,
    HASH_TOTAL,
    BALANCE_TOTAL,
    ANTI_JOIN,
    LEDGER,
    SAMPLE,
    INVARIANT,
    WATERMARK_COMPLETENESS
}

run_manifest
  -> input_manifest
  -> transform_manifest
  -> output_manifest
  -> reconciliation_result
  -> publication_decision

Because run R processed source snapshot S, transform version V, produced output snapshot O,
passed contract C, passed quality gate Q, passed reconciliation result K,
and publication event P promoted O at time T.

expected_count == actual_count

source_count = count(*) from source_table where updated_at >= T1 and updated_at < T2
sink_count   = count(*) from target_table where updated_at >= T1 and updated_at < T2

source accepted records == canonical records + rejected records

-- Source ingestion ledger
select count(*) as source_records
from ingestion_record_ledger
where run_id = :run_id
  and decision in ('ACCEPTED', 'REJECTED');

-- Canonical output
select count(*) as canonical_records
from canonical_case_event
where run_id = :run_id;

-- Rejections
select count(*) as rejected_records
from ingestion_rejection
where run_id = :run_id;

source_records = canonical_records + rejected_records

public enum CardinalityMode {
    ONE_TO_ONE,
    ZERO_OR_ONE,
    ONE_TO_MANY,
    MANY_TO_ONE,
    MANY_TO_MANY
}

public record TransformContract(
    String transformName,
    String version,
    CardinalityMode cardinalityMode,
    boolean requiresTerminalDecisionForEveryInput
) {}

source_count == target_count
source_checksum == target_checksum

JSON string as received

canonical key + canonical payload fields + source version + business effective time

import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.time.Instant;
import java.util.HexFormat;
import java.util.List;
import java.util.Objects;

public final class StableHash {
    private StableHash() {}

    public static String sha256(List<String> fields) {
        try {
            MessageDigest digest = MessageDigest.getInstance("SHA-256");
            for (String field : fields) {
                String normalized = normalize(field);
                digest.update(normalized.getBytes(StandardCharsets.UTF_8));
                digest.update((byte) 0x1F); // unit separator
            }
            return HexFormat.of().formatHex(digest.digest());
        } catch (NoSuchAlgorithmException e) {
            throw new IllegalStateException("SHA-256 unavailable", e);
        }
    }

    private static String normalize(String value) {
        return Objects.toString(value, "<NULL>").trim();
    }

    public static String caseEventHash(
        String caseId,
        String eventType,
        String status,
        Instant businessEffectiveAt,
        String schemaVersion
    ) {
        return sha256(List.of(
            caseId,
            eventType,
            status,
            businessEffectiveAt.toString(),
            schemaVersion
        ));
    }
}

sum(hash(case_id))

count_rows
count_distinct_keys
sum_hash_key
sum_hash_payload
min_event_time
max_event_time

import java.math.BigInteger;
import java.time.Instant;

public record PartitionFingerprint(
    long rowCount,
    long distinctKeyCount,
    BigInteger keyHashTotal,
    BigInteger payloadHashTotal,
    Instant minEventTime,
    Instant maxEventTime
) {
    public boolean sameAs(PartitionFingerprint other) {
        return rowCount == other.rowCount
            && distinctKeyCount == other.distinctKeyCount
            && keyHashTotal.equals(other.keyHashTotal)
            && payloadHashTotal.equals(other.payloadHashTotal)
            && minEventTime.equals(other.minEventTime)
            && maxEventTime.equals(other.maxEventTime);
    }
}

sum(source.penalty_amount) == sum(gold.penalty_amount)
sum(opening_balance + debits - credits) == closing_balance
sum(case_obligation_created) - sum(case_obligation_resolved) == active_obligation_balance

For every enforcement case in report month M:
  total_assessed_penalty
  - total_withdrawn_penalty
  - total_reduced_penalty
  + total_increased_penalty
  = currently_enforceable_penalty

public enum AmountEffect {
    INCREASE,
    DECREASE,
    NEUTRAL
}

public record AmountFact(
    String factId,
    String caseId,
    String amountType,
    AmountEffect effect,
    long amountMinorUnits,
    String currency,
    Instant effectiveAt
) {
    public long signedAmount() {
        return switch (effect) {
            case INCREASE -> amountMinorUnits;
            case DECREASE -> -amountMinorUnits;
            case NEUTRAL -> 0L;
        };
    }
}

-- Source keys missing from target
select s.case_id
from source_keys s
left join target_keys t on t.case_id = s.case_id
where t.case_id is null;

-- Target keys not explainable by source
select t.case_id
from target_keys t
left join source_keys s on s.case_id = t.case_id
where s.case_id is null;

1. Compute partition fingerprints.
2. If fingerprints match, mark partition reconciled.
3. If fingerprints differ, run anti-join drilldown for that partition.
4. Store first N differences plus difference counts.
5. Open incident or block publication depending on severity.

Every input receives a terminal accounting state.

create table pipeline_effect_ledger (
    ledger_id uuid primary key,
    pipeline_id text not null,
    run_id text not null,
    source_system text not null,
    source_position text not null,
    source_record_id text not null,
    idempotency_key text not null,
    terminal_state text not null,
    target_asset text,
    target_key text,
    output_fingerprint text,
    error_code text,
    error_message text,
    transform_version text not null,
    recorded_at timestamptz not null,
    unique (pipeline_id, idempotency_key)
);

For every source input in scope,
there exists exactly one terminal ledger state for the active transform version and run mode.

public interface EffectLedger {
    LedgerDecision recordSuccess(EffectSuccess success);
    LedgerDecision recordRejection(EffectRejection rejection);
    LedgerDecision recordQuarantine(EffectQuarantine quarantine);
    boolean alreadyRecorded(String pipelineId, String idempotencyKey);
}

public record EffectSuccess(
    String pipelineId,
    String runId,
    String sourcePosition,
    String sourceRecordId,
    String idempotencyKey,
    String targetAsset,
    String targetKey,
    String outputFingerprint,
    String transformVersion,
    Instant recordedAt
) {}

public enum LedgerDecision {
    INSERTED,
    ALREADY_EXISTS_SAME_EFFECT,
    CONFLICT_DIFFERENT_EFFECT
}

For partition P and offsets [A..B], what happened to each record?

PROCESSED_TO_SINK
REJECTED_TO_DLQ
QUARANTINED
SKIPPED_BY_RULE
DUPLICATE_ALREADY_APPLIED
FAILED_RETRYABLE
FAILED_NON_RETRYABLE

public record OffsetEffect(
    String topic,
    int partition,
    long offset,
    String eventId,
    String effectState,
    String sinkAsset,
    String sinkKey,
    String errorCode,
    Instant processedAt
) {}

select
    topic,
    partition,
    min(offset) as min_offset,
    max(offset) as max_offset,
    count(*) as accounted_offsets,
    count(*) filter (where effect_state = 'PROCESSED_TO_SINK') as sink_success,
    count(*) filter (where effect_state = 'QUARANTINED') as quarantined,
    count(*) filter (where effect_state like 'FAILED%') as failed
from kafka_offset_effect_ledger
where topic = :topic
  and partition = :partition
  and offset between :from_offset and :to_offset
group by topic, partition;

{
  "batchId": "batch-2026-07-04-001",
  "producer": "case-management-exporter",
  "files": [
    {
      "path": "landing/cases-0001.csv",
      "rowCount": 250000,
      "sha256": "...",
      "schemaVersion": "case-export.v3"
    }
  ]
}

manifest_file_count == discovered_file_count
manifest_row_count == parsed_row_count
manifest_hash == actual_file_hash
parsed_row_count == accepted + rejected
accepted == canonical_output + skipped_with_reason

create table file_import_ledger (
    batch_id text not null,
    file_path text not null,
    file_hash text not null,
    declared_row_count bigint not null,
    parsed_row_count bigint,
    accepted_row_count bigint,
    rejected_row_count bigint,
    import_status text not null,
    imported_at timestamptz,
    archived_at timestamptz,
    primary key (batch_id, file_path)
);

source table snapshot count == target snapshot count
source table key hash == target key hash
source max primary key / source commit position recorded

source log position range -> CDC events -> canonical events -> sink effects

select count(*) from source_table where not deleted
==
select count(*) from target_current_table where is_current = true

count + key hash + payload hash + delete count + max source position

select count(*) from cdc_event
where op = 'DELETE'
  and source_lsn between :from and :to;

select count(*) from target_current
where deleted_at is not null
  and source_lsn between :from and :to;

public record WindowAggregateOutput(
    String windowId,
    Instant windowStart,
    Instant windowEnd,
    Instant watermarkAtEmission,
    long inputEventCount,
    long distinctBusinessKeyCount,
    long rejectedEventCount,
    long lateEventCount,
    String aggregateFingerprint,
    String transformVersion,
    boolean finalForWatermark,
    boolean subjectToCorrection
) {}

For window W:
  input events assigned to W
  - rejected events
  - duplicates
  + valid correction effects
  = aggregate contribution count

PRELIMINARY
WATERMARK_FINAL
BUSINESS_FINAL
RESTATED
SUPERSEDED

backfill_campaign_id
source_scope
transform_version
target_scope
expected_mutation
staging_location
validation_result
reconciliation_result
approval_record
publication_event
rollback_or_supersession_plan

source fingerprint == staged output explainability fingerprint
old output is not mutated directly
staged output passes quality and reconciliation
publication is atomic at partition/snapshot boundary

asset: gold.enforcement_monthly_report
reconciliation:
  required: true
  failClosed: true
  checks:
    - method: COUNT
      tolerance: 0
      severityOnFailure: P1
    - method: BALANCE_TOTAL
      fields: [assessed_penalty_minor_units]
      tolerance: 0
      severityOnFailure: P0
    - method: CHECKSUM
      tolerance: 0
      severityOnFailure: P1

public interface ReconciliationCheck {
    ReconciliationMethod method();
    ReconciliationScope scope();
    MetricSet expected(ReconciliationContext context);
    MetricSet actual(ReconciliationContext context);
    ReconciliationDecision compare(MetricSet expected, MetricSet actual);
}

public record ReconciliationContext(
    String pipelineId,
    String runId,
    String assetName,
    String transformVersion,
    Map<String, String> parameters
) {}

public record MetricSet(Map<String, MetricValue> values) {}

public sealed interface MetricValue permits LongMetric, DecimalMetric, StringMetric, HashMetric {}

public record LongMetric(long value) implements MetricValue {}
public record DecimalMetric(java.math.BigDecimal value) implements MetricValue {}
public record StringMetric(String value) implements MetricValue {}
public record HashMetric(String algorithm, String value) implements MetricValue {}

public record ReconciliationDecision(
    ReconciliationStatus status,
    List<ReconciliationDifference> differences,
    String explanation
) {}

public record ReconciliationDifference(
    String metricName,
    String expected,
    String actual,
    String severity,
    String explanation
) {}

import java.util.ArrayList;
import java.util.List;

public final class ReconciliationRunner {
    private final ReconciliationResultStore resultStore;

    public ReconciliationRunner(ReconciliationResultStore resultStore) {
        this.resultStore = resultStore;
    }

    public List<ReconciliationResult> run(
        ReconciliationContext context,
        List<ReconciliationCheck> checks
    ) {
        List<ReconciliationResult> results = new ArrayList<>();

        for (ReconciliationCheck check : checks) {
            MetricSet expected = check.expected(context);
            MetricSet actual = check.actual(context);
            ReconciliationDecision decision = check.compare(expected, actual);

            ReconciliationResult result = new ReconciliationResult(
                java.util.UUID.randomUUID().toString(),
                context.pipelineId(),
                context.runId(),
                context.assetName(),
                check.scope(),
                check.method(),
                decision.status(),
                expected.values(),
                actual.values(),
                decision.differences(),
                context.transformVersion(),
                java.time.Instant.now()
            );

            resultStore.save(result);
            results.add(result);
        }

        return results;
    }
}

public interface ReconciliationResultStore {
    void save(ReconciliationResult result);
}

overwrite production partition
then reconcile
then discover mismatch

write staging partition
reconcile staging against source
publish atomically
store publication event

case_id: CASE-2026-00931
source events:
  - CASE_OPENED at valid time T1, recorded time R1
  - CASE_ESCALATED at valid time T2, recorded time R2
  - PENALTY_ASSESSED at valid time T3, recorded time R3
transform version: enforcement-report-transform:4.2.1
reporting rule version: monthly-enforcement-rules:2026.07
window: 2026-07-01T00:00:00Z to 2026-08-01T00:00:00Z
output row fingerprint: sha256:...
reconciliation status: PASSED
publication snapshot: iceberg snapshot 918273645

case_id: CASE-2026-00412
terminal decision: SKIPPED
reason: case did not reach reportable status before business cutoff
rule version: monthly-enforcement-rules:2026.07
validated_at: 2026-08-01T03:01:12Z

[ ] Does every critical asset have declared reconciliation methods?
[ ] Is reconciliation scope explicit and tied to run manifest?
[ ] Are source positions recorded?
[ ] Are transform and contract versions recorded?
[ ] Are source, target, and exception counts compared?
[ ] Are checksums/fingerprints used where counts are insufficient?
[ ] Are business balances reconciled for amount-bearing facts?
[ ] Are deletes/retractions/corrections reconciled separately?
[ ] Is reconciliation performed before destructive publication?
[ ] Are failed reconciliations tied to block/warn/quarantine policy?
[ ] Are differences stored with enough examples to debug?
[ ] Are results queryable by asset, run, partition, time, and source position?
[ ] Is reconciliation evidence retained according to audit policy?
[ ] Are backfills reconciled more strictly than normal runs?
[ ] Are known exceptions explicit, approved, and expiring?

A pipeline should not merely move data.
It should produce evidence that the movement was complete, correct, bounded, and explainable.

Trust me, I ran.

Here is the input scope, output scope, transform version, reconciliation proof, exceptions, and publication evidence.