Gowtham Potureddi

Posted on Aug 29

Apache Fluss: Streaming Storage Purpose-Built for Flink & the Real-Time Lakehouse

#python #sql #interview #dataengineering

Apache Fluss is what happens when you stop treating a message log as an analytics store and build a streaming layer for the job Flink actually does — a columnar stream with primary keys, real-time updates, native lookups, and a path into the lakehouse — instead of a row-oriented, append-only pipe that forces you to bolt an external key-value store, a separate history table, and a stack of workarounds around it. The hard problem was never "move the events"; Kafka moves events beautifully. The problem is everything that comes after the pipe: a Flink job that needs three columns out of forty still pays to ship all forty, a dimension table that changes still can't be updated in the log, an enrichment join still has to reach out to Redis or HBase, and the same data has to be copied into a warehouse or lakehouse to be queried historically — the row-log tax that every real-time analytics team eventually pays twice.

This guide is the senior-data-engineering walkthrough of the streaming storage layer that closes that gap — built for Flink and the real-time lakehouse, framed the way an interviewer probes it: why a Kafka topic is not an analytics stream, how Fluss splits Log Tables (append-only, columnar) from PrimaryKey Tables (upsertable, emitting a changelog), how columnar storage turns a wide stream read into a projection pushdown instead of a whole-record scan, how Flink reads, writes, and runs streaming lookup joins straight against Fluss with no external dimension store, and how tiering compacts aged data into Paimon or Iceberg so one logical table serves both the fresh tail and the full history. Each section pairs a teaching block with a Solution-Tail interview answer — code, a step-by-step trace, an output table, then a concept-by-concept breakdown of why it works.

When you want hands-on reps immediately after reading, drill the streaming practice library →, rehearse serving patterns on the real-time analytics practice library →, and sharpen the architecture axis with the system design practice library →.

On this page

Why Apache Fluss exists — the streaming-storage gap Kafka leaves
Fluss architecture — log tables, changelog, primary-key tables
The columnar stream — reads, column pruning, updates
Flink integration — streaming reads, writes, lookup joins
Lakehouse tiering — Paimon, Iceberg, and cost vs Kafka
Cheat sheet — streaming storage for Flink
Frequently asked questions
Practice on PipeCode

1. Why Apache Fluss exists — the streaming-storage gap

The consumption gap — Kafka moves events; a streaming store must also be read, updated, and queried

The one-sentence invariant: Apache Fluss is streaming storage built for the way Flink consumes data — a columnar stream that supports primary-key updates, point lookups, and lakehouse tiering — because a message log like Kafka is optimised to transport rows in order, not to be read analytically (only some columns), updated in place (a changing dimension), looked up by key (enrichment), or queried historically (the lake), and every one of those four needs is a workaround you graft onto Kafka but a first-class feature in Fluss. Point a real-time analytics platform at a raw topic and you inherit a row tax, an update gap, an external lookup store, and a second copy of the data in a lakehouse; put a purpose-built streaming store in the middle and Flink reads exactly the columns it needs, updates dimensions in place, looks them up directly, and reads history and the fresh tail as one table.

The four axes interviewers actually probe.

Record format — row vs columnar. How is a record physically stored, and what does a reader pay to read three of forty columns? A Kafka record is a row-oriented blob; you deserialize the whole thing even to read one field. The senior answer names columnar storage (Arrow) as the enabler of projection pushdown, and explains that analytics streams are read narrowly, not wholly.
Mutability — append-only vs primary-key updates. Can a record be updated, or only appended? A Kafka topic is an immutable append log; a changing dimension row means log compaction hacks or a separate store. The senior answer names PrimaryKey Tables and the changelog an upsert emits, and knows why "update-in-place" is the feature that lets a stream carry mutable state.
Read shape — scan vs lookup. Can a consumer look up a single key cheaply, or only scan the log? Kafka has no key-point-read; enrichment joins reach an external HBase/Redis. The senior answer names Fluss's native lookup on primary-key tables and the streaming lookup join that removes the external store.
Storage split — stream store vs lakehouse. Where does history live, and is it a second system? Kafka retention plus a separate warehouse/lakehouse is the Lambda tax — two copies, two pipelines, two truths. The senior answer names lakehouse tiering (to Paimon/Iceberg) and Union Read, one logical table over a fresh tier and a historical tier.

The 2026 reality — the streaming store is its own layer, not a topic.

Fluss is columnar streaming storage. Records live in Apache Arrow columnar format, so a reader fetches only the columns it projects — the single biggest cost difference from a row log when the consumer is an analytics job.
Two table types. Log Tables are append-only (the topic analogue); PrimaryKey Tables are upsertable and emit a changelog (+I/-U/+U/-D), so mutable state — dimensions, aggregates, the latest row per key — lives in the stream itself.
Native lookups and lookup joins. A PrimaryKey Table supports fast point reads by key, so Flink streaming lookup joins enrich a stream directly from Fluss with no bolted-on key-value store.
Lakehouse tiering built in. Fluss keeps the fresh tail in its own fast tier and tiers aged data into a lakehouse table (Apache Paimon, and Iceberg) on a schedule, then serves both as one table via Union Read — the stream store and the lake stop being two systems.

What interviewers listen for.

Do you say a Kafka topic is a row-oriented transport, not an analytics store and name the column tax unprompted? — senior signal.
Do you name primary-key updates and the changelog as how a stream carries mutable state, not "compact the topic"? — required answer.
Do you know lookup joins hit Fluss directly instead of an external HBase/Redis? — senior signal.
Do you frame history as lakehouse tiering + Union Read, not a second copy in a warehouse? — required answer.
Do you position Fluss as the streaming-storage layer under Flink, not a Kafka replacement for every use case? — senior signal.

Worked example — the Kafka-gap decision table

Detailed explanation. The most useful artifact for a streaming-storage interview is a memorised mapping of need → what Kafka forces → what Fluss gives. Every senior discussion converges on it: given a real-time requirement, does the row log cost you a workaround, and does a purpose-built streaming store remove it? Walk through building the table for a real-time analytics platform enriching and serving an order stream.

The needs. Read a few columns of a wide event; keep a mutable customer dimension; enrich orders by customer key; query the last 90 days.
The tension. Kafka satisfies transport, but each analytics need becomes an add-on system or a hack.
The rule. Match each need to the storage feature that serves it natively instead of a bolt-on.

Question. For each need, name what a raw Kafka topic forces and what Fluss provides natively.

Input.

Need	Kafka forces	Fluss provides
Read 3 of 40 columns	ship/deserialize the whole record	columnar projection pushdown
Update a dimension row	compaction hack or external store	PrimaryKey Table upsert + changelog
Enrich by key (lookup)	external HBase/Redis	native lookup join on a PK table
Query 90 days of history	separate warehouse/lakehouse copy	lakehouse tiering + Union Read

Code.

-- The Fluss way: one streaming store expresses all four needs.
-- 1) A columnar Log Table (append-only) — the "topic", but readers prune columns.
CREATE TABLE orders_log (
  order_id   BIGINT,
  cust_id    BIGINT,
  amount_cents BIGINT,
  channel    STRING,
  order_ts   TIMESTAMP(3)
  -- ... 35 more columns
) WITH ('bucket.num' = '8');

-- 2) A PrimaryKey Table (upsertable) — a mutable dimension living in the stream.
CREATE TABLE cust_dim (
  cust_id BIGINT,
  name    STRING,
  tier    STRING,
  PRIMARY KEY (cust_id) NOT ENFORCED
) WITH ('bucket.num' = '8');

-- 3) Enrichment is a lookup join straight against Fluss (see section 4) — no Redis.
-- 4) History is tiered to Paimon (see section 5) — no separate warehouse copy.

Step-by-step explanation.

The wide orders_log is a Log Table — the append-only, topic-like surface — but because Fluss stores it columnar, a reader that projects order_id, amount_cents pays for two columns, not forty. On Kafka the same read deserializes the whole record.
cust_dim is a PrimaryKey Table: a customer's tier can be updated by upserting the same cust_id, and Fluss emits a changelog so downstream jobs see the change. On Kafka a mutable dimension needs log compaction or an external database.
Enrichment (need 3) becomes a streaming lookup join that reads cust_dim by key directly from Fluss — the external key-value store Kafka pipelines bolt on (HBase, Redis) disappears.
History (need 4) is handled by tiering orders_log into Paimon on a schedule and Union-Reading it — one logical table for the fresh tail and the 90-day history, instead of copying the topic into a warehouse.
The mistake is treating each need as a new system: a topic, a compacted topic, a Redis, and a warehouse — four stores, four pipelines. The table is the antidote — one streaming store expresses all four because it was designed for how Flink reads, updates, looks up, and queries.

Output.

Need	Bolt-on with Kafka	Native with Fluss
Narrow column read	full-record deserialize	projection pushdown
Mutable dimension	compaction / external DB	PK upsert + changelog
Enrichment lookup	HBase / Redis	streaming lookup join
Historical query	warehouse/lakehouse copy	tiering + Union Read

Rule of thumb. Enumerate the analytics needs on top of the transport — narrow reads, updates, lookups, history — and count how many separate systems Kafka forces you to add for each. A purpose-built streaming store earns its place exactly when that count is more than one, because it collapses the bolt-ons into features of a single store.

Worked example — what interviewers actually probe

Detailed explanation. The senior streaming-storage interview has a predictable escalation: an ambiguous opener ("we're on Kafka + Flink, why change?"), then progressive narrowing to test whether you understand the row tax, mutability, lookups, and the storage split. The candidates who name columnar pushdown, primary-key changelogs, native lookups, and tiering score highest.

Ambiguous opener. "Kafka and Flink already work. Why add Fluss?"
Follow-up 1. "Our jobs read a handful of columns from fat events. What's the cost?" — probes columnar/pushdown.
Follow-up 2. "A dimension changes hourly. Where does it live?" — probes PK tables/changelog.
Follow-up 3. "Enrichment joins hammer Redis. Can we drop it?" — probes lookup joins.
Follow-up 4. "We keep 90 days in Kafka and a copy in the lake. Necessary?" — probes tiering/Union Read.

Question. Draft a five-point senior answer that pre-empts all four follow-ups without waiting to be asked.

Input.

Interview signal	Weak answer	Senior answer
Why change	"Fluss is newer/faster"	"Kafka transports; Fluss is analytics-shaped streaming storage"
Fat events	"make events smaller"	"columnar storage + projection pushdown reads only needed columns"
Mutable dim	"compact the topic"	"PrimaryKey Table upsert emits a changelog"
Enrichment	"scale Redis"	"native lookup join on a Fluss PK table — drop the KV store"
History	"raise Kafka retention"	"tier to Paimon, Union-read fresh + historical as one table"

Code.

Senior "why Fluss" answer template (5 points)
=============================================

1 — name the gap up front
  "Kafka is a row-oriented transport. Fluss is streaming storage shaped
   for how Flink consumes: columnar reads, primary keys, lookups, and a
   lakehouse tier. The change is about consumption, not transport."

2 — columnar / column pruning
  "Our jobs read a few of many columns; a columnar stream lets Flink
   push the projection down and fetch only those columns off the wire."

3 — primary-key updates + changelog
  "Changing dimensions live in a PrimaryKey Table: an upsert updates the
   row in place and Fluss emits a changelog so downstream jobs react."

4 — native lookup joins
  "Enrichment is a streaming lookup join straight against a Fluss PK
   table by key — we delete the external HBase/Redis dimension store."

5 — lakehouse tiering + Union Read
  "History tiers into Paimon on a schedule; one logical table serves the
   fresh Fluss tail and the historical lake via Union Read — no Lambda."

Step-by-step explanation.

Point 1 frames the whole answer around consumption, not speed. Weak candidates call Fluss "newer"; naming "Kafka transports, Fluss is analytics-shaped streaming storage" signals you understand the layer boundary, not just the tool.
Point 2 pre-empts the fat-event follow-up by naming projection pushdown — the columnar property that makes narrow reads cheap and is the single most concrete Kafka-vs-Fluss difference for analytics jobs.
Point 3 pre-empts the mutability follow-up. Naming the PrimaryKey Table and the changelog it emits is the difference between "compact the topic" and "the stream carries mutable state as a first-class feature."
Point 4 volunteers dropping the external store before the interviewer raises Redis — showing you know a Fluss PK table serves point lookups, so the enrichment join no longer needs a separate KV system.
Point 5 closes on the storage split and the Lambda tax — one store, fresh plus historical via Union Read — the sentence that separates a platform engineer from someone who just raises Kafka retention.

