nordisk/knut

Getting rows out of the lake and into the engine — for idiots

Class: idiot · Function axis: f0 (zero → rows arriving in Doris). No prior knowledge assumed. The fuller version is in doris-sink-guide.md; if you have never met bifröst at all, read cdc-for-idiots.md first.

The problem, in one picture

The same two billion people are stored twice.

• In a lake. Parquet files sitting in object storage (MinIO). Cheap, open, anything can read it.
• In an engine. Apache Doris, on its own local disks. Expensive, closed, extremely fast.

Ask "who has personnummer X?" and you get:

That is not a small difference. So the question is: how do you keep the engine's copy up to date without redoing the whole thing every night?

Why not just ask Doris to copy it

Doris can keep a copy of a lake table, called a materialized view. The problem used to be that refreshing it over an Iceberg lake table meant rebuilding the whole copy. If a thousand new people arrive today, Doris would not add a thousand rows — it would throw the copy away and rewrite all two billion, about 81 GB, to pick up those thousand.

This turned out to be out of date, and it matters. Doris learned to refresh such a copy partition-by-partition in version 3.1, and the cluster we measured runs 4.0.8. So Doris can do the incremental thing now — with a condition: the lake table has to be partitioned on exactly one column, and ours is not.

So there are two honest answers, and the cheap one should be tried first:

1. Partition the lake table. Then Doris refreshes only the partitions that changed, and you may not need anything else at all. (The lake table wants partitioning anyway, for unrelated reasons.)
2. Ship the changes yourself. Work out what changed and push exactly those rows into Doris. That is what this document describes.

Option 2 is more general — it works on unpartitioned tables, it handles deletions, and it is not tied to Doris. But it is a moving part you have to run, and moving parts break at 3 a.m. Do not build it because it is more interesting.

What the lake gives us for free

Iceberg — the format the lake table uses — never overwrites files. Every write makes a new snapshot, like a save point, and records which files that snapshot added.

So you can say "compare snapshot 1042 with snapshot 1043" and get back only the files that appeared in between. No guessing, no timestamps, no scanning two billion rows to find the thousand new ones.

knut already knows how to do this. The library called skade computes the difference, and bifröst already polls for it, reads it, and pushes it into a graph database. All of that exists and runs.

What was actually missing

Bifröst could push changes into a graph (nodes, edges, relationships). It could not push changes into a table (rows and columns). Those are different enough shapes that pretending they are the same makes both worse.

So bifröst gains a second kind of destination. The graph one is called GraphSink; the new one is called TableSink, and the first thing that implements it talks to Doris.

That is the entire change. Not a new program, not a new library — one more kind of destination in a program that already does the hard parts.

The part people get wrong

Two steps happen every round:

1. Send the changed rows to Doris.
2. Write down "I have now delivered everything up to snapshot 1043."

Step 2 is called the cursor, and it is how the program knows where to pick up after a restart.

The order matters and it is not obvious which way. If you write down the cursor first and the machine dies before the rows are sent, the next run starts at 1044 and those rows are gone. Forever. No error, no warning — they are simply not there, and nobody finds out until someone asks a question and gets the wrong answer.

So: send first, write the cursor second. Always.

Now the other worry: if you send the rows and then die before writing the cursor, the next run sends the same rows again. Do you get duplicates?

No, and the reason is neat.

Why sending it twice is harmless

Every upload to Doris carries a name you choose. Doris refuses an upload whose name it has already seen.

The name is built from the snapshot numbers:

skade-fake:personer2g-1042-1043

Same snapshots in, same name out. So the second attempt is rejected on purpose, and nothing is duplicated. That is the whole trick, and it is why nobody needs to build a deduplication layer on top.

This was tested against the real cluster rather than taken on faith: the same three rows sent twice under one name landed once; sent again under a different name they landed twice. The protection is entirely in the name being worked out from the snapshot numbers, never invented.

Two honest footnotes, because the earlier version of this document skipped them:

• Doris only remembers those names for a while — about 12 hours by default. If the program has been switched off for longer than that and resumes in the middle of a window, the protection has expired. In practice a crash restarts in seconds, so this is a corner, not a hole. But it is a real corner.
• If someone cancels an upload inside Doris, its name becomes free again.

Does bifröst write to Doris's disk?

No, and it must never be allowed to. It sends rows over the network and Doris decides what to do with them on its own storage.

This sounds like a detail. It is not. The moment one program writes into another program's storage directory, you own both of them forever — every upgrade, every format change, every bug in either one becomes a joint problem. A network protocol between them keeps them separable.

The trap that wastes an afternoon

Doris answers this kind of upload on one machine (the "frontend") and then immediately says "actually, send it to that other machine over there" (a "backend"). Both have to be reachable.

If your loader can reach the frontend but not the backend, you get no error at all. The upload just sits there. Meanwhile the frontend keeps cheerfully reporting that it is healthy, because it is — it is the other machine that is unreachable.

That is worse than an error, because a job that is stuck looks a lot like a job that has nothing to do. So the sink puts a stopwatch on every request and turns the silence into a message that names the machine it could not reach.

When it gives up

Sometimes the lake does something the "what changed?" question cannot answer — somebody rewrites rows in place instead of appending new ones. Then the only correct answer is "reload everything".

On a two-billion-row table, doing that quietly is how a helpful background job becomes an outage.

So it does not do it quietly. It stops, it reports a red status, and it waits for a human to say "yes, reload it". Silent full reloads are how a streamer becomes an incident.

There is a subtlety here that the first draft of this design got wrong, and it is worth knowing about: the "give up" decision has to be made before reading the data, not after. Asking for the changes already triggers the enormous read. So the check happens against the lake's bookkeeping first — which costs almost nothing — and only then does anything get read.

Should we build it?

The seam — a table-shaped destination for bifröst — is worth having regardless. It is small, it has no cost when switched off, and without it bifröst can only ever talk to a graph.

Whether to run it in production for this particular table is a different question, and it is genuinely open. Try partitioning the lake table and letting Doris refresh itself first. If that works, this stays a capability rather than a deployment, and that is a fine outcome.

Measure first.