The Maple Syrup Guide to Data Lakes: Tap, Boil, Grade

There is no better Sunday ritual than a stack of waffles and a slow pour of warm, amber maple syrup.

It's easy to forget what that syrup used to be. A few weeks before it reached a bottle, it was sap: a clear, watery liquid that runs when spring nights still freeze and the days begin to thaw. Sap from a sugar maple is only about 2% sugar. Sugar makers use a rule of thumb called the rule of 86: divide 86 by the sugar percentage, and you get the litres of sap needed for one litre of syrup. At 2%, that's about 43. Nearly everything a sugar maker does is removing water without ruining the sugar.

Then they check. Syrup must finish between 66 and 68.9 °Brix. Too thin, and it ferments in the jug. Too thick, and it crystallizes. Every batch is also graded by how much light passes through it, into four Grade A colour classes, from Golden to Very Dark.

Now a story about what happens when you skip the check.

In July 2012, an accountant was taking the annual inventory at a warehouse in Saint-Louis-de-Blandford, Quebec. It held part of the province's strategic maple syrup reserve: thousands of barrels stacked floor to ceiling. He climbed onto a stack, and a barrel that should have weighed a few hundred pounds rocked under his feet. It was empty. Others held water.

Over several months, thieves had trucked barrels out, siphoned off the syrup, refilled them with water, and put them back. About 3,000 tonnes went missing, worth roughly C$18.7 million. The barrels were in the right place, sealed and stacked. Counting them said everything was fine. Weighing them didn't. The reserve only did a full physical check once a year, so nobody weighed them for months.

Data platforms fail the same way. The tables are in the right schema, row counts are green, the dashboard loads in 200 milliseconds, and the number inside is water.

Modern data engineering has a name for the sugar shack's layout: the medallion architecture, with bronze, silver and gold layers. Databricks popularized the name, but the idea works on any lake. Most write-ups stop at the layout. This one is about the two jobs the layout exists for: taking the water out, and weighing the barrels.

The first half is the sugar shack: what each layer is for, and whether you need one at all. The second half is a Tuesday morning, a revenue number that's off by C$330,000, and a working lakehouse called sugarshack that accounts for every cent of it.

Reading the sugarbush

A family with ten taps boils sap on the kitchen stove. An operation with ten thousand taps builds a sugar shack. Before you build one, three questions decide which you are.

Will you ever need to read something again?

Sources change their minds. An API reshapes its payload. A database replays a day of changes after an outage. A bank republishes yesterday's exchange rate with a correction. If your pipeline parses data on arrival and keeps only the result, every one of those becomes a data loss you can't undo. If it keeps what arrived, each one becomes a replay.

Do you control the sources?

If all your data comes from one database your team owns, you can fix problems at the source. Once you're receiving webhooks from a payment processor, change streams from someone else's database and files from a partner, you can't. You can only defend yourself.

What does a wrong number cost, and who would notice?

An internal dashboard that's off by 2% is an annoyance. A revenue figure that goes to a board, a regulator or a bank is not. The more a number is trusted, the more it needs to be checked against something you didn't compute yourself.

	The stove	The sugar shack
Sources	One or two databases you own	Many, including APIs, change streams and files you don't control
When a source changes	Fix it upstream	Keep the raw data, add a parser, replay
Checks	Tests on the final tables	Checks at every layer, and against an outside source of truth
Good fit for	Internal analytics, early-stage products	Finance, billing, anything reconciled or audited

If you're on the stove, stay there. A warehouse and a handful of dbt models is a perfectly good way to make syrup for ten taps. The rest of this post is for when it isn't.

From tree to table

A sugar shack is a sequence of rooms, and each one has a single job.

The tubing is ingestion. Sugar makers once hung buckets on trees and collected them on a schedule. That's batch ingestion: files land in cloud storage and wait to be swept up. Most now run tubing from tree to shack, which is streaming. Many also put the lines under vacuum to pull sap as soon as it runs, which is closer to change data capture (CDC): a database's own change log, streamed as it happens. Sap also doesn't flow evenly. It runs hard on a thawing afternoon and stops overnight, so the shack keeps a tank that absorbs the surge. Your landing zone does the same for traffic spikes and backfills.

Bronze is the sap tank. Sap goes into the tank exactly as it came out of the tree, still 98% water. Nothing is filtered, boiled or thrown away. In a lake, bronze keeps every record exactly as the source sent it, plus when and how it arrived. It is your only way back if a later step gets something wrong.

Silver is the reverse osmosis unit and the evaporator. Reverse osmosis pushes sap through membranes to remove much of the water before it meets any heat. The evaporator boils off most of the rest. Along the way, minerals called sugar sand settle out and get filtered. Silver does the equivalent work on data: it parses, types, deduplicates, and turns change logs into current state. Records it can't use are set aside with a reason attached, not thrown away. A sugar maker watches what the filter catches, because a sudden change there means something changed in the sap. We'll do the same.

Gold is the finishing pan, and then the grader. Late in the boil, the Maillard reaction and caramelization turn sweet water into something that tastes like maple. Gold is where business meaning appears: facts joined to dimensions, currencies converted, metrics defined. But syrup isn't barrelled because the pan says it's done. It's measured for density and graded for colour first. Gold data shouldn't be published because the SQL succeeded, either. It should be graded first. Most medallion diagrams leave this step out, and it's the one that would have caught the heist.