Output.

Grading criterion	Weak score	Senior score
Frames Fluss as consumption-shaped storage	rare	mandatory
Names columnar projection pushdown	occasional	mandatory
PK updates + changelog for mutable state	rare	senior signal
Lookup join replaces external KV	rare	senior signal
Tiering + Union Read kills the Lambda split	rare	senior signal

Rule of thumb. The senior "why Fluss" answer is a five-point monologue — consumption not transport, columnar pushdown, primary-key changelogs, native lookup joins, and tiering with Union Read — delivered without waiting for the follow-ups. Rehearse it once; it pre-empts the whole escalation.

Worked example — row log vs columnar stream, concretely

Detailed explanation. A common trap is "isn't a columnar stream just Parquet on Kafka?" The senior answer distinguishes transport format from storage-and-read model: Fluss is columnar and streaming and keyed, so a reader can subscribe, prune columns, look up a key, and update a row — properties a row log cannot offer at once. Contrast the two for one wide event stream.

The row log. Each record is a serialized row; a subscriber reads records in order and deserializes all fields.
The columnar stream. Records are stored column-wise (Arrow); a subscriber reads a range and fetches only projected columns, and keyed tables allow lookups/updates.
The decision. Row logs win on tiny, whole-record events; columnar streams win when reads are narrow, keys matter, or history is queried.

Question. Contrast a row-oriented log and a columnar stream on a narrow read, a keyed update, and a historical scan of the same wide event.

Input.

Dimension	Row log (Kafka-style)	Columnar stream (Fluss)
Physical layout	row-serialized record	column-wise (Arrow)
Read 3 of 40 cols	deserialize all 40	fetch 3 (pushdown)
Update by key	append-only (no update)	PK upsert + changelog
Point lookup	none (scan only)	native key lookup
History query	copy to warehouse	tier to Paimon, Union Read

Code.

-- Same logical stream, two storage models.

-- Row-log mental model: a consumer must read the whole record to get 2 fields.
--   consume(topic) -> bytes -> deserialize 40 columns -> use 2.   (row tax)

-- Fluss columnar Log Table: the reader projects, Fluss ships 2 columns.
SELECT order_id, amount_cents          -- 2 of 40 columns
FROM   orders_log;                     -- projection pushed into the columnar scan

-- Fluss PrimaryKey Table: the same data keyed → lookups AND updates.
SELECT name, tier FROM cust_dim WHERE cust_id = 42;   -- point lookup (not a scan)
INSERT INTO cust_dim (cust_id, tier) VALUES (42, 'gold');  -- upsert → changelog

Step-by-step explanation.

In the row-log model, the physical unit is a serialized record, so even a two-field read pays the full deserialization of all forty columns — the "row tax" that makes analytics-on-Kafka expensive at width.
Fluss stores the same Log Table column-wise, so SELECT order_id, amount_cents pushes the projection into the scan and only two columns cross the wire — the read cost tracks the columns projected, not the record width.
Keying the data as a PrimaryKey Table adds two abilities a log cannot have: a point lookup by cust_id (read one row without scanning) and an upsert that updates the row in place and emits a changelog.
The historical query is where the split shows: a row log needs a copy in a warehouse to be queried at scale, while the columnar stream tiers into Paimon and is Union-Read — the same table, no second copy.
The senior framing is not "columnar is better" but "columnar + streaming + keyed is a different storage model": it serves narrow reads, keyed lookups, updates, and history that a row transport structurally cannot serve together — which is exactly why Fluss is a layer, not a topic.

Output.

Operation	Row log	Columnar stream (Fluss)
Narrow read	O(record width)	O(columns projected)
Update by key	not possible	upsert + changelog
Point lookup	full scan	O(1) key lookup
History	warehouse copy	tier + Union Read

Rule of thumb. Distinguish transport format from storage model: a columnar stream is not "Parquet on a topic" but a store that is columnar, subscribable, and keyed at once — so it serves narrow reads, updates, lookups, and history together. Reach for it when your consumers do more than read whole records in order.

Senior interview question on choosing a streaming store

A senior interviewer often opens with: "Your platform runs Flink on Kafka. Jobs read a few columns from fat events, a customer dimension changes hourly, enrichment joins hammer a Redis cluster, and you keep 90 days in Kafka plus a copy in the lake for history. Decide whether to introduce Apache Fluss: what specifically it removes, how you'd model the order stream and the dimension, and why this is a streaming-storage decision — not just a faster broker."

Solution Using a columnar Log Table, a PrimaryKey dimension, native lookups, and tiering

-- 1) The order stream as a COLUMNAR Log Table — narrow reads become pushdowns.
CREATE TABLE orders_log (
  order_id   BIGINT,
  cust_id    BIGINT,
  amount_cents BIGINT,
  channel    STRING,
  order_ts   TIMESTAMP(3)
  -- ... many more columns (fat event)
) WITH (
  'bucket.num' = '16',                 -- parallelism
  'table.datalake.enabled' = 'true'    -- tier history to the lakehouse (section 5)
);

-- 2) The customer dimension as a PrimaryKey Table — update in place, emit a changelog.
CREATE TABLE cust_dim (
  cust_id BIGINT,
  name    STRING,
  tier    STRING,
  PRIMARY KEY (cust_id) NOT ENFORCED
) WITH ('bucket.num' = '16');

-- 3) Enrichment is a streaming lookup join straight against Fluss — Redis is gone.
SELECT o.order_id, o.amount_cents, c.name, c.tier
FROM   orders_log AS o
JOIN   cust_dim FOR SYSTEM_TIME AS OF o.proc_time AS c   -- lookup by key on Fluss
  ON   o.cust_id = c.cust_id;

# 4) What Fluss removes from the old topology, system by system.
BEFORE (Kafka + Flink):
  topic(orders, row)  +  compacted topic OR external DB(cust_dim)
  +  Redis(dim lookups)  +  Kafka 90d retention  +  lake copy(history)
AFTER (Fluss + Flink):
  Log Table(orders, columnar)  +  PrimaryKey Table(cust_dim, changelog)
  +  native lookup join         +  tiering to Paimon + Union Read
  => four bolt-on systems collapse into features of one streaming store.

Step-by-step trace.

Decision	Before (Kafka + bolt-ons)	After (Fluss)
Fat-event read	deserialize all columns	project columns (pushdown)
Dimension change	compaction / external DB	PK upsert + changelog
Enrichment lookup	Redis/HBase cluster	native lookup join
90-day history	Kafka retention + lake copy	tier to Paimon + Union Read
Systems to run	4 stores + pipelines	1 streaming store
Source of truth	scattered	one table, two tiers

After the change, orders_log is a columnar Log Table so a job reading order_id, amount_cents fetches two columns, not the full fat event; cust_dim is a PrimaryKey Table so an hourly change is an upsert that emits a changelog downstream jobs consume; enrichment is a lookup join served directly by Fluss, so the Redis cluster is decommissioned; and table.datalake.enabled tiers history into Paimon, Union-Read with the fresh tail — the Kafka retention plus lake copy becomes one logical table. Four bolt-on systems collapse into features of one store.

Output:

Metric	Kafka + Flink (bolt-ons)	Fluss + Flink
Bytes read per narrow query	full record width	projected columns only
Mutable dimension	external store / compaction	PK upsert + changelog
Enrichment infra	Redis/HBase cluster	none (native lookup)
History storage	Kafka retention + lake copy	tiered Paimon, one table
Distinct systems	~4	1

Why this works — concept by concept:

Columnar Log Table — storing the append-only stream column-wise makes a narrow read a projection pushdown, so read cost scales with the columns a job needs, not the width of a fat event — the row tax disappears.
PrimaryKey Table + changelog — keying the dimension turns an immutable log into upsertable state that updates in place and emits +I/-U/+U/-D, so a changing dimension lives in the stream and downstream jobs see every delta.
Native lookup join — because a PK table serves point reads by key, enrichment is a streaming lookup join against Fluss itself, deleting the external HBase/Redis store an ordinary Kafka pipeline must run and scale.
Tiering + Union Read — aged data compacts into Paimon and one logical table serves the fresh Fluss tail plus the historical lake, so 90-day history stops being a second copy and a second pipeline.
Cost — one streaming store versus a topic, a compacted topic/external DB, a Redis cluster, and a lake copy — projected columnar reads and O(1) key lookups instead of full-record scans and remote round-trips. The eliminated cost is three bolt-on systems and their pipelines — O(columns) reads and one store instead of O(record) reads across four.

Streaming
Topic — streaming
Streaming problems on log stores and event pipelines

Practice →

Design Topic — design Design problems on streaming-storage and architecture trade-offs

Practice →

2. Fluss architecture — log tables, changelog, primary-key tables

A coordinator over tablet servers holds bucketed tables; primary keys turn a log into upsertable state

The mental model in one line: Apache Fluss runs a CoordinatorServer for metadata and cluster control over a set of TabletServers that store the actual data as buckets of tables inside databases, keeping a hot local log tier durably backed by remote storage (S3), and exposing two table shapes — a Log Table that appends columnar records like a topic, and a PrimaryKey Table that upserts by key and emits a changelog (+I/-U/+U/-D) — so the same cluster serves both an append-only event stream and mutable, keyed state that downstream Flink jobs consume as change deltas. Get the table type and bucketing right and the cluster scales like Kafka but reads like a columnar store; get them wrong and you have a topic that can't be looked up or a keyed table that can't parallelise.

The cluster anatomy.

CoordinatorServer. The control plane: holds table metadata, assigns buckets to tablet servers, orchestrates rebalancing and failover. It is not on the data hot path — clients talk to tablet servers for reads and writes.
TabletServer. The data plane: stores bucket replicas, serves appends, reads, and key lookups. Scaling throughput means adding tablet servers and buckets, the same horizontal story as Kafka brokers/partitions.
Databases → tables → buckets. A table lives in a database and is split into buckets (the partition/parallelism unit); each bucket is an ordered log with its own offsets. More buckets = more parallel readers and writers.
Local + remote tiers. Recent data sits in a fast local log on the tablet servers; Fluss tiers it to remote storage (S3-compatible) for durability and cheap retention, so a server loss doesn't lose data and cold data isn't pinned to local disk.

Log Tables — the append-only, columnar surface.

Topic-like semantics. A Log Table is append-only with per-bucket offsets; a consumer subscribes from an offset (earliest/latest/timestamp) and reads forward — the familiar streaming contract.
Columnar on disk. Records are stored column-wise (Arrow), which is what enables projection pushdown on read (section 3) — the property a row-oriented topic lacks.
No key required. Log Tables model raw events (clicks, orders, logs) where every record is a new fact, not an update to a prior one.

PrimaryKey Tables — upsertable, keyed state.

Upsert by key. Writing a row with an existing primary key updates it; a new key inserts. The table always holds the latest row per key — a dimension, a materialized aggregate, the current state of an entity.
The changelog. Every upsert produces change events — +I (insert), -U/+U (the before/after of an update), -D (delete) — so a downstream Flink job reads the stream of changes, exactly like consuming a CDC feed.
Point lookups. Because it is keyed, a PrimaryKey Table serves a single-key read without scanning — the basis for streaming lookup joins (section 4).
Partial updates. An upsert can set a subset of columns, leaving the rest intact — useful when different producers own different columns of the same keyed row.

The changelog is the bridge.

Why it matters. The changelog is how mutable state propagates: a job that maintains a PrimaryKey Table of "latest order status per order" emits a clean delta stream, so consumers converge without replaying the whole table.
Retract semantics. -U/+U pairs let downstream aggregations retract the old value and apply the new one, keeping sums and counts correct as rows change — the mechanism behind correct streaming aggregates.
Feeds tiering. The same changelog/log is what the tiering service compacts into Paimon (section 5), so the lakehouse copy stays consistent with the stream.

The failure modes senior engineers pre-empt.

Under-bucketing. Too few buckets caps parallelism — a hot table serialises on a handful of tablet servers. Mitigation: size bucket.num to target parallelism/throughput up front; buckets are hard to change later.
Wrong table type. Modelling a mutable dimension as a Log Table means no updates and no lookups; modelling raw events as a PrimaryKey Table forces a key that doesn't exist and collapses distinct events. Mitigation: Log for facts, PrimaryKey for state.
Ignoring the remote tier. Relying on local disk for retention makes storage expensive and recovery slow. Mitigation: configure remote (S3) storage so cold data tiers off local disk and durability is decoupled from a single server.

Common interview probes on Fluss architecture.

