luoyuxia commented on code in PR #10: URL: https://github.com/apache/fluss-blog/pull/10#discussion_r3346756771
########## blog/2026-06-10-tiering-service-part2.md: ########## @@ -0,0 +1,319 @@ +--- +slug: fluss-tiering-service-deep-dive-part2 +title: "Tiering Service Deep Dive Part 2: Tuning" +date: 2026-06-10 +authors: [giannis] +image: ./assets/tiering_service_dd_part2/banner.png +--- + + + +[Part 1]() built the mental model. What tiering is, who does what, how the round runs end-to-end. + +This part adds the dials. Buckets and splits determine how a round parallelizes. +Log and PK tables behave so differently on round one that the difference deserves its own treatment. +The freshness setting, the one knob most users actually touch, does two different jobs that share the same value. +Once a single job is handling many tables, queue position starts to dominate effective freshness more than any per-table setting. +And once that happens, you have a deployment-shape decision: stay with one job, or scale out. By the end, you'll know which levers matter most and how to use them. + +<!-- truncate --> + +## Buckets, Splits, And How The Work Gets Divided +In Part 1, we had a table with 4 buckets that ended up with 4 splits. +That's not accidental: the rule is one split per bucket. +Let's slow down and look at the parallelism story end-to-end, because this is where bucket count actually starts to matter. + +The mental model is straightforward: **buckets** are how the data is physically partitioned, **splits** are the chunks of work the tiering job hands out, and **readers** are the workers that consume splits. +The number of buckets is fixed at table creation time. The number of readers is your Flink job parallelism. The number of splits per round equals the number of buckets. + + + +What happens when the number of readers doesn't match the number of buckets? Two cases: + +**More readers than buckets**. A table with 4 buckets and 16 readers means only 4 readers can work on *that* table. Bucket count caps how many readers a single table's round can use. The other 12 don't necessarily sit idle, though: if other tables are waiting in the queue, the enumerator hands those spare readers the next table's splits, and they start reading ahead (we'll see exactly how in the multi-table section). They only truly idle when there's no other work queued. + +**Fewer readers than buckets**. A table with 16 buckets and 4 readers means each reader will process 4 splits sequentially. Reader 0 takes one split, finishes it, asks for the next, takes another, and so on. The whole round takes roughly 4× longer than it would with 16 readers. This is the common production case, and it's fine, just slower. + +One more thing: the splits are shuffled (randomized) before being assigned. This is a small but important detail. Without it, reader 0 would always get bucket 0, reader 1 would always get bucket 1, and so on. If bucket 0 is consistently the heaviest (because of a bad key distribution), reader 0 would consistently be the bottleneck. Shuffling spreads the bad luck around: over many rounds, every reader gets stuck with the heavy bucket about equally often. + +## The Slow-Bucket Effect +In the normal commit path, the commit operator waits for every split of a table before committing. +If 3 readers finish quickly and the 4th is still processing a much larger bucket, the table's commit can't happen until that last bucket lands. The 3 fast readers move on to other queued tables (or idle, if there's nothing else queued), but this table's round isn't done until its slowest bucket is. +The lake commit itself is always atomic. Readers can't see a half-written snapshot, and most of the time every bucket of a round lands in the same commit. + +**The end result:** the slowest bucket determines how long a round takes. Skewed bucket keys can hurt you here. (There's one exception: when freshness fires and the round force-finishes, whatever each bucket has tiered so far gets committed, and the unread remainder rolls to the next round. We'll cover this in the freshness section.) + + +## Two Kinds of Tables: Log vs. Primary-Key Tables + +### Log Tables: The Simple Case +A log table is exactly what it sounds like: an append-only stream. Records arrive, they get assigned an offset, and they sit there. You can't update or delete a record; you can only append. Examples: a clickstream, an event bus, an audit log. + +Tiering a log table is straightforward. Every round, the Flink job reads "everything since the last commit" (just the new offsets) and writes those records as Parquet files in the lake. +The lake table grows monotonically. The work per round is proportional to how much you wrote since the last round. + +### Primary-Key Tables: The Trickier Case +A primary-key (PK) table is like a row-keyed materialized view. +Each record has a primary key and the latest value for that key replaces any previous value. +Internally, Fluss stores PK tables in two parts: **a changelog** (a log of all the upserts and deletes that have happened) and a **KV state** (the current value for each key, derived from the changelog). It's similar to how a database has a write-ahead log plus the actual table. + +The changelog records every event using four operation codes you'll see throughout the Flink and Fluss world: `+I` for an insert, `-U` and `+U` together for an update (the old row going out, the new row coming in), and `-D` for a delete. +Apply all of those to an empty state in order and you get the current row for every key; that's the KV state. Here's a small example to make this concrete: + + + + +Now think about what "tier this to the lake" means for a PK table. +The lake needs the current state of every row, not just the log of what changed. +So the very first round of a PK table has to copy the entire KV state to the lake. +That can be huge: think 200 GB if you have a big customer-profile table. +After that first round, subsequent rounds only need the changelog records since the last snapshot offset, because the lake's merge engine knows how to apply upserts and deletes to existing rows. + + + +The numbers here are illustrative. A 200 GB PK table might take 10+ minutes to read on its first round (because all 200 GB must be copied). +The second round, where only the upserts and deletes from the last few minutes need to be applied, might take 30 seconds. +This is an important difference, and the single most important thing to understand about PK tiering: the first round is in a different category from every round that follows. + +### The Summary Table + +| Aspect | Log Table | Primary-Key Table | +|---|---|---| +| **Round 1 reads** | Whatever offsets exist so far | Full KV state, up to 100s of GB | +| **Round 2+ reads** | New offsets since round 1 | Changelog since round 1's snapshot offset | +| **What the lake stores** | Append-only Parquet | Snapshot rows + applied upserts/deletes | +| **Cost stability** | Roughly constant per round | Round 1 huge, rest tiny | +| **Partial progress useful?** | Yes: committed buckets advance the lake; the rest roll to the next round | Round 1 snapshot: no (force-finish is ignored for snapshot splits, so they run to completion rather than committing partially). Round 2+: yes | + +## The Freshness Knob +Every table that's enabled for tiering has a setting called `table.datalake.freshness`. +It's a duration, something like `5min` or `30s`, and the default is 3 minutes. +This single number is the most common source of confusion in the tiering service, because it does two different jobs that share the same value. + +### Job 1: How Often Rounds Start +The obvious meaning. +After a round completes, the coordinator schedules the next round one full freshness interval later, measured from the moment that round *finished*. So with 5-minute freshness, the next round becomes eligible 5 minutes after the previous round committed. The round's own duration is *not* subtracted. + +Under the hood the scheduler computes `freshness − (now − last_completion_time)`, which looks like it deducts elapsed time. But `last_completion_time` is reset to the instant the round just finished, so the subtracted term is ~0 for back-to-back rounds. That subtraction only bites after a coordinator restart, where the last completion can be well in the past and the next round may fire immediately. + +The practical consequence: the effective start-to-start cadence is `round_duration + freshness`, not `freshness` on its own. A table with 5-minute freshness and 90-second rounds runs a round roughly every 6.5 minutes, not every 5. + +#### If You're Coming From a Flink Streaming Background +Tiering cadence is driven by freshness; it is *not* tied to Flink checkpoints. +A common mental model from Flink-CDC-style pipelines is "data lands in the sink when a checkpoint completes". +That's not quite how the tiering service works. A tiering round commits to the lake when all the round's bucket results have arrived at the commit operator, which is driven by the round's own progress rather than by external checkpoint cadence. +Whatever checkpoint interval the Flink tiering job has configured doesn't bound when the lake sees new data. +The round itself does. + +The correctness story has two layers worth separating. +Atomicity comes from the lake's own commit primitive (Paimon's snapshot commit, Iceberg's metadata swap). Consistency across attempts comes from the epoch fencing mechanism the coordinator uses, introduced in **Part 1**, in the heartbeat section, which rejects any commit from a stale attempt (an epoch-fencing error). +So you don't need to tune Flink checkpoint settings to get correct tiering; the defaults are fine. + +What you do still need is a healthy checkpoint cycle. +The Flink tiering job is configured with a full-restart strategy because the commit operator is stateless: it holds the per-table bucket results it has collected so far only in memory (nothing is checkpointed), so if it fails those in-flight committables are lost and the whole job restarts to collect them fresh. + +That's deliberate. It keeps the all-or-nothing commit semantics simple. So checkpoints aren't your tuning knob, but they do need to complete; treat the tiering job's checkpoint cycle as load-bearing infrastructure rather than something you can ignore. + +### Job 2: The Ceiling On A Round's Wall-clock Duration +The less-obvious meaning. The same value also caps how much wall-clock time a single round of this table is allowed to consume before something gives. When freshness elapses mid-round, the Flink enumerator fires a force-finish: every log split that has started commits whatever it has read so far (advancing that bucket's offset), and buckets that hadn't started yet are skipped. The skipped buckets, and the remaining tail of any partially-read bucket, get re-read in the next round (which starts immediately, since the table is re-queued straight away). + +This is an important nuance and worth slowing down on: **force-finish is not a failure**. It's a partial commit. On the coordinator side, the table transitions `Tiering` → `Tiered` (with a `force_finished` flag) → `Pending`. It is not marked `Failed`. It does take a fresh tiering epoch on the way back into the queue: the `Tiered` → `Pending` transition bumps the epoch exactly as any normal re-enqueue does, but that's ordinary re-attempt bookkeeping, not a failure recorded against the table. +The lake snapshot that lands contains every bucket that tiered any data this round, each up to wherever it reached, and that snapshot is still atomic. Readers never see a partial commit. Buckets that never started just roll forward to the next round. + + +**One exception to highlight:** for the first round of a PK table, assuming a KV snapshot already exists (the normal case), that round reads the snapshot, and force-finish is suppressed for snapshot splits. (If no KV snapshot has been taken yet, the first round instead reads the changelog from the earliest offset as ordinary log splits, and force-finish applies as usual.) +The reason force-finish is suppressed is that a snapshot split has to be read in full to yield the log offset that the snapshot ends at, and that offset is what anchors the next incremental round. +Truncate the snapshot read part-way and you lose that offset. There's nothing for the following round to start from. +So when freshness fires during a PK snapshot round, the force-finish signal is sent but the snapshot reader simply ignores it; the splits keep running until the snapshot finishes naturally. +The lake commit lands whenever that happens, no matter how much longer it takes. Force-finish only meaningfully truncates work on log splits, not snapshot splits. + +### What The Freshness Gives You And What It Doesn't +Freshness is a target scheduling cadence, not a guarantee of how stale the lake is. +The actual lake lag is the time between when a record is written to Fluss and when it appears in the lake, and it oscillates in a sawtooth. The floor is one `round_duration`: a record written just before a round's cutoff lands in the lake about one round-length later. The peak is roughly `freshness + 2 × round_duration`: a record written just *after* a round's cutoff has to wait a full freshness interval for the next round to start, then that round's duration to commit. So a table with 5-minute freshness and 90-second tiering rounds gives you a lake that's between ~90 seconds and ~8 minutes behind. + + + +So the rule of thumb to take away is: floor lag ≈ `round_duration`, and peak lag ≈ `freshness + ~2 × round_duration`, meaningfully more than the configured freshness once rounds get long. +If your stakeholders say "the lake must never be more than 10 minutes behind", work backwards from the peak: with 90-second rounds, a freshness around 7 minutes keeps peak lag near 10 minutes (7 min + ~3 min). Leave yourself margin. + +### The Other Timeout You Should Know About +There's a second timeout, set on the coordinator side and applied to every table regardless of its freshness. It's a 2-minute window, but the thing it measures is easy to misread. It's *not* a ceiling on how long a round can run; rather, it's a ceiling on how long the coordinator will wait between heartbeats from the Flink job about a given in-flight table. + +The coordinator runs a background sweep every 15 seconds. +For every table currently in `Tiering`, it checks `currentTime - lastHeartbeat`. +If that delta exceeds 2 minutes, the table is fenced and transitioned through `Failed` back to `Pending`, epoch bumped, any uncommitted lake files orphaned. +The `lastHeartbeat` timestamp gets refreshed on every heartbeat from the Flink job that mentions the table, which by default is every 30 seconds (controlled by `tiering.poll.table.interval`). + +So in a healthy Flink job, the 2-minute timer never fires for round duration. +The job heartbeats four times in every 2-minute window, each heartbeat lists the in-flight tables, the coordinator's `lastHeartbeat` stays fresh, and a single round can keep running for an hour without the coordinator ever raising an eyebrow. +The 2-minute window only catches Flink jobs that have genuinely stopped heartbeating; GC pause long enough to look like unavailable, network partition, JobManager crash, that kind of thing. + +### The Two Ceilings, Side By Side +**Freshness** is set per table, fully tunable, and triggers a force-finish on log splits when it fires mid-round. +For PK snapshot splits, force-finish is ignored and those splits run to completion regardless, because the snapshot read has to finish to produce the log offset that anchors the next incremental round. +**The 2-minute coordinator window** is hardcoded and applies to every table, but it measures heartbeat liveness, not round duration. +A healthy Flink job that heartbeats every 30 seconds can spend hours on a single round without ever triggering it. +The 2-minute window is the coordinator's safety net for a job that has actually stopped responding, which is the only failure mode that needs it. + +## Multiple Tables: How The Queue Actually Works +Up until now we've talked about one table. Real deployments have many. So what happens when you have, say, three tables (two log tables and one PK table) all sharing a single tiering Flink job? + + +The coordinator keeps one FIFO queue of pending tables. When a Flink job asks for work via heartbeat, the coordinator pops the table at the front of the queue and hands it over. +The Flink job works that table's buckets in parallel and asks for the next table once its splits are all handed out. The coordinator gives out work one table per request. As we'll see, a job's spare readers can start a second table's reads before the first table's commit lands; what stays strictly one-at-a-time is the commit. + +This means the effective freshness for any single table is not just its own configured freshness. +It's also a function of where it sits in the queue and how long the tables ahead of it take. + + +## A Concrete Walkthrough: Three Tables, One Job +Here's the setup. Three tables sharing one Flink tiering job that's running with parallelism 16 (so up to 16 readers can run in parallel at any moment). +Each table has its own bucket count, freshness target, and expected per-round duration: + +| Table | Type | Buckets | Freshness | Round duration | +|---|---|---|---|---| +| clicks | Log | 16 | 1 min | ~30 s | +| orders | Log | 8 | 2 min | ~90 s | +| customers | PK | 12 | 5 min | ~100 s first round, ~20 s after | + +Each table generates one split per bucket per round, so a clicks round produces 16 splits, an orders round produces 8 splits, and a customers round produces 12 splits. +When the Flink job processes clicks, all 16 of its readers are busy on it at once. +orders has only 8 buckets, so at most 8 readers can work on orders itself, but the other 8 don't idle: once orders' splits are all assigned and none are left pending, the enumerator pulls the next queued table and hands its splits to those readers, which start reading ahead. Same with customers (12 buckets): the 4 spare readers pick up whatever is next in the queue. +The bucket count is a per-table cap on *intra-table* parallelism (how many readers one table's round can use), not a cap on how busy the job is. The spare readers pipeline onto the next table (detailed in *Reads Pipeline: Commits Serialize*, below). What stays strictly one-table-at-a-time is the *commit*, not the reads. + + +### What Determines Round Duration? +Before we walk through the example, let's pin down what "30 seconds" or "100 seconds" actually means, because those numbers aren't magic. +They emerge from three concrete inputs that every round has to negotiate with: + +1. **How much data the round has to read**. For a log table, this is the volume of writes since the last commit, roughly `write_rate × time_since_last_commit`. The longer the gap between rounds, the more data each round has to move. For a PK table's first round, the data is the entire KV state; every key in the table, regardless of how long it's been since the last anything. For a PK table's incremental rounds (round 2 and beyond), it's just the changelog records since the last snapshot offset, which is typically much smaller than the full state. +2. **How parallel the work is**. Bucket count bounds the parallelism within a single round. With 16 buckets, the work is split 16 ways. With 4 buckets, it's split 4 ways, no matter how many Flink readers you have. The Flink job's parallelism setting only matters up to the bucket count of the table currently being tiered for that table's own splits; any extra readers don't wait around: they pick up the next queued table's splits (more on that in a moment). +3. **How fast each reader can move data**. Bounded by network throughput from the Fluss tablet servers, deserialization speed, and write bandwidth to the lake (S3, GCS, etc.). Real-world numbers are usually a few tens of MB/s per reader. + +**Putting it together:** `round_duration ≈ data_to_read / (bucket_count × per_reader_throughput)`. +For our clicks table, which has been collecting writes for ~1 minute, has 16 buckets, and each reader pulls maybe 30 MB/s, a round of a few hundred MB of new clicks events takes roughly 30 seconds. +For customers' first round, the full KV state, let's say a few GB, 12 readers at 30 MB/s would take around 100 seconds. The exact numbers depend on your workload, but the shape of the formula is what matters: more buckets shortens a round, more data lengthens it. + + +**One important thing this formula leaves out:** the slowest bucket determines when a round can commit, because the commit operator waits for all of them. +If your bucketing key has bad distribution and one bucket holds 3× the data of the others, that one bucket sets the round duration: the other 15 readers finish their splits early and move on to the next queued table, but this table's commit still waits on that one straggler. +The `shuffle` on split assignment doesn't fix this; it just makes the bad bucket land on a different reader each round. + +A newly created table doesn't tier immediately; it waits one freshness interval before its first round (the coordinator schedules it: `New` → `Scheduled` → `Pending`). So picture **T=0** as the moment all three tables have passed that initial wait and are sitting in the coordinator's pending queue, each awaiting its first-ever round. They entered the queue as each table's first scheduling timer fired, in ascending freshness order, so it reads `[clicks, orders, customers]`. + +The Flink job's first heartbeat asks for work and gets clicks. +The enumerator generates 16 splits (one per bucket), shuffles them, and assigns one to each of the 16 readers. +All readers work in parallel; clicks finishes at **T=30s**. + +At **T=30s**, the Flink job asks for more work. It gets orders. The enumerator generates 8 splits and assigns them to 8 readers. +The other 8 readers don't sit idle: with orders' 8 splits assigned and nothing left pending, the enumerator immediately pulls the next table and those readers begin reading ahead (this is the read pipelining we detail below). What we're tracking in this walkthrough is the *commit* order, and orders is the table committing next: it tiers for about 90 seconds and its commit lands at **T=120s**, comfortably inside its 2-minute freshness window. + + +Now customers is next in the commit order. The enumerator generates 12 splits (since this is round 1 of a PK table, these are snapshot splits reading the full KV state). 12 readers can work on its splits at once; the 4 spare readers, again, pick up whatever's next in the queue rather than idling. The round takes roughly 100 seconds, and customers' commit lands at **T=220s**, well inside the configured 5-minute freshness. + + +But notice one thing here. clicks committed at **T=30s**; with a 1-minute freshness, its next round was scheduled for **T=90s**, so it's been waiting in the queue since **T=90s**. Its reads may well start earlier (spare readers can pick up clicks round 2 while customers is still being read), but its *commit* can't land until orders and customers have committed ahead of it. Review Comment: The article says commits happen “in queue order”, but the current implementation doesn't guarantees that. The commit operator is single-parallelism, so commits are serialized, but `TieringCommitOperator` commits a table once it has collected all write results for that table. It does not appear to maintain the coordinator pending-queue order across in-flight tables. A later-assigned but faster table could therefore reach the committer first and be committed before an earlier table. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