Tuesday, 9:12 a.m.

Saltbox Market is a made-up online marketplace in Halifax that sells work by independent makers across Canada and the US. Canadian customers pay in CAD. American customers pay in USD through a separate US account with the same payment processor.

It's Tuesday, September 8, 2026, the morning after Labour Day. The CFO forwards a screenshot with one line: Can someone explain this before Thursday's board deck?

The revenue dashboard says August brought in C$2,864,598.75. Finance has summed the processor's daily settlement reports and converted the USD at each day's exchange rate: C$3,195,004.53. The two numbers are C$330,405.78 apart.

Ten minutes later, the warehouse lead writes to the same channel. The operations dashboard shows 54 orders that were paid more than three days ago and still haven't shipped. The team has been hunting for them since 8 a.m. They've found four.

The pipeline behind both dashboards was built quickly, a year ago, and it has been fine. Let's call it the bucket brigade. It reads the landing folder directly:

# sugarshack/bucket.py (the old pipeline)
def naive_revenue(through: datetime, month: str) -> float:
    """'Revenue' as the dashboard computed it: every charge in the month, from the files."""
    with duckdb.connect() as con:
        con.execute("set TimeZone = 'UTC'")
        return con.execute("""
            select sum(try_cast(json->>'$.data.object.amount' as bigint)) / 100.0
            from read_json_objects(?, format = 'newline_delimited', ignore_errors = true)
            where json->>'$.type' = 'charge.succeeded'
              and strftime(to_timestamp((json->>'$.created')::bigint), '%Y-%m') = ?
        """, [landed("payments", through), month]).fetchone()[0] or 0.0


def naive_queue(through: datetime) -> tuple[int, int]:
    """Orders 'awaiting shipment': the last change we received for each order says 'paid'.
    Returns (waiting, waiting for more than 72 hours)."""
    latest = {}
    for path in landed("orders", through):         # files in arrival order...
        for line in open(path):
            change = json.loads(line)
            if change["after"]:                     # ...deletes have no 'after', so skip them
                latest[change["after"]["order_id"]] = change["after"]
    ...

It reads well. It's wrong in five ways, and they don't all push in the same direction:

1. Re-sent files. The landing job sometimes retries an upload and drops an identical file under a new name. Both get read. (−C$16,287.17)
2. Webhooks delivered twice. The processor promises at-least-once delivery: if it isn't sure you received an event, it sends it again. About one in forty events arrives twice. (−C$72,401.70)
3. A payload that changed shape. On August 17, the processor moved Saltbox's US account to a new API version. Amounts became objects, {"value": 8167, "currency": "USD"}, instead of plain integers. try_cast didn't fail. It quietly returned null, and sum quietly skipped it. Two weeks of US sales vanished without an error. (+C$319,482.85)
4. Refunds. The dashboard was built when finance handled refunds in a spreadsheet, so it only ever counted charges. (−C$149,038.45)
5. Currencies. US dollars were added as if they were Canadian. (+C$248,650.25)

Mistakes that push in opposite directions are the dangerous kind. A 10% gap is noticeable. Five errors that net to 10% are much harder to find, because fixing any one of them doesn't close the gap, and sometimes widens it.

The warehouse's 54 phantom orders have a different cause. Saltbox's order database streams its changes through CDC. Every so often, part of that stream falls a day or two behind, and the changes in it arrive after newer ones. Here's one order:

The bucket brigade keeps the last change it received, so this order has been "paid, not shipped" since August 4. Of the 54 orders on the list, 46 had shipped and 4 had been cancelled. Only 4 were really waiting.

None of this is exotic. Every source system does at least one of these things. So let's build the shack. I've called it sugarshack.

A shack you can carry

The version of this I'd run in production is Spark on Databricks, with Delta Lake or Iceberg tables, Auto Loader for ingestion, and dbt for silver and gold. That's also the version you can't run on a laptop in two minutes. It needs a JVM, a cluster or at least a local Spark install, and usually a cloud account before you've seen a single row.

sugarshack runs on DuckDB and dbt, and that's a deliberate choice. pip install is the whole setup: no Java, no Docker, no cloud account. Generating the season and building the lake takes well under a minute; replaying it one day at a time, as this post does, takes about two. Every layer is a file or a table you can open. The sources are generated from a fixed seed, so you should get exactly the numbers in this post. And the ideas that make a medallion architecture work don't depend on the engine. Ordering by commit sequence and grading before you publish look the same in DuckDB as they do on a cluster.

The trade-offs are real. DuckDB is one process on one machine, with one writer at a time. Bronze's commit log is a JSON-lines file I wrote myself, standing in for what Delta Lake's transaction log or Iceberg's metadata gives you with much stronger guarantees. Here's how each piece maps to its production version:

In sugarshack	At lake scale
Landing folder, sugarshack tap	Auto Loader, or a Kafka Connect sink to object storage
Parquet files plus _manifest.jsonl	A Delta Lake or Iceberg table (the transaction log is the manifest)
dbt-duckdb incremental merge	dbt-databricks, or Spark MERGE INTO on Delta
Folding CDC by LSN	Auto CDC (formerly APPLY CHANGES) in Lakeflow pipelines, sequenced by the LSN
silver.quarantine	Expectations in Lakeflow pipelines (formerly Delta Live Tables), plus a quarantine table
ASOF JOIN for exchange rates	A range join against rates with valid-from and valid-to dates
gold_build, graded, then published to gold	Write-audit-publish with Iceberg branches, or build-then-swap schemas on Delta

Tap, boil, grade: the build, layer by layer

The code lives in my blog-tools repository, next to sumbisori from the last post. The README takes you from an empty machine to every result below.

git clone https://github.com/Kunal-Vyas/blog-tools.git
cd blog-tools/sugarshack
python3 -m venv .venv && source .venv/bin/activate
pip install -e .

sugarshack sugarbush                          # generate the season's source files
sugarshack run --through 2026-08-16           # tap -> boil -> grade, as of that moment

Here's the whole shack, and what can stop a batch at each step:

1. The sugarbush: make the mess on purpose

You can't test a lake against clean data. sugarshack sugarbush writes the sources for August and September 2026, with the problems built in:

Source	Shape	What goes wrong
orders	CDC events from Postgres, one JSON line per change	About 2% arrive hours or days late. Test orders are created, then deleted.
payments	Processor webhooks	About 2.5% delivered twice. The US account changes API version on August 17. A crashed receiver writes a few truncated lines.
fx_rates	One CSV per business day	No rates on weekends. August 12 is republished with a correction.
settlements	The processor's daily totals per currency, in cents	Nothing. This is the independent scale.

On top of that, the landing job re-sends seven files byte for byte.

Every file name carries the time it arrived. Every other command takes --through, and only sees files that had arrived by then. That's how the post can replay the season one morning at a time.

The API change is worth seeing, because the silver layer will have to deal with it:

// api_version 2025-11-01: every CAD event, and USD before August 17
{"id": "evt_…", "type": "charge.succeeded", "api_version": "2025-11-01", "created": 1786712452,
 "data": {"object": {"id": "ch_…", "order_id": "ord_…", "amount": 8167, "currency": "usd"}}}

// api_version 2026-08-01: USD from August 17
{"id": "evt_…", "type": "charge.succeeded", "api_version": "2026-08-01", "created": 1787237997,
 "data": {"object": {"id": "ch_…", "amount": {"value": 8167, "currency": "USD"},
                     "metadata": {"order_id": "ord_…"}}}}

Three things moved at once: the amount, the currency and its capitalization, and the order ID. That's typical of a real version change, and it's why "just coalesce the two paths" rarely survives contact with one.

2. Bronze: the sap tank

Bronze has three rules, and nearly all of its code is about keeping them when something fails halfway through.

1. Never parse. Each line is stored as a string. A payload the parser can't read today will still be there tomorrow, when it can.
2. Never take the same file twice. Files are identified by a hash of their contents, so a re-sent file is recognized even under a new name.
3. A batch exists only once the log says so. Parquet is written first. The manifest line is written last.

# sugarshack/tap.py (excerpt)
def tap(through: datetime) -> dict:
    BRONZE.mkdir(parents=True, exist_ok=True)
    manifest = read_manifest()
    seen = {m["sha256"] for m in manifest}
    tapped = {m["file"] for m in manifest}
    committed = {m["batch_id"] for m in manifest if m["batch_id"]}
    next_seq = max((m["batch_seq"] for m in manifest if m["batch_seq"]), default=0) + 1
    report = {"orphans_removed": sweep_orphans(committed), "batches": []}
    ingested_at = datetime.now(UTC)
    duplicates = []

    for source in SOURCES:
        new, rows = [], []
        for path in sorted((LANDING / source).glob("*/*.*")):
            name = str(path.relative_to(LANDING))
            when = arrived_at(path)
            if when > through or name in tapped:
                continue
            digest = hashlib.sha256(path.read_bytes()).hexdigest()
            if digest in seen:                      # same bytes as a file we already have
                duplicates.append({"source": source, "batch_id": None, "batch_seq": None,
                                   "parquet": None, "ingested_at": ingested_at.isoformat(),
                                   "file": name, "sha256": digest})
                continue
            seen.add(digest)
            new.append({"file": name, "sha256": digest})
            for n, line in enumerate(lines_of(path), start=1):
                rows.append({"_raw": line, "_landing_file": name,
                             "_line": n, "_file_sha256": digest, "_arrived_at": when.isoformat()})
        if not new:
            continue

        # Batch id from content: the same set of files always gets the same name.
        hashes = "".join(sorted(n["sha256"] for n in new))
        batch_id = hashlib.sha256(hashes.encode()).hexdigest()[:16]
        folder = BRONZE / source / f"ingest_date={ingested_at:%Y-%m-%d}"
        folder.mkdir(parents=True, exist_ok=True)
        target = folder / f"{batch_id}.parquet"
        _write_parquet(rows, target, batch_id, next_seq, ingested_at)

        # Commit: one manifest line per landing file, appended and flushed to disk.
        with MANIFEST.open("a") as log:
            for n in new:
                log.write(json.dumps({"source": source, "batch_id": batch_id,
                                      "batch_seq": next_seq,
                                      "parquet": str(target.relative_to(BRONZE.parent)),
                                      "ingested_at": ingested_at.isoformat(), **n}) + "\n")
            log.flush()
            os.fsync(log.fileno())
        report["batches"].append({"source": source, "batch_seq": next_seq,
                                  "files": len(new), "rows": len(rows)})
        next_seq += 1

    if duplicates:                      # remember them, so they are not re-hashed every run
        with MANIFEST.open("a") as log:
            log.writelines(json.dumps(d) + "\n" for d in duplicates)
    report["skipped_duplicates"] = len(duplicates)
    return report