"What are the server roles?" — a CoordinatorServer for metadata/control, TabletServers for data and lookups.
"What's the parallelism unit?" — buckets within a table; size them for throughput.
"Log Table vs PrimaryKey Table?" — append-only facts vs upsertable keyed state that emits a changelog.
"How is data durable?" — a hot local log tier backed by remote (S3) storage.

Worked example — create a Log Table and a PrimaryKey Table

Detailed explanation. The core modelling decision is per table: append-only facts or upsertable state. Create both in a Fluss database — a clicks Log Table for raw events and a product_dim PrimaryKey Table for a mutable dimension — and note what each enables.

Log Table. clicks — every row is a new event; no key; readers prune columns.
PrimaryKey Table. product_dim — keyed by product_id; upsertable; emits a changelog; supports lookups.
Buckets. Sized for parallelism on both.

Question. Create a Log Table for raw clicks and a PrimaryKey Table for a product dimension, and state what each table type makes possible.

Input.

Table	Type	Key	Enables
`clicks`	Log Table	none	append, subscribe, column pruning
`product_dim`	PrimaryKey Table	`product_id`	upsert, changelog, point lookup

Code.

CREATE DATABASE retail;
USE retail;

-- Log Table: append-only raw events (facts). No primary key.
CREATE TABLE clicks (
  event_id   BIGINT,
  user_id    BIGINT,
  product_id BIGINT,
  url        STRING,
  event_ts   TIMESTAMP(3)
) WITH ('bucket.num' = '8');           -- 8-way parallel append/read

-- PrimaryKey Table: upsertable, keyed state (a dimension). Emits a changelog.
CREATE TABLE product_dim (
  product_id BIGINT,
  name       STRING,
  category   STRING,
  price_cents BIGINT,
  PRIMARY KEY (product_id) NOT ENFORCED
) WITH ('bucket.num' = '8');           -- keyed by product_id, bucketed by it

Step-by-step explanation.

clicks has no primary key, so it is a Log Table: writes append new events and readers subscribe from an offset. It is the topic-shaped surface — but columnar, so a reader that projects user_id, product_id prunes the rest.
product_dim declares PRIMARY KEY (product_id) NOT ENFORCED, making it a PrimaryKey Table: writing an existing product_id updates that product's row in place rather than appending a duplicate.
Every write to product_dim produces changelog events, so a downstream job can consume the changes to products (a price update arrives as a -U/+U pair) without re-reading the whole dimension.
Because product_dim is keyed, SELECT ... WHERE product_id = 99 is a point lookup, not a scan — which is what lets a Flink lookup join enrich a click stream with product attributes directly from Fluss.
bucket.num on each table sets parallelism: eight buckets means up to eight parallel writers/readers per table. Choosing it up front matters because buckets define the ordering/parallelism unit and are not cheaply changed later.

Output.

Capability	`clicks` (Log)	`product_dim` (PrimaryKey)
Write semantics	append	upsert (update-in-place)
Read semantics	subscribe from offset	subscribe + point lookup
Emits changelog	no (append log)	yes (+I/-U/+U/-D)
Column pruning	yes	yes

Rule of thumb. Model raw, immutable facts as Log Tables and mutable, keyed entities as PrimaryKey Tables — the key is what unlocks upserts, the changelog, and point lookups. Decide bucket.num from target throughput at create time, because buckets are the parallelism unit and are painful to resize later.

Worked example — the changelog a primary-key upsert emits

Detailed explanation. The feature that makes a PrimaryKey Table more than a keyed store is the changelog: each upsert emits change events so downstream aggregations stay correct as rows mutate. Trace the changelog for a sequence of upserts to a latest_status table and show how a downstream count stays right.

The table. order_status keyed by order_id, holding the latest status.
The sequence. Insert an order as pending, then update it to paid.
The changelog. +I(pending), then -U(pending) + +U(paid).

Question. Show the changelog Fluss emits for an insert then an update of the same key, and explain how a downstream count by status stays correct.

Input.

Step	Write	Changelog emitted
1	upsert (1, pending)	`+I (1, pending)`
2	upsert (1, paid)	`-U (1, pending)`, `+U (1, paid)`
3	delete (1)	`-D (1, paid)`

Code.

-- A PrimaryKey Table holding the LATEST status per order.
CREATE TABLE order_status (
  order_id BIGINT,
  status   STRING,
  PRIMARY KEY (order_id) NOT ENFORCED
) WITH ('bucket.num' = '4');

-- Writes (upserts) over time:
INSERT INTO order_status VALUES (1, 'pending');   -- emits +I (1, pending)
INSERT INTO order_status VALUES (1, 'paid');      -- emits -U (1, pending), +U (1, paid)

# A downstream Flink aggregate consuming the CHANGELOG stays correct:
#   SELECT status, count(*) FROM order_status GROUP BY status;
#
#  event            pending  paid
#  +I (1, pending)     1       0     <- order 1 counted as pending
#  -U (1, pending)     0       0     <- retract the old value
#  +U (1, paid)        0       1     <- apply the new value
#
# Without retractions, "pending" would stay at 1 forever (double counting).

Step-by-step explanation.

The first upsert of key 1 has no prior row, so Fluss emits +I (1, pending) — a pure insert. A downstream count by status increments pending to 1.
The second upsert of the same key updates the row, so Fluss emits a pair: -U (1, pending) retracts the old value and +U (1, paid) applies the new one. This is the retract-then-apply that keeps aggregates correct.
The downstream count processes the retraction first — pending goes back to 0 — then the apply — paid goes to 1 — so the totals reflect the current state, not the history of writes.
Without a changelog (a plain append log), the consumer would see two independent rows and count both, leaving pending stuck at 1 — the classic double-count bug of aggregating a mutable stream that only appends.
A -D (1, paid) (delete) would retract the last value, dropping paid back to 0 — so deletes propagate as cleanly as updates, and any downstream materialized view converges to exactly the set of live keys.

Output.

After event	`pending` count	`paid` count
`+I (1, pending)`	1	0
`-U (1, pending)`	0	0
`+U (1, paid)`	0	1
`-D (1, paid)`	0	0

Rule of thumb. A PrimaryKey Table's changelog emits -U/+U pairs (and -D on delete) so downstream aggregations retract the stale value before applying the new one — the mechanism that keeps streaming counts and sums correct as rows mutate. Consume the changelog, not a re-scan, to react to changes cheaply.

Worked example — bucketing for parallelism and lookup distribution

Detailed explanation. Buckets are Fluss's parallelism and key-distribution unit. Too few and a hot table serialises; the wrong bucket key and lookups skew. Size and key buckets for a high-throughput orders_log and a lookup-heavy cust_dim.

The throughput table. orders_log — bucket for parallel append/read.
The lookup table. cust_dim — bucket by the lookup key so a lookup hits one bucket.
The rule. Bucket count follows throughput; bucket key follows the access pattern.

Question. Choose bucket.num and the bucket key for a high-write log and a lookup-heavy dimension, and explain the parallelism and skew consequences.

Input.

Table	Access pattern	bucket.num	bucket key
`orders_log`	high append + parallel read	32	default (round-robin/hash)
`cust_dim`	point lookups by `cust_id`	16	`cust_id` (the PK)

Code.

-- High-throughput Log Table: many buckets for parallel writers/readers.
CREATE TABLE orders_log (
  order_id BIGINT, cust_id BIGINT, amount_cents BIGINT, order_ts TIMESTAMP(3)
) WITH ('bucket.num' = '32');           -- 32-way parallelism for a hot stream

-- Lookup-heavy PrimaryKey Table: bucketed by the primary key so a lookup
-- for a given cust_id is served by exactly ONE bucket (no fan-out).
CREATE TABLE cust_dim (
  cust_id BIGINT, name STRING, tier STRING,
  PRIMARY KEY (cust_id) NOT ENFORCED
) WITH ('bucket.num' = '16');           -- PK tables bucket by the key by default

Step-by-step explanation.

orders_log sets 32 buckets because it is a hot append stream: 32 buckets allow up to 32 parallel writers and readers across tablet servers, so throughput scales instead of serialising on a few buckets.
Under-bucketing here is the classic mistake — 4 buckets on a stream that needs 32 caps you at 4-way parallelism regardless of how many tablet servers you add, and you cannot cheaply raise it later.
cust_dim is a PrimaryKey Table, so Fluss buckets it by the primary key: all rows for a given cust_id land in one bucket, and a lookup by cust_id is routed to exactly that bucket — no scatter-gather across buckets.
This key-aligned bucketing is what makes streaming lookup joins efficient: each lookup is a single-bucket point read, so lookup latency stays low and predictable even as the dimension grows.
Skew is the risk to watch: if the bucket key is low-cardinality or hot (say most orders share one cust_id), one bucket becomes a hotspot. Choose a high-cardinality, evenly distributed key for keyed tables, and size bucket count to spread the load.

Output.

Choice	Effect	Failure if wrong
32 buckets on `orders_log`	32-way parallel throughput	under-bucketing caps parallelism
PK-keyed buckets on `cust_dim`	single-bucket lookups	scatter-gather if not key-aligned
high-cardinality bucket key	even distribution	hot-bucket skew
set at create time	stable ordering unit	costly to resize later

Rule of thumb. Size bucket.num from target throughput and let a PrimaryKey Table bucket by its key so each lookup is a single-bucket point read. Pick a high-cardinality, evenly distributed key to avoid hot buckets, and decide up front — buckets are the ordering and parallelism unit and are expensive to change.

Senior interview question on Fluss table modelling

A senior interviewer might ask: "Model a real-time retail platform on Fluss: a raw order-event stream, a mutable customer dimension that changes throughout the day, and a downstream job that keeps a live count of orders by status. Choose table types and bucketing, show the changelog the dimension and the status table emit, and explain how the downstream aggregate stays correct as rows change."

Solution Using a Log Table, a PrimaryKey dimension, a changelog-emitting status table, and sized buckets

-- 1) Raw order events → Log Table (append-only facts, columnar, 32 buckets).
CREATE TABLE orders_log (
  order_id BIGINT, cust_id BIGINT, status STRING,
  amount_cents BIGINT, order_ts TIMESTAMP(3)
) WITH ('bucket.num' = '32');

-- 2) Mutable customer dimension → PrimaryKey Table (upsert + changelog + lookup).
CREATE TABLE cust_dim (
  cust_id BIGINT, name STRING, tier STRING,
  PRIMARY KEY (cust_id) NOT ENFORCED
) WITH ('bucket.num' = '16');

-- 3) Live "latest status per order" → PrimaryKey Table; its changelog drives the count.
CREATE TABLE order_status (
  order_id BIGINT, status STRING,
  PRIMARY KEY (order_id) NOT ENFORCED
) WITH ('bucket.num' = '16');

-- 4) Maintain latest status by upserting; each change emits -U/+U so the count is correct.
INSERT INTO order_status SELECT order_id, status FROM orders_log;   -- upsert by order_id

-- 5) The live aggregate consumes the CHANGELOG (retract + apply), not a re-scan.
SELECT status, count(*) AS n
FROM   order_status
GROUP  BY status;

# Changelog for order 1 as it moves pending -> paid, and the running count:
#   +I (1, pending)   -> pending=1
#   -U (1, pending)   -> pending=0     (retract stale)
#   +U (1, paid)      -> paid=1        (apply new)
# The count reflects CURRENT state, never double-counting the transition.

Step-by-step trace.

Layer	Component	Purpose
Facts	`orders_log` (Log Table)	append raw events, 32-way parallel
Dimension	`cust_dim` (PrimaryKey)	upsertable, changelog, lookup by key
State	`order_status` (PrimaryKey)	latest status per order, emits changelog
Aggregate	`count by status`	retract/apply keeps totals correct
Parallelism	bucket sizing	throughput on the log, key lookups on dims
Durability	local + remote (S3) tier	hot local, durable/cheap remote

After deployment, raw orders append to the 32-bucket orders_log; cust_dim upserts customer changes and emits a changelog so enrichment and downstream jobs see every delta; order_status upserts the latest status per order, and its -U/+U changelog lets the count by status retract the stale value and apply the new one, so the live count reflects the current state without double-counting the pending→paid transition. Log Tables carry facts, PrimaryKey Tables carry mutable state, and bucketing sizes throughput and lookup distribution.

Output:

Metric	Naive (append-only)	Fluss (typed tables + changelog)
Mutable dimension	duplicate rows	update-in-place (upsert)
Downstream count	double-counts transitions	correct (retract + apply)
Point lookup on a key	full scan	single-bucket lookup
Read width	whole record	projected columns
Throughput ceiling	few partitions	sized buckets

