10 rows a day · 30 days · three tables

Medallion architecture on ten rows a day, in Delta Lake

A pipeline small enough to print every row of, and complete enough to have every problem a real one has: duplicates, late arrivals, unparseable numbers, and a report someone needs to reproduce exactly as it looked last Tuesday.

day 1 · 10 rows landedorders_raw.csv

One file a day, and the report that cannot be reproduced

A supplier drops a CSV into a folder every morning. Ten order lines, always. Somebody wrote a script that reads it, tidies it up, and refreshes a dashboard. The script has one table and it overwrites it.

It works until a Thursday when three things happen at once. Finance asks why Tuesday's revenue figure changed. The supplier resends Monday's file with two rows corrected. And a row arrives with the price written as 19,99 instead of 19.99, which the script silently turns into a null that nobody sees for a fortnight.

None of those are exotic. They are the normal weather of data work, and a single table cannot survive any of them, because it holds one state and no history. What it needs is not more cleaning logic. It needs somewhere to put the raw truth, somewhere to put the cleaned version, and somewhere to put the answer, with each one derivable from the one before.

Figure 1 · what a single table losesone table
Press resend in both modes. In one, Monday's original numbers are gone. In the other they are still on disk and still queryable.

That is the whole of the medallion idea, and the names are only names: bronze for exactly what arrived, silver for the cleaned and conformed version, gold for the shapes the business actually reads. The value is not in the word medallion. It is in the rule that each layer is rebuildable from the one upstream, so a mistake in cleaning costs you a reprocessing job rather than a lost history.

picture it

A kitchen with three surfaces. Deliveries land on the first one in their boxes, untouched. Prep happens on the second: washed, trimmed, weighed, labelled. Plating happens on the third. Nobody plates onto the delivery bench, and nobody throws away the boxes until the meal has gone out.

You can always go back a surface. That is the property being bought, and it is worth more than any individual cleaning rule.

Three tables need a table format that can append without rewriting, update without corrupting, and be read while it is being written. A folder of CSVs cannot do that.

A Delta table is Parquet files plus a log that says which ones count

Open a Delta table on disk and there is nothing mysterious in it. A directory of Parquet files, and a subdirectory called _delta_log containing numbered JSON files. Each JSON file is one version: a list of files added, files removed, and some statistics.

Reading the table means reading the log first, working out which Parquet files are live at the version you asked for, and then reading only those. That indirection is the entire trick, and everything else follows from it. Writers never edit a Parquet file: they write new ones and record the swap in a new log entry. A reader that started before the swap keeps reading the old list and sees a consistent table.

Figure 2 · the log and the filesversion 0
Step through the versions. Files in solid outline are live at the selected version. Faded ones are still on disk and no longer part of the table.

Three consequences worth naming, because they are the reason this piece uses Delta rather than a folder of Parquet.

Appends are cheap and atomic: a new file, one log entry, and either both happened or neither did. Updates and deletes are possible at all, by rewriting the affected files and recording the swap, which plain Parquet has no way to express. And every past version remains addressable, so versionAsOf gives you the table exactly as it was, which is how Tuesday's number gets reproduced on Thursday.