The details that matter:

• The hash is the identity, not the name. The landing job's re-sends are skipped because their bytes match a file already in bronze. The manifest records them too, so they aren't re-hashed on every run.
• Metadata goes on every row. _landing_file, _line, _arrived_at, _batch_seq and _ingested_at answer the question you'll eventually be asked: where did this number come from?
• Writes are atomic. Parquet goes to a temporary name and is renamed into place, so nobody reads half a file. The manifest is appended and flushed to disk only after that.
• A crash leaves an orphan, not a duplicate. If the process dies between the Parquet write and the manifest write, the next run deletes the unlisted file before doing anything else.

The last point needs one more piece. Between the crash and the next run, that orphan file sits in the folder. So the readers don't list the folder. They read the log:

-- sugarshack/shack/macros/bronze.sql (excerpt)
{#
  Readers list the log, not the folder.

  A bronze view selects exactly the Parquet files the manifest has committed. A file a
  crashed run left behind is invisible here even before tap.py sweeps it away. Table
  formats (Delta, Iceberg) work the same way: the transaction log decides what a table is.
#}
{% macro bronze_source(source) %}
    {%- set files = [] -%}
    {%- if execute -%}
        {%- set q -%}
            select distinct parquet
            from read_json('{{ var("lake") }}/bronze/_manifest.jsonl',
                           columns = {source: 'VARCHAR', parquet: 'VARCHAR'})
            where source = '{{ source }}' and parquet is not null
            order by 1
        {%- endset -%}
        {%- set files = run_query(q).columns[0].values() -%}
    {%- endif %}
    {%- if files | length > 0 %}
    select *, _landing_file || ':' || _line as _bronze_row
    from read_parquet([
        {%- for f in files %}
        '{{ var("lake") }}/{{ f }}'{{ "," if not loop.last }}
        {%- endfor %}
    ])
    {%- else %}
    select null::varchar as _raw, null::varchar as _landing_file, null::integer as _line,
           null::varchar as _file_sha256, null::timestamptz as _arrived_at,
           null::varchar as _batch_id, null::bigint as _batch_seq,
           null::timestamptz as _ingested_at, null::varchar as _bronze_row
    where false
    {%- endif %}
{% endmacro %}

This is the idea that Delta Lake and Iceberg are built on. A table isn't whatever happens to be in a directory. It's whatever the transaction log says it is. Writing the small version once makes it much easier to understand what the real ones are protecting you from.

The first run, as of the end of August 16:

as of 2026-08-16 23:59 UTC
  tapped   orders       batch   1   384 files   43,035 rows
  tapped   payments     batch   2   384 files   15,382 rows
  tapped   fx_rates     batch   3    12 files       12 rows
  tapped   settlements  batch   4    15 files       30 rows
  skipped  3 file(s) with the same bytes as one already in bronze

Run it again with the same --through and it says nothing new has arrived. Taking data in should be idempotent, so that a scheduler can retry it without anyone worrying.

3. Silver: reverse osmosis and the evaporator

Silver has four jobs, and each one maps to a line in the Tuesday post-mortem.

Parse by version, and quarantine the rest. Each API version the pipeline understands gets an entry in one place:

-- sugarshack/shack/macros/parsers.sql
{#
  One entry per payment API version the pipeline understands. Each field is a SQL
  expression over `j`, the parsed JSON event.

  Supporting a new version means adding an entry here and replaying from bronze. Until
  then, events in that version are quarantined, never guessed at.
#}
{% macro payment_parsers() %}
    {{ return({
        "2025-11-01": {
            "charge_id": "j->>'$.data.object.id'",
            "refund_id": "j->>'$.data.object.refund.id'",
            "order_id":  "j->>'$.data.object.order_id'",
            "amount":    "coalesce(j->>'$.data.object.refund.amount',
                                   j->>'$.data.object.amount')",
            "currency":  "j->>'$.data.object.currency'",
        },
        "2026-08-01": {
            "charge_id": "coalesce(j->>'$.data.object.charge', j->>'$.data.object.id')",
            "refund_id": "case when j->>'$.data.object.object' = 'refund'
                                then j->>'$.data.object.id' end",
            "order_id":  "j->>'$.data.object.metadata.order_id'",
            "amount":    "j->>'$.data.object.amount.value'",
            "currency":  "j->>'$.data.object.amount.currency'",
        },
    }) }}
{% endmacro %}


{% macro enabled_parsers() %}
    {{ return(var('payment_parsers').split(',') | map('trim') | list) }}
{% endmacro %}


{# CASE api_version WHEN <v> THEN <this field's expression in v> ... END, enabled versions only #}
{% macro parse(field) %}
    {%- set all = payment_parsers() -%}
    case api_version
    {%- for version in enabled_parsers() if version in all %}
        when '{{ version }}' then {{ all[version][field] }}
    {%- endfor %}
    end
{%- endmacro %}

The parsing model uses those expressions, and gives every unusable row a reason instead of a null:

-- sugarshack/shack/models/silver/stg_payment_events.sql (excerpt)
select
    event_id,
    case event_type_raw when 'charge.succeeded' then 'capture'
                        when 'charge.refunded'  then 'refund' end as event_type,
    api_version, charge_id, refund_id, order_id, amount_minor, currency, created_at,
    case
        when j is null                                        then 'unparseable_json'
        when api_version not in ('{{ enabled_parsers() | join("', '") }}')
                                                              then 'unsupported_api_version'
        when event_type_raw not in ('charge.succeeded', 'charge.refunded')
                                                              then 'unknown_event_type'
        when event_id is null or created_at is null           then 'missing_envelope'
        when order_id is null                                 then 'missing_order_id'
        when amount_minor is null or amount_minor <= 0        then 'bad_amount'
        when currency not in ('cad', 'usd')                   then 'unknown_currency'
        when event_type_raw = 'charge.refunded' and refund_id is null
                                                              then 'missing_refund_id'
    end as quarantine_reason,
    _raw, _bronze_row, _batch_seq, _arrived_at, _ingested_at
from parsed

Now replay August 17 as the pipeline would have seen it before anyone knew about the new version. SUGAR_PARSERS limits silver to the old one:

$ SUGAR_PARSERS=2025-11-01 sugarshack run --through 2026-08-17T12:00
as of 2026-08-17 12:00 UTC
  tapped   orders       batch   5    13 files    1,049 rows
  tapped   payments     batch   6    13 files      284 rows
  tapped   settlements  batch   7     1 files        2 rows
  boiled   15 models, 17 tests
  warn     relationships_silver_payments_order_id__order_id__ref_silver_orders_: 1 rows
  BLOCK  quarantine rate  orders 0.00% of 1,049; payments 20.42% of 284 (mostly unsupported_api_version)
  PASS   conservation     15,224 events, same count and same net in silver and gold
  PASS   fx coverage      every USD event has a rate; the oldest used is 2 days old
  WARN   reconciliation   still settling: 2026-08-16 CAD net off by -298.01

Not published. gold still shows the build as of 2026-08-16 23:59 UTC.

This is the moment the bucket brigade got wrong. The same change arrives, but instead of turning into a quiet null, a fifth of the morning's payment events land in silver.quarantine labelled unsupported_api_version. Grading blocks the publish. The dashboard keeps showing August 16, correctly labelled as August 16, instead of a wrong number for August 17. Someone gets paged at breakfast rather than finding out from the CFO three weeks later.

The fix is a parser entry, then a replay from bronze:

$ sugarshack run --through 2026-08-17T12:00 --replay
as of 2026-08-17 12:00 UTC
  nothing new has arrived
  replayed 15 models, 17 tests
  PASS   quarantine rate  orders 0.00% of 1,049; payments 0.00% of 284
  PASS   conservation     15,281 events, same count and same net in silver and gold
  ...
Published gold as of 2026-08-17 12:00 UTC.

Nothing had to be re-requested from the processor. Everything it sent was already in the tank. That's what bronze is for.

Deduplicate on the event's own ID. At-least-once delivery means the event ID is the idempotency key:

-- sugarshack/shack/models/silver/silver_payments.sql
-- One row per payment event, however many times it was delivered.
--
-- The processor delivers at least once, so the event id is the idempotency key. Within
-- the new batches, keep the first delivery; against the table, insert only ids we have
-- never seen. Nothing is ever updated: a payment event is a fact, not a state.
{{ config(
    materialized='incremental',
    incremental_strategy='merge',
    unique_key='event_id',
    merge_clauses={'when_not_matched': [{'action': 'insert', 'mode': 'by_name'}]},
) }}

select
    event_id, event_type, api_version, charge_id, refund_id, order_id,
    amount_minor,                       -- integer minor units (cents): money is never a float
    currency, created_at,
    _bronze_row, _batch_seq, _arrived_at, _ingested_at
from {{ ref('stg_payment_events') }}
where quarantine_reason is null
  and {{ new_batches() }}
qualify row_number() over (partition by event_id
                           order by _batch_seq, _arrived_at, _bronze_row) = 1

Within the new batches, keep the first delivery. Against the existing table, only insert IDs you've never seen. Nothing is ever updated, because a payment event is a fact, not a state. The qualify clause deduplicates the new rows against each other; the merge deduplicates them against history. You need both.

Fold change logs by commit order, not arrival order. This is the fix for the warehouse's phantom orders:

-- sugarshack/shack/models/silver/silver_orders.sql
-- The current state of every order, folded from its change log.
--
-- The rule that makes this correct: the newest change is the one with the highest LSN,
-- not the one that arrived last. Late and replayed events arrive after newer ones all the
-- time, so the merge only overwrites a row when the incoming LSN is higher.
-- Deletes are kept as tombstones (is_deleted) so a late update can't resurrect the row.
{{ config(
    materialized='incremental',
    incremental_strategy='merge',
    unique_key='order_id',
    merge_update_condition='DBT_INTERNAL_SOURCE._lsn > DBT_INTERNAL_DEST._lsn',
) }}

select
    order_id, customer_id, status, currency, total_minor, ship_region,
    created_at, updated_at,
    op = 'd'  as is_deleted,
    lsn       as _lsn,
    _batch_seq, _arrived_at, _ingested_at
from {{ ref('stg_order_changes') }}
where quarantine_reason is null
  and {{ new_batches() }}
qualify row_number() over (partition by order_id order by lsn desc) = 1

A log sequence number (LSN) is the database's own record of the order in which changes were committed. The merge only overwrites a row when the incoming LSN is higher, so a late "paid" can't overwrite an earlier-arrived "shipped". The easy mistake is order by _ingested_at desc. It looks reasonable, passes every test written with tidy data, and is exactly the bucket brigade's bug.

Two details make it hold up:

• Deletes become tombstones. A deleted order stays in the table with is_deleted = true. If it vanished instead, a late update for it would bring it back.
• It relies on full row images. Debezium-style change events carry the whole row after each change, so the highest LSN wins outright. If your source only sends the columns that changed, you have to merge column by column instead.

Keep money in integers, and test each row. Amounts stay in cents (amount_minor bigint) until the last possible moment. Floats can't represent most decimal amounts exactly, and a reconciliation that has to allow for rounding differences can't tell a rounding error from a real one.

Row-level checks are ordinary dbt tests:

# sugarshack/shack/models/silver/silver.yml (excerpt)
# Record-level checks: is each row syrup? (Batch-level checks live in grade.py.)
models:
  - name: silver_payments
    columns:
      - name: event_id
        data_tests: [unique, not_null]
      - name: event_type
        data_tests:
          - accepted_values:
              arguments: {values: [capture, refund]}
      - name: currency
        data_tests:
          - accepted_values:
              arguments: {values: [cad, usd]}
      - name: amount_minor
        data_tests: [not_null, positive]
      - name: order_id
        data_tests:
          # CDC can lag the webhook by hours, so a payment may briefly precede its order.
          - relationships:
              arguments: {to: ref('silver_orders'), field: order_id}
              config: {severity: warn}

  - name: silver_orders

The relationships test is set to warn on purpose. CDC can lag the webhooks by hours, so for a while, a payment can exist before its order does. That's normal, and a test that fails on normal behaviour gets switched off. Name the lag, give it a severity, and move on.

4. Gold: the finishing pan

One fact table, two dates. Every money movement in gold carries the day the processor recorded it and the day the customer ordered:

-- sugarshack/shack/models/gold/fct_payments.sql
-- Every money movement, signed, with both of its dates and its CAD value.
--
--   processor_date  the UTC day the processor booked it. Settlement reports use this,
--                   so reconciliation must too.
--   order_date      the Halifax day the customer ordered. Product and marketing use this.
--
-- USD converts at the rate for the processor date: when the money moved. Weekends and
-- holidays have no rate, so the ASOF join takes the latest rate on or before that day,
-- and fx_rate_date says which one it used.

with payments as (
    select *, (created_at at time zone 'UTC')::date as processor_date
    from {{ ref('silver_payments') }}
),

usd_cad as (
    select rate_date, rate
    from {{ ref('silver_fx_rates') }}
    where base = 'USD' and quote = 'CAD'
),

orders as (
    select order_id, ship_region,
           (created_at at time zone '{{ var("business_tz") }}')::date as order_date
    from {{ ref('silver_orders') }}
    where not is_deleted
)

select
    p.event_id,
    p.event_type,
    p.charge_id,
    p.refund_id,
    p.order_id,
    p.currency,
    case p.event_type when 'capture' then p.amount_minor else -p.amount_minor end as amount_minor,
    p.created_at,
    p.processor_date,
    o.order_date,
    o.ship_region,
    case p.currency when 'cad' then p.processor_date else fx.rate_date end   as fx_rate_date,
    case p.currency when 'cad' then 1.0 else fx.rate end::decimal(12, 6)     as fx_rate,
    round(case p.event_type when 'capture' then 1 else -1 end
          * p.amount_minor / 100.0
          * case p.currency when 'cad' then 1.0 else fx.rate end, 2)::decimal(18, 2)
                                                                             as amount_cad,
    p._batch_seq,
    p._ingested_at
from payments p
asof left join usd_cad fx
    on p.processor_date >= fx.rate_date
left join orders o
    using (order_id)

The processor date is in UTC, because the settlement reports are. The order date is in Halifax time, because that's how the business thinks about its day. A capture at 10 p.m. in Halifax on August 31 belongs to September 1 in UTC. Neither date is wrong, but mixing them up is. Make both explicit, and name them for what they are.

The ASOF JOIN takes the latest exchange rate on or before each event's date. There are no rates on weekends, so a Saturday sale uses Friday's rate, and fx_rate_date records which rate was used.

Two right answers. From that fact table, gold builds two revenue views:

• mart_daily_cash groups by processor date. It answers how much money moved in August, and it's what finance reconciles.
• mart_order_revenue groups by order date. It answers how much did August's orders end up being worth, and it's what product and marketing use.

They disagree, and they should. Here's August through both views as refunds trickle in:

As of	August cash (by processor date)	August orders (by order date)
Sep 8	C$3,195,004.53	C$3,161,389.10
Sep 15	C$3,195,004.53	C$3,143,144.34
Sep 22	C$3,195,004.53	C$3,132,979.77
Sep 30	C$3,195,004.53	C$3,129,185.18
Oct 15	C$3,195,004.53	C$3,126,378.11

A refund on September 20 for an order placed on August 9 is September cash, but it changes the value of August's orders. The bug isn't that these numbers differ. It's a dashboard that shows one while labelling it as the other.

Rebuild only the days that changed. That late refund raises a practical question. How does an incremental model know to go back and change August 9?

-- sugarshack/shack/models/gold/mart_order_revenue.sql
-- Net revenue by the day customers ordered: the view product and marketing use.
--
-- A refund three weeks later changes an old day. So each run finds the order dates that
-- new data touches (a payment event, or a change to the order itself) and rebuilds just
-- those dates: delete them, insert them again. Everything else stays as it was.
--
-- Conversion uses the order-date rate for captures and refunds alike, so a full refund
-- takes an order to exactly zero. (The cash view uses the rate on the day money moved.)
{{ config(
    materialized='incremental',
    incremental_strategy='delete+insert',
    unique_key='order_date',
) }}

{%- set watermark -%}
    (select coalesce(max(_built_through_seq), 0) from {{ this }})
{%- endset %}

with affected as (
    select distinct order_date
    from {{ ref('fct_payments') }}
    where order_date is not null
    {%- if is_incremental() %}
      and _batch_seq > {{ watermark }}
    union
    select distinct (created_at at time zone '{{ var("business_tz") }}')::date
    from {{ ref('silver_orders') }}
    where _batch_seq > {{ watermark }}
    {%- endif %}
),

usd_cad as (
    select rate_date, rate from {{ ref('silver_fx_rates') }} where base = 'USD' and quote = 'CAD'
),

events as (
    select f.order_date, f.ship_region, f.order_id, f.event_type, f.currency, f.amount_minor
    from {{ ref('fct_payments') }} f
    where f.order_date in (select order_date from affected)
),

converted as (
    select e.*,
           round(e.amount_minor / 100.0
                 * case e.currency when 'cad' then 1.0 else fx.rate end, 2) as amount_cad
    from events e
    asof left join usd_cad fx on e.order_date >= fx.rate_date
),

built_through as (
    select greatest(
        (select coalesce(max(_batch_seq), 0) from {{ ref('fct_payments') }}),
        (select coalesce(max(_batch_seq), 0) from {{ ref('silver_orders') }})
    ) as seq
)

select
    order_date,
    ship_region,
    count(distinct order_id) filter (where event_type = 'capture')        as paid_orders,
    coalesce(sum(amount_cad) filter (where event_type = 'capture'), 0)::decimal(18, 2)
                                                                          as gross_cad,
    coalesce(-sum(amount_cad) filter (where event_type = 'refund'), 0)::decimal(18, 2)
                                                                          as refunds_cad,
    sum(amount_cad)::decimal(18, 2)                                       as net_cad,
    (select seq from built_through)                                       as _built_through_seq
from converted
group by all

Each run finds the order dates that new data touches, whether through a payment event or a change to the order itself. It rebuilds only those dates: delete+insert keyed on order_date replaces whole days. The common alternative is a fixed look-back window ("always rebuild the last 7 days"). That's simpler, and it's wrong for any refund older than the window. Tracking the dates that actually changed costs one extra query, and it stays correct however late the data arrives.

To check that incremental logic is right, compare it against a full rebuild. The test suite runs the pipeline daily and then rebuilds everything in one pass. Every gold table matches, row for row.

5. Grade before you barrel

dbt tests ask whether each row is syrup. Grading asks whether the batch is. Gold is built into a schema called gold_build. Four audits read it, and only if none of them blocks does the build get copied to gold, in a single transaction. This is the write-audit-publish pattern.

# sugarshack/grade.py (excerpt)
def audit_reconciliation(con, as_of: datetime) -> Result:
    """Weigh the barrels: gold cash per day and currency against the processor's settlement."""
    rows = con.execute("""
        select s.settlement_date, s.currency,
               s.gross_minor, s.refunds_minor, s.net_minor,
               coalesce(c.captured_minor, 0), coalesce(c.refunded_minor, 0),
               coalesce(c.net_minor, 0)
        from silver.silver_settlements s
        left join gold_build.mart_daily_cash c
               on c.processor_date = s.settlement_date and c.currency = s.currency
        order by 1, 2
    """).fetchall()
    settling, broken = [], []
    for day, cur, sg, sr, sn, gg, gr, gn in rows:
        if (sg, sr, sn) == (gg, gr, gn):
            continue
        closed_at = datetime.combine(day + timedelta(days=1), datetime.min.time(), as_of.tzinfo)
        gap = f"{day} {cur.upper()} net off by {(gn - sn) / 100:+,.2f}"
        (settling if as_of - closed_at < timedelta(hours=SETTLING_HOURS) else broken).append(gap)
    if broken:
        more = " ..." if len(broken) > 3 else ""
        return Result("reconciliation", "BLOCK", "; ".join(broken[:3]) + more)
    if settling:
        return Result("reconciliation", "WARN", "still settling: " + "; ".join(settling[:3]))
    return Result("reconciliation", "PASS", f"{len(rows)} settlement lines match to the cent")


def grade(as_of: datetime) -> tuple[list[Result], dict | None]:
    with duckdb.connect(str(WAREHOUSE)) as con:
        previous = last_published(con)
        since = previous["through_seq"] if previous else 0
        results = [
            audit_quarantine(con, since),
            audit_conservation(con),
            audit_fx(con),
            audit_reconciliation(con, as_of),
        ]
    return results, previous


def publish(as_of: datetime, results: list[Result]) -> None:
    summary = "; ".join(f"{r.name}: {r.status}" for r in results)
    with duckdb.connect(str(WAREHOUSE)) as con:
        through_seq = con.execute("""select greatest(
                (select coalesce(max(_batch_seq), 0) from silver.stg_payment_events),
                (select coalesce(max(_batch_seq), 0) from silver.stg_order_changes))
            """).fetchone()[0]
        con.execute("begin transaction")
        try:
            con.execute("create schema if not exists gold")
            for table in GOLD_TABLES:
                con.execute(f"create or replace table gold.{table} as "
                            f"select * from gold_build.{table}")
            con.execute("""create table if not exists gold._published (
                               as_of timestamptz, through_seq bigint,
                               published_at timestamptz, audits varchar)""")
            con.execute("insert into gold._published values (?, ?, now(), ?)",
                        [as_of, through_seq, summary])
            con.execute("commit")
        except Exception:
            con.execute("rollback")
            raise

The four audits, and why each one exists:

• Quarantine rate. More than 2% of the rows that arrived since the last publish went to quarantine. Like syrup density, this needs limits on both sides. Set it too loose, and a changed API flows straight through. Set it too tight, and the pipeline stops for every bad line. It also needs a minimum count. An earlier version blocked a quiet Saturday because 1 row out of 39 was a truncated line: 2.6% of almost nothing.
• Conservation. Silver and gold must agree on both the number of payment events and their net total. A join that loses rows, or fans them out, is caught here before anyone downstream sees it.
• FX coverage. Every USD event has a rate.
• Reconciliation. Gold's daily cash has to match the processor's settlement report, in cents, for every day and currency. This is weighing the barrels.

Reconciliation is the audit that matters most, and it has one nuance. Some webhooks arrive more than a day late, so a day's cash can be slightly short when its settlement report lands. Gaps on days that closed less than 48 hours ago are warnings (still settling). Older gaps block. That's why the runs above show WARN reconciliation for the previous day, and why those warnings are gone by the next morning.

A blocked publish is a feature. It means the dashboard shows the last graded build, with its as-of time, instead of a number nobody has checked.

6. Weigh the barrels

Back to Tuesday:

$ sugarshack compare --through 2026-09-08T12:00
2026-08 net revenue, as of 2026-09-08 12:00 UTC

  bucket brigade dashboard                           C$2,864,598.75
  re-sent files counted twice       -   16,287.17    C$2,848,311.58
  webhook retries counted twice     -   72,401.70    C$2,775,909.88
  v2 amounts read as null           +  319,482.85    C$3,095,392.73
  refunds never subtracted          -  149,038.45    C$2,946,354.28
  USD added as if it were CAD       +  248,650.25    C$3,195,004.53

  gold.mart_daily_cash C$3,195,004.53 (matches)

  awaiting shipment          all   over 72h
  bucket brigade           1,505         54
  gold                     1,481          4

Every cent of the C$330,405.78 is accounted for, and the gold number matches the processor's settlement reports. The warehouse has four orders to find, not 54.

The test suite checks the claims this post makes, end to end, in about a minute and a half:

• Tapping twice changes nothing, and re-sent files are recognized.
• A batch file a crash left behind is invisible to readers, then removed.
• The API change blocks publishing until the parser is added and bronze is replayed.
• Running daily produces exactly the same gold tables as one full rebuild.
• Every order's state in silver matches its highest-LSN change.
• The comparison ends exactly at gold.

The daily-versus-rebuild test is the one teams skip, and the one I'd keep if I could only have one. Incremental models don't fail loudly. They drift a little each day.

If you'd rather start from the working version than from the listings above, clone sugarshack and follow the README.

Knowing when to boil

The shack costs more than the stove. Bronze keeps everything, so you pay to store data you may never read. You now have a quarantine table, and someone has to own it: a label nobody reads is just a slower way of dropping data. Replays have to be planned once the data is large. And audits that block will sometimes block at 7 a.m. on a day you'd rather they didn't.

In exchange, a changed API becomes a paged engineer instead of a board-deck correction. A late refund changes the right day, and only that day. Every number in gold has been weighed against a scale you didn't build yourself, and you can trace any row back to the file and line it came from.

It's worth being precise about which part does what. The layers take the water out. The grading is what makes the result trustworthy. A lake with bronze, silver and gold but no grading step has the same problem the reserve had: barrels in the right place, sealed, stacked and counted, and nobody weighing them.

That's all a sugar maker's hydrometer is: a cheap, independent check, used on every batch, before the barrel is sealed. Weigh yours on every run.