Why this works — concept by concept:

Log Table for facts — modelling raw, immutable order events as an append-only columnar Log Table gives topic-like subscribe semantics and projection pushdown, so the fact stream is both streamable and cheap to read narrowly.
PrimaryKey Table for state — keying the dimension and the status table turns append-only logs into upsertable state, so a changing customer or order status updates in place instead of accumulating duplicates.
Changelog retract/apply — the -U/+U pairs a PrimaryKey Table emits let downstream aggregations retract the stale value before applying the new one, which is exactly what keeps a live count-by-status correct across transitions.
Bucket sizing — many buckets on the hot log give parallel throughput while key-aligned buckets on the dimensions make each lookup a single-bucket point read, so the same cluster scales writes and serves lookups.
Cost — one cluster with a hot local log tier and a durable remote (S3) tier, columnar projected reads, and single-bucket lookups, versus a topic plus an external mutable store plus a scan-based count. The eliminated cost is duplicate-row storage and re-scan aggregation — O(columns) reads and O(1) lookups instead of O(record) reads and O(table) recomputes.

Design
Topic — design
Design problems on table modelling and partitioning

Practice →

Streaming Topic — streaming Streaming problems on changelogs and upsert state

Practice →

3. The columnar stream — reads, column pruning, updates

Fetch only the columns you read; look up and update by key — the analytics-streaming difference

The mental model in one line: Fluss stores stream records in Apache Arrow columnar format, which changes three things at once — a reader that projects a few columns triggers projection pushdown so only those columns cross the wire, a filter can use predicate/partition pruning to skip data it does not need, and because PrimaryKey Tables are keyed, a consumer can do an O(1) point lookup by key and a producer can do a partial update of some columns — so a columnar stream serves the narrow, filtered, keyed access patterns that analytics and enrichment jobs actually issue, instead of forcing the whole-record read a row log imposes. The width of your events stops mattering; only the columns you touch do.

Columnar storage and projection pushdown.

Arrow columns on the wire. Because records are stored column-by-column, Fluss can serve just the projected columns to a reader — the read cost is proportional to columns touched, not record width.
Projection pushdown. When Flink's plan projects a, b from a 40-column table, Fluss reads and ships only a, b. On a row log the broker ships the whole record and the client discards 38 columns.
The win scales with width. The wider the event and the narrower the read, the larger the saving — analytics streams (fat events, few columns per job) are exactly this shape.

Predicate and partition pruning.

Predicate pushdown. Filters (WHERE region = 'EU') can be pushed toward the scan so non-matching data is skipped earlier, reducing bytes processed.
Partitioning. A partitioned table (e.g. by date) lets a reader prune whole partitions it does not need, so a query for today never touches last month's data.
Combined with projection. Pruning columns and rows compounds: a narrow, filtered read over a wide, partitioned stream touches a small fraction of the data.

Primary-key point lookups.

O(1) by key. A PrimaryKey Table answers WHERE pk = ? as a single-bucket point read, not a scan — the property that makes it usable as a dimension store.
Latest-row semantics. The lookup returns the current row for the key (upserts updated it in place), so enrichment always sees fresh state.
The join enabler. These lookups are what section 4's streaming lookup joins call under the hood — no external KV needed.

Real-time and partial updates.

Update in place. An upsert to an existing key replaces its row and emits a changelog — the stream carries mutable state, not just facts.
Partial-column updates. A writer can update a subset of columns (e.g. only price_cents), leaving other columns untouched — so independent producers can own different columns of the same keyed row without clobbering each other.
Correctness downstream. Because updates emit -U/+U, downstream aggregates retract and reapply (section 2), staying correct as columns change.

The failure modes senior engineers pre-empt.

SELECT * on a wide stream. Projecting every column defeats pushdown and ships the full record — the row tax, reintroduced. Mitigation: project only needed columns in every streaming read.
No partitioning on a huge history. Without partitions, a filtered/historical read scans everything. Mitigation: partition by a natural filter key (date/tenant) so pruning applies.
Full-row writes when partial suffices. Overwriting all columns on every update forces every producer to know every column and risks clobbering. Mitigation: use partial-column updates so producers own their columns.

Common interview probes on the columnar stream.

"Why is a columnar stream cheaper to read?" — projection pushdown ships only projected columns; cost tracks columns, not record width.
"How do you make a filtered read cheap?" — predicate pushdown + partition pruning skip non-matching data.
"How does a dimension lookup avoid a scan?" — a PrimaryKey Table serves an O(1) point read by key.
"What is a partial update?" — upserting a subset of columns without touching the rest.

Worked example — projection pushdown on a wide log

Detailed explanation. The clearest columnar win is a narrow read of a fat event. Show a 40-column events_wide Log Table where a job needs two columns, and reason about what crosses the wire with columnar pushdown versus a row log.

The table. events_wide — 40 columns, high volume.
The read. A job that needs user_id, amount_cents only.
The saving. ~2/40 of the bytes versus a whole-record read.

Question. For a 40-column stream where a job reads two columns, contrast bytes shipped with columnar projection pushdown versus a row log.

Input.

Aspect	Row log	Columnar stream
Columns read	40 (whole record)	2 (projected)
Pushdown	none	projection pushed to scan
Bytes over wire	~full record	~2 columns
Client discards	38 columns	0

Code.

-- A wide event Log Table (40 columns). Storage is columnar (Arrow).
CREATE TABLE events_wide (
  event_id BIGINT, user_id BIGINT, amount_cents BIGINT,
  c4 STRING, c5 STRING, /* ... */ c40 STRING,
  event_ts TIMESTAMP(3)
) WITH ('bucket.num' = '16');

-- Narrow streaming read: project ONLY the two columns the job needs.
-- Fluss pushes the projection into the scan and ships 2 columns, not 40.
SELECT user_id, amount_cents
FROM   events_wide;

-- Anti-pattern: SELECT * defeats pushdown and ships the full 40-column record.
-- SELECT * FROM events_wide;   -- row tax reintroduced

Step-by-step explanation.

events_wide stores its 40 columns column-wise, so each column is a separate run of values on disk — the physical property that lets Fluss serve a subset without reading the rest.
The narrow query projects user_id, amount_cents; Flink's planner pushes that projection into the Fluss scan, so the tablet server reads and ships only those two columns' data.
On a row log the same logical read is impossible to optimise: the record is one serialized blob, so the broker ships all 40 columns and the client deserializes and discards 38 — bytes and CPU wasted proportional to the unused width.
The saving scales with the ratio of unused to used columns: at 2 of 40, you move roughly a twentieth of the column payload; the fatter the event and the narrower the read, the bigger the win.
The anti-pattern is SELECT *, which projects every column and defeats pushdown entirely — reintroducing the row tax on a columnar store. Disciplined narrow projection is what realises the columnar benefit in practice.

Output.

Read	Columns shipped	Relative bytes
Row log (any read)	40	1.0x
Fluss `SELECT user_id, amount_cents`	2	~0.05x
Fluss `SELECT *`	40	~1.0x (no benefit)
Fluss project 8 of 40	8	~0.2x

Rule of thumb. Project only the columns a streaming job needs so Fluss's columnar pushdown ships a fraction of a wide record — read cost tracks columns touched, not event width. Never SELECT * a wide stream; it defeats pushdown and reintroduces the row tax on a columnar store.

Worked example — a partial-column update on a PrimaryKey Table

Detailed explanation. Partial updates let independent producers own different columns of the same keyed row. Show a product_dim where a pricing job updates price_cents and a catalog job updates name/category, without either clobbering the other.

The row. product_dim keyed by product_id.
Producer A. Pricing — updates only price_cents.
Producer B. Catalog — updates only name, category.

Question. Show two producers partially updating disjoint columns of the same primary-key row and explain why neither clobbers the other's columns.

Input.

Producer	Updates	Leaves intact
Pricing	`price_cents`	`name`, `category`
Catalog	`name`, `category`	`price_cents`
Result	merged latest row	—

Code.

CREATE TABLE product_dim (
  product_id BIGINT,
  name       STRING,
  category   STRING,
  price_cents BIGINT,
  PRIMARY KEY (product_id) NOT ENFORCED
) WITH ('bucket.num' = '8');

-- Producer A (pricing) upserts ONLY the key + price → other columns untouched.
INSERT INTO product_dim (product_id, price_cents) VALUES (99, 1499);

-- Producer B (catalog) upserts ONLY the key + name/category → price untouched.
INSERT INTO product_dim (product_id, name, category) VALUES (99, 'Widget', 'tools');

-- Final row for product 99 is the MERGE of both partial updates:
--   (99, name='Widget', category='tools', price_cents=1499)

Step-by-step explanation.

product_dim is a PrimaryKey Table keyed by product_id, so any write with an existing key updates that row rather than appending a new one — the precondition for merging partial updates.
Producer A upserts only (product_id, price_cents). A partial update sets the listed columns and leaves the others as they were, so name and category are not overwritten with nulls.
Producer B upserts only (product_id, name, category). Again the unlisted price_cents is preserved, so A's price survives B's write.
The final materialised row for key 99 is the merge of both producers' columns — name='Widget', category='tools', price_cents=1499 — even though no single writer ever supplied all four fields.
Without partial updates, each producer would have to read-modify-write the whole row (racing and clobbering each other) or you would split the dimension into separate tables and join them — partial updates let disjoint owners share one keyed row safely.

Output.

After write	name	category	price_cents
A: price only	(unset)	(unset)	1499
B: name/category	Widget	tools	1499
merged row	Widget	tools	1499
(no clobber)	✓	✓	✓

Rule of thumb. Use partial-column updates so independent producers can each own a subset of a keyed row's columns without read-modify-write races or clobbering — the merged row reflects every producer's latest columns. It is the clean alternative to splitting a dimension into many tables just because different jobs write different fields.

Worked example — a primary-key point lookup vs a scan

Detailed explanation. The lookup is what makes a PrimaryKey Table a dimension store. Contrast an O(1) point read by key against a scan filter on the same data, and connect it to enrichment.

The lookup. WHERE cust_id = 42 on a keyed table → single-bucket point read.
The scan. The same filter on a non-keyed log → read and filter many rows.
The use. Enrichment reads the current row for a key with low latency.

Question. Contrast a primary-key point lookup and a scan-with-filter for reading one customer's current row, and explain the latency difference.

Input.

Approach	Table type	Work to read key 42
Point lookup	PrimaryKey Table	route to one bucket, read one row
Scan + filter	Log Table	read rows, filter for `cust_id=42`

Code.

-- PrimaryKey Table: O(1) point lookup, served by the bucket owning cust_id=42.
SELECT name, tier
FROM   cust_dim
WHERE  cust_id = 42;          -- single-bucket point read, latest row

-- Log Table (no key): the same "lookup" is a scan + filter over the stream.
-- SELECT name, tier FROM cust_log WHERE cust_id = 42;   -- reads/filters many rows

Step-by-step explanation.

On cust_dim, cust_id is the primary key and the bucket key, so a read for cust_id = 42 is routed directly to the single bucket that owns that key — no other bucket is touched.
That bucket returns the current row for key 42 (upserts kept it up to date), so the lookup is both O(1) and fresh — exactly what an enrichment join needs.
On a non-keyed Log Table, "read customer 42" has no index or routing: it is a scan that reads records and filters for the matching cust_id, with cost proportional to how much of the log you scan.
The latency gap is the difference between a routed single-row read and a stream scan — milliseconds versus seconds-to-minutes on a large log, and the gap widens as the data grows.
This is why enrichment against Fluss uses PrimaryKey Tables: the streaming lookup join (section 4) issues one point lookup per incoming row, and only a keyed table makes that cheap enough to do per-event at high throughput.

Output.

Read pattern	Cost	Freshness
PK point lookup	O(1), one bucket	current row
Scan + filter	O(scanned)	depends on scan window
At 1 row / event	cheap per event	live
At scale	flat	degrades

Rule of thumb. Serve per-key reads from a PrimaryKey Table so each is an O(1) single-bucket point lookup of the current row, not a scan-and-filter over a log. It is the property that makes a keyed Fluss table usable as a low-latency dimension store for per-event enrichment.

Senior interview question on columnar reads and keyed access

A senior interviewer might ask: "Your Flink jobs read a handful of columns from very wide events, filter by date, enrich per event by a customer key, and let independent producers update different columns of the customer row. Explain how a columnar streaming store makes each of these cheap — projection, filtering, keyed lookups, and partial updates — and show the reads and writes you'd issue."