from pyspark.sql import SparkSession from delta import configure_spark_with_delta_pip builder = (SparkSession.builder.appName("medallion") .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")) spark = configure_spark_with_delta_pip(builder).getOrCreate() BRONZE = "/lake/bronze/orders" SILVER = "/lake/silver/orders" GOLD = "/lake/gold/daily_revenue" # the table as it is right now spark.read.format("delta").load(BRONZE).count() # the table as it was three versions ago (spark.read.format("delta") .option("versionAsOf", 3) .load(BRONZE).count()) # or by wall clock (spark.read.format("delta") .option("timestampAsOf", "2024-05-07 09:00:00") .load(BRONZE).count())
picture it

A library where nobody writes in the books. Corrections are printed as new pages, and a single ledger at the front says which pages are current as of each date. Ask for today and you follow today's ledger entry. Ask for last March and you follow March's, and the pages it points to are still on the shelf.

The ledger is the table. The pages are just storage.

With that in place, the first layer is almost embarrassingly simple, and the discipline is in what it refuses to do rather than what it does.

Bronze appends what arrived and changes nothing

The bronze table is the raw feed, landed as it came, with every column read as a string so that nothing can fail to parse and nothing can be silently coerced. The only additions are ingestion metadata: which file this row came from, when it was loaded, and which batch it belonged to.

Ten rows a day, appended. After thirty days the table holds three hundred rows and thirty versions, and every one of those days can be isolated by its batch column without reading the file system.

Figure 3 · bronze, day by dayday 1 · 10 rows
Run a few days, then press re-run. With the idempotency guard on, a repeated file adds nothing. The row count and the version counter tell you which happened.
from pyspark.sql import functions as F from delta.tables import DeltaTable def ingest_bronze(spark, path_csv: str, batch_id: str): # everything as string: bronze must never fail on a bad value raw = (spark.read .option("header", True) .schema("order_id STRING, customer_id STRING, product STRING, " "qty STRING, unit_price STRING, country STRING, event_ts STRING") .csv(path_csv)) stamped = (raw .withColumn("_source_file", F.input_file_name()) .withColumn("_ingested_at", F.current_timestamp()) .withColumn("_batch_id", F.lit(batch_id))) # idempotency: if this batch is already in, do nothing if DeltaTable.isDeltaTable(spark, BRONZE): already = (spark.read.format("delta").load(BRONZE) .where(F.col("_batch_id") == batch_id).limit(1).count()) if already: print(f"batch {batch_id} already ingested, skipping") return 0 (stamped.write.format("delta").mode("append") .partitionBy("_batch_id") .save(BRONZE)) return stamped.count() ingest_bronze(spark, "/landing/orders_2024-05-01.csv", "2024-05-01")

The guard matters more than it looks. Pipelines get re-run: a scheduler retries, an engineer reruns yesterday after a fix, a file gets redelivered. Without a check on idempotency, every retry doubles a day. With one, re-running is free and safe, which is what lets you rerun without a meeting first.

picture it

A goods-in book at a warehouse door. You write down what turned up, the time, and the delivery note number. You do not open the boxes, you do not reject anything, and you do not write down what you think should have arrived.

Bronze is a receipt, not an opinion. If a value is wrong, that is information about the supplier, and destroying it at the door means never being able to prove it.

Which leaves everything wrong with the data still in the table. That is deliberate, and it is the next layer's job.

Silver, first pass: type it, and put the failures somewhere

The first of the four things silver does here is typing: turn the strings into the types the rest of the pipeline needs, and decide what happens to rows that will not convert.

Casting in Spark returns null on failure rather than raising, which is the most dangerous default in the whole stack, because a failed cast and a legitimately empty field look identical downstream. So cast into a new column, compare, and route the mismatches to a quarantine table with the reason attached. Nothing is dropped. Somebody can look at the quarantine on Monday and tell the supplier about the commas.

Figure 4 · casting, and what fails itclean 8 · quarantined 2
Switch to the silent version. The row counts are identical and the revenue total is not, which is the failure mode this rule exists to prevent.
from pyspark.sql import functions as F, types as T def silver_typed(bronze_df): # normalise the obvious formatting damage before casting fixed = (bronze_df .withColumn("unit_price_clean", F.regexp_replace(F.trim("unit_price"), r"[£$,\s]", "")) .withColumn("qty_clean", F.trim("qty"))) typed = (fixed .withColumn("qty_int", F.col("qty_clean").cast(T.IntegerType())) .withColumn("unit_price_dec", F.col("unit_price_clean").cast(T.DecimalType(10, 2))) .withColumn("event_ts_t", F.to_timestamp("event_ts", "yyyy-MM-dd HH:mm:ss"))) # a cast that returned null from a non-null input is a failure, not a blank reason = (F.when(F.col("order_id").isNull(), "missing order_id") .when(F.col("qty_clean").isNotNull() & F.col("qty_int").isNull(), "qty not an integer") .when(F.col("unit_price_clean").isNotNull() & F.col("unit_price_dec").isNull(), "unit_price not a decimal") .when(F.col("event_ts").isNotNull() & F.col("event_ts_t").isNull(), "event_ts unparseable") .when(F.col("qty_int") <= 0, "qty must be positive") .otherwise(F.lit(None))) marked = typed.withColumn("_reject_reason", reason) clean = marked.where(F.col("_reject_reason").isNull()) reject = marked.where(F.col("_reject_reason").isNotNull()) return clean, reject clean, reject = silver_typed(spark.read.format("delta").load(BRONZE) .where(F.col("_batch_id") == batch_id)) (reject.write.format("delta").mode("append").save("/lake/quarantine/orders"))
picture it

A returns desk rather than a bin. The item that does not scan is put on the shelf behind the counter with a note saying why, and somebody deals with the shelf on Friday. Throwing it in the bin makes the shift look tidier and makes the stock count wrong forever.

A row you cannot process is a fact about your data, and it belongs somewhere you can count it.

The rows that survive are now correctly typed and still full of duplicates, because a supplier resending Monday means Monday's orders are in the table twice.

Silver, second pass: one row per order, however many times it arrives

Duplicates arrive for dull reasons. A supplier resends a file, a retry fires twice, an upstream system emits a correction as a fresh event with the same identifier. Bronze holds all of them, correctly. Silver must hold one.

Which one is a decision, not a technicality. Pick a business key, here the order id, and a rule for choosing among the copies: usually the latest by event time, with ingestion time breaking ties. Number the rows within each key and keep the first.

Then write it with MERGE rather than an append, so a key already in silver is updated in place and a new one is inserted. That single operation is what makes the whole layer re-runnable: run the same day twice and the second run changes nothing.

Figure 5 · three copies, one survivororder 1043
Switch the tie-break rule. The surviving row changes, and so does the revenue it contributes, which is why this choice belongs in a written specification rather than in someone's head.
from pyspark.sql import Window def dedupe(df): w = (Window.partitionBy("order_id") .orderBy(F.col("event_ts_t").desc(), F.col("_ingested_at").desc())) return (df.withColumn("_rn", F.row_number().over(w)) .where(F.col("_rn") == 1) .drop("_rn")) def merge_silver(spark, batch_df): if not DeltaTable.isDeltaTable(spark, SILVER): batch_df.write.format("delta").partitionBy("order_date").save(SILVER) return tgt = DeltaTable.forPath(spark, SILVER) (tgt.alias("t") .merge(batch_df.alias("s"), "t.order_id = s.order_id") # only overwrite when the incoming row is genuinely newer .whenMatchedUpdateAll(condition="s.event_ts_t > t.event_ts_t") .whenNotMatchedInsertAll() .execute()) merge_silver(spark, dedupe(clean))

Note the condition on the update. Without it, a replayed older copy would overwrite a newer correction, and replays are exactly the situation this is meant to survive. With it, the merge is not merely idempotent, it is order-independent: any sequence of the same batches produces the same silver table.

picture it

A guest list on the door. The same name turning up three times does not mean three guests. You decide, in advance, that the most recent booking wins, and you write that on the sheet so the person on the door at midnight makes the same call as the one at eight.

Deduplication is only half a mechanism. The other half is a stated rule for which copy is the truth.

One row per order, correctly typed. It is still written in whatever shape the supplier felt like using.

Silver, third pass: make every row mean the same thing

Standardisation is the unglamorous work that decides whether anyone can join your table to anything else. Country written as gb, GB and United Kingdom is three countries to a group-by. Prices in three currencies are not comparable until they are one.

Follow one real row through the whole silver stage, all four operations, with the values at each point.

Figure 6 · one row through silveras it landed
Click any step to jump to it. Every value shown is produced by the same rules the code beside this figure describes.

As it landed in bronze

Every column a string, exactly as the supplier wrote it, including the price with a currency symbol glued to the front and the country in lower case.

Typed, and it survived

Quantity becomes an integer, price a decimal with two places, the timestamp a real timestamp. Had any of those failed, this row would be in quarantine instead and the rest of these steps would never run on it.

Deduplicated against its own key

This order arrived twice. The copy with the later event time wins and the earlier one is dropped, silently but reproducibly.

Standardised into the house style

Country upper-cased and mapped to its two-letter code, product name trimmed and title-cased, currency converted to the reporting currency at the rate for that date.

Enriched with what it did not carry

A derived line total, the order date pulled out of the timestamp for partitioning, and the region joined in from the country dimension. Nothing here came from the supplier.

Merged into silver

One row, one order, fully typed, conformed to the vocabulary the rest of the warehouse uses. This is the table analysts should be allowed to query.

DIM_COUNTRY = spark.createDataFrame( [("GB", "United Kingdom", "EMEA", "GBP"), ("DE", "Germany", "EMEA", "EUR"), ("US", "United States", "AMER", "USD")], "country_code STRING, country_name STRING, region STRING, local_ccy STRING") FX = spark.createDataFrame( [("2024-05-01", "EUR", 0.855), ("2024-05-01", "USD", 0.794), ("2024-05-01", "GBP", 1.000)], "rate_date STRING, ccy STRING, to_gbp DOUBLE") def standardise_and_enrich(df): std = (df # 1. standardise: one spelling per concept .withColumn("country_code", F.upper(F.trim(F.col("country"))).substr(1, 2)) .withColumn("product", F.initcap(F.trim(F.col("product")))) .withColumn("order_date", F.to_date("event_ts_t")) # 2. derive: columns nobody sent us .withColumn("line_total_local", (F.col("qty_int") * F.col("unit_price_dec")).cast("decimal(12,2)"))) # 3. enrich: conform to the shared dimensions enriched = (std .join(F.broadcast(DIM_COUNTRY), "country_code", "left") .join(F.broadcast(FX), (F.col("local_ccy") == F.col("ccy")) & (F.col("order_date").cast("string") == F.col("rate_date")), "left") .withColumn("line_total_gbp", (F.col("line_total_local") * F.coalesce("to_gbp", F.lit(1.0))) .cast("decimal(12,2)")) .withColumn("_silver_at", F.current_timestamp())) return enriched.select( "order_id", "customer_id", "product", "qty_int", "unit_price_dec", "country_code", "region", "order_date", "event_ts_t", "line_total_local", "local_ccy", "line_total_gbp", "_batch_id", "_silver_at")
picture it

Three teams filing expenses, one in pounds, one in euros, one writing GB and one writing UK. Nobody is wrong. The total is meaningless until someone decides on a single vocabulary and converts everything into it.

Standardisation is the decision, written down as code, about what the warehouse's words mean. It is where most of the arguments in a data team actually live.

One thing has been quietly assumed through all of that: the country dimension and the customer record are fixed. Customers move.

Silver, fourth pass: keep the history of things that change

A customer's country changes in June. What was the revenue for EMEA in May?

If the customer dimension holds only the current country, May's answer changes retroactively every time somebody moves, and the same query run twice gives two answers. Overwriting the old value is called a type 1 dimension and it is the right choice for correcting typos and the wrong one for real changes.

A type 2 dimension instead closes the old row and opens a new one, each stamped with the window during which it was true. The customer now has two rows, one flagged current, and any fact can be joined to whichever version was live when it happened.

Figure 7 · the customer who movedtype 2
Move the customer, then ask for May. Under type 1 the May figure moves with them. Under type 2 it does not, which is the entire point.
DIM = "/lake/silver/dim_customer" def scd2_customer(spark, incoming, as_of: str): # incoming: customer_id, country_code, segment (today's snapshot) if not DeltaTable.isDeltaTable(spark, DIM): (incoming .withColumn("valid_from", F.lit(as_of).cast("date")) .withColumn("valid_to", F.lit(None).cast("date")) .withColumn("is_current", F.lit(True)) .write.format("delta").save(DIM)) return dim = DeltaTable.forPath(spark, DIM) current = (spark.read.format("delta").load(DIM) .where("is_current").alias("c")) # rows whose tracked attributes actually changed changed = (incoming.alias("n").join(current, "customer_id") .where("n.country_code <> c.country_code OR n.segment <> c.segment") .select("n.*")) # step one: close the old version (dim.alias("t").merge(changed.alias("s"), "t.customer_id = s.customer_id") .whenMatchedUpdate( condition="t.is_current = true", set={"is_current": F.lit(False), "valid_to": F.lit(as_of).cast("date")}) .execute()) # step two: open the new one, plus any customer we have never seen brand_new = incoming.join(current, "customer_id", "left_anti") (changed.unionByName(brand_new) .withColumn("valid_from", F.lit(as_of).cast("date")) .withColumn("valid_to", F.lit(None).cast("date")) .withColumn("is_current", F.lit(True)) .write.format("delta").mode("append").save(DIM)) # joining a fact to the version that was live when it happened fact_with_dim = (silver.alias("f").join( spark.read.format("delta").load(DIM).alias("d"), (F.col("f.customer_id") == F.col("d.customer_id")) & (F.col("f.order_date") >= F.col("d.valid_from")) & (F.col("d.valid_to").isNull() | (F.col("f.order_date") < F.col("d.valid_to"))), "left"))
picture it

An address book where you cross out the old address and write the new one, against one where you add a new card and date the old one. Only the second can answer where you posted the parcel in March.

Type 2 is not more thorough bookkeeping, it is a different question being answerable. It also costs a join with two inequality conditions on every fact query, which is why nobody does it for every column.

Four kinds of processing, one table, and still nothing a business user would recognise. Silver is correct, not useful.

Gold is the shape a question is asked in

Silver has one row per order. Nobody asks a question with one row per order. They ask for revenue by day and region, orders per customer per month, the top ten products this week. Each of those is a gold table with its own grain, built by aggregating silver.

The temptation with three hundred rows is to rebuild the whole thing every night, and for three hundred rows that is correct: it costs nothing and it cannot drift. The reason to learn the incremental version is that at three hundred million rows a full rebuild stops finishing before the morning, and the incremental logic is easier to write on day one than to retrofit.

Figure 8 · silver rolled up into goldfull rebuild
Advance a few days in both modes. The gold table is identical either way; the rows read to produce it are not, and that difference is the whole argument.
def build_gold_full(spark): silver = spark.read.format("delta").load(SILVER) gold = (silver.groupBy("order_date", "region") .agg(F.countDistinct("order_id").alias("orders"), F.sum("line_total_gbp").cast("decimal(14,2)").alias("revenue_gbp"), F.countDistinct("customer_id").alias("customers"), F.max("_silver_at").alias("_built_from"))) (gold.write.format("delta").mode("overwrite") .option("overwriteSchema", "true").save(GOLD)) def build_gold_incremental(spark, affected_dates: list): # only the days this run actually touched, including late arrivals silver = (spark.read.format("delta").load(SILVER) .where(F.col("order_date").isin(affected_dates))) slice_ = (silver.groupBy("order_date", "region") .agg(F.countDistinct("order_id").alias("orders"), F.sum("line_total_gbp").cast("decimal(14,2)").alias("revenue_gbp"), F.countDistinct("customer_id").alias("customers"), F.max("_silver_at").alias("_built_from"))) tgt = DeltaTable.forPath(spark, GOLD) (tgt.alias("t").merge(slice_.alias("s"), "t.order_date = s.order_date AND t.region = s.region") .whenMatchedUpdateAll() .whenNotMatchedInsertAll() .execute())

The subtlety is the list of affected dates. It is not today. A late-arriving row with an event time from three days ago changes a gold row for three days ago, and an incremental job that only ever rebuilds today will leave that number wrong forever while reporting success every night.

picture it

A scoreboard rather than a match report. It answers one question instantly and it cannot answer anything else, so you keep the match report and rebuild the scoreboard from it whenever the report changes.

Gold is disposable by design. Anything in gold that cannot be rebuilt from silver is a bug.

Which makes the size of that rebuild window the most consequential number in the pipeline, and it is a number somebody picks.

The restatement window decides which late rows are ever counted

One dial: how many days back each nightly run reprocesses. At zero, only today's date is rebuilt. At seven, the last week is rebuilt every night, so anything that turned up late for any of those days is picked up.

Below is the thirty day feed with its real distribution of late arrivals. Set the window, run the month, and read the two numbers that matter: how much revenue ends up correctly counted, and how many gold rows had to be rewritten to get it.

Figure 9 · thirty nights, one windowwindow 0 days
0 days35710 days
Run the month at 0, then at 3, then at 7. The missed rows and the rewrite cost are both counted from the same simulated feed.

At a window of 0, 64 of the month's rows never reach gold, worth £5,229 of £19,240. They sit correctly in silver, where anybody querying silver would find them, and the dashboard built on gold is quietly short by 27 percent. Nothing errors, and that is the failure mode worth fearing, because it is invisible from the inside.

At 3 the miss falls to 41 rows and £3,028. At 7 it falls to 18 rows and £1,249, and the cost is that every night rewrites eight days of gold partitions instead of one.

Past 7 the numbers stop moving, and it is worth understanding why rather than turning the dial further. Those last 18 rows are not slightly late, they are resends of orders more than ten days old, and no window anyone would run nightly will ever catch them. They need a deliberate backfill, triggered by someone noticing, which is an argument for monitoring the age of incoming events rather than for a bigger window.

The right answer is not a number, it is a measurement: look at your own distribution of event time against ingestion time, pick the window that covers the percentile you can defend, and then say out loud that anything beyond it needs a manual backfill.

picture it

A monthly ledger you close on the third of the following month. Invoices arriving on the second are counted. One arriving on the tenth needs a decision by a person, and the reason you close at all is that leaving every month open forever means never being able to sign anything off.

The window is not a technical setting, it is a promise about how correct the numbers are and when.

from datetime import date, timedelta RESTATEMENT_DAYS = 3 def affected_dates(spark, batch_id: str, window_days: int = RESTATEMENT_DAYS): # dates actually touched by this batch, however old their events are touched = (spark.read.format("delta").load(SILVER) .where(F.col("_batch_id") == batch_id) .select("order_date").distinct() .rdd.flatMap(lambda r: [r[0]]).collect()) run_day = date.fromisoformat(batch_id) window = {run_day - timedelta(days=i) for i in range(window_days + 1)} return sorted(window.union(set(touched))) # the nightly job, end to end def run_day(spark, batch_id: str): ingest_bronze(spark, f"/landing/orders_{batch_id}.csv", batch_id) bronze_slice = (spark.read.format("delta").load(BRONZE) .where(F.col("_batch_id") == batch_id)) clean, reject = silver_typed(bronze_slice) reject.write.format("delta").mode("append").save("/lake/quarantine/orders") merge_silver(spark, standardise_and_enrich(dedupe(clean))) build_gold_incremental(spark, affected_dates(spark, batch_id))

That is the pipeline, complete. What remains is the housekeeping that decides whether it still works in a year.

Ten rows a night is a file problem, not a data problem

Ten rows is a Parquet file of a few kilobytes. Three hundred and sixty five nights is three hundred and sixty five of them, plus a log entry each, plus whatever the merges rewrote. The data is trivial and the metadata is not: every query now opens hundreds of files to read a few hundred kilobytes, and each open costs more than the read.

Two maintenance commands handle it. OPTIMIZE rewrites many small files into few large ones, changing nothing about the contents. VACUUM deletes files that no live version needs any more, subject to a retention period, and it is the operation that ends time travel beyond that horizon.

Figure 10 · a year of small filesday 30 · 30 files
Run a few months, then compact. Watch what OPTIMIZE does to the file count, and what VACUUM does to how far back you can still travel.
from delta.tables import DeltaTable # weekly: many tiny files become few large ones spark.sql(f"OPTIMIZE delta.`{BRONZE}`") # with ordering, if queries filter on a column spark.sql(f"OPTIMIZE delta.`{SILVER}` ZORDER BY (order_date, customer_id)") # monthly: drop files no version within retention still needs spark.sql(f"VACUUM delta.`{BRONZE}` RETAIN 720 HOURS") # 30 days # what the log currently holds spark.sql(f"DESCRIBE HISTORY delta.`{SILVER}`").show(truncate=False)

VACUUM is the one to be careful with. Its retention period is the real limit on time travel: once those files are gone, versionAsOf for anything older fails, and the promise made in the first chapter about reproducing Tuesday's number quietly expires. Set it against how far back anyone might actually need to reproduce a figure, not against how much storage costs.

picture it

A filing room where every day's paperwork goes into its own envelope. After a year the room is envelopes. Consolidating twelve months into twelve folders changes nothing about what is written, and it is the difference between finding something in a minute and in an afternoon.

Compaction is not an optimisation you do when you have time. On a small daily feed it is the difference between a table that stays usable and one that degrades on a schedule.

Three honest limits before you take this anywhere real.

This shape is overkill for genuinely small and simple problems. Ten rows a day that arrive clean, are never resent, and feed one chart do not need three layers; they need a table and a backup. The architecture earns its cost when there are multiple sources, multiple consumers, or a legal need to explain a number from six months ago.

Deletion for privacy requests is real work rather than a delete statement. A DELETE removes a row from the current version, but the old files still hold it until VACUUM passes the retention horizon, and it is still sitting in bronze. Any credible right-to-erasure process has to reach every layer and account for retention.

And nothing here validates meaning. Every rule in this piece checks form: types, uniqueness, spelling, arithmetic. A supplier who sends ten perfectly formed rows with the wrong prices will pass every check in the pipeline, and the only defences are expectations about volume and distribution, checked and alerted on, which is a separate discipline that starts where this one stops.

Land what arrived and never touch it. Make one row mean one thing. Aggregate into the shape of the question. Keep every layer rebuildable from the one above it, and the pipeline becomes something you can fix rather than something you must not disturb.

Now break it

The full thirty day feed with every dial exposed: how dirty the source is, how often it resends, how late events run, and how far back each night reprocesses. Run the month and read what reaches gold.

the whole pipelineday 0 of 30
bad rows resends late window
Push bad rows to 40 percent and run. Bronze never shrinks, silver does, and the gap between them is the quarantine you would be explaining to the supplier.