Solution Using projection pushdown, partition pruning, point lookups, and partial updates

-- 1) Wide events, partitioned by date → narrow reads prune columns AND partitions.
CREATE TABLE events_wide (
  event_id BIGINT, user_id BIGINT, cust_id BIGINT, amount_cents BIGINT,
  /* ...many columns... */ dt STRING
) PARTITIONED BY (dt)
  WITH ('bucket.num' = '32');

-- Narrow + filtered read: project 3 columns, prune to one date partition.
SELECT user_id, cust_id, amount_cents
FROM   events_wide
WHERE  dt = '2026-08-26';        -- projection pushdown + partition pruning

-- 2) Customer dimension as a keyed table → O(1) enrichment lookups + partial updates.
CREATE TABLE cust_dim (
  cust_id BIGINT, name STRING, tier STRING, credit_cents BIGINT,
  PRIMARY KEY (cust_id) NOT ENFORCED
) WITH ('bucket.num' = '16');

-- Pricing/risk producer updates ONLY credit; catalog producer updates ONLY name/tier.
INSERT INTO cust_dim (cust_id, credit_cents) VALUES (42, 500000);   -- partial
INSERT INTO cust_dim (cust_id, name, tier)   VALUES (42, 'Acme', 'gold');  -- partial

-- 3) Per-event enrichment is a point lookup by key (single-bucket read).
SELECT e.event_id, e.amount_cents, c.name, c.tier
FROM   events_wide AS e
JOIN   cust_dim FOR SYSTEM_TIME AS OF e.proc_time AS c
  ON   e.cust_id = c.cust_id;     -- one O(1) lookup per event

Step-by-step trace.

Access pattern	Fluss mechanism	Effect
Read 3 of many columns	projection pushdown	ships 3 columns
Filter by date	partition pruning	skips other partitions
Read one customer	PK point lookup	O(1) single-bucket read
Update disjoint columns	partial update	no clobber, merged row
Enrich per event	lookup join	one point lookup / row

After deployment, the wide-event read projects three columns and prunes to a single date partition, so it touches a small fraction of a large, wide table; the customer dimension takes partial updates from a risk producer and a catalog producer without either clobbering the other's columns; and per-event enrichment is a lookup join that issues one O(1) point read per event against the keyed dimension. Columnar storage makes the read narrow, partitioning makes it shallow, keys make the lookup flat, and partial updates keep producers independent.

Output:

Metric	Row log + external store	Fluss columnar stream
Bytes for a narrow read	full record width	projected columns
Filtered historical read	full scan	pruned partitions
Per-key read	scan/filter or external KV	O(1) point lookup
Multi-producer updates	clobber / many tables	partial updates, one row
Enrichment infra	external KV cluster	native lookups

Why this works — concept by concept:

Projection pushdown — columnar storage lets Fluss ship only the columns a query projects, so a narrow read of a wide event costs the columns touched, not the record width — the core analytics-streaming saving.
Partition pruning — partitioning by a natural filter key lets a filtered or historical read skip whole partitions, compounding with projection so the scan is both narrow and shallow.
Primary-key point lookups — a keyed table answers a per-key read as an O(1) single-bucket lookup of the current row, which is what makes per-event enrichment against Fluss cheap enough to replace an external key-value store.
Partial-column updates — upserting a subset of columns lets independent producers own disjoint fields of one keyed row without read-modify-write races, so a shared dimension stays a single merged row.
Cost — projected, pruned reads and O(1) lookups against one store, versus full-record scans and a separate KV cluster. The eliminated cost is the bytes of unused columns, the rows of unpruned partitions, and an entire external lookup system — O(columns × matching partitions) reads and O(1) lookups instead of O(record × all data) reads plus remote calls.

Optimization
Topic — optimization
Optimization problems on projection pushdown and pruning

Practice →

Real-time analytics Topic — real-time-analytics Real-time analytics problems on columnar reads and lookups

Practice →

4. Flink integration — streaming reads, writes, lookup joins

Flink reads, upserts, and lookup-joins straight against Fluss — no external dimension store

The mental model in one line: Apache Fluss is wired into Flink through a Fluss catalog, so a Flink SQL job treats Fluss tables as first-class streaming sources and sinks — it reads a Log or PrimaryKey Table as a streaming source (subscribing from an offset and consuming the changelog for keyed tables), it writes by appending to a Log Table or upserting a PrimaryKey Table, and — the headline feature — it runs a streaming lookup join that enriches a stream by issuing a point lookup per row directly against a Fluss PrimaryKey Table, so the external HBase/Redis dimension store an ordinary Flink pipeline needs simply disappears. The join that used to be a network call to a separate cluster becomes a keyed read of the same streaming store.

The Fluss catalog in Flink.

Register once. A CREATE CATALOG ... WITH ('type' = 'fluss', ...) points Flink at the Fluss cluster; every Fluss database/table becomes addressable Flink SQL, no per-table connector boilerplate.
Tables are streaming by default. A Fluss source is unbounded — a Flink streaming job subscribes and reads forward; the same table can also be read in batch mode for a bounded snapshot.
DDL flows through. Creating a table in the Fluss catalog from Flink creates it in Fluss, so schema lives in one place.

Streaming reads (source).

Subscribe from an offset. scan.startup.mode = earliest / latest / from a timestamp — the standard streaming-consumer contract over a Log Table's buckets.
Changelog for keyed tables. Reading a PrimaryKey Table as a source yields its changelog (+I/-U/+U/-D), so a downstream job reacts to every row change with correct retract semantics.
Columnar pushdown applies. The source honours projection and filter pushdown (section 3), so a streaming read is as narrow as the query.

Streaming writes (sink).

Append to a Log Table. INSERT INTO log_table SELECT ... continuously appends events — the streaming ingest path.
Upsert a PrimaryKey Table. INSERT INTO pk_table SELECT ... upserts by key, maintaining latest-row state and emitting a changelog for downstream consumers.
Exactly-once. With Flink checkpointing, writes are consistent across failures, so the streaming pipeline is end-to-end reliable.

Streaming lookup joins — the external store, deleted.

FOR SYSTEM_TIME AS OF. Flink's temporal lookup-join syntax enriches each streamed row by looking up the current dimension row for its key at processing time.
Served by Fluss. The lookup is a point read on a Fluss PrimaryKey Table — no HBase/Redis; the dimension is the same streaming store, kept fresh by upserts.
Fresh by construction. Because the dimension updates in place and the lookup reads the current row, enrichment never uses stale attributes — and the changelog can also drive delta/lookup-style joins that keep less state than a dual-stream join.

The failure modes senior engineers pre-empt.

Lookup join against a non-keyed table. Enriching from a Log Table has no point lookup, so the join degrades to a scan or fails. Mitigation: dimensions are PrimaryKey Tables keyed by the join key.
Wide streaming source. A SELECT * source ships full records through the job. Mitigation: project columns in the source query so pushdown applies.
Unbounded dual-stream join state. Joining two high-volume streams with regular equi-joins materialises huge state. Mitigation: model the smaller/slower side as a keyed Fluss table and use a lookup (or delta) join so state stays bounded.

Common interview probes on Flink + Fluss.

"How does Flink see Fluss tables?" — through a Fluss catalog; tables are streaming sources and sinks.
"How do you enrich a stream without Redis?" — a streaming lookup join against a Fluss PrimaryKey Table.
"How do downstream jobs react to updates?" — read the PrimaryKey Table's changelog (retract/apply).
"How do you keep join state bounded?" — lookup/delta joins against a keyed table instead of a dual-stream equi-join.

Worked example — streaming write then streaming read

Detailed explanation. The base pipeline: a Flink job appends a source stream into a Fluss Log Table (sink), and another job subscribes and reads it forward (source). Wire both and note the offset and pushdown behaviour.

The sink. INSERT INTO orders_log SELECT ... — continuous append.
The source. SELECT ... FROM orders_log with a startup mode.
The contract. Unbounded, offset-based, projection-pruned.

Question. Write the Flink SQL to continuously append an external order stream into a Fluss Log Table and to read it back as a projected streaming source from the earliest offset.

Input.

Direction	Statement	Behaviour
Write (sink)	`INSERT INTO orders_log SELECT ...`	append, exactly-once with checkpoints
Read (source)	`SELECT cols FROM orders_log`	subscribe from offset, prune columns
Startup	`scan.startup.mode = earliest`	read history then tail

Code.

-- Register the Fluss catalog once, then address Fluss tables directly.
CREATE CATALOG fluss_catalog WITH (
  'type' = 'fluss',
  'bootstrap.servers' = 'fluss-coordinator:9123'
);
USE CATALOG fluss_catalog;
USE retail;

-- Streaming WRITE (sink): continuously append the source stream into a Log Table.
INSERT INTO orders_log
SELECT order_id, cust_id, amount_cents, order_ts
FROM   kafka_orders_source;      -- some upstream source table

-- Streaming READ (source): subscribe from earliest, projecting two columns.
SELECT order_id, amount_cents
FROM   orders_log
/*+ OPTIONS('scan.startup.mode' = 'earliest') */;   -- read history, then tail live

Step-by-step explanation.

The CREATE CATALOG registers Fluss with Flink; after USE CATALOG, orders_log is addressable directly — no per-table connector DDL, because the catalog resolves schema and location from Fluss.
The INSERT INTO orders_log SELECT ... is an unbounded streaming write: as rows arrive on the upstream source, Flink appends them to the Log Table's buckets, and with checkpointing the append is exactly-once across failures.
The reading job issues SELECT order_id, amount_cents FROM orders_log — a streaming source because the table is unbounded; it does not terminate, it tails new appends.
scan.startup.mode = earliest tells the source to begin at the start of the log (read all retained history) and then continue live; latest would read only new rows, and a timestamp mode would start at a point in time.
The projection to two columns is pushed into the Fluss scan (section 3), so even though orders_log may be wide, the source ships only order_id, amount_cents — the streaming read is as narrow as the query.

Output.

Job	Role	Semantics
`INSERT INTO orders_log`	sink	continuous append, exactly-once
`SELECT ... FROM orders_log`	source	unbounded, tails the log
`scan.startup.mode=earliest`	offset	history then live
projected columns	pushdown	narrow read

Rule of thumb. Register a Fluss catalog once and treat tables as ordinary Flink streaming sinks (INSERT appends/upserts) and sources (SELECT tails), choosing scan.startup.mode for where to begin and projecting columns so pushdown keeps the read narrow. Checkpointing makes the writes exactly-once end to end.

Worked example — a streaming lookup join that drops the external store

Detailed explanation. The headline pattern: enrich an order stream with customer attributes via a temporal lookup join against a Fluss PrimaryKey Table — no Redis/HBase. Build it and trace one row through the lookup.

The stream. orders_log with a processing-time attribute.
The dimension. cust_dim PrimaryKey Table, kept fresh by upserts.
The join. FOR SYSTEM_TIME AS OF o.proc_time — current-row lookup per event.

Question. Write a streaming lookup join enriching orders with customer name/tier from a Fluss PrimaryKey Table, and trace one order through the point lookup.

Input.

Piece	Value
Stream	`orders_log` (+ `proc_time AS PROCTIME()`)
Dimension	`cust_dim` (PK `cust_id`)
Join	temporal lookup on `cust_id`
Store removed	external HBase/Redis

Code.

-- The stream needs a processing-time attribute for the temporal lookup.
CREATE TABLE orders_log (
  order_id BIGINT, cust_id BIGINT, amount_cents BIGINT, order_ts TIMESTAMP(3),
  proc_time AS PROCTIME()                 -- processing-time column
) WITH ('bucket.num' = '16');

-- The dimension is a Fluss PrimaryKey Table (fresh via upserts).
CREATE TABLE cust_dim (
  cust_id BIGINT, name STRING, tier STRING,
  PRIMARY KEY (cust_id) NOT ENFORCED
) WITH ('bucket.num' = '16');

-- Streaming LOOKUP JOIN: one point lookup per order, served by Fluss itself.
SELECT o.order_id, o.amount_cents, c.name, c.tier
FROM   orders_log AS o
JOIN   cust_dim FOR SYSTEM_TIME AS OF o.proc_time AS c
  ON   o.cust_id = c.cust_id;

# Trace one order (cust_id=42) through the lookup:
#  order arrives:  (order_id=1001, cust_id=42, amount_cents=4200)
#  lookup:         cust_dim[cust_id=42] -> (name='Acme', tier='gold')   # O(1) point read on Fluss
#  emit:           (1001, 4200, 'Acme', 'gold')
# No Redis/HBase call — the dimension IS the Fluss PK table, kept fresh by upserts.

Step-by-step explanation.

orders_log declares proc_time AS PROCTIME(), the processing-time attribute Flink's temporal lookup join requires — it anchors "look up the dimension row as of now."
cust_dim is a PrimaryKey Table keyed by cust_id, so it supports the O(1) point lookup the join issues per order, and upserts keep each customer's row current.
JOIN cust_dim FOR SYSTEM_TIME AS OF o.proc_time is the temporal lookup: for each incoming order, Flink asks Fluss for the current cust_dim row matching cust_id, and joins it in.
Tracing order 1001 (cust_id 42): the join issues a point lookup for key 42, Fluss returns ('Acme','gold'), and the enriched row (1001, 4200, 'Acme', 'gold') is emitted — one keyed read, no scan.
Crucially there is no external store: the dimension is a Fluss table, so the lookup that used to be a network call to Redis/HBase is a point read of the same streaming store — one fewer system to run, scale, and keep in sync.

Output.

Order in	Lookup	Enriched out
(1001, 42, 4200)	cust_dim[42] → (Acme, gold)	(1001, 4200, Acme, gold)
(1002, 7, 990)	cust_dim[7] → (Globex, silver)	(1002, 990, Globex, silver)
store used	Fluss PK table	—
external KV	none	—

Rule of thumb. Enrich a stream with a Flink temporal lookup join (FOR SYSTEM_TIME AS OF proc_time) against a Fluss PrimaryKey Table keyed by the join key — each event triggers one O(1) point lookup of the current dimension row, and the external HBase/Redis store disappears. Keep the dimension fresh with upserts so enrichment never reads stale attributes.

Worked example — consuming a changelog downstream

Detailed explanation. A PrimaryKey Table read as a source yields a changelog, so a downstream job can maintain a correct aggregate as rows change. Build a downstream "revenue by tier" that reacts to customer tier changes via the changelog.

The upstream. cust_dim (PK) whose tier changes over time.
The downstream. An aggregate that must move revenue between tiers when a customer's tier changes.
The mechanism. The changelog's -U/+U retract and reapply.

Question. Show how a downstream aggregate stays correct when a customer's tier changes, by consuming the PrimaryKey Table's changelog.

Input.

Event on `cust_dim`	Changelog	Downstream effect
insert (42, gold)	`+I`	+revenue to gold
update (42, → platinum)	`-U`,`+U`	move revenue gold→platinum
delete (42)	`-D`	remove revenue

Code.

-- Downstream job reads cust_dim AS A STREAM → it receives the CHANGELOG.
-- Joined with per-customer revenue, a tier change retracts and reapplies.
SELECT c.tier, sum(r.revenue_cents) AS revenue
FROM   cust_dim AS c                       -- changelog source (+I/-U/+U/-D)
JOIN   cust_revenue AS r ON r.cust_id = c.cust_id
GROUP  BY c.tier;

# Customer 42 (revenue 10,000c) changes tier gold -> platinum:
#   +I (42, gold)       -> gold     += 10,000
#   -U (42, gold)       -> gold     -= 10,000   (retract)
#   +U (42, platinum)   -> platinum += 10,000   (apply)
# Result: revenue moves cleanly from gold to platinum, no double counting.

Step-by-step explanation.

Reading cust_dim as a streaming source yields its changelog rather than a static snapshot — the downstream job sees each change as +I, -U/+U, or -D, not just the final state.
When customer 42 is first inserted as gold, the +I adds their revenue to the gold bucket of the aggregate.
When 42's tier updates to platinum, the changelog emits -U (42, gold) then +U (42, platinum); the aggregate retracts 42's revenue from gold and applies it to platinum.
The net effect is that revenue moves between tiers exactly once — no double counting, no stale contribution left in gold — because retract-then-apply is atomic from the aggregate's perspective.
A -D (customer removed) would retract their revenue entirely, so the aggregate always reflects the current set of customers and their current tiers — the changelog keeps a downstream materialized view convergent with the source table.

Output.

After event	gold	platinum
`+I (42, gold)`	10,000	0
`-U (42, gold)`	0	0
`+U (42, platinum)`	0	10,000
`-D (42)`	0	0

Rule of thumb. Consume a PrimaryKey Table as a changelog source so downstream aggregates retract the stale contribution and apply the new one when a row changes — revenue and counts move correctly between groups without re-scanning. The changelog is how mutable state stays consistent across an entire streaming DAG.

Senior interview question on Flink enrichment without an external store

A senior interviewer might ask: "Your Flink pipeline enriches a high-throughput order stream with a customer dimension via a Redis lookup, and a downstream job aggregates revenue by customer tier. The Redis cluster is expensive and sometimes stale. Redesign on Fluss: the streaming read and write, the lookup join that removes Redis, and how the downstream aggregate reacts correctly when a customer's tier changes."

Solution Using a Fluss catalog, a lookup join, upsert sink, and changelog consumption

-- 1) Fluss catalog + tables: a Log stream, a keyed dimension, a keyed enriched sink.
CREATE CATALOG fluss_catalog WITH ('type' = 'fluss', 'bootstrap.servers' = 'fluss:9123');
USE CATALOG fluss_catalog; USE retail;

CREATE TABLE orders_log (
  order_id BIGINT, cust_id BIGINT, amount_cents BIGINT, order_ts TIMESTAMP(3),
  proc_time AS PROCTIME()
) WITH ('bucket.num' = '32');

CREATE TABLE cust_dim (               -- the dimension, fresh via upserts (replaces Redis)
  cust_id BIGINT, name STRING, tier STRING,
  PRIMARY KEY (cust_id) NOT ENFORCED
) WITH ('bucket.num' = '32');

CREATE TABLE orders_enriched (        -- keyed sink so downstream gets a clean changelog
  order_id BIGINT, cust_id BIGINT, amount_cents BIGINT, tier STRING,
  PRIMARY KEY (order_id) NOT ENFORCED
) WITH ('bucket.num' = '32');

-- 2) Enrich via a LOOKUP JOIN against Fluss (no Redis) and upsert the result.
INSERT INTO orders_enriched
SELECT o.order_id, o.cust_id, o.amount_cents, c.tier
FROM   orders_log AS o
JOIN   cust_dim FOR SYSTEM_TIME AS OF o.proc_time AS c    -- O(1) point lookup on Fluss
  ON   o.cust_id = c.cust_id;

-- 3) Downstream revenue-by-tier consumes the changelog → correct on tier changes.
SELECT tier, sum(amount_cents) AS revenue
FROM   orders_enriched                 -- changelog source: -U/+U move revenue between tiers
GROUP  BY tier;

# Redis removed; Fluss serves the lookup. A tier change propagates cleanly:
#   cust_dim: (42, gold) -> (42, platinum)   emits -U/+U
#   orders_enriched re-emits affected rows' changelog
#   revenue-by-tier: retract from gold, apply to platinum   (no double count)

Step-by-step trace.

Layer	Component	Purpose
Stream	`orders_log` (Log Table)	high-throughput order events
Dimension	`cust_dim` (PrimaryKey)	fresh, keyed — replaces Redis
Enrich	lookup join `FOR SYSTEM_TIME AS OF`	one O(1) lookup per order
Sink	`orders_enriched` (PrimaryKey)	keyed, emits a clean changelog
Aggregate	revenue by tier	retract/apply on tier changes
Infra removed	external Redis cluster	none

After the redesign, orders stream through orders_log; each order is enriched by a lookup join that issues one O(1) point read against the cust_dim PrimaryKey Table — the Redis cluster is deleted, and because the dimension updates in place, enrichment is never stale. The enriched rows land in a keyed orders_enriched sink whose changelog drives a revenue-by-tier aggregate; when a customer moves from gold to platinum, the -U/+U pair retracts their revenue from gold and applies it to platinum, so totals stay correct without re-scanning. Flink reads, enriches, upserts, and aggregates entirely against Fluss.

Output:

Metric	Flink + Redis	Flink + Fluss
Dimension store	external Redis cluster	Fluss PK table (native)
Lookup freshness	cache TTL / stale risk	current row (upsert-fresh)
Enrichment call	network to Redis	O(1) point lookup on Fluss
Tier-change correctness	manual/late	changelog retract/apply
Systems to operate	Flink + Kafka + Redis	Flink + Fluss

Why this works — concept by concept:

Fluss catalog + streaming source/sink — registering the catalog makes Fluss tables first-class Flink sources and sinks, so the whole pipeline — read, enrich, write — runs against one streaming store with projection pushdown on every read.
Lookup join replaces the KV store — the temporal lookup join issues an O(1) point read against a keyed Fluss table per event, so the external Redis/HBase dimension store is removed while enrichment stays per-event and low-latency.
Upsert-fresh dimension — because cust_dim updates in place, the lookup always returns the current row, eliminating the stale-cache class of bugs a TTL'd external store suffers.
Changelog-driven aggregate — the keyed sink emits a changelog whose -U/+U pairs let the revenue-by-tier aggregate retract and reapply on a tier change, so totals converge to current state without re-scanning.
Cost — one streaming store serving reads, lookups, and history, versus Flink plus Kafka plus a Redis cluster with its own scaling, sync, and staleness. The eliminated cost is the external dimension system and its failure modes — O(1) in-store lookups instead of remote calls, and retract/apply instead of full recomputes.

Streaming
Topic — streaming
Streaming problems on lookup joins and enrichment

Practice →

Event processing Topic — event-processing Event processing problems on changelogs and stateful joins

Practice →

5. Lakehouse tiering — Paimon, Iceberg, and cost vs Kafka

Tier aged data to Paimon; Union-Read the fresh tail with the history as one table

The mental model in one line: Apache Fluss makes the real-time lakehouse one system instead of two — enabling table.datalake.enabled runs a tiering (compaction) service that continuously moves aged Fluss data into a lakehouse table format (Apache Paimon, and Iceberg) on the object store, and a Union Read serves a single logical table by combining the fresh, seconds-old tail still in Fluss with the historical rows already in the lakehouse — so a streaming job reads the live edge, a batch/OLAP query reads deep history, and neither needs a separate pipeline or a second copy, collapsing the Lambda split (a streaming store plus a warehouse) that Kafka-based architectures pay for. The stream and the lake become tiers of one table, not two systems to keep in sync.

How tiering works.

Enable per table. table.datalake.enabled = 'true' marks a table for tiering; a tiering service (a background compaction job) reads aged Fluss data and writes it into the lakehouse format (Paimon by default; Iceberg supported).
Fresh stays in Fluss. The most recent, seconds-old data lives in Fluss's fast tier for low-latency streaming reads and lookups; only aged data is compacted down.
Columnar all the way. Because Fluss is already columnar (Arrow) and Paimon/Iceberg are columnar (Parquet/ORC), tiering is a format-aligned compaction, and the tiered data is directly queryable by any lakehouse engine.
Freshness knob. A freshness/compaction cadence controls how quickly data becomes available in the lakehouse tier and how much stays hot in Fluss.

Union Read — one logical table, two tiers.

Streaming read = fresh tail. A streaming query reads the live Fluss tier, seeing data seconds old — the real-time path.
Batch/OLAP read = history + tail. A batch query over the same table reads the historical lakehouse unioned with the fresh Fluss tail, so it sees complete, up-to-the-second data without a separate pipeline.
Direct lakehouse access. The tiered data is a normal Paimon/Iceberg table, so Spark, Trino, StarRocks, or Flink batch can read the history directly (e.g. via a $lake suffix in Fluss, or the Paimon catalog) — the lake is not locked inside Fluss.

Why this beats the Kafka + warehouse split.

One copy, one truth. Instead of Kafka retention and a copied lakehouse table maintained by a separate ingestion job, tiering keeps one table with a hot and a cold tier — no dual pipeline, no drift.
Cheaper retention. Long history lives in cheap object storage as compacted columnar files, not in expensive broker-attached storage sized for row-log retention.
Consistent semantics. The tiered lake is derived from the same changelog/log, so history and the live stream agree — no reconciliation between a stream and a nightly batch load.

Cost and operations vs Kafka.

Storage. Kafka retains rows on broker/tiered storage sized for the retention window; Fluss keeps a small hot tier and compacts the rest into cheap columnar object storage — lower cost at long retention.
Systems. Kafka + external KV + separate lakehouse ingestion vs one Fluss cluster with a tiering service — fewer moving parts to run and monitor.
Read efficiency. Analytics reads are columnar and pruned in both the hot and cold tiers, versus row-log reads plus a warehouse copy — less compute per query.

Use cases where this wins.

Real-time lakehouse dashboards. Live tiles from the Fluss tail, historical trends from the Paimon tier, one table.
Streaming enrichment + history. Lookup joins on the fresh dimension, batch reprocessing on the tiered history.
CDC-into-lakehouse. PrimaryKey Tables' changelogs land as upserts in Paimon, keeping a queryable, mutable historical table in sync with the stream.

The failure modes senior engineers pre-empt.

Tiering off for high-retention tables. Keeping long history only in Fluss's hot tier is expensive. Mitigation: enable tiering so aged data compacts to cheap object storage.
Assuming the lake is real-time. The tiered tier lags by the compaction cadence; a query needing the live edge must Union-Read (or read Fluss directly), not the lake alone. Mitigation: use Union Read / streaming reads for freshness-critical queries.
Ignoring compaction cost. The tiering service consumes resources; unbounded small files hurt. Mitigation: size compaction cadence and file targets, and monitor the tiering job like any pipeline.

Common interview probes on tiering.

"How does Fluss serve history without a warehouse?" — tier aged data into Paimon/Iceberg and Union-Read it with the fresh tail.
"What is Union Read?" — one logical table = fresh Fluss tier + historical lakehouse, combined at query time.
"Why is this cheaper than Kafka?" — small hot tier + compacted columnar object storage vs row-log retention + a separate lake copy.
"How fresh is the lake tier?" — bounded by the compaction cadence; use Union/streaming reads for the live edge.

Worked example — enable tiering to Paimon

Detailed explanation. Turning a Fluss table into a tiered lakehouse table is a table property plus a running tiering service. Enable it on orders_log and note what the tiering service does.

The property. table.datalake.enabled = 'true' (+ a freshness setting).
The service. A background compaction job moves aged data into Paimon.
The result. Fresh in Fluss, history in Paimon — one table.

Question. Enable lakehouse tiering on a Fluss Log Table and explain what the tiering service does with fresh versus aged data.

Input.

Setting	Value	Effect
`table.datalake.enabled`	`true`	mark table for tiering
`table.datalake.freshness`	`3min`	how fresh the lake tier is
tiering service	running	compacts aged data → Paimon

Code.

-- Enable tiering at table creation (or ALTER an existing table).
CREATE TABLE orders_log (
  order_id BIGINT, cust_id BIGINT, amount_cents BIGINT, order_ts TIMESTAMP(3)
) WITH (
  'bucket.num' = '16',
  'table.datalake.enabled'   = 'true',   -- tier aged data into the lakehouse (Paimon)
  'table.datalake.freshness' = '3min'    -- target lag for the lakehouse tier
);

-- A tiering/compaction service runs alongside the cluster and continuously
-- moves aged Fluss data into the Paimon table backing orders_log.
-- Fresh (seconds-old) data stays in Fluss for low-latency streaming reads.

Step-by-step explanation.

table.datalake.enabled = 'true' marks orders_log as a tiered table: Fluss now maintains a backing Paimon table and a tiering job that keeps it populated with aged data.
table.datalake.freshness = '3min' sets the target lag — how far behind the live edge the lakehouse tier is allowed to be — trading compaction frequency (cost) against how current the lake tier is.
The tiering service runs continuously in the background, reading aged Fluss log/changelog data and writing it into the Paimon table as compacted columnar files — it is a managed compaction, not a separate ingestion pipeline you write.
The freshest data (seconds old) stays in Fluss's fast tier so streaming reads and lookup joins keep their low latency; only data past the hot window is compacted down to the lake tier.
The outcome is one table with two physical tiers — a hot Fluss tier and a cold Paimon tier — instead of a Kafka topic plus a separately maintained lakehouse copy, and the Paimon table is directly readable by lakehouse engines.

Output.

Data age	Physical tier	Read path
seconds	Fluss hot tier	streaming read / lookup
minutes+ (past freshness)	Paimon (object store)	batch / OLAP / Union
whole table	both tiers	Union Read
pipeline to maintain	tiering service (managed)	not hand-written

Rule of thumb. Turn on table.datalake.enabled (with a freshness target) so a managed tiering service compacts aged data into Paimon while the fresh tail stays hot in Fluss — you get a lakehouse copy without writing or operating a separate ingestion pipeline. Size the freshness cadence to trade compaction cost against how current the lake tier must be.

Worked example — a Union Read of fresh plus historical

Detailed explanation. The payoff of tiering is reading one logical table across both tiers. Show a streaming read that sees the fresh tail, a direct lakehouse read of history, and a batch Union Read that sees everything.

Streaming read. Live Fluss tier — seconds old.
Lakehouse read. orders_log$lake — the Paimon history directly.
Union Read. Batch over the table — history + fresh tail.

Question. Show the three read paths over a tiered Fluss table — streaming (fresh), lakehouse-only (history), and batch Union Read (both) — and what each returns.

Input.

Read	Path	Sees
streaming	`SELECT ... FROM orders_log` (streaming)	fresh Fluss tail
lakehouse-only	`SELECT ... FROM orders_log$lake`	compacted history (Paimon)
batch union	`SELECT ... FROM orders_log` (batch)	history + fresh tail

Code.

-- (A) Streaming read → the live Fluss tier, data seconds old.
SELECT order_id, amount_cents
FROM   orders_log
/*+ OPTIONS('scan.startup.mode' = 'latest') */;     -- real-time edge

-- (B) Lakehouse-only read → the Paimon history directly (via the $lake suffix).
--     Readable by Flink batch, Spark, Trino, StarRocks against the Paimon table.
SELECT count(*), sum(amount_cents)
FROM   orders_log$lake;                              -- historical (compacted) tier

-- (C) Batch Union Read over the SAME table → history UNIONED with the fresh tail.
SET 'execution.runtime-mode' = 'batch';
SELECT count(*), sum(amount_cents)
FROM   orders_log;                                   -- history + fresh = complete

Step-by-step explanation.

Read (A) is a streaming query with scan.startup.mode = latest: it tails the Fluss hot tier and sees data seconds old — the real-time path for live dashboards and enrichment.
Read (B) targets orders_log$lake, the backing Paimon table, directly — this is the history tier, compacted columnar files on object storage, and it is readable not just by Flink but by any lakehouse engine (Spark, Trino, StarRocks).
Read (B) alone lags the live edge by the freshness cadence, so it is right for heavy historical/OLAP scans but wrong when you need the last few minutes.
Read (C) runs a batch query over the plain orders_log: Fluss performs a Union Read, combining the Paimon history with the fresh Fluss tail, so the aggregate is complete and current — history plus the live edge — from one table.
The three paths share one table and one copy of the data: real-time from the hot tier, deep history from the lake tier, and a complete picture via Union Read — none of which requires a separate warehouse load or a reconciliation between a stream and a batch table.

Output.

Read	Freshness	Best for
(A) streaming	seconds	live tiles, enrichment
(B) `$lake` only	~freshness cadence behind	heavy historical scans
(C) batch union	complete (history + tail)	correct point-in-time totals
all three	one table, one copy	no Lambda split

Rule of thumb. Read the fresh tail with a streaming query, deep history directly from the $lake (Paimon) tier, and a complete picture with a batch Union Read over the same table — one logical table serves real-time, historical, and point-in-time-complete needs without a second copy. Use Union or streaming reads whenever freshness matters; the lake-only tier lags by the compaction cadence.

Worked example — the cost and ops comparison vs Kafka

Detailed explanation. Put the architectures side by side: Kafka's row retention plus external stores plus a separate lakehouse ingestion, versus one Fluss cluster with tiering. Reason about storage, systems, and reads.

Kafka stack. Row-log retention + external KV + separate lakehouse ingestion.
Fluss stack. One cluster, hot tier + tiered Paimon, native lookups.
The axes. Storage cost, number of systems, read efficiency.

Question. Compare the Kafka-based and Fluss-based real-time-lakehouse architectures on storage cost, systems to operate, and read efficiency.

Input.

Axis	Kafka stack	Fluss stack
History storage	row retention on brokers/tiered	compacted columnar on object store
Enrichment store	external HBase/Redis	native PK lookups
Lakehouse copy	separate ingestion pipeline	built-in tiering service
Analytics reads	row + warehouse copy	columnar pruned, both tiers

Code.

# Architecture cost/ops comparison for a real-time lakehouse.

KAFKA-BASED (Lambda-ish):
  Kafka topics (row) ── retain 90d on brokers/tiered  ── $$$ row retention
     ├─ external KV (Redis/HBase) for dim lookups      ── a cluster to run
     └─ separate ingestion → warehouse/lakehouse copy   ── a pipeline + drift
  reads: row deserialize + a second copy for analytics

FLUSS-BASED (unified):
  Fluss cluster
     ├─ hot tier (seconds)  ── small, fast
     ├─ tiering service → Paimon (90d) on object store ── cheap columnar
     ├─ native PK lookups   ── no external KV
     └─ Union Read           ── one table, fresh + history
  reads: columnar pruned in BOTH tiers, one copy

Net: fewer systems, cheaper long retention, one source of truth.

Step-by-step explanation.

On storage, Kafka retains 90 days as row-oriented records on broker or tiered storage sized for the retention window, while Fluss keeps only a small hot tier and compacts the rest into columnar files on cheap object storage — long retention is materially cheaper.
On enrichment, the Kafka stack runs an external HBase/Redis cluster for dimension lookups, whereas Fluss serves those lookups natively from PrimaryKey Tables — one entire system removed.
On the lakehouse copy, Kafka needs a separate ingestion pipeline to land data in the warehouse/lakehouse (and that copy can drift from the stream), while Fluss's tiering service maintains the Paimon tier from the same log — one copy, no drift, no hand-written pipeline.
On reads, the Kafka path deserializes row records and maintains a second analytics copy, while Fluss reads columnar-pruned data in both the hot and cold tiers — less compute per query and no duplication.
Net, the Fluss architecture is fewer systems (no external KV, no separate ingestion), cheaper at long retention (columnar object storage), and a single source of truth (Union Read over one table) — the concrete case for the unified real-time lakehouse.

Output.

Axis	Kafka stack	Fluss stack
Long-retention storage	expensive (row)	cheap (compacted columnar)
Systems to operate	topic + KV + ingestion + lake	one cluster + tiering
Copies of the data	≥ 2 (stream + lake)	1 (two tiers)
Analytics read cost	row + second copy	columnar pruned, one copy

Rule of thumb. Count storage, systems, and copies: a Kafka real-time-lakehouse pays for row retention, an external lookup store, and a separate lakehouse ingestion, while a Fluss cluster with tiering collapses those into one store with a cheap columnar history tier and native lookups. The unified architecture wins on cost and operational surface exactly when you need enrichment and long history.

Senior interview question on designing a real-time lakehouse

A senior interviewer might ask: "Design a real-time lakehouse for an order platform: live dashboards needing seconds-fresh data, historical analytics over 90 days, and streaming enrichment — on Fluss with Paimon. Cover how history is tiered, how one logical table serves both fresh and historical reads, how enrichment avoids an external store, and why this is cheaper and simpler than a Kafka-plus-warehouse architecture."

Solution Using tiering to Paimon, Union Read, native lookups, and one source of truth

-- 1) Tiered tables: fresh in Fluss, aged compacted into Paimon automatically.
CREATE TABLE orders_log (
  order_id BIGINT, cust_id BIGINT, amount_cents BIGINT, order_ts TIMESTAMP(3)
) WITH (
  'bucket.num' = '32',
  'table.datalake.enabled'   = 'true',   -- tier history to Paimon
  'table.datalake.freshness' = '3min'
);

CREATE TABLE cust_dim (                   -- fresh, keyed dimension for enrichment
  cust_id BIGINT, name STRING, tier STRING,
  PRIMARY KEY (cust_id) NOT ENFORCED
) WITH ('bucket.num' = '32', 'table.datalake.enabled' = 'true');

-- 2) Live dashboard: streaming read of the fresh Fluss tier (seconds old).
SELECT window_start, sum(amount_cents) AS revenue
FROM   TABLE(TUMBLE(TABLE orders_log, DESCRIPTOR(order_ts), INTERVAL '1' MINUTE))
GROUP  BY window_start;                   -- real-time tiles

-- 3) Historical analytics: batch Union Read → history (Paimon) + fresh tail, complete.
SET 'execution.runtime-mode' = 'batch';
SELECT cust_id, sum(amount_cents)
FROM   orders_log                         -- Union Read over 90d history + live edge
WHERE  order_ts >= now() - INTERVAL '90' DAY
GROUP  BY cust_id;

-- 4) Enrichment: lookup join against the Fluss PK dimension — no external KV.
SELECT o.order_id, o.amount_cents, c.tier
FROM   orders_log AS o
JOIN   cust_dim FOR SYSTEM_TIME AS OF o.proc_time AS c
  ON   o.cust_id = c.cust_id;

Step-by-step trace.

Requirement	Mechanism	Tier read
Seconds-fresh dashboard	streaming read	Fluss hot tier
90-day analytics	batch Union Read	Paimon + fresh tail
Direct lake access	`orders_log$lake`	Paimon (any engine)
Enrichment	lookup join	Fluss PK dimension
History storage	tiering service	compacted columnar (object store)
Source of truth	one table, two tiers	no separate copy

After deployment, live dashboards read the fresh Fluss tier and see seconds-old data; 90-day analytics run as a batch Union Read over the same table, combining the Paimon history with the live tail for complete totals; heavy historical jobs can hit orders_log$lake (the Paimon table) directly from Spark/Trino; and enrichment is a lookup join against the keyed cust_dim with no external store. The tiering service maintains the Paimon history from the same log, so there is one copy and one source of truth — the Kafka topic, external KV, and separate warehouse ingestion of the old architecture are all gone.

Output:

Metric	Kafka + warehouse	Fluss + Paimon
Dashboard freshness	minutes (batch load)	seconds (streaming tier)
Historical query	warehouse copy	Union Read, one table
Enrichment store	external KV cluster	native PK lookups
Copies of data	stream + warehouse	one (hot + cold tiers)
Long-retention cost	row retention	compacted columnar object store
Systems to operate	4+	1 cluster + tiering

Why this works — concept by concept:

Tiering to Paimon — a managed tiering service compacts aged Fluss data into a columnar Paimon table on cheap object storage, so long history is retained without expensive row-log storage or a hand-written ingestion pipeline.
Union Read — one logical table combines the fresh Fluss tail with the historical Paimon tier at query time, so real-time and historical reads hit the same table and a batch query still sees the live edge — the Lambda split disappears.
Native lookups for enrichment — the keyed dimension serves O(1) point lookups for the streaming lookup join, so enrichment needs no external KV cluster and never reads stale attributes.
One source of truth — because both tiers derive from the same log/changelog, the live stream and the historical lake agree, eliminating the drift and reconciliation between a stream and a separate warehouse copy.
Cost — one cluster with a small hot tier, a cheap compacted columnar history tier, and native lookups, versus a topic plus external KV plus a separate lakehouse ingestion and copy. The eliminated cost is a whole tier of systems and a duplicate dataset — one store, columnar-pruned reads, and O(1) lookups instead of row retention, remote lookups, and a second pipeline.

Design
Topic — design
Design problems on real-time lakehouse architecture

Practice →

Optimization
Topic — optimization
Optimization problems on tiering, compaction, and retention cost

Practice →

Cheat sheet — streaming storage for Flink

The consumption gap. Kafka is a row-oriented transport optimised to move events in order; a streaming store must also be read narrowly (few columns), updated by key, looked up by key, and queried historically. Fluss is that store — columnar, keyed, tierable — purpose-built for how Flink consumes. Reach for it when your analytics needs more than "read whole records in order."
Log Table vs PrimaryKey Table. Model immutable facts (clicks, orders, logs) as Log Tables — append-only, columnar, subscribe-from-offset. Model mutable, keyed entities (dimensions, latest-state, aggregates) as PrimaryKey Tables — upsert-in-place, serve O(1) point lookups, and emit a changelog (+I/-U/+U/-D). The key is what unlocks updates, lookups, and the changelog.
Columnar reads. Fluss stores records column-wise (Arrow), so a narrow read triggers projection pushdown — only projected columns cross the wire, cost tracks columns not record width. Partition tables by a filter key for partition pruning. Never SELECT * a wide stream; it defeats pushdown and reintroduces the row tax.
Updates and lookups. Upsert a PrimaryKey Table to update state in place; use partial-column updates so independent producers own disjoint columns of one keyed row without clobbering. Read one key as an O(1) point lookup (single bucket), not a scan — the property that makes a Fluss table a dimension store.
Flink integration. Register a Fluss catalog once; tables become streaming sources (SELECT tails from scan.startup.mode) and sinks (INSERT appends to Log / upserts PrimaryKey). Enrich with a streaming lookup join — JOIN dim FOR SYSTEM_TIME AS OF o.proc_time ON o.k = dim.k — one O(1) lookup per event against a Fluss PK table, so the external HBase/Redis dimension store is deleted. Consume a PK table's changelog downstream so aggregates retract/apply correctly on change.
Changelog correctness. A PrimaryKey Table's -U/+U pairs (and -D) let downstream count/sum retract the stale value before applying the new one — the mechanism that keeps streaming aggregates correct as rows mutate. Consume the changelog, don't re-scan.
Lakehouse tiering. Set table.datalake.enabled = 'true' (+ a freshness cadence) so a tiering service compacts aged data into Paimon (or Iceberg) on object storage. Fresh, seconds-old data stays in Fluss's hot tier; only aged data compacts down. The tiered table is a normal Paimon table readable by Spark/Trino/StarRocks.
Union Read. One logical table = fresh Fluss tail + historical lakehouse. Streaming read → the live edge; table$lake → history only (lags by the compaction cadence); batch read → Union Read of history + fresh tail (complete). Use Union/streaming reads whenever freshness matters.
Bucketing. Buckets are the parallelism and key-distribution unit. Size bucket.num from target throughput on hot logs; PrimaryKey Tables bucket by the key so a lookup hits one bucket. Pick a high-cardinality, evenly distributed key to avoid hot buckets; decide at create time — buckets are costly to resize.
Durability. A hot local log tier is backed by remote (S3) storage, so durability and cheap retention are decoupled from any single TabletServer; a CoordinatorServer holds metadata and control off the data hot path.
Cost vs Kafka. Kafka real-time-lakehouse = row retention + external KV + separate lakehouse ingestion (≥ 2 copies, ≥ 4 systems). Fluss = one cluster, small hot tier + compacted columnar history, native lookups, Union Read (1 copy, 2 tiers). The unified stack wins on cost and operational surface when you need enrichment and long history.
Kafka or Fluss? Fluss is the streaming-storage layer under Flink, not a drop-in Kafka replacement for every use case. Fluss can sit alongside Kafka (ingest from a topic) or replace it where the workload is Flink-centric analytics with updates, lookups, and a lakehouse — pick per workload, not by fashion.

Frequently asked questions

What is Apache Fluss and how is it different from Kafka?

Apache Fluss is a streaming storage layer purpose-built for Flink and the real-time lakehouse. Where Kafka is a row-oriented, append-only message log optimised to transport events in order, Fluss is designed for how Flink consumes data: records are stored in a columnar format (Apache Arrow), so a reader can push down a projection and fetch only the columns it needs; tables can be keyed as PrimaryKey Tables that update in place and emit a changelog; those keyed tables serve O(1) point lookups so enrichment joins hit Fluss directly instead of an external store; and aged data tiers into a lakehouse format (Paimon/Iceberg) so history and the live stream are one table. In short, Kafka moves rows, Fluss stores a stream you can read narrowly, update, look up, and query historically — the analytics-shaped needs Kafka forces you to bolt other systems on for.

Log Table vs PrimaryKey Table — which do I use?

Use a Log Table for immutable facts — raw events like clicks, orders, or logs where every record is a new occurrence, not a change to a prior one. A Log Table is append-only, columnar, and consumed by subscribing from an offset, exactly like a topic but with column pruning. Use a PrimaryKey Table for mutable, keyed state — a dimension, the latest row per entity, or a materialized aggregate — where writing an existing key should update the row rather than append a duplicate. PrimaryKey Tables add three things a log cannot: upsert-in-place, a changelog (+I/-U/+U/-D) so downstream jobs react to changes with correct retract semantics, and O(1) point lookups by key that power streaming lookup joins. The rule of thumb: facts are Log Tables, state is PrimaryKey Tables.

How does Fluss make streaming reads cheaper (columnar / column pruning)?

Fluss stores stream records column-by-column in Apache Arrow, so a read that projects a few columns only touches those columns' data — projection pushdown. On a row-oriented log, a record is one serialized blob, so even reading two fields deserializes the whole record; the cost scales with the record's width. On Fluss the cost scales with the columns you actually project, so a narrow read of a wide, fat event moves a small fraction of the bytes. This compounds with predicate and partition pruning: filtering by a partition key (say date) lets a query skip whole partitions, so a narrow, filtered read over a wide, partitioned stream touches only a sliver of the data. The one discipline that realises this is projecting explicit columns — a SELECT * on a wide stream defeats pushdown and reintroduces the row tax on a columnar store.

How do streaming lookup joins work on Fluss?

A streaming lookup join enriches each row of a stream with the current matching row from a dimension, using Flink's temporal syntax JOIN dim FOR SYSTEM_TIME AS OF stream.proc_time ON stream.key = dim.key. When the dimension is a Fluss PrimaryKey Table keyed by the join key, each incoming event triggers a single O(1) point lookup served directly by Fluss — so the external HBase or Redis dimension store an ordinary Flink pipeline needs is removed entirely. Because the dimension is a PrimaryKey Table updated in place by upserts, the lookup always returns the freshest attributes, eliminating the stale-cache bugs a TTL'd external store suffers. The dimension being a first-class streaming table also means it stays in sync via its changelog, and — for high-volume both-sides joins — lookup/delta-style joins keep far less state than a dual-stream equi-join.

What is lakehouse tiering and Union Read (Paimon/Iceberg)?

Lakehouse tiering is Fluss's built-in path to history: set table.datalake.enabled = 'true' and a background tiering (compaction) service continuously moves aged data from Fluss's fast tier into a lakehouse table format — Apache Paimon by default, with Iceberg supported — as compacted columnar files on object storage. The freshest, seconds-old data stays in Fluss for low-latency streaming reads and lookups; only aged data compacts down. Union Read is how you query across both tiers as one logical table: a streaming read sees the fresh Fluss tail, a direct read of the table$lake (Paimon) tier sees compacted history, and a batch read performs a Union Read that combines history with the live tail for a complete, up-to-the-second result. The tiered data is a normal Paimon/Iceberg table, so Spark, Trino, or StarRocks can read the history directly — the lake is not locked inside Fluss.

Does Fluss replace Kafka, or sit alongside it?

It depends on the workload, and the senior answer resists a blanket claim. Fluss is a streaming-storage layer purpose-built under Flink for analytics that need columnar reads, primary-key updates, native lookups, and a lakehouse tier — so where your streaming is Flink-centric and hits those needs, Fluss can replace a Kafka-plus-external-KV-plus-warehouse stack and collapse it into one store. But Kafka remains a superb general-purpose event bus with a vast ecosystem of producers, connectors, and non-Flink consumers, so many architectures keep Kafka at the edge for ingestion and integration and use Fluss as the Flink-facing storage layer, ingesting from topics into Fluss tables. Choose per workload: Fluss where the value is analytics-shaped streaming storage with updates and a lakehouse; Kafka where the value is broad, language-agnostic event distribution — and often both, each doing what it is best at.

Practice on PipeCode

Drill the streaming practice library → for the log-store, changelog, and lookup-join problems that Log Tables and PrimaryKey Tables make concrete.
Rehearse serving patterns on the real-time analytics practice library → for the columnar-read, freshness, and Union-Read scenarios where the fresh-tier-vs-lakehouse decision earns its keep.
Sharpen the architecture axis with the system design practice library → for the table-modelling, bucketing, tiering, and stream-vs-lake trade-offs a real-time lakehouse must get right.
Stack the prerequisites against PipeCode's broader 450+ data-engineering catalogue to anchor the columnar-stream, primary-key-changelog, lookup-join, and tiering patterns against real graded inputs — Flink SQL, streaming state, and lakehouse design.

Lock in streaming-storage muscle memory

Docs explain Fluss's features. PipeCode drills explain the decision — when a `columnar stream` beats a row log, when a `PrimaryKey Table` and its changelog beat an append-only topic, when a `lookup join` against Fluss deletes the external store, and when `tiering` to Paimon with a union read beats a Kafka-plus-warehouse split. Pipecode.ai is Leetcode for Data Engineering — streaming-storage practice tuned for the production trade-offs senior data engineers actually face.

Practice streaming problems →
Practice real-time analytics problems →