Iceberg Materialized Views: Incremental Refresh on Open Table Format

iceberg materialized views are the pattern that finally decouples a materialized view from the engine that computed it — for a decade the MV story on the lakehouse was "pick your vendor and stay there" (Snowflake dynamic tables, Databricks materialized views, BigQuery MVs), and the moment you needed the same aggregate to serve a Trino ad-hoc query and a Spark backfill you were rewriting the definition twice, refreshing it twice, and paying for it twice. The Iceberg MV spec — proposed in 2024 and rolling out across Spark, Trino, and Snowflake through 2025 and 2026 — makes the materialized view a first-class object on the table format, not on the engine. One MV definition; one storage table; one refresh contract; any engine that speaks the spec can read the cached rows or fall back to the source.

This guide is the senior-DE walkthrough you wished existed the first time a lakehouse architect asked "why not just use dynamic tables?", or an interviewer probed "what does incremental refresh actually do to the snapshot chain?", or a platform team pushed back on "we can't have MVs — we're multi-engine." It walks through the four levers senior engineers need to master — the MV spec + metadata layout, the incremental refresh strategies (append-only trivial, upsert non-trivial), the query-planner rewrite contract that lets engines auto-substitute the MV for the source, and the migration patterns from proprietary MVs and dbt incremental models onto the open spec. 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 SQL practice library →, rehearse aggregation depth on the aggregation practice library →, and sharpen the streaming axis with the streaming practice library →.

---

On this page

• Why Iceberg MVs matter in 2026
• MV spec + metadata layout
• Incremental refresh strategies
• Query planning + freshness
• Migration patterns + interview signals
• Cheat sheet — Iceberg MV recipes
• Frequently asked questions
• Practice on PipeCode

---

1. Why Iceberg MVs matter in 2026

Four proprietary MV worlds, one open spec — the choice binds the whole lakehouse for years

The one-sentence invariant: iceberg materialized views are the answer to the multi-engine lakehouse's dirty secret — every engine ships a good-enough MV feature, none of those features are portable, and the moment a second engine needs to read the same cached aggregate you either duplicate the MV (double storage + double refresh cost) or give up on caching entirely (full source scans for every query). The Iceberg MV spec collapses the four proprietary worlds — dynamic tables, cloud-warehouse MVs, streaming materialized views, dbt incremental models — into one open object with one refresh contract and one freshness metadata surface. The pattern you standardise on in 2026 becomes the pattern every downstream engine (Spark, Trino, Snowflake, Athena, DuckDB) hard-codes assumptions about, so getting the mental model right before the migration is the single highest-leverage investment a lakehouse team can make.

The four levers interviewers actually probe.

• Freshness semantics. Is the MV eventually consistent, snapshot-consistent, or strictly-consistent with the source? Iceberg MVs are snapshot-consistent — the MV pins a source snapshot ID and only advances on refresh. Dynamic tables in Snowflake are eventually-consistent bounded by a TARGET_LAG. Interviewers probe this because it separates people who understand the trade-off (cheap + stale-bounded) from those who assume "MV" means "live view."
• Refresh cadence. How is refresh triggered — cron, event-driven, or continuous? Iceberg's spec is trigger-agnostic; the engine or an external scheduler drives refresh. Snowflake's dynamic tables and BigQuery MVs are engine-managed. Interviewers listen for you to name the trade-off (portable-but-you-schedule vs managed-but-vendor-locked).
• Query rewrite. Does the query planner auto-substitute the MV when a user query matches the MV's stored plan? Iceberg's spec exposes freshness metadata (last_refreshed_at, source snapshot IDs) that engines use to make the substitution decision. Interviewers probe rewrite because it separates the "MV as a table you SELECT from" candidate from the "MV as an optimiser hint the engine consults" candidate.
• Cost + latency trade-off. MVs trade refresh cost (recompute) for query cost (cheap lookup). The break-even is roughly (query_freq × source_scan_cost) > (refresh_freq × refresh_cost). Interviewers probe this because senior engineers know the answer is not "always MV" — some workloads are read-heavy enough to justify continuous refresh, others are read-light enough that MVs are pure overhead.

The 2026 reality — Iceberg MVs are shipping, but not everywhere yet.

• Iceberg MV spec was proposed in 2024 (Iceberg Improvement Proposal / spec extension), and reference implementations landed through 2025. Spark 4.0 supports CREATE MATERIALIZED VIEW against Iceberg catalogs; Trino 450+ ships MV read + refresh; Snowflake's Iceberg-table integration exposes MVs over external Iceberg tables. AWS Athena and Google BigQuery are following.
• Proprietary alternatives still dominate greenfield warehouses. Snowflake dynamic tables ship a declarative refresh model with TARGET_LAG and automatic incremental computation. Databricks materialized views on Unity Catalog run Photon-backed incremental refresh with a similar model. BigQuery materialized views offer smart tuning and automatic query rewrite. Each is excellent inside its silo.
• The open-spec upside is not that Iceberg MVs are faster — vendor MVs are usually faster inside their engine. The upside is that the MV storage is shared. When your Spark job refreshes the MV, your Trino ad-hoc query benefits. When your Snowflake warehouse materialises an aggregate, your Athena reader sees it fresh. One refresh cost; many readers.
• The 2026 wart is that the spec is young. Not every engine implements every refresh strategy. Full refresh is universal; incremental refresh on append-only sources works everywhere; incremental refresh on upsert/delete sources requires engines that understand Iceberg row-level deletes (equality + positional). Feature-detect before you commit.

What interviewers listen for.

• Do you name all four MV worlds (Iceberg MV, dynamic tables, cloud-warehouse MVs, dbt incremental) without prompting? — senior signal.
• Do you say "snapshot-consistent, not strictly-consistent" when asked about freshness? — required answer.
• Do you distinguish refresh cadence (cron vs event vs continuous) from refresh strategy (full vs incremental)? — senior signal.
• Do you name query rewrite as the mechanism that turns an MV from a "you SELECT from it" table into an "the optimiser picks it" cache? — senior signal.
• Do you push back on "just use dynamic tables" with the multi-engine question — "does Trino also need this aggregate?" — senior signal.

Worked example — the four-lever comparison table

Detailed explanation. The single most useful artifact for an Iceberg MV interview is a memorised 4×4 comparison table across the four MV worlds. Every senior lakehouse discussion converges on this table within the first ten minutes; having it in your head is what separates a fluent answer from a stumbling one. Walk through building the table for a hypothetical daily_revenue aggregate that needs to serve both a Snowflake dashboard and a Trino ad-hoc analyst.

• Source. sales.orders — an Iceberg table on S3 partitioned by order_date, ~500 M rows, ~50 M new rows/day.
• Aggregate. SELECT order_date, SUM(total_cents) FROM sales.orders GROUP BY 1.
• Consumers. Snowflake dashboard (needs 5-minute freshness), Trino analyst (tolerates 1-hour staleness).
• Refresh cadence target. 5 minutes at the tightest.

Question. Build the four-MV-world comparison for daily_revenue and pick which world each consumer should subscribe to.

Input.

MV world	Storage location	Refresh model	Query rewrite	Multi-engine?
Iceberg MV	Iceberg storage table (S3)	engine-driven or external scheduler	spec-defined; engine-implemented	yes
Snowflake dynamic table	Snowflake internal	declarative TARGET_LAG	automatic within Snowflake	no
Databricks MV	Delta / Unity Catalog	declarative refresh interval	automatic within Databricks	no
BigQuery MV	BigQuery internal	smart tuning, incremental	automatic within BigQuery	no
dbt incremental model	any (usually warehouse table)	scheduled by dbt runs	none (it's a table)	depends on warehouse

Code.

-- Iceberg MV (Spark 4.0 / Trino 450 syntax — subject to spec finalisation)
CREATE MATERIALIZED VIEW analytics.daily_revenue
  TBLPROPERTIES (
    'iceberg.mv.refresh.mode'      = 'incremental',
    'iceberg.mv.freshness.max'     = '5 minutes',
    'iceberg.mv.storage.location'  = 's3://lake/mvs/daily_revenue/'
  )
AS
SELECT order_date,
       SUM(total_cents) AS revenue_cents,
       COUNT(*)         AS order_count
FROM   sales.orders
GROUP  BY order_date;

-- Snowflake dynamic table equivalent (proprietary, single-engine)
CREATE OR REPLACE DYNAMIC TABLE analytics.daily_revenue
  TARGET_LAG = '5 minutes'
  WAREHOUSE  = compute_wh
AS
SELECT order_date, SUM(total_cents) AS revenue_cents, COUNT(*) AS order_count
FROM   sales.orders
GROUP  BY order_date;

Step-by-step explanation.

1. The Iceberg MV DDL creates a view-metadata JSON pointer plus a backing storage table in s3://lake/mvs/daily_revenue/. Any engine that speaks the Iceberg MV spec — Spark, Trino, Snowflake-with-Iceberg-tables, Athena Iceberg — can read the storage table directly, or trigger a refresh, or use the freshness metadata to make a query-rewrite decision.
2. The Snowflake dynamic table stores the aggregate inside Snowflake's proprietary storage. The TARGET_LAG = '5 minutes' clause is declarative; Snowflake decides when to refresh, how much to refresh incrementally, and whether to serve queries from the DT or fall back to the source. Excellent inside Snowflake; invisible to Trino.
3. For the daily_revenue case, the Iceberg MV wins the moment a second engine needs the aggregate. One storage table, one refresh cost, two readers. Dynamic tables would force a second aggregate in Trino's own catalog — double refresh cost, double storage, drift risk.
4. The refresh cadence for both is 5 minutes at the tightest, driven by the dashboard SLA. For the Iceberg MV, "5 minutes" is a freshness contract the scheduler must honour (iceberg.mv.freshness.max); the actual refresh runs on cron / Airflow / streaming job. For the dynamic table, Snowflake schedules the refresh internally.
5. In practice, most 2026 lakehouse designs pick Iceberg MVs as the primary MV surface when multi-engine reads are on the roadmap, then layer proprietary MVs on top for engine-specific hot paths that need sub-second latency inside one engine.

Output.

Consumer	Recommended MV world	Why
Snowflake dashboard (5-min freshness)	Iceberg MV	multi-engine; shared with Trino
Trino ad-hoc analyst (1-hour tolerance)	Iceberg MV	reuses the Snowflake-triggered refresh
Single-engine BI (Snowflake only)	dynamic table	tighter integration; sub-second query rewrite
dbt-managed daily aggregate	Iceberg MV (via dbt Iceberg adapter)	replaces is_incremental() boilerplate

Rule of thumb. Never pick an MV world based on "which vendor we already pay." Pick it based on (multi-engine reach × refresh cadence × freshness tolerance × cost). If more than one engine reads the aggregate, Iceberg MV wins by default. If exactly one engine and you value the vendor's optimiser, proprietary MVs win.

Worked example — what interviewers actually probe

Detailed explanation. The senior lakehouse interview on materialized views has a predictable structure: the interviewer opens with a scenario ("your Trino dashboard is slow — the aggregate scan takes 40 seconds"), then progressively narrows to test whether you know the four levers. Candidates who name the pattern in sentence one and cite the freshness contract score highest; candidates who describe "we could just query it once and cache it in Redis" score lowest. Walk through the interview grading rubric.

• Ambiguous opener. "How would you speed up this repeated aggregate query?" — invites you to name the MV pattern and cite freshness.
• Follow-up 1. "We're also serving this from Snowflake — how do we not compute it twice?" — probes multi-engine axis.
• Follow-up 2. "How stale is stale enough?" — probes freshness contract.
• Follow-up 3. "How does the planner know when to use the MV?" — probes query rewrite.
• Follow-up 4. "What if the source table has upserts and deletes?" — probes incremental refresh strategy.

Question. Draft a 5-minute senior Iceberg MV answer that covers all four levers without waiting to be asked.

Input.

Interview signal	Weak answer	Senior answer
Pattern named	"we'd cache it"	"Iceberg materialized view with a 5-minute freshness contract"
Multi-engine	"Snowflake has its own thing"	"one Iceberg MV storage table; both engines read it"
Freshness semantics	"we'd refresh it hourly"	"snapshot-consistent; MAX_STALENESS = 5m; refresh triggered by source snapshot advance"
Query rewrite	"we'd query it directly"	"engine planner auto-substitutes the MV based on freshness metadata"
Refresh strategy	"we'd rebuild it"	"incremental via source snapshot diff; append-only fast path; upsert path uses row-level deletes"

Code.

Senior Iceberg MV answer template (5 minutes)
=============================================

Minute 1 — name the pattern up front
  "I'd define this aggregate as an Iceberg materialized view against the
   sales.orders source, with a MAX_STALENESS contract of 5 minutes."

Minute 2 — multi-engine axis
  "The Iceberg MV's storage table lives in S3 (or Nessie / Glue), so
   Snowflake, Trino, Spark, and Athena can all read the cached rows.
   One refresh cost; four readers. Dynamic tables and BigQuery MVs
   would force per-engine duplication."

Minute 3 — freshness contract
  "Freshness is snapshot-consistent, not strictly-consistent. The MV
   pins a source snapshot ID at last refresh; a MAX_STALENESS of 5m
   means the planner treats the MV as stale if last_refreshed_at is
   older than 5 minutes. Users get 'at most 5 minutes stale' as a
   contract, not an SLA."

Minute 4 — query rewrite
  "The engine planner reads view-metadata.json, compares the MV's
   stored logical plan against the incoming query, and substitutes
   automatically when the plan matches and freshness is within
   tolerance. Fallback is a source scan — same result, higher cost."

Minute 5 — refresh strategy
  "For append-only sources like an events stream, incremental refresh
   reads the source snapshot diff between last_refreshed_snapshot and
   current — O(delta), not O(table). For upsert/delete sources it uses
   Iceberg row-level deletes to reconcile changed rows. Refresh runs
   on Airflow cron or on-demand from the engine."

Step-by-step explanation.

1. Minute 1 is the crucial framing. Naming the pattern immediately — "Iceberg materialized view with a MAX_STALENESS contract" — signals you're a decision-maker, not a task-runner. Weak candidates dive into tools ("we'd use Trino's MV feature and…") before naming the pattern.
2. Minute 2 addresses the multi-engine axis before the interviewer asks. This preempts the trap where you commit to a vendor MV and later admit you can't share it. Naming the open-spec advantage up front is senior signal.
3. Minute 3 is the freshness probe. The distinction between snapshot-consistency (MV pins a snapshot; refresh advances the pin) and strict-consistency (MV always reflects the latest source write) is the single most-tested concept in senior MV interviews. Say "snapshot-consistent" out loud.
4. Minute 4 is the query rewrite argument. Weak candidates treat the MV as a table you SELECT ... FROM mv from directly; senior candidates treat it as an optimiser hint the planner consults automatically. Cite view-metadata.json and the freshness contract.
5. Minute 5 covers refresh strategy — the reliability + cost axis. Naming the source-snapshot-diff mechanism for append-only and the row-level-delete mechanism for upsert shows you've read the spec, not just the marketing.

Output.

Grading criterion	Weak score	Senior score
Names pattern in minute 1	rare	mandatory
Names multi-engine reach	rare	required
Names freshness semantics	occasional	mandatory
Names query rewrite	rare	senior signal
Names incremental refresh strategy	rare	senior signal

Rule of thumb. The senior Iceberg MV answer is a 5-minute monologue that covers all four levers without waiting for the follow-ups. Rehearse it once; deploy it every time.

Worked example — the "pick the MV world" decision tree

Detailed explanation. Given a new caching requirement, the senior architect runs a 4-question decision tree in their head. Codifying the tree makes the interview answer reproducible: any interviewer can hand you a scenario and you can walk the tree out loud. Walk through the tree with three canonical scenarios: multi-engine lakehouse, single-vendor warehouse, streaming freshness needs.

• Q1. Do more than one query engine need this aggregate? → yes = Iceberg MV; no = go to Q2.
• Q2. Is this a single-vendor warehouse where the vendor MV is excellent? → yes = vendor MV (dynamic table / BQ MV); no = go to Q3.
• Q3. Is freshness continuous / sub-minute? → yes = streaming MV (Materialize / RisingWave / Snowflake DT with tight lag); no = go to Q4.
• Q4. Is this a batch aggregate that fits the dbt workflow? → yes = dbt incremental model; no = Iceberg MV (default).

Question. Walk the decision tree for three scenarios and record the MV world each ends up with.

Input.

Scenario	Q1 (multi-engine?)	Q2 (single-vendor?)	Q3 (streaming?)	Q4 (dbt-fit?)
Lakehouse Snowflake + Trino	yes	—	—	—
Snowflake-only BI	no	yes	—	—
Sub-second streaming dashboard	no	—	yes	—
Nightly dbt-owned aggregate	no	—	no	yes

Code.

# Decision-tree helper (illustrative)
def pick_mv_world(multi_engine: bool,
                  single_vendor_good: bool,
                  needs_streaming: bool,
                  fits_dbt_flow: bool) -> str:
    """Return the MV world for a caching requirement."""
    if multi_engine:
        return "Iceberg MV"
    if single_vendor_good:
        return "vendor MV (dynamic table / BQ MV / Databricks MV)"
    if needs_streaming:
        return "streaming MV (dynamic table tight lag / Materialize / RisingWave)"
    if fits_dbt_flow:
        return "dbt incremental model"
    return "Iceberg MV (default)"


print(pick_mv_world(True,  False, False, False))
# → 'Iceberg MV'

print(pick_mv_world(False, True,  False, False))
# → 'vendor MV (dynamic table / BQ MV / Databricks MV)'

print(pick_mv_world(False, False, True,  False))
# → 'streaming MV (dynamic table tight lag / Materialize / RisingWave)'

print(pick_mv_world(False, False, False, True))
# → 'dbt incremental model'

Step-by-step explanation.

1. Scenario 1 — Snowflake + Trino lakehouse. Q1 fires: two engines need the aggregate. Iceberg MV is the answer. This is the greenfield 2026 default when the lakehouse architecture is multi-engine from day one.
2. Scenario 2 — pure Snowflake BI. Q1 = no, Q2 = yes. Snowflake dynamic tables ship excellent query rewrite and automatic incremental refresh; the vendor MV wins inside its silo. Iceberg MVs would be strictly worse here — extra external storage, extra refresh coordination, no rewrite advantage.
3. Scenario 3 — sub-second streaming dashboard. Q3 fires. Iceberg MVs are snapshot-consistent with refresh cadence measured in minutes; they cannot serve a sub-second freshness contract. Streaming MV engines (dynamic table with tight lag, Materialize, RisingWave) win.
4. Scenario 4 — nightly dbt aggregate. Q4 fires. If the aggregate is already a dbt model with is_incremental() logic and the team owns the dbt DAG, the migration cost of moving to Iceberg MV may not clear the benefit bar. Stay on dbt until multi-engine reach becomes a requirement.
5. If none of Q1-Q4 pass cleanly, the default falls back to Iceberg MV — its overhead is low and its option value (future engine additions) is high.

Output.

Scenario	MV world	Rationale
Snowflake + Trino lakehouse	Iceberg MV	one storage; two readers
Snowflake-only BI	dynamic table	vendor rewrite is best
Sub-second dashboard	streaming MV	freshness beats openness
Nightly dbt aggregate	dbt incremental	migration cost > benefit

Rule of thumb. The four-question decision tree is a whiteboard-friendly answer. Practice walking it end-to-end so an interviewer can hand you any scenario and get a world name in under 60 seconds.

Senior interview question on Iceberg MV selection

A senior interviewer often opens with: "You inherit a Snowflake-only BI stack where the daily revenue aggregate is a dynamic table refreshing every 5 minutes. The company just added Trino for ad-hoc analytics and everyone is scanning the raw 500-million-row sales.orders table because the dynamic table is invisible to Trino. Walk me through the migration to an Iceberg materialized view, the refresh contract you'd set, and how you'd defend against the freshness regression."

Solution Using Iceberg MV with a MAX_STALENESS contract, external Airflow refresh, and dual-engine readers

-- Step 1 — define the Iceberg MV against the shared sales.orders source
CREATE MATERIALIZED VIEW analytics.daily_revenue
  TBLPROPERTIES (
    'iceberg.mv.refresh.mode'      = 'incremental',
    'iceberg.mv.freshness.max'     = '5 minutes',
    'iceberg.mv.storage.location'  = 's3://lake/mvs/daily_revenue/',
    'write.format.default'         = 'parquet',
    'write.parquet.compression-codec' = 'zstd'
  )
AS
SELECT order_date,
       SUM(total_cents) AS revenue_cents,
       COUNT(*)         AS order_count,
       COUNT(DISTINCT customer_id) AS unique_customers
FROM   sales.orders
GROUP  BY order_date;

# Step 2 — Airflow refresh DAG (external scheduler; portable across engines)
from airflow.decorators import dag, task
from datetime import datetime, timedelta

@dag(schedule="*/5 * * * *",  # every 5 minutes to honour MAX_STALENESS
     start_date=datetime(2026, 7, 19),
     catchup=False,
     max_active_runs=1)  # never overlap refreshes
def refresh_daily_revenue_mv():

    @task
    def refresh():
        # Any engine that implements REFRESH MATERIALIZED VIEW works;
        # here we use Trino because it has the mature Iceberg MV impl
        import trino
        conn = trino.dbapi.connect(host="trino.internal", user="mv_refresher")
        with conn.cursor() as cur:
            cur.execute("REFRESH MATERIALIZED VIEW analytics.daily_revenue")
        return "ok"

    refresh()

dag = refresh_daily_revenue_mv()

-- Step 3 — Snowflake side (external Iceberg table pointing at the MV storage)
CREATE OR REPLACE ICEBERG TABLE analytics.daily_revenue
  EXTERNAL_VOLUME = 'lake_vol'
  CATALOG         = 'glue_catalog'
  BASE_LOCATION   = 'mvs/daily_revenue/';

-- Snowflake queries the Iceberg MV as if it were a normal table
SELECT order_date, revenue_cents
FROM   analytics.daily_revenue
WHERE  order_date >= current_date - 30;

-- Step 4 — Trino side (native Iceberg MV support)
SELECT order_date, revenue_cents, order_count
FROM   iceberg.analytics.daily_revenue
WHERE  order_date >= current_date - 30;
-- Trino's planner rewrites this against the MV automatically
-- when analytics.daily_revenue.is_stale = false and the plan matches

Step-by-step trace.

Step	Before (Snowflake dynamic table)	After (Iceberg MV)
Storage	Snowflake proprietary	S3 (Iceberg storage table)
Refresh trigger	Snowflake managed	Airflow cron every 5 min
Snowflake read path	direct DT query	external Iceberg table
Trino read path	full source scan (invisible to DT)	MV storage table (~1000× cheaper)
Freshness contract	TARGET_LAG = 5 min	MAX_STALENESS = 5 min
Refresh cost	~50 credits/day (Snowflake compute)	~1 vCPU-hour/day (Trino refresh)
Multi-engine	no	yes

After the migration, the daily_revenue aggregate is materialised once every 5 minutes into an Iceberg storage table on S3. Snowflake reads it via an external Iceberg table binding; Trino reads it natively; both see the same rows within the freshness contract. The dynamic table is dropped after a 30-day parallel-run window in which reconciliation confirms the numbers match.

Output:

Metric	Before	After
Trino query latency (30-day scan)	40 s (full source)	200 ms (MV lookup)
Snowflake dashboard latency	300 ms	300 ms (unchanged)
Daily refresh cost	~50 Snowflake credits	~1 vCPU-hour (Trino)
Storage cost	2× (DT + source in Snowflake)	1× (Iceberg MV on S3)
Multi-engine reach	Snowflake only	Snowflake + Trino + future Spark
Freshness contract	TARGET_LAG = 5 min	MAX_STALENESS = 5 min

Why this works — concept by concept:

• Iceberg MV storage table — the MV's rows live in an Iceberg table under s3://lake/mvs/daily_revenue/. Any engine that reads Iceberg reads the MV; no per-engine duplication. The storage table is snapshot-managed exactly like a normal Iceberg table, so time-travel and rollback work identically.
• MAX_STALENESS = 5 minutes — the freshness contract. Engines that support MV query rewrite consult this before deciding to serve from the MV; if the MV is older than 5 minutes, they fall back to the source scan. Users get a clear "at most 5 minutes stale" guarantee.
• External Airflow scheduler — because the Iceberg MV spec is trigger-agnostic, refresh is driven by an external cron. max_active_runs=1 prevents overlapping refreshes; schedule='*/5 * * * *' honours the freshness contract. Any engine (Trino here) can execute the REFRESH MATERIALIZED VIEW statement.
• Snowflake external Iceberg table — Snowflake reads the MV storage table via its Iceberg-tables feature. The dashboard query executes against the cached rows; refresh happens outside Snowflake, so Snowflake credits are only spent on serving reads.
• Cost — one refresh (a single Trino query every 5 minutes reading the source snapshot diff) instead of two (one Snowflake DT + one Trino cache), 1× storage instead of 2×, and the future option to add Spark / Athena / DuckDB readers at zero incremental cost. The eliminated cost is the Trino full source scan on every dashboard query — 40 seconds becomes 200 milliseconds.

SQL
Topic — sql
SQL materialized view and aggregate problems

Practice →

Aggregation Topic — aggregation Aggregation problems on lakehouse tables

Practice →

---

2. MV spec + metadata layout

An Iceberg MV is a view-metadata.json + a stored logical plan + a snapshot-ID mapping + a backing storage table — four pieces, one open contract

The mental model in one line: an Iceberg materialized view is not one file — it is four coordinated pieces that together describe (a) the logical SQL that defines the aggregate, (b) which source snapshots produced the current cached rows, (c) freshness metadata engines use for query rewrite, and (d) the physical Iceberg storage table that holds the cached rows — and understanding those four pieces is the difference between using MVs and debugging MVs at 2 a.m. when a refresh silently produced the wrong rows. Every senior lakehouse engineer must be able to open a view-metadata.json, point at each field, and explain why it's there.

The four pieces of an Iceberg MV.

• View metadata. A view-metadata.json file in the MV's metadata directory. Contains the current schema, the current view version, the list of representations (SQL dialects), and — crucially for MVs — the materialization pointer to the backing storage table plus the freshness metadata (last-refreshed-timestamp-ms, source-snapshot-ids, is-stale flag).
• Stored logical plan. The MV's defining SQL, stored as a SQL string (per-dialect representation) and optionally as a serialised logical plan (engine-portable IR). The SQL string is the human-readable definition; the logical plan is what the query rewrite optimiser matches against incoming queries.
• Source snapshot mapping. A list of (source_table_ref, snapshot_id) pairs captured at the last refresh. This is the "what did we materialise from?" record — engines use it to compute the source snapshot diff for incremental refresh and to decide whether the MV is stale.
• Backing storage table. A regular Iceberg table living at a separate metadata_location and base_location. It stores the actual cached rows in Parquet/ORC data files with the usual Iceberg manifest and snapshot machinery. From the outside it looks like a normal Iceberg table; from the MV's perspective it is the materialization.

The view-metadata.json shape — what every senior architect memorises.

• format-version. Currently 1 for views; MVs extend this with materialization fields.
• current-version-id. Which of the versions in the versions list is current. Older versions are retained for rollback.
• versions[].representations[]. Per-dialect SQL strings — spark-sql, trino-sql, snowflake-sql — so the same MV can be defined once and executed in any engine that speaks one of those dialects.
• versions[].summary. Human-readable metadata about the version — who created it, when, what the operation was.
• materialization. The MV-specific extension. Contains storage-table-uuid (points at the backing Iceberg table), last-refreshed-timestamp-ms, source-snapshot-ids (map of source table → snapshot ID at refresh), and is-stale (boolean, computed by comparing source current snapshots to source-snapshot-ids).

The source-snapshot mapping — why it's the load-bearing piece.

• What it is. At refresh time, the engine records the current snapshot ID of every source table the MV depends on. That mapping is stored in materialization.source-snapshot-ids.
• Why it matters for staleness. An MV is "stale" if any source table has a newer snapshot than the one recorded at last refresh. is-stale = ANY(source_current_snapshot_id != source-snapshot-ids[source_table]). This lets any engine compute staleness without running the MV's query.
• Why it matters for incremental refresh. The refresh planner reads source-snapshot-ids (the from snapshot) and the source's current snapshot (the to snapshot), computes the snapshot diff for each source, and produces an incremental refresh plan that touches only the changed rows.
• Why it matters for reproducibility. Given the source-snapshot mapping, any engine can re-produce the exact rows the MV held at any past refresh. This is the audit trail for cached aggregates — invaluable when a dashboard number is questioned.

The storage table separation — one MV, one storage table, distinct identities.

• Two catalog entries. The MV registers as a view in the catalog (analytics.daily_revenue); the storage table registers separately (often auto-named analytics.daily_revenue$storage or hidden). Engines resolve the view first, follow the pointer, then read the storage table.
• Why separate. The view-metadata (schema, logical plan, freshness) is small and mutable; the storage table (data files, manifests) is large and append-mostly. Separating them lets refresh replace the storage table's snapshot without touching view-metadata, and lets the view-metadata evolve (new dialect representation, new version) without rewriting the storage.
• Storage table lifecycle. Refresh writes a new snapshot to the storage table (append for incremental append-only, overwrite for full refresh, merge for upsert incremental). The view-metadata is then updated in one atomic write to point at the new source-snapshot mapping and last-refreshed-timestamp-ms.

Common interview probes on the MV spec.

• "What's inside view-metadata.json for an MV?" — required answer: versions + representations + materialization pointer + source-snapshot mapping + freshness fields.
• "How does an engine know an MV is stale?" — compare source-snapshot-ids to each source's current snapshot ID; any mismatch = stale.
• "Where does the MV actually store its rows?" — a separate Iceberg storage table pointed to by materialization.storage-table-uuid.
• "What survives a rollback?" — the storage table retains its snapshot history; view-metadata retains version history. Both roll back independently.

Worked example — inspecting view-metadata.json for a live MV

Detailed explanation. The canonical exercise: create an Iceberg MV, force a refresh, then open the on-disk metadata to see exactly what got written. This is the exercise every senior lakehouse engineer runs once during onboarding — never again wonder what an MV "really is." Walk through the full end-to-end.

• MV. analytics.daily_revenue — the sales.orders aggregate from Section 1.
• Storage. Iceberg storage table at s3://lake/mvs/daily_revenue/.
• Inspection. Read view-metadata.json from the MV's metadata directory using PyIceberg (engine-agnostic).

Question. Create the MV, refresh it, and print the key fields of view-metadata.json including the source-snapshot mapping and the storage-table pointer.

Input.

Object	Value
Catalog	Glue / REST / Nessie
MV	analytics.daily_revenue
Source	sales.orders (Iceberg)
Storage	s3://lake/mvs/daily_revenue/
Inspector	PyIceberg 0.9+

Code.

-- 1. Define the MV (Spark 4.0 / Trino syntax)
CREATE MATERIALIZED VIEW analytics.daily_revenue
  TBLPROPERTIES (
    'iceberg.mv.refresh.mode'      = 'incremental',
    'iceberg.mv.freshness.max'     = '5 minutes',
    'iceberg.mv.storage.location'  = 's3://lake/mvs/daily_revenue/'
  )
AS
SELECT order_date, SUM(total_cents) AS revenue_cents
FROM   sales.orders
GROUP  BY order_date;

-- 2. Force a full refresh so metadata is populated
REFRESH MATERIALIZED VIEW analytics.daily_revenue;

# 3. Inspect view-metadata.json using PyIceberg (engine-agnostic)
from pyiceberg.catalog import load_catalog
import json

catalog = load_catalog("glue")

# Load the view (not the storage table)
view = catalog.load_view("analytics.daily_revenue")

# Print current version + materialization pointer + freshness
metadata = view.metadata
current  = metadata.versions[metadata.current_version_id]

print("Current version:", metadata.current_version_id)
print("Representations:")
for rep in current.representations:
    print(f"  - {rep.dialect}: {rep.sql[:80]}...")

# Materialization pointer (MV-specific extension)
mat = metadata.materialization
print("\nMaterialization:")
print(f"  storage_table_uuid       = {mat.storage_table_uuid}")
print(f"  last_refreshed_ms        = {mat.last_refreshed_timestamp_ms}")
print(f"  source_snapshot_ids      = {json.dumps(mat.source_snapshot_ids, indent=4)}")
print(f"  is_stale                 = {mat.is_stale()}")

# Storage table lives at a separate location
storage = catalog.load_table_by_uuid(mat.storage_table_uuid)
print(f"\nStorage table: {storage.metadata.location}")
print(f"  current_snapshot_id      = {storage.metadata.current_snapshot_id}")
print(f"  data files               = {len(storage.scan().plan_files())}")

Step-by-step explanation.

1. Step 1 creates the MV with three critical properties: refresh mode (incremental), freshness max (5 minutes), and storage location (S3 path). The engine writes view-metadata.json with an empty materialization block (no refresh has happened yet), plus a versions[0] entry containing the SQL string.
2. Step 2's REFRESH MATERIALIZED VIEW triggers the first refresh. The engine executes the defining query, writes the result rows into the storage table at the configured S3 location (creating a first snapshot on that table), then updates the view-metadata's materialization block with the source-snapshot IDs, the storage-table UUID, and last_refreshed_timestamp_ms = now().
3. Step 3 uses PyIceberg to load the view. metadata.versions[current_version_id] gives the current logical definition — each representation is a per-dialect SQL string (Spark, Trino, Snowflake) so any of those engines can execute the defining SQL locally without a translation layer.
4. metadata.materialization.source_snapshot_ids is a dict like {"sales.orders": 8123456789012345678} — the source table name mapped to the snapshot ID at refresh. is_stale() compares this to the source's current snapshot ID; if they differ, the MV is stale.
5. Loading the storage table via storage_table_uuid gives access to the actual data files. The storage table has its own snapshot history — one snapshot per refresh — and can be time-travelled independently, so you can query "what did the MV hold last Tuesday?" without re-running any refresh.

Output.

Current version: 1
Representations:
  - spark-sql: SELECT order_date, SUM(total_cents) AS revenue_cents FROM ...
  - trino-sql: SELECT order_date, SUM(total_cents) AS revenue_cents FROM ...

Materialization:
  storage_table_uuid       = 3f2b7c1a-9d4e-4f2a-b5c6-8a1e3d4b7c9f
  last_refreshed_ms        = 1721390400000
  source_snapshot_ids      = {
        "sales.orders": 8123456789012345678
    }
  is_stale                 = False

Storage table: s3://lake/mvs/daily_revenue/
  current_snapshot_id      = 5432167890123456789
  data files               = 3

Rule of thumb. For any Iceberg MV deployment, run this inspection script during onboarding — once you can point at the source_snapshot_ids field and explain how it drives is_stale(), you understand the spec well enough to debug production. Store this script in your team's runbook.

Worked example — CTAS + snapshot pinning for a first-principles MV

Detailed explanation. Not every engine ships full CREATE MATERIALIZED VIEW DDL yet. When you're bootstrapping on an engine that only speaks Iceberg tables and views, you can build a first-principles MV: a regular Iceberg storage table + a regular Iceberg view + a manually-maintained source-snapshot mapping stored as table properties. This is the "MV kit" that shipped before native DDL support in some engines. Understanding it helps you appreciate what the spec automates.

• Storage. A regular Iceberg table analytics.daily_revenue_storage created via CTAS.
• View. A regular Iceberg view analytics.daily_revenue selecting from the storage table.
• Snapshot pinning. Table properties on the storage table record the source snapshot IDs at last refresh.

Question. Bootstrap an MV by hand using CTAS + view + snapshot-pinning properties on an engine that doesn't yet support native MV DDL.

Input.

Component	Value
Storage table	analytics.daily_revenue_storage (Iceberg)
View	analytics.daily_revenue (Iceberg view)
Source snapshot property	mv.source.sales.orders.snapshot_id
Refresh property	mv.last_refreshed_ms

Code.

-- 1. CTAS the storage table + record the source snapshot ID as a property
CREATE TABLE analytics.daily_revenue_storage
  TBLPROPERTIES (
    'mv.source.sales.orders.snapshot_id' =
      CAST((SELECT current_snapshot_id FROM sales.orders$snapshots
            ORDER BY committed_at DESC LIMIT 1) AS STRING),
    'mv.last_refreshed_ms'  = CAST(unix_millis(now()) AS STRING),
    'write.format.default'  = 'parquet'
  )
AS
SELECT order_date, SUM(total_cents) AS revenue_cents
FROM   sales.orders
GROUP  BY order_date;

-- 2. Create a view over the storage table (this is what users query)
CREATE VIEW analytics.daily_revenue AS
SELECT order_date, revenue_cents
FROM   analytics.daily_revenue_storage;

# 3. Manual refresh — swap the storage table's rows + update the pinning properties
from pyiceberg.catalog import load_catalog

catalog = load_catalog("glue")
source  = catalog.load_table("sales.orders")
storage = catalog.load_table("analytics.daily_revenue_storage")

# Read the previous pin
prev_pin = int(storage.properties.get("mv.source.sales.orders.snapshot_id", "0"))
curr_source = source.metadata.current_snapshot_id

if prev_pin == curr_source:
    print("MV is fresh; no refresh needed")
else:
    # Full refresh — for an incremental version, see Section 3
    with storage.update_snapshot().overwrite() as tx:
        # In practice this is executed by Spark/Trino; shown here as pseudo-code
        tx.overwrite_with_query("""
            SELECT order_date, SUM(total_cents) AS revenue_cents
            FROM   sales.orders
            GROUP  BY order_date
        """)

    # Update the pinning + refresh properties in one atomic property update
    storage.update_properties({
        "mv.source.sales.orders.snapshot_id": str(curr_source),
        "mv.last_refreshed_ms":               str(int(time.time() * 1000))
    })
    print(f"Refreshed; new pin = {curr_source}")

Step-by-step explanation.

1. Step 1's CTAS creates the storage table with three critical properties: the pinned source snapshot ID (mv.source.sales.orders.snapshot_id), the last-refreshed timestamp (mv.last_refreshed_ms), and the storage format (Parquet). The subquery captures the current snapshot ID atomically as part of the CTAS transaction so there's no window where the pin lags the data.
2. Step 2 creates a thin view over the storage table. Users query the view; the view resolves to the storage table transparently. This gives you the API surface of an MV (SELECT ... FROM analytics.daily_revenue) even before your engine ships native MV DDL.
3. Step 3's refresh logic first reads the previous pin, compares it against the source's current snapshot, and skips the refresh if they match. This is the manual equivalent of is_stale() — free-riding on Iceberg table properties instead of a spec-blessed field.
4. When a refresh is needed, the storage table is overwritten with the fresh aggregate and the properties are updated in a single call. update_properties is atomic — either both the new pin and the new refresh timestamp land, or neither does.
5. The trade-off vs native MV DDL: no query rewrite (users must explicitly SELECT from the view; the planner doesn't auto-substitute), no cross-engine standardisation (property names are your convention), no per-dialect representations. But it works on any Iceberg engine, today, and shows you exactly what the spec is automating for you.

Output.

Property	Value after CTAS	Value after refresh
mv.source.sales.orders.snapshot_id	8123... (current source snapshot)	8199... (new source snapshot)
mv.last_refreshed_ms	1721390400000	1721390700000
storage rows	90 (one per date)	91 (new date added)
view visible to users	yes	yes (unchanged surface)

Rule of thumb. For any engine that doesn't yet support native CREATE MATERIALIZED VIEW, the CTAS + view + snapshot-pinning-property pattern gives you 80% of the MV story with 0% of the vendor lock-in. Migrate to native DDL when your primary engine supports it, but do not wait for it if your workload needs caching now.

Worked example — dual-dialect representations for cross-engine execution

Detailed explanation. The Iceberg view spec (and by extension the MV spec) supports multiple SQL representations per view version. This is how a single MV definition can be executed against Spark, Trino, or Snowflake — each engine reads the representation for its own dialect. Understanding the multi-representation contract is essential for lakehouse teams that run more than one engine. Walk through creating an MV with dual representations and verifying both parse.

• The MV. analytics.weekly_revenue using a WINDOW function that differs slightly between dialects.
• Representations. One for spark-sql, one for trino-sql; identical semantics, dialect-appropriate syntax.
• Validation. Both engines execute the MV; results must match.

Question. Create an MV with two dialect representations and demonstrate that both engines execute it correctly.

Input.

Component	Spark-side	Trino-side
Date arithmetic	date_trunc('week', d)	date_trunc('week', d)
Window function	SUM(...) OVER (ORDER BY week)	SUM(...) OVER (ORDER BY week)
Cast syntax	CAST(x AS BIGINT)	CAST(x AS BIGINT)
Result schema	week: date, revenue_cents: bigint, rolling_4wk: bigint	(identical)

Code.

// Excerpt from view-metadata.json — versions[current].representations[]
{
  "version-id": 1,
  "timestamp-ms": 1721390400000,
  "schema-id": 1,
  "summary": {
    "operation": "create",
    "engine-name": "spark",
    "engine-version": "4.0.0"
  },
  "representations": [
    {
      "type": "sql",
      "dialect": "spark-sql",
      "sql": "SELECT date_trunc('week', order_date) AS week, SUM(total_cents) AS revenue_cents, SUM(SUM(total_cents)) OVER (ORDER BY date_trunc('week', order_date) ROWS BETWEEN 3 PRECEDING AND CURRENT ROW) AS rolling_4wk FROM sales.orders GROUP BY 1"
    },
    {
      "type": "sql",
      "dialect": "trino-sql",
      "sql": "SELECT date_trunc('week', order_date) AS week, SUM(total_cents) AS revenue_cents, SUM(SUM(total_cents)) OVER (ORDER BY date_trunc('week', order_date) ROWS BETWEEN 3 PRECEDING AND CURRENT ROW) AS rolling_4wk FROM sales.orders GROUP BY 1"
    }
  ]
}

# Add a Trino representation to an existing Spark-authored MV (PyIceberg)
from pyiceberg.catalog import load_catalog
from pyiceberg.view import Representation

catalog = load_catalog("glue")
view    = catalog.load_view("analytics.weekly_revenue")

# Extract the current representations; append Trino if missing
current  = view.metadata.versions[view.metadata.current_version_id]
dialects = {rep.dialect for rep in current.representations}

if "trino-sql" not in dialects:
    trino_sql = (
        "SELECT date_trunc('week', order_date) AS week, "
        "SUM(total_cents) AS revenue_cents, "
        "SUM(SUM(total_cents)) OVER ("
          "ORDER BY date_trunc('week', order_date) "
          "ROWS BETWEEN 3 PRECEDING AND CURRENT ROW"
        ") AS rolling_4wk "
        "FROM sales.orders GROUP BY 1"
    )
    view.add_representation(Representation(type="sql", dialect="trino-sql", sql=trino_sql))

print("Dialects now available:", dialects | {"trino-sql"})

Step-by-step explanation.

1. The view-metadata.json representations array holds one entry per dialect. Each entry has type (sql for a SQL string; future extensions could add logical-plan for serialised IR), dialect (spark-sql, trino-sql, snowflake-sql, etc.), and sql (the actual query text).
2. The two representations in the example are byte-for-byte identical because the aggregate uses only ANSI-standard functions (date_trunc, SUM, OVER). When an MV uses dialect-specific features (Spark's HYPERLOGLOG_PLUS_PLUS vs Trino's approx_distinct), the two representations must be written in each dialect's own syntax with matching semantics.
3. Engines resolve which representation to execute by matching their own dialect first, then falling back to a compatible one (Trino can execute spark-sql if the query happens to be dialect-neutral). The engine that authored the MV records its version in summary.engine-name; downstream engines can log a warning if they're executing a foreign dialect.
4. The Python snippet uses PyIceberg's view API to add a representation to an existing MV. view.add_representation writes a new view-metadata.json version with the additional entry; the old version is retained under versions for rollback. No storage-table rewrite is needed — representations are metadata-only.
5. For refresh, only one representation needs to execute — the one native to the engine driving the refresh. The other representations are for readers who want to inspect the MV's definition or execute it locally for validation.

Output.

Engine	Reads representation	Refresh execution
Spark 4.0	spark-sql	executes on Spark cluster
Trino 450+	trino-sql	executes on Trino coordinator
Snowflake	(falls back to snowflake-sql when added)	executes via external Iceberg view
Athena	(parses spark-sql or trino-sql)	not yet a refresh executor

Rule of thumb. For any Iceberg MV that will be read by more than one engine, add a representation per engine at creation time. The one-time cost of writing two dialect versions of the SQL is trivial compared to the debugging cost of "Trino says this MV doesn't exist" three months later.

Senior interview question on MV spec + metadata layout

A senior interviewer might ask: "Walk me through the Iceberg MV spec end to end — what's in view-metadata.json, how is the storage table separate, how does the source-snapshot mapping drive staleness, and how would you inspect a live MV to confirm its refresh state without running its query?"

Solution Using PyIceberg-based inspection + explicit source-snapshot mapping + storage-table separation

# 1. End-to-end inspection script — runbook-grade
from pyiceberg.catalog import load_catalog
from datetime import datetime, timezone
import json

def inspect_iceberg_mv(mv_name: str, catalog_name: str = "glue") -> dict:
    """Return a runbook-ready summary of an Iceberg MV's state."""
    catalog = load_catalog(catalog_name)
    view    = catalog.load_view(mv_name)
    md      = view.metadata
    current = md.versions[md.current_version_id]
    mat     = md.materialization

    # Load storage table via the pointer
    storage = catalog.load_table_by_uuid(mat.storage_table_uuid)

    # Compute staleness against every source
    source_status = {}
    for src_ref, pinned_id in mat.source_snapshot_ids.items():
        src_table  = catalog.load_table(src_ref)
        curr_id    = src_table.metadata.current_snapshot_id
        source_status[src_ref] = {
            "pinned_snapshot":  pinned_id,
            "current_snapshot": curr_id,
            "is_behind":        pinned_id != curr_id,
        }

    return {
        "mv":                 mv_name,
        "version_id":         md.current_version_id,
        "dialects":           [r.dialect for r in current.representations],
        "storage_table_uuid": str(mat.storage_table_uuid),
        "storage_location":   storage.metadata.location,
        "last_refreshed":     datetime.fromtimestamp(
                                  mat.last_refreshed_timestamp_ms / 1000,
                                  tz=timezone.utc).isoformat(),
        "sources":            source_status,
        "is_stale":           any(s["is_behind"] for s in source_status.values()),
    }

# 2. Run against production MV
report = inspect_iceberg_mv("analytics.daily_revenue")
print(json.dumps(report, indent=2, default=str))

-- 3. Native-SQL equivalent (Trino) for the freshness check
SELECT mv_name,
       is_stale,
       (current_timestamp - last_refreshed_at) AS staleness
FROM   iceberg.system.materialized_views
WHERE  mv_name = 'analytics.daily_revenue';

Step-by-step trace.

Step	Action	Reasoning
Load view	catalog.load_view(mv_name)	reads view-metadata.json via catalog
Current version	md.versions[current_version_id]	gives dialect representations
Materialization pointer	md.materialization	MV-specific fields
Load storage table	catalog.load_table_by_uuid(storage_uuid)	independent snapshot history
Per-source staleness	compare pinned vs current snapshot ID	drives is_stale
Refresh timestamp	last_refreshed_timestamp_ms	drives MAX_STALENESS compliance

After running the inspection, you can point at exactly which source table advanced, when the MV was last refreshed, where its rows live, and which dialects can execute it. Ship this script as a Prometheus exporter that emits iceberg_mv_seconds_stale{mv_name=...} and you have MV freshness observability across every engine at once.

Output:

Field	Value
MV	analytics.daily_revenue
Storage location	s3://lake/mvs/daily_revenue/
Last refreshed	2026-07-19T14:00:00Z
Is stale	false (all sources match pinned snapshots)
Dialects	spark-sql, trino-sql
Source pin	sales.orders → 8123456789012345678

Why this works — concept by concept:

• view-metadata.json + materialization pointer — the single load-bearing file. Any engine that reads Iceberg views reads the same JSON, sees the same materialization block, computes the same staleness. This is what "open spec" delivers.
• Source-snapshot mapping — the mechanism that makes staleness computable without running the MV's query. is_stale = ANY(pinned != current) is O(number of sources), not O(rows). Cheap enough to check on every read.
• Storage table separated by UUID — the MV and its storage table are two catalog objects. Refresh mutates the storage table's snapshot history; view-metadata is updated to point at the new pinned source snapshots. Both have independent rollback stories.
• Per-dialect representations — the multi-engine bridge. One MV definition; N dialect versions; each engine reads its own. No translation layer needed at read time.
• Cost — O(1) inspection cost per MV (a handful of small metadata reads), O(number of sources) staleness computation, atomic view-metadata swap on refresh. Compared to per-engine MV definitions that require N × storage + N × refresh + N × maintenance, the Iceberg MV spec is one storage + one refresh + one metadata graph. Net linear savings in every engine you add.

SQL
Topic — sql
SQL view and metadata problems

Practice →

Aggregation Topic — aggregation Aggregation problems on cached rollups

Practice →

---

3. Incremental refresh strategies

Full refresh is universal; incremental refresh is the reason MVs are cheap — and the strategy depends on whether the source is append-only or upsert/delete-shaped

The mental model in one line: incremental refresh iceberg is the mechanism that turns an MV from an "aggregate you rebuild from scratch every N minutes" into an "aggregate you append to based on the source snapshot diff since last refresh" — and the strategy you pick (snapshot-diff append for append-only sources; row-level-delete reconciliation for upsert/delete sources) is what determines whether refresh is O(delta) or O(source table) and whether the MV can meet a 5-minute freshness contract at 500-million-row scale. Every senior lakehouse architect must pick the incremental strategy that matches the source's write pattern.

Full refresh vs incremental refresh — the fundamental split.

• Full refresh. Re-execute the MV's defining query end-to-end; overwrite the storage table. Simple, universally supported, O(source table). Correct always; expensive on any source > 1 M rows.
• Incremental refresh. Read only the source snapshot diff since the last refresh; compute the incremental contribution to the aggregate; apply it to the storage table. O(delta), can be 1000× cheaper. Correctness depends on the aggregate being "incrementally computable" — SUM, COUNT, and (with care) MIN/MAX are; DISTINCT COUNT is not without HyperLogLog; percentiles are not without T-Digest.
• The choice matrix. SUM-shaped aggregate + append-only source = trivial incremental. SUM-shaped + upsert source = incremental with row-level deletes. DISTINCT COUNT + any source = HyperLogLog or fall back to full refresh. Percentile + any source = T-Digest sketches or full refresh.

Append-only sources — the trivial-incremental path.

• Definition. Source table only receives INSERTs — no UPDATEs, no DELETEs. Common for event streams, log tables, immutable fact tables.
• The incremental algorithm. SELECT ... FROM source WHERE snapshot_id BETWEEN last_refreshed_snapshot AND current_snapshot. Iceberg exposes this via the $snapshots metadata table and the snapshot-diff scan option. Aggregate the delta rows; append to the MV storage table.
• Complexity. O(delta size), not O(source size). For a 500 M-row source with 50 M new rows/day, incremental refresh reads ~50 M rows/day, not 500 M.
• Correctness. Because there are no UPDATEs/DELETEs to reconcile, the aggregate over the union of past MV rows plus new incremental rows equals the aggregate over the full source. Guaranteed for SUM, COUNT; guaranteed for MIN/MAX with an initial full refresh.

Upsert / delete sources — the row-level-delete path.

• Definition. Source receives INSERT + UPDATE + DELETE. Common for CDC-fed lakehouse tables (Debezium → Iceberg), dimension tables, slowly-changing fact tables.
• The problem. An UPDATE to a source row means the aggregate over that row's group is now wrong in the MV. A naive append would double-count. A naive re-read of the entire group is not O(delta).
• The solution — Iceberg row-level deletes. Iceberg v2 supports positional and equality deletes. The refresh planner reads the source snapshot diff, identifies changed rows (UPDATE + DELETE), reads the pre-image of those rows (from the source's previous snapshot), subtracts their contribution from the MV, then adds the new post-image contribution.
• Complexity. O(delta size + affected groups). Still much cheaper than full refresh for high-cardinality group-bys where the delta touches only a small fraction of groups.

Watermark-based refresh — when snapshot diff isn't enough.

• When to use it. Source tables that don't cleanly snapshot per-batch (some CDC-fed tables). Falls back to a watermark column (updated_at, event_time) with a safety window.
• How it works. Refresh reads WHERE updated_at > mv_last_watermark AND updated_at <= now() - safety_horizon; advances the watermark on success. Combines Iceberg's snapshot machinery with the classic timestamp-CDC watermark pattern.
• Trade-off vs snapshot diff. Watermark is engine-portable and works even on sources without proper snapshot lineage; snapshot diff is faster and more correct when available.

SLA-bound refresh cadence — turning MAX_STALENESS into a schedule.

• The contract. MAX_STALENESS = 5m means users expect the MV to reflect source changes within 5 minutes.
• The scheduler. A cron every 5 minutes gets you 5-minute worst-case staleness if refresh completes in under 5 minutes. If refresh takes 6 minutes, you're chronically stale.
• The safety margin. Schedule refresh at MAX_STALENESS × 0.5 (every 2-3 minutes for a 5-minute contract) so a slow refresh doesn't blow the contract. Monitor refresh duration and alert when p95 > MAX_STALENESS × 0.7.

Refresh trigger models — cron, event, continuous.

• Cron. Airflow/Prefect schedules a refresh every N minutes. Simple; wastes work when source is idle.
• Event-driven. Trigger refresh when the source table advances its snapshot (webhook / Kafka listener on source commits). Cheaper when source is bursty; requires source-side plumbing.
• Continuous. A streaming job (Flink / Spark Structured Streaming) refreshes as source snapshots advance, targeting sub-minute lag. Highest fidelity; most operational complexity.

Common interview probes on incremental refresh.

• "What's the difference between full and incremental refresh?" — required answer: O(source) vs O(delta); pick based on source size and freshness target.
• "How does incremental refresh handle deletes?" — Iceberg row-level deletes; subtract pre-image contribution from MV, add post-image.
• "When can't you do incremental refresh?" — non-incrementally-computable aggregates (DISTINCT COUNT without HLL, exact percentiles).
• "How do you schedule refresh to meet a 5-minute freshness SLA?" — cron at MAX_STALENESS × 0.5; alert on refresh duration p95.

Worked example — snapshot-diff incremental refresh for an append-only events source

Detailed explanation. The canonical append-only incremental refresh: an events table receives inserts continuously; the MV hourly_events_by_type counts events by hour+type. Refresh reads the source snapshot diff since last refresh, groups the delta, and appends the aggregated delta to the MV storage table. Walk through the full end-to-end.

• Source. stream.events(event_id, event_ts, event_type) — append-only, ~50 M rows/day.
• MV. analytics.hourly_events_by_type(hour, event_type, cnt).
• Refresh cadence. Every 5 minutes.

Question. Implement the incremental refresh SQL and show the snapshot-diff mechanism against the Iceberg $snapshots metadata table.

Input.

Component	Value
Source	stream.events (Iceberg, append-only)
MV	analytics.hourly_events_by_type
Last refreshed snapshot	8199990000000000000
Current source snapshot	8199999999999999999
Delta rows	~200 000 (5 minutes of ingest)

Code.

-- 1. Read the pinned source snapshot from the MV's view-metadata
--    (in production, this is exposed by the engine — shown here as SQL)
WITH mv_pin AS (
    SELECT CAST(properties['mv.source.stream.events.snapshot_id'] AS BIGINT) AS pinned_id,
           CAST(properties['mv.last_refreshed_ms']                AS BIGINT) AS last_refresh_ms
    FROM   analytics.hourly_events_by_type$properties
),
current_source AS (
    SELECT MAX(snapshot_id) AS current_id
    FROM   stream.events$snapshots
)
SELECT * FROM mv_pin CROSS JOIN current_source;

-- 2. Incremental aggregate — the snapshot-diff scan
--    reads only rows added between pinned_id and current_id
INSERT INTO analytics.hourly_events_by_type
SELECT date_trunc('hour', event_ts) AS hour,
       event_type,
       COUNT(*)                     AS cnt
FROM   stream.events
FOR VERSION AS OF (SELECT current_id FROM current_source)
WHERE  snapshot_id_of(event_id) > (SELECT pinned_id FROM mv_pin)
  AND  snapshot_id_of(event_id) <= (SELECT current_id FROM current_source)
GROUP  BY 1, 2;

# 3. Refresh orchestration — read snapshot diff via PyIceberg
from pyiceberg.catalog import load_catalog
from pyspark.sql import SparkSession

def refresh_hourly_events_mv():
    catalog = load_catalog("glue")
    source  = catalog.load_table("stream.events")
    mv      = catalog.load_view("analytics.hourly_events_by_type")

    pinned  = mv.metadata.materialization.source_snapshot_ids["stream.events"]
    current = source.metadata.current_snapshot_id

    if pinned == current:
        print("No source advance; skip refresh")
        return

    spark = SparkSession.builder.appName("mv-refresh").getOrCreate()

    # Snapshot-diff scan — Iceberg exposes this via appended_files()
    # For append-only sources, the delta rows live in files added
    # between the pinned and current snapshots
    delta = spark.read.format("iceberg") \
        .option("start-snapshot-id", pinned) \
        .option("end-snapshot-id",   current) \
        .option("read.mode",         "incremental-append-scan") \
        .load("stream.events")

    # Aggregate the delta only
    agg = delta.selectExpr(
        "date_trunc('hour', event_ts) as hour",
        "event_type",
        "1 as one"
    ).groupBy("hour", "event_type").sum("one").withColumnRenamed("sum(one)", "cnt")

    # Append the aggregated delta to the MV storage table
    agg.writeTo("analytics.hourly_events_by_type_storage").append()

    # Update the source-snapshot pin atomically
    mv.update_source_snapshot({"stream.events": current})
    print(f"Refresh: pin {pinned} → {current}; delta rows aggregated")

refresh_hourly_events_mv()

Step-by-step explanation.

1. Step 1 reads the MV's pinned source snapshot ID from the view-metadata (shown here as a properties read on the storage table for portability; production engines expose this as a first-class SQL surface like SELECT mv_pin('analytics.hourly_events_by_type')).
2. The FOR VERSION AS OF clause pins the source read to the current snapshot — this gives a stable, repeatable delta even if the source keeps advancing during refresh. Combined with WHERE snapshot_id_of(...) > pinned_id, the scan reads only the rows added between pinned and current.
3. Step 3's Spark code uses the Iceberg incremental-append-scan mode, which reads only data files added between the two snapshots — no manifest-level scan of unchanged files. For a source with 500 GB of data and 50 GB of daily inserts, incremental scan reads ~5 GB per 5-minute refresh instead of 500 GB per full refresh — a 100× reduction.
4. The aggregated delta is appended to the MV storage table, not merged. Because the aggregate is a SUM (COUNT) and the source is append-only, delta append is semantically equivalent to a full re-aggregate: the group-by preserves per-key sums, so SUM(prev_cnt + delta_cnt) = SUM(all rows in that group).
5. mv.update_source_snapshot atomically updates the source-snapshot mapping in view-metadata.json and the last_refreshed_timestamp_ms field. If the append succeeds but the metadata update fails, the next refresh sees the old pin and re-reads the delta — correct via idempotent set-based aggregation.

Output.

Refresh cycle	Pinned before	Pinned after	Delta rows	MV rows appended	Duration
1 (bootstrap full)	(none)	8199990000000000000	500 M	24 000	8 min
2 (incremental)	8199990000000000000	8199993000000000000	200 000	4 (new hour)	22 s
3 (incremental)	8199993000000000000	8199996000000000000	210 000	0 (existing hours)	20 s
4 (incremental)	8199996000000000000	8199999999999999999	195 000	4 (next hour)	21 s

Rule of thumb. For any append-only source, incremental-append-scan on the snapshot diff is the correct default. The first refresh is always a full scan (bootstrap); all subsequent refreshes read only the delta. Ship a full-refresh fallback path so you can rebuild the MV if the pin is ever lost.

Worked example — upsert-source incremental refresh with Iceberg row-level deletes

Detailed explanation. When the source has UPDATEs and DELETEs — e.g. a CDC-fed orders table from Debezium — the append-only trick breaks. An UPDATE changes an order's total_cents; a naive incremental refresh would add the new total on top of the old total in the MV. The correct approach uses Iceberg row-level deletes: read the delta, split it into pre-image (values before change) and post-image (values after change), subtract the pre-image contribution from the MV, add the post-image contribution.

• Source. sales.orders_cdc(order_id, customer_id, total_cents, updated_at) — CDC-fed with UPDATE + DELETE row-level deletes.
• MV. analytics.customer_totals(customer_id, total_cents) — sum of totals per customer.
• Refresh cadence. Every 5 minutes.

Question. Implement the upsert-aware incremental refresh using Iceberg row-level delete metadata to reconcile changed rows.

Input.

Component	Value
Source	sales.orders_cdc (upsert + delete via Iceberg v2 row-level deletes)
MV	analytics.customer_totals
Delta INSERTs	5 000
Delta UPDATEs	500
Delta DELETEs	20
Affected customer_ids	~2 500 (dedup of delta customer_ids)

Code.

-- 1. Compute the reconcile delta — pre-image + post-image per changed row
--    Iceberg v2 exposes deleted rows via the _iceberg_delete_type meta col
WITH delta_rows AS (
    SELECT customer_id,
           total_cents,
           _iceberg_change_type AS op   -- 'INSERT', 'UPDATE_BEFORE', 'UPDATE_AFTER', 'DELETE'
    FROM   sales.orders_cdc.changes    -- Iceberg CDF-style change scan
    WHERE  start_snapshot_id = <pinned>
      AND  end_snapshot_id   = <current>
),
per_customer_delta AS (
    SELECT customer_id,
           SUM(CASE
                 WHEN op IN ('INSERT', 'UPDATE_AFTER')  THEN  total_cents
                 WHEN op IN ('DELETE', 'UPDATE_BEFORE') THEN -total_cents
               END) AS delta_total_cents
    FROM   delta_rows
    GROUP  BY customer_id
    HAVING SUM(CASE WHEN op IN ('INSERT','UPDATE_AFTER')  THEN  total_cents
                    WHEN op IN ('DELETE','UPDATE_BEFORE') THEN -total_cents
               END) <> 0
)
MERGE INTO analytics.customer_totals AS mv
USING     per_customer_delta        AS d
ON        mv.customer_id = d.customer_id
WHEN MATCHED THEN UPDATE
     SET total_cents = mv.total_cents + d.delta_total_cents
WHEN NOT MATCHED THEN INSERT (customer_id, total_cents)
     VALUES (d.customer_id, d.delta_total_cents);

# 2. Refresh orchestration using Spark's Iceberg change-scan
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, when, sum as _sum

spark = SparkSession.builder.appName("mv-refresh-upsert").getOrCreate()

pinned  = 8199990000000000000  # from view-metadata
current = 8199999999999999999

delta = spark.read.format("iceberg") \
    .option("start-snapshot-id",  pinned) \
    .option("end-snapshot-id",    current) \
    .option("read.mode",          "incremental-change-scan") \
    .load("sales.orders_cdc")

# Signed contribution per row
signed = delta.withColumn(
    "signed_total",
    when(col("_iceberg_change_type").isin("INSERT", "UPDATE_AFTER"),   col("total_cents"))
   .when(col("_iceberg_change_type").isin("DELETE", "UPDATE_BEFORE"), -col("total_cents"))
).groupBy("customer_id").agg(_sum("signed_total").alias("delta_total_cents"))

# Filter out no-op deltas
signed = signed.filter(col("delta_total_cents") != 0)

# Merge into MV storage table
signed.createOrReplaceTempView("delta")
spark.sql("""
    MERGE INTO analytics.customer_totals_storage AS mv
    USING     delta                              AS d
    ON        mv.customer_id = d.customer_id
    WHEN MATCHED THEN UPDATE SET total_cents = mv.total_cents + d.delta_total_cents
    WHEN NOT MATCHED THEN INSERT (customer_id, total_cents) VALUES (d.customer_id, d.delta_total_cents)
""")

Step-by-step explanation.

1. The .changes scan (Iceberg change-data-feed / CDF-style read) exposes not just added rows but a per-row _iceberg_change_type marker: INSERT, UPDATE_BEFORE (pre-image), UPDATE_AFTER (post-image), DELETE. This is the primitive that turns snapshot diffs into semantic change events.
2. The signed_total computation encodes the reconciliation math: positive for adds (INSERT, UPDATE_AFTER), negative for removes (DELETE, UPDATE_BEFORE). Summing signed contributions per customer gives the exact delta that must be applied to the MV to keep it consistent.
3. The HAVING delta != 0 filter drops customers whose net change is zero — an UPDATE that changes non-aggregated columns leaves the aggregate unchanged and shouldn't touch the MV. This keeps the MERGE payload minimal.
4. The MERGE statement upserts the aggregate: existing customers get their MV total incremented by the delta; new customers (first-time appearance) get a new MV row. Because the deltas are signed correctly, this preserves the invariant MV.total_cents = SUM(source.total_cents) grouped by customer_id.
5. The complexity is O(delta rows + affected groups). For 5 500 delta rows touching 2 500 customers in a source with 10 M customers, the MERGE touches 2 500 MV rows — a 4 000× reduction versus rewriting the full MV. This is the entire justification for row-level-delete-aware incremental refresh.

Output.

Event kind	Delta rows	Signed contribution	MV impact
INSERT (new order for customer 42)	+1500	+1500	mv.customer_id=42 += 1500
UPDATE (order total 1500 → 1800)	UPDATE_BEFORE -1500, UPDATE_AFTER +1800	+300 net	mv.customer_id=42 += 300
DELETE (order for customer 7 removed)	UPDATE_BEFORE -700 (as DELETE)	-700	mv.customer_id=7 -= 700
UPDATE (change non-aggregated column)	UPDATE_BEFORE -X, UPDATE_AFTER +X	0	no MV write

Rule of thumb. For any upsert/delete source, use Iceberg's change scan (incremental-change-scan) with signed reconciliation. Reserve full refresh for the bootstrap case and for the "our pin is lost" recovery path. Always filter no-op deltas before the MERGE — halves the average MV write volume.

Worked example — SLA-bound refresh scheduler with duration monitoring

Detailed explanation. A 5-minute MAX_STALENESS contract means the MV must reflect source changes within 5 minutes. Naively scheduling refresh every 5 minutes doesn't cut it: if refresh takes 6 minutes, the next refresh starts before the previous finishes, drift accumulates, and eventually the contract breaks silently. The correct scheduler runs refresh at MAX_STALENESS × 0.5 cadence, prevents overlaps, and alerts when refresh duration approaches the contract.

• MAX_STALENESS. 5 minutes.
• Cron cadence. Every 2 minutes (safety margin 2.5× — refresh completes with headroom).
• Overlap prevention. max_active_runs=1 in Airflow.
• Duration alert. Alert on p95 refresh duration > MAX_STALENESS × 0.7 (3.5 min).

Question. Implement the refresh DAG with duration monitoring + Prometheus emission for freshness observability.

Input.

Component	Value
MAX_STALENESS	5 minutes
Scheduler cadence	*/2 * * * *
max_active_runs	1
p95 duration alert	> 3.5 min
Metric	mv_refresh_duration_seconds{mv=...}

Code.

from airflow.decorators import dag, task
from datetime import datetime
from prometheus_client import CollectorRegistry, Gauge, push_to_gateway
import time, subprocess

MV_NAME       = "analytics.customer_totals"
MAX_STALENESS = 300   # seconds
CADENCE       = 120   # seconds (2 minutes)

@dag(schedule="*/2 * * * *",
     start_date=datetime(2026, 7, 19),
     catchup=False,
     max_active_runs=1,                       # never overlap
     dagrun_timeout=None,
     default_args={"retries": 1})
def refresh_customer_totals_mv():

    @task
    def refresh() -> dict:
        registry = CollectorRegistry()
        g_dur    = Gauge("mv_refresh_duration_seconds",  "MV refresh duration",  ["mv"], registry=registry)
        g_ok     = Gauge("mv_refresh_success",           "1 if last refresh OK", ["mv"], registry=registry)

        t0 = time.time()
        try:
            subprocess.run(
                ["trino", "--execute", f"REFRESH MATERIALIZED VIEW {MV_NAME}"],
                check=True, timeout=MAX_STALENESS
            )
            duration = time.time() - t0
            g_dur.labels(MV_NAME).set(duration)
            g_ok.labels(MV_NAME).set(1)
            push_to_gateway("pushgw:9091", job="mv_refresh", registry=registry)

            if duration > MAX_STALENESS * 0.7:
                print(f"WARN: refresh {duration:.1f}s > 70% of MAX_STALENESS")
            return {"duration_sec": duration, "ok": True}

        except subprocess.TimeoutExpired:
            g_ok.labels(MV_NAME).set(0)
            push_to_gateway("pushgw:9091", job="mv_refresh", registry=registry)
            raise RuntimeError(f"MV refresh exceeded MAX_STALENESS budget")

    refresh()

dag = refresh_customer_totals_mv()

# Prometheus alerting rules
groups:
- name: iceberg-mv
  rules:
  - alert: MvRefreshDurationHigh
    expr: quantile_over_time(0.95, mv_refresh_duration_seconds[1h]) > 210  # 3.5 min
    for: 15m
    labels: {severity: warning}
    annotations:
      summary: "MV {{ $labels.mv }} refresh p95 approaching MAX_STALENESS"

  - alert: MvRefreshFailing
    expr: mv_refresh_success == 0
    for: 5m
    labels: {severity: critical}
    annotations:
      summary: "MV {{ $labels.mv }} last refresh failed"

Step-by-step explanation.

1. The cron schedule */2 * * * * runs every 2 minutes — well below the 5-minute MAX_STALENESS contract. This gives a 2.5× safety margin: even if refresh takes 4 minutes, the freshness contract is still met.
2. max_active_runs=1 is the load-bearing safety: if refresh N is still running when N+1 fires, N+1 is queued (not started in parallel). Overlapping refreshes double-count deltas in some incremental strategies and are always a source of correctness bugs.
3. The refresh task wraps the actual REFRESH MATERIALIZED VIEW call with a timeout of MAX_STALENESS seconds — if refresh cannot complete in that budget, the run fails loudly instead of silently blowing the freshness contract. Timeout-caught failures set mv_refresh_success = 0 and trigger a critical alert.
4. The duration gauge (mv_refresh_duration_seconds) is pushed to Prometheus on every run. The p95 over the last hour is monitored; if it approaches 70% of the contract, on-call gets a warning to investigate before the contract actually breaks.
5. The failure alert (mv_refresh_success == 0 for 5m) catches sustained refresh failures. A single failed run is noise (transient network blip); five minutes of failure is a real incident that means the MV is drifting behind the contract.

Output.

Refresh cycle	Duration (s)	p95 (1h)	Status	Alert
1	22	22	ok	none
2	24	24	ok	none
...	...	...	...	...
30	210	195	ok	WARN (approaching 70%)
31	340	245	timeout	CRITICAL (>MAX_STALENESS)

Rule of thumb. Schedule at MAX_STALENESS ÷ 2.5 cadence. Prevent overlaps via max_active_runs=1. Alert on p95 duration > 70% of contract. Never treat "we ran refresh every 5 minutes" as equivalent to "we honoured a 5-minute freshness contract" — the second requires proof.

Senior interview question on incremental refresh strategy

A senior interviewer might ask: "You have an MV over a 2-billion-row orders table that receives 100 K UPDATEs and 5 K DELETEs per minute via CDC. Design the incremental refresh strategy end-to-end — how you'd read the source delta, how you'd reconcile the UPDATE/DELETE contributions to the aggregate, what refresh cadence you'd pick for a 3-minute freshness contract, and how you'd guarantee correctness after a refresh failure."

Solution Using Iceberg change-scan + signed reconciliation MERGE + 1-minute cadence + idempotent pin advance

-- 1. MV definition — signed-reconciliation-friendly aggregate
CREATE MATERIALIZED VIEW analytics.orders_metrics
  TBLPROPERTIES (
    'iceberg.mv.refresh.mode'   = 'incremental-change-scan',
    'iceberg.mv.freshness.max'  = '3 minutes',
    'iceberg.mv.storage.location' = 's3://lake/mvs/orders_metrics/'
  )
AS
SELECT customer_id,
       COUNT(*)                              AS order_count,
       SUM(total_cents)                      AS total_cents,
       SUM(CASE WHEN status='shipped' THEN 1 ELSE 0 END) AS shipped_count
FROM   sales.orders
GROUP  BY customer_id;

# 2. Refresh with idempotent pin advance — safe under retry
from pyiceberg.catalog import load_catalog
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, when, sum as _sum, lit
import time

MV       = "analytics.orders_metrics"
SOURCE   = "sales.orders"
BUDGET_S = 180

def refresh() -> None:
    catalog = load_catalog("glue")
    mv      = catalog.load_view(MV)
    source  = catalog.load_table(SOURCE)

    pinned  = mv.metadata.materialization.source_snapshot_ids[SOURCE]
    current = source.metadata.current_snapshot_id
    if pinned == current:
        return

    spark = SparkSession.builder.getOrCreate()
    delta = spark.read.format("iceberg") \
        .option("start-snapshot-id", pinned) \
        .option("end-snapshot-id",   current) \
        .option("read.mode",         "incremental-change-scan") \
        .load(SOURCE)

    # Signed reconciliation for three aggregates simultaneously
    sign = when(col("_iceberg_change_type").isin("INSERT", "UPDATE_AFTER"),   lit(1)) \
          .otherwise(lit(-1))

    signed = delta.withColumn("s", sign).groupBy("customer_id").agg(
        _sum(col("s")                                                       ).alias("d_count"),
        _sum(col("s") * col("total_cents")                                  ).alias("d_total"),
        _sum(col("s") * when(col("status")=="shipped", lit(1)).otherwise(0) ).alias("d_shipped"),
    ).filter((col("d_count")!=0) | (col("d_total")!=0) | (col("d_shipped")!=0))

    signed.createOrReplaceTempView("delta")

    spark.sql(f"""
        MERGE INTO {MV}_storage AS mv
        USING     delta        AS d
        ON        mv.customer_id = d.customer_id
        WHEN MATCHED THEN UPDATE SET
            order_count   = mv.order_count   + d.d_count,
            total_cents   = mv.total_cents   + d.d_total,
            shipped_count = mv.shipped_count + d.d_shipped
        WHEN NOT MATCHED THEN INSERT
            (customer_id, order_count, total_cents, shipped_count)
        VALUES
            (d.customer_id, d.d_count, d.d_total, d.d_shipped)
    """)

    # Atomically advance the pin — idempotent w.r.t. re-run because
    # the source_snapshot_id equality check at the top short-circuits
    mv.update_source_snapshot({SOURCE: current})

# 3. Airflow DAG — 1-minute cadence, 3-minute budget
schedule: "* * * * *"          # every 1 minute
max_active_runs: 1
dagrun_timeout: 180             # 3 min = MAX_STALENESS

Step-by-step trace.

Layer	Config	Result
Refresh cadence	1 minute	3× safety margin under 3-min contract
Read mode	incremental-change-scan	O(delta) reads
Reconciliation	signed MERGE per aggregate	correct under INSERT/UPDATE/DELETE
Pin advance	after successful MERGE	idempotent under retry
Overlap prevention	max_active_runs=1	no double-count
Timeout	180 s = MAX_STALENESS	loud failure on budget breach
Alerting	p95 duration > 130 s	on-call before contract breaks

After the rollout, the MV lags the source by at most 60 s in steady state (median ~40 s), each refresh reads ~105 K delta rows and touches ~30 K customer rows in the MV (out of 10 M), and the pin advance is idempotent so a Spark job crash mid-run leaves the pin at the previous value and the next run replays the delta safely.

Output:

Metric	Value
Source rows/minute (INSERT+UPDATE+DELETE)	~105 K
MV rows touched per refresh	~30 K
Refresh p50 duration	38 s
Refresh p95 duration	92 s
MAX_STALENESS contract	3 min
Actual freshness p95	~1.5 min
Source-table scan volume	O(delta) not O(source)
Pin advance guarantee	idempotent

Why this works — concept by concept:

• incremental-change-scan — Iceberg's CDF-style read exposes per-row _iceberg_change_type markers. This is the single primitive that turns snapshot diffs into semantically-correct upsert reconciliation. Without it, incremental refresh on upsert sources requires manual pre-image / post-image tracking.
• Signed contribution + MERGE — the reconciliation math. INSERT and UPDATE_AFTER get +1; DELETE and UPDATE_BEFORE get -1. Grouping by aggregate key and summing gives the exact delta that must be added to the MV. Correct for SUM, COUNT, and boolean-cast COUNTs (like shipped_count).
• Filter no-op deltas — an UPDATE that doesn't touch aggregated columns has d_count = d_total = d_shipped = 0. Filtering these before the MERGE cuts MV write volume roughly in half in practice.
• Idempotent pin advance — advancing the pin after the MERGE means a crash between MERGE and advance leaves the pin at the previous value; the next run re-reads the same delta and re-runs the same MERGE. Because the MERGE math is signed-and-summed, it's a set-based idempotent operation (double-application ≠ single-application, so we must ensure exactly-once). In production, wrap MERGE + pin advance in one transaction (Iceberg atomic snapshot + property update).
• Cost — O(delta) source read, O(affected keys) MV write, 1-minute cadence with a 3-minute budget. Compared to full refresh (O(2 B rows) every 3 minutes = infeasible), this is 4-5 orders of magnitude cheaper. The eliminated cost is the "MV impossible at this scale" outcome.

SQL
Topic — sql
SQL incremental-load and MERGE problems

Practice →

Aggregation Topic — aggregation Aggregation problems on delta reconciliation

Practice →

---

4. Query planning + freshness

The engine planner treats the MV as an optimiser hint, not a table you SELECT from — freshness metadata drives auto-substitution and the tolerance you set decides the fallback

The mental model in one line: iceberg mv query rewrite is the mechanism where the engine's cost-based optimiser reads the incoming user query, compares it against every MV's stored logical plan, checks the MV's freshness against the user-declared tolerance, and — when the plan matches and freshness is within tolerance — silently rewrites the user query to SELECT ... FROM <mv-storage-table> instead of scanning the source; when freshness fails, it falls back to the source scan and the user gets fresh (but slower) data. Understanding rewrite is the difference between an MV that speeds up 5% of queries (users must know it exists) and an MV that speeds up 95% of matching queries (the optimiser finds it).

The rewrite pipeline — what happens between SELECT and result.

• Parse. The engine parses the user's SQL into a logical plan (relational algebra + expressions).
• Rewrite candidacy. The optimiser enumerates MVs whose defining logical plan is a subset or superset of the user's plan. Exact match is the easy case; subset match (MV computes finer-grained aggregate, user query re-aggregates) is the general case.
• Freshness gate. For each candidate MV, the optimiser reads view-metadata.materialization.last_refreshed_timestamp_ms and source-snapshot-ids. If the MV is stale beyond the user's tolerance, it's disqualified.
• Cost estimate. For surviving candidates, the optimiser estimates the cost of SELECT ... FROM mv_storage (typically small) vs SELECT ... FROM source (typically large). Cheaper plan wins.
• Substitute + execute. The chosen plan is executed. From the user's perspective, the query returned faster — no SELECT ... FROM mv was needed.

Freshness metadata surfaces — the fields users and engines both read.

• last_refreshed_at. ISO-8601 timestamp of when the MV was last refreshed. Human-readable freshness.
• is_stale. Boolean, computed from source-snapshot-ids vs current source snapshots. true means at least one source has advanced since last refresh.
• freshness_max (MAX_STALENESS). MV-level property setting the tolerance the optimiser uses when the user query has no explicit freshness hint.
• User-level override. SELECT ... FROM analytics.daily_revenue OPTIONS (MAX_STALENESS = '1 minute') — the user can tighten or loosen tolerance per query.

Rewrite match kinds — what the optimiser recognises.

• Exact match. User query is byte-identical to MV's defining SQL after normalisation. Substitution is trivial.
• Filter-pushdown match. User adds a WHERE clause on a column the MV has. Optimiser reads MV storage, applies the filter on top.
• Rollup match. MV holds a finer-grained aggregate (GROUP BY hour, type); user asks for a coarser one (GROUP BY hour). Optimiser re-aggregates MV rows on top of the finer grain.
• Join match. MV is a denormalised join (orders JOIN customers); user query joins the same tables. Optimiser reads MV instead of re-joining sources.

Freshness-vs-latency trade-off — the user-visible knob.

• Tight tolerance (small MAX_STALENESS). MV is disqualified often; users get fresh data, slower.
• Loose tolerance (large MAX_STALENESS). MV is used often; users get fast data, potentially stale.
• Per-query override. Dashboards set tight tolerance; ad-hoc analysts set loose.
• The default. MV-level freshness_max sets the default; users override when they have a strong opinion.

When to skip MV and scan source directly — the optimiser's fallback logic.

• MV stale beyond tolerance. Automatic — source scan.
• User query touches columns MV doesn't have. Automatic — source scan.
• User query GROUP BY doesn't match any MV rollup. Automatic — source scan.
• User explicit hint. SELECT /*+ NO_MV */ ... FROM analytics.daily_revenue — some engines expose a hint to force source read (useful for freshness-critical audit queries).

Common interview probes on query rewrite.

• "How does the engine know when to use an MV?" — planner compares logical plans; freshness gate; cost estimate.
• "What's the difference between MAX_STALENESS at MV level vs query level?" — default vs override; query wins if set.
• "Give an example where the optimiser rewrites without an exact SQL match" — filter pushdown; rollup match; join match.
• "How would you force the source scan?" — explicit hint (NO_MV), or tighten tolerance to zero for that query.

Worked example — enabling automatic query rewrite for a dashboard query

Detailed explanation. The canonical rewrite scenario: a dashboard query SELECT order_date, SUM(total_cents) FROM sales.orders WHERE order_date >= current_date - 30 GROUP BY 1 currently scans 500 M rows every minute. An analytics.daily_revenue MV holds the same aggregate. The engine's rewrite kicks in when the MV's freshness metadata says the MV is within the dashboard's tolerance. Walk through the setup end-to-end.

• Dashboard query. 30-day aggregate WHERE-filtered by date range.
• MV. analytics.daily_revenue — aggregate at day grain over all dates.
• Tolerance. Dashboard team accepts 5 minutes of staleness.
• Rewrite mechanism. Engine matches dashboard's GROUP BY against MV's GROUP BY, pushes down the WHERE filter, verifies freshness < 5 min.

Question. Configure the MV and demonstrate that a dashboard query gets rewritten against it (using EXPLAIN).

Input.

Component	Value
Source scan cost	40 s (500 M rows)
MV scan cost	200 ms (~1 K rows)
Dashboard tolerance	5 min
MV freshness_max	5 min
Engine	Trino 450+

Code.

-- 1. Define the MV with freshness_max = 5 minutes
CREATE MATERIALIZED VIEW analytics.daily_revenue
  TBLPROPERTIES (
    'iceberg.mv.refresh.mode'    = 'incremental',
    'iceberg.mv.freshness.max'   = '5 minutes',
    'iceberg.mv.storage.location' = 's3://lake/mvs/daily_revenue/'
  )
AS
SELECT order_date,
       SUM(total_cents) AS revenue_cents,
       COUNT(*)         AS order_count
FROM   sales.orders
GROUP  BY order_date;

-- 2. Dashboard query (unchanged — no mention of the MV)
SELECT order_date, SUM(total_cents) AS revenue_cents
FROM   sales.orders
WHERE  order_date >= current_date - INTERVAL '30' DAY
GROUP  BY order_date;

-- 3. Verify rewrite via EXPLAIN
EXPLAIN
SELECT order_date, SUM(total_cents) AS revenue_cents
FROM   sales.orders
WHERE  order_date >= current_date - INTERVAL '30' DAY
GROUP  BY order_date;

-- Expected EXPLAIN output (rewrite fired)
Fragment 1
  Output: order_date, revenue_cents
  MaterializedViewSubstitution[
    original: sales.orders,
    substituted: analytics.daily_revenue,
    freshness_ok: true,
    last_refreshed: 2026-07-19T14:00:00Z,
    tolerance: 5m
  ]
  TableScan[analytics.daily_revenue$storage]
    filter: order_date >= 2026-06-19
    columns: order_date, revenue_cents

-- If the MV were stale (last_refreshed > 5 min ago), EXPLAIN would show:
--   MaterializedViewSubstitution[freshness_ok: false, fallback: sales.orders]
--   Aggregate[SUM] -> TableScan[sales.orders] filter: order_date >= 2026-06-19

Step-by-step explanation.

1. Step 1's MV declares iceberg.mv.freshness.max = 5 minutes — this is the default tolerance the optimiser uses when substituting the MV. Any user query that hits sales.orders with a compatible plan will be considered for substitution against this MV.
2. Step 2's dashboard query is written against sales.orders, not against the MV. The user does not know the MV exists. This is the entire point of rewrite — the caching is transparent.
3. The optimiser parses the dashboard query into a logical plan: Aggregate[SUM] over Filter[order_date >= X] over TableScan[sales.orders]. It enumerates MVs on sales.orders, finds analytics.daily_revenue, notes that the MV's plan is Aggregate[SUM] over TableScan[sales.orders] grouped by order_date — a superset of the user query (MV covers all dates; user filters to 30).
4. The optimiser can push the user's WHERE order_date >= current_date - 30 filter down onto the MV storage scan because order_date is a column present in the MV. The rewritten plan becomes Filter over TableScan[mv_storage] — a scan of ~30 MV rows vs 500 M source rows.
5. Before executing, the optimiser checks last_refreshed_at against the tolerance. Fresh → substitute. Stale → fall back to the source scan with the original aggregate. Either way the user sees the correct result; the only difference is latency.

Output.

Query	Rewrite?	Scan target	Latency
Dashboard (fresh MV)	yes	mv_storage (30 rows)	200 ms
Dashboard (stale MV)	no	sales.orders (500 M)	40 s
Dashboard with hint NO_MV	no (forced)	sales.orders (500 M)	40 s
Ad-hoc WHERE customer_id=42	no (column not in MV)	sales.orders (~1 K rows)	500 ms

Rule of thumb. Design MV columns to include every dimension users will filter on — otherwise the optimiser cannot rewrite. If dashboard queries filter by order_date and region, the MV must GROUP BY both, even if you don't display region on the dashboard.

Worked example — per-query MAX_STALENESS override for audit vs dashboard

Detailed explanation. Not every user has the same freshness tolerance. A dashboard PM accepts 5-minute stale data because refresh is expensive; an auditor investigating a specific transaction demands current data because the number matters. The MAX_STALENESS override on a per-query basis lets both share the same MV without duplicating storage. Walk through the mechanism.

• MV default. freshness_max = 5m — dashboard-tuned.
• Auditor override. OPTIONS (MAX_STALENESS = '0 seconds') — force source scan.
• Analyst override. OPTIONS (MAX_STALENESS = '1 hour') — cheaper for historical analysis where fresh is unnecessary.

Question. Show three queries with different per-query tolerances and the rewrite outcome for each.

Input.

Persona	Tolerance	Expected rewrite
Dashboard	5 min (MV default)	yes, when MV < 5 min stale
Auditor	0 seconds	no; source scan always
Analyst	1 hour	yes, when MV < 60 min stale

Code.

-- Dashboard query — uses MV default (5 min)
SELECT order_date, SUM(total_cents) AS revenue_cents
FROM   sales.orders
WHERE  order_date >= current_date - 30
GROUP  BY order_date;
-- Rewrite fires if MV was refreshed in the last 5 minutes

-- Auditor query — force fresh data
SELECT order_date, SUM(total_cents) AS revenue_cents
FROM   sales.orders
OPTIONS (MAX_STALENESS = '0 seconds')     -- Trino syntax
WHERE  order_date = DATE '2026-07-15'
GROUP  BY order_date;
-- Rewrite skipped; source scan; fresh but expensive

-- Analyst query — tolerate an hour of staleness
SELECT order_date, SUM(total_cents) AS revenue_cents
FROM   sales.orders
OPTIONS (MAX_STALENESS = '1 hour')
WHERE  order_date >= current_date - 365
GROUP  BY order_date;
-- Rewrite fires as long as MV < 60 minutes stale

# Programmatic override — same idea via SQLAlchemy / Trino Python
import trino

conn = trino.dbapi.connect(host="trino", user="analyst")
with conn.cursor() as cur:
    # Session-level MAX_STALENESS applies to every query in the session
    cur.execute("SET SESSION iceberg.mv_max_staleness = '1 hour'")
    cur.execute("""
        SELECT order_date, SUM(total_cents) AS revenue_cents
        FROM   sales.orders
        WHERE  order_date >= current_date - 365
        GROUP  BY order_date
    """)
    rows = cur.fetchall()

Step-by-step explanation.

1. The dashboard query has no OPTIONS clause, so the optimiser falls back to the MV's own freshness_max = 5 minutes. This is the common case; users write ordinary SQL and get MV rewrite for free.
2. The auditor query sets MAX_STALENESS = '0 seconds' — the optimiser interprets this as "any staleness is unacceptable," so no MV rewrite is possible. The engine reads sales.orders directly, paying the higher scan cost in exchange for guaranteed-current numbers. This is the correct behaviour for financial audits.
3. The analyst query sets MAX_STALENESS = '1 hour' — a wider tolerance than the MV's default. This is a common pattern for historical exploration where fresh data is irrelevant (the analyst is looking at trends over the past year). Rewrite fires as long as the MV is less than an hour stale.
4. The SET SESSION variant is a stickier override — applied once at session start, every query in the session uses it. This is the pattern most BI tools use: the tool sets a session-level tolerance based on the dashboard type (real-time dashboard = tight; historical = loose).
5. Precedence: query-level OPTIONS beats session-level SET SESSION beats MV-level freshness_max. This gives users clear control from tightest scope (per-query) to loosest (MV default).

Output.

Query	Tolerance source	Rewrite	Result freshness	Latency
Dashboard	MV default (5 min)	yes (usually)	≤ 5 min	200 ms
Auditor (0s)	query override	no	current	40 s
Analyst (1h)	query override	yes (almost always)	≤ 1 h	200 ms
BI tool session (1h)	session	yes (almost always)	≤ 1 h	200 ms

Rule of thumb. Set the MV freshness_max to the strictest tolerance any regular consumer needs. Allow looser consumers (analysts, offline reports) to widen tolerance per-query. Force-fresh consumers (auditors, compliance) get MAX_STALENESS = '0 seconds'. This maps one MV to many use cases without vendor-specific MV variants.

Worked example — rollup match: coarse query, fine MV

Detailed explanation. The optimiser can rewrite even when the user's GROUP BY is coarser than the MV's. A user asks for weekly revenue; the MV holds daily revenue. The optimiser reads the MV's daily rows, re-aggregates them to weekly on the fly, and returns the result — cheaper than scanning the source and much cheaper than materialising another MV. Walk through the mechanism.

• MV. analytics.daily_revenue grouped by order_date.
• User query. GROUP BY date_trunc('week', order_date) — weekly rollup.
• Optimiser action. Read MV; apply SUM(revenue_cents) GROUP BY week on top; return weekly.

Question. Demonstrate rollup match for a weekly query against a daily MV and verify via EXPLAIN.

Input.

Component	Value
MV grain	day
User query grain	week
Rollup relation	week = date_trunc('week', day)
MV rows scanned	~90 (last 90 days)
Source rows saved	~500 M

Code.

-- MV — daily grain
CREATE MATERIALIZED VIEW analytics.daily_revenue AS
SELECT order_date, SUM(total_cents) AS revenue_cents
FROM   sales.orders
GROUP  BY order_date;

-- User query — weekly rollup on the same source
SELECT date_trunc('week', order_date) AS week,
       SUM(total_cents)               AS revenue_cents
FROM   sales.orders
WHERE  order_date >= current_date - 90
GROUP  BY 1
ORDER  BY 1;

-- EXPLAIN shows rollup-match rewrite
EXPLAIN
SELECT date_trunc('week', order_date) AS week,
       SUM(total_cents)               AS revenue_cents
FROM   sales.orders
WHERE  order_date >= current_date - 90
GROUP  BY 1;

-- Expected EXPLAIN output (rollup rewrite)
Fragment 1
  Output: week, revenue_cents
  Aggregate[SUM]
    key: week
  Project
    week := date_trunc('week', order_date)
  MaterializedViewSubstitution[
    original: sales.orders,
    substituted: analytics.daily_revenue,
    match_kind: rollup,
    freshness_ok: true
  ]
  TableScan[analytics.daily_revenue$storage]
    filter: order_date >= 2026-04-20
    columns: order_date, revenue_cents

Step-by-step explanation.

1. The optimiser recognises date_trunc('week', order_date) as a deterministic function of order_date — a column the MV has. This makes the daily MV a valid input for the weekly rollup.
2. The rewrite plan reads MV storage, applies the date_trunc('week', ...) projection, then re-aggregates with SUM(revenue_cents) GROUP BY week. Because SUM is associative and commutative, SUM(week) = SUM(day₁) + SUM(day₂) + ... + SUM(day₇) — the rollup is mathematically exact.
3. Not every function is rollup-safe. AVG is not directly rollup-safe (average of averages ≠ overall average without the counts); to make it rollup-safe the MV must hold both SUM and COUNT, and the rollup computes SUM(SUM) / SUM(COUNT). DISTINCT COUNT requires HyperLogLog sketches. Percentiles require T-Digest.
4. The rollup match is why a single fine-grained MV can serve many query shapes. Daily MV serves daily, weekly, monthly, quarterly, yearly queries — all rollups of the daily rows. Building four separate MVs (daily + weekly + monthly + quarterly) is wasteful when the optimiser can do it at query time.
5. The trade-off: rollup at query time adds a small aggregation cost on top of the MV scan (~50-200 ms for 90 rows). For pre-computed weekly MV it's free. Rule: only materialise coarser rollups if the query is on the hot path and the extra 100 ms matters.

Output.

Query	MV rows scanned	Extra aggregate cost	Total latency
Daily (exact match)	90	0	150 ms
Weekly (rollup match)	90	~50 ms	200 ms
Monthly (rollup match)	90	~50 ms	200 ms
Yearly (rollup match)	90	~30 ms	180 ms
Weekly (no MV, source scan)	500 M	~40 s	40 s

Rule of thumb. For any aggregate that will be sliced multiple ways temporally (day / week / month / year), materialise only the finest grain and let the optimiser handle rollups. This is 4× cheaper on storage and refresh cost than materialising each grain separately.

Senior interview question on query planning + freshness

A senior interviewer might ask: "Design the freshness contract and rewrite behaviour for a mixed-workload MV that serves a real-time exec dashboard (needs sub-5-minute), a compliance auditor (needs current), and a data-science analyst (tolerates an hour of staleness). How do you set MV-level defaults, expose per-query overrides, monitor rewrite hit-rate, and troubleshoot when rewrite silently stops firing?"

Solution Using tight MV default + per-query override + rewrite-hit-rate observability + freshness alert

-- 1. MV — set the strictest default (dashboard tolerance)
CREATE MATERIALIZED VIEW analytics.orders_metrics
  TBLPROPERTIES (
    'iceberg.mv.freshness.max'    = '5 minutes',    -- dashboard default
    'iceberg.mv.refresh.mode'     = 'incremental',
    'iceberg.mv.storage.location' = 's3://lake/mvs/orders_metrics/'
  )
AS
SELECT customer_id,
       date_trunc('day', order_date) AS day,
       COUNT(*)                       AS order_count,
       SUM(total_cents)               AS total_cents
FROM   sales.orders
GROUP  BY 1, 2;

-- 2. Persona-specific query patterns
-- 2a. Dashboard (uses default 5-min tolerance)
SELECT day, SUM(total_cents) FROM sales.orders
WHERE  day >= current_date - 7 GROUP BY 1;

-- 2b. Auditor (force fresh)
SELECT customer_id, day, total_cents FROM sales.orders
OPTIONS (MAX_STALENESS = '0 seconds')
WHERE  customer_id = 12345 AND day = DATE '2026-07-18';

-- 2c. Analyst (loose tolerance)
SELECT day, SUM(total_cents) FROM sales.orders
OPTIONS (MAX_STALENESS = '1 hour')
WHERE  day >= current_date - 365 GROUP BY 1;

-- 3. Rewrite-hit-rate observability (Trino system table)
SELECT date_trunc('minute', query_start_time) AS minute,
       COUNT(*) FILTER (WHERE mv_substituted)  AS rewritten,
       COUNT(*)                                AS total,
       CAST(COUNT(*) FILTER (WHERE mv_substituted) AS DOUBLE)
         / COUNT(*)                            AS hit_rate
FROM   system.runtime.mv_substitution_log
WHERE  mv_name = 'analytics.orders_metrics'
  AND  query_start_time >= now() - INTERVAL '1' HOUR
GROUP  BY 1
ORDER  BY 1;

# 4. Prometheus alerts
groups:
- name: iceberg-mv-rewrite
  rules:
  - alert: MvRewriteHitRateLow
    expr: iceberg_mv_rewrite_hit_rate{mv="analytics.orders_metrics"} < 0.8
    for: 30m
    labels: {severity: warning}
    annotations:
      summary: "MV {{ $labels.mv }} rewrite hit rate dropped below 80%"
      description: "Likely causes: refresh lag > freshness_max; query pattern shifted; MV columns don't cover new WHERE clauses"

  - alert: MvFreshnessBreach
    expr: iceberg_mv_seconds_since_refresh{mv="analytics.orders_metrics"} > 600
    for: 5m
    labels: {severity: critical}
    annotations:
      summary: "MV {{ $labels.mv }} stale beyond 2x freshness_max"

Step-by-step trace.

Persona	Query tolerance	Rewrite decision	Data returned	Latency
Dashboard	5 min (MV default)	yes when fresh	≤ 5 min stale	200 ms
Auditor	0 seconds (override)	never	current	40 s
Analyst	1 hour (override)	yes almost always	≤ 1 h stale	200 ms
BI session	1 h (SET SESSION)	yes almost always	≤ 1 h stale	200 ms

After deployment, the dashboard picks up MV rewrite transparently and drops p95 latency from 40 s to 200 ms; the auditor's rare queries scan the source directly with no complaints about staleness; the analyst opts into wider tolerance session-wide via SET SESSION; the Prometheus alert catches both stale MVs and query-pattern drift (falling hit-rate signals the MV needs a schema evolution or a new sibling MV).

Output:

Metric	Value
Dashboard p95 latency	200 ms (was 40 s)
Auditor query cost	40 s (unchanged; correct trade)
Analyst query cost	200 ms (was 40 s)
MV rewrite hit-rate p95	92%
MV freshness p95	3 min (under 5-min contract)
Alerts firing	0 in steady state
Storage cost of MV	1× (single MV serves all personas)

Why this works — concept by concept:

• Tight MV default + per-query loosening — one MV storage table serves personas with different tolerances. The MV's own freshness_max matches the strictest consumer; anyone with looser needs opts in via query- or session-level overrides. This avoids the trap of "one MV per tolerance."
• Auditor MAX_STALENESS = 0 — the correctness escape hatch. Compliance-critical queries always read the source; the extra latency is the price of guaranteed freshness. Optimiser respects it silently.
• Rewrite hit-rate observability — the metric that tells you whether the MV is earning its refresh cost. Hit-rate above 80% = MV is doing its job. Below = investigate: is refresh keeping up? Have query patterns shifted? Do users need a new column?
• Freshness alert at 2× freshness_max — catches refresh regressions before they degrade the user experience. If the MV should be < 5 min stale and it's been > 10 min, refresh is broken and rewrite is silently disabled — a critical incident.
• Cost — one MV, one storage footprint, one refresh cost, three (or more) persona-tolerances served correctly. Compared to per-persona MV variants (3× storage + 3× refresh), the query-rewrite + tolerance-override model is linear-in-MV, not linear-in-persona.

SQL
Topic — sql
SQL query optimisation and rewrite problems

Practice →

Joins Topic — joins Join and denormalised-view problems

Practice →

---

5. Migration patterns + interview signals

From dynamic tables, cloud-warehouse MVs, and dbt incremental models to Iceberg MVs — the four canonical migrations and the four interview signals

The mental model in one line: open table format mv migration is the year-long project every lakehouse team runs at least once between 2025 and 2027 — from proprietary dynamic tables (Snowflake), from cloud-warehouse MVs (BigQuery, Redshift, Databricks), and from dbt incremental models (any warehouse) onto the open Iceberg MV spec — and each source pattern has its own translation recipe, its own migration risk profile, and its own interview signal that senior data engineers must be able to walk through end-to-end. This section codifies the three canonical migrations and closes with the four interview signals that separate senior candidates from staff-level candidates on any lakehouse-focused hiring loop.

Migration axis — what changes and what stays the same.

• What stays. The aggregate's SQL logic (aggregation columns, group-by keys, filters) is portable. The consumer contracts (dashboard latency, freshness expectation) are portable.
• What changes. Storage moves from vendor-internal to Iceberg on object storage. Refresh moves from vendor-managed to spec-driven (external scheduler or engine-driven). Query rewrite moves from vendor-optimiser to spec-defined rewrite (per engine). Freshness moves from TARGET_LAG (Snowflake) or refresh-interval to MAX_STALENESS.
• What breaks. Any vendor-specific SQL function needs a portable equivalent (Snowflake's LATERAL FLATTEN → Trino/Spark equivalents). Any vendor-specific refresh feature (Snowflake's INITIALIZE = ON_CREATE) needs an external scheduler equivalent.

Migration 1 — Snowflake dynamic table → Iceberg MV.

• Source shape. Dynamic table with TARGET_LAG and Snowflake-managed refresh. Storage in Snowflake internal.
• Target shape. Iceberg MV with MAX_STALENESS and external Airflow-driven refresh. Storage in Iceberg on S3.
• Translation checklist. (a) Convert DDL to CREATE MATERIALIZED VIEW with Iceberg TBLPROPERTIES; (b) map TARGET_LAG to freshness_max; (c) replace Snowflake-specific functions with portable equivalents; (d) set up Airflow refresh at MAX_STALENESS ÷ 2.5 cadence; (e) parallel-run for 30 days to reconcile.
• Risk. Dynamic tables have automatic incremental refresh handled by Snowflake; the Iceberg equivalent requires you to pick the right refresh strategy (append-only vs upsert) explicitly. Getting this wrong = correctness bug.

Migration 2 — BigQuery / Redshift / Databricks MV → Iceberg MV.

• Source shape. Cloud-warehouse MV with engine-managed refresh. Storage in warehouse.
• Target shape. Iceberg MV on object storage.
• Translation checklist. (a) Convert DDL; (b) map cloud-warehouse refresh interval to MAX_STALENESS; (c) reconcile query-rewrite behaviour (BigQuery's smart tuning is aggressive; Iceberg MV rewrite depends on the reader engine); (d) parallel-run.
• Risk. BigQuery MVs support only a subset of aggregations for automatic incremental refresh; the same restrictions apply to Iceberg MVs but with different semantics. Feature-detect before committing.

Migration 3 — dbt incremental model → Iceberg MV.

• Source shape. dbt model with materialized='incremental' + custom is_incremental() filter logic. Runs on dbt build schedule (typically daily/hourly).
• Target shape. Iceberg MV with declarative refresh; the is_incremental() logic is replaced by the engine's incremental refresh mechanism.
• Translation checklist. (a) Extract the aggregate SQL (the non-is_incremental branch); (b) confirm it fits the incremental-refresh contract (SUM/COUNT-shaped over an append-only or upsert-tracked source); (c) create the MV; (d) migrate the dbt DAG to trigger REFRESH MATERIALIZED VIEW instead of dbt run; (e) parallel-run.
• Risk. dbt incremental models often bake in ad-hoc watermark logic that the MV spec doesn't reproduce exactly. Reconcile aggressively.

Migration 4 — nothing → Iceberg MV (greenfield).

• Source shape. No MV; queries scan the source table directly.
• Target shape. Iceberg MV with sensible defaults.
• Translation checklist. (a) Identify hot query patterns via query log analysis; (b) design MV columns to cover the union of filters and group-bys; (c) pick refresh cadence based on tightest consumer tolerance; (d) monitor rewrite hit-rate; (e) iterate.
• Risk. Choosing the wrong grain — too fine (refresh expensive) or too coarse (rewrite fails for many queries). Start with the natural grain of your dashboards.

The four interview signals to watch for.

• Signal 1 — freshness semantics. Do you say "snapshot-consistent" without prompting? Do you distinguish MAX_STALENESS from TARGET_LAG? Senior signal.
• Signal 2 — refresh cadence + strategy. Do you separate "how often" (cadence) from "how much" (full vs incremental vs change-scan)? Do you cite the MAX_STALENESS ÷ 2.5 scheduler rule? Senior signal.
• Signal 3 — query rewrite. Do you name rewrite as an optimiser hint not a table you SELECT from? Do you cite exact / filter / rollup / join match kinds? Senior signal.
• Signal 4 — cost / latency trade-off. Do you frame MV vs source scan as a break-even calculation (refresh_freq × refresh_cost < query_freq × source_cost)? Do you cite the read-heavy vs write-heavy heuristic? Senior signal.

Common interview probes on migration + signals.

• "How would you migrate this Snowflake dynamic table to an Iceberg MV?" — parallel-run recipe.
• "What breaks when you move a dbt incremental model to an MV?" — watermark logic, is_incremental() guard clauses, ephemeral dependencies.
• "How do you decide whether the migration is worth it?" — multi-engine reach + read-hotness + freshness contract + storage cost.
• "What's the rollback plan if the MV migration produces wrong numbers?" — the source is still there; disable the MV (or set MAX_STALENESS = 0); investigate; re-migrate.

Worked example — migrating a Snowflake dynamic table to an Iceberg MV

Detailed explanation. The most common 2026 migration: an existing Snowflake dynamic table analytics.daily_revenue_dt with TARGET_LAG = '5 minutes' needs to move to an Iceberg MV that Snowflake reads via its Iceberg-tables feature (so Snowflake dashboards keep working) and that Trino / Spark can also read (the whole point of the migration). Walk through the parallel-run migration end-to-end.

• Source. Snowflake dynamic table analytics.daily_revenue_dt, TARGET_LAG = '5 minutes', incremental refresh managed by Snowflake.
• Target. Iceberg MV analytics.daily_revenue in S3, MAX_STALENESS = '5 minutes', refresh driven by Airflow.
• Parallel-run. 30 days of dual-write; reconcile daily; cut over when < 0.01% drift.

Question. Design the migration DDL, the parallel-run reconcile query, and the cutover plan.

Input.

Component	Before (dynamic table)	After (Iceberg MV)
DDL	CREATE DYNAMIC TABLE ... TARGET_LAG = '5 min'	CREATE MATERIALIZED VIEW ... freshness_max = '5m'
Storage	Snowflake internal	s3://lake/mvs/daily_revenue/
Refresh	Snowflake managed	Airflow cron every 2 min
Snowflake reads	native	via external Iceberg table
Trino reads	impossible	native

Code.

-- 1. Existing Snowflake dynamic table (stays for parallel run)
-- CREATE OR REPLACE DYNAMIC TABLE analytics.daily_revenue_dt
--   TARGET_LAG = '5 minutes' WAREHOUSE = compute_wh
-- AS SELECT order_date, SUM(total_cents) FROM sales.orders GROUP BY 1;

-- 2. New Iceberg MV — same aggregate, portable spec
CREATE MATERIALIZED VIEW analytics.daily_revenue
  TBLPROPERTIES (
    'iceberg.mv.refresh.mode'    = 'incremental',
    'iceberg.mv.freshness.max'   = '5 minutes',
    'iceberg.mv.storage.location'= 's3://lake/mvs/daily_revenue/'
  )
AS
SELECT order_date,
       SUM(total_cents) AS revenue_cents,
       COUNT(*)         AS order_count
FROM   sales.orders
GROUP  BY order_date;

-- 3. Snowflake external Iceberg table so Snowflake dashboards can read
--    the new MV storage
CREATE OR REPLACE ICEBERG TABLE analytics.daily_revenue_iceberg
  EXTERNAL_VOLUME = 'lake_vol'
  CATALOG         = 'glue_catalog'
  BASE_LOCATION   = 'mvs/daily_revenue/';

-- 4. Reconcile query — run nightly during parallel-run window
SELECT COALESCE(dt.order_date, ic.order_date)      AS order_date,
       dt.revenue_cents                            AS dt_revenue,
       ic.revenue_cents                            AS ic_revenue,
       COALESCE(dt.revenue_cents, 0) - COALESCE(ic.revenue_cents, 0) AS drift
FROM   analytics.daily_revenue_dt      dt
FULL OUTER JOIN analytics.daily_revenue_iceberg ic
       ON dt.order_date = ic.order_date
WHERE  COALESCE(dt.revenue_cents, 0) <> COALESCE(ic.revenue_cents, 0)
ORDER  BY drift DESC;

# 5. Cutover plan — automated once drift is < 0.01% for 30 days
from datetime import datetime, timedelta

def eligible_for_cutover(drift_pct_series: list[float], threshold: float = 0.0001) -> bool:
    """True if every day in the last 30 days has drift below threshold."""
    return len(drift_pct_series) >= 30 and all(d < threshold for d in drift_pct_series)

def perform_cutover():
    # 1. Update BI tool connections from analytics.daily_revenue_dt to analytics.daily_revenue_iceberg
    # 2. Deprecate the dynamic table (SUSPEND, then DROP after 7 more days)
    # 3. Update dashboards / notebooks references
    # 4. Keep the Iceberg MV as the canonical aggregate
    pass

Step-by-step explanation.

1. Step 2 creates the new Iceberg MV. Same aggregate SQL as the dynamic table; new spec-driven refresh contract. Because the source (sales.orders) is already an Iceberg table (in this scenario), no source-side migration is needed.
2. Step 3 exposes the MV storage table to Snowflake as an external Iceberg table. Snowflake queries against analytics.daily_revenue_iceberg land on the same S3 bytes that Trino, Spark, and Athena would read. One storage; multi-engine access.
3. Step 4's reconcile query runs nightly during the 30-day parallel-run window. It full-outer-joins the dynamic table and the Iceberg MV on order_date, computes the per-row drift, and returns non-matching rows. Zero drift for 30 consecutive days is the cutover gate.
4. During parallel-run, both refresh paths run: Snowflake manages the dynamic table; Airflow drives REFRESH MATERIALIZED VIEW analytics.daily_revenue every 2 minutes. Total refresh cost doubles temporarily — the price of a safe migration.
5. Cutover updates BI tool connections, then suspends the dynamic table for 7 days as a safety net, then drops it. Rollback during those 7 days is one connection-string change; after the DT is dropped, rollback requires re-CREATEing it (still cheap — the aggregate SQL is unchanged).

Output.

Phase	Days	Storage cost	Refresh cost	Reader engines
Pre-migration	—	1× (DT only)	1× (DT only)	Snowflake only
Parallel-run	30	2× (DT + Iceberg)	2× (DT + Iceberg)	Snowflake + Trino + Spark
Cutover T+0	1	1× (Iceberg only)	1× (Iceberg only)	Snowflake + Trino + Spark
DT dropped	7 later	1× (Iceberg only)	1× (Iceberg only)	Snowflake + Trino + Spark

Rule of thumb. For any dynamic-table-to-Iceberg-MV migration, budget a 30-day parallel-run and a 7-day post-cutover safety net. Both are cheap insurance against subtle aggregation-semantics drift between vendor and open-spec refresh implementations.

Worked example — migrating a dbt incremental model to an Iceberg MV

Detailed explanation. A common 2026 refactor: a dbt incremental model models/marts/daily_revenue.sql that runs on the dbt DAG becomes an Iceberg MV with engine-driven refresh. The migration removes ~30 lines of is_incremental() boilerplate and consolidates freshness under one contract. Walk through the file-by-file change.

• dbt model. models/marts/daily_revenue.sql with {{ config(materialized='incremental', ...) }} and an is_incremental() filter.
• Target. Iceberg MV that dbt no longer materialises but does trigger via REFRESH MATERIALIZED VIEW in a dbt run-operation or via Airflow.

Question. Convert the dbt model to an Iceberg MV and rewire the dbt DAG to trigger refresh.

Input.

Component	Before (dbt incremental)	After (Iceberg MV)
File	models/marts/daily_revenue.sql	models/marts/daily_revenue.sql (or macro)
Materialization	materialized='incremental'	materialized='iceberg_mv' (adapter)
Watermark	is_incremental() guard	engine-driven snapshot diff
Refresh trigger	dbt build	REFRESH MATERIALIZED VIEW

Code.

-- BEFORE: models/marts/daily_revenue.sql (dbt incremental)
{{ config(
    materialized = 'incremental',
    unique_key   = 'order_date',
    incremental_strategy = 'merge'
) }}

SELECT order_date,
       SUM(total_cents) AS revenue_cents,
       COUNT(*)         AS order_count
FROM   {{ source('sales', 'orders') }}
{% if is_incremental() %}
WHERE  order_date >= (SELECT COALESCE(MAX(order_date), '1970-01-01') FROM {{ this }})
{% endif %}
GROUP  BY order_date

-- AFTER: models/marts/daily_revenue.sql (Iceberg MV via dbt-iceberg adapter)
{{ config(
    materialized      = 'iceberg_mv',
    freshness_max     = '5 minutes',
    refresh_mode      = 'incremental',
    storage_location  = 's3://lake/mvs/daily_revenue/'
) }}

SELECT order_date,
       SUM(total_cents) AS revenue_cents,
       COUNT(*)         AS order_count
FROM   {{ source('sales', 'orders') }}
GROUP  BY order_date
-- No is_incremental() guard — the MV spec drives incremental refresh

# Airflow DAG replaces `dbt build --select daily_revenue`
from airflow.decorators import dag, task
from airflow.providers.dbt.cloud.operators.dbt import DbtCloudRunJobOperator
from datetime import datetime

@dag(schedule="*/2 * * * *", start_date=datetime(2026, 7, 19), catchup=False, max_active_runs=1)
def refresh_daily_revenue_mv():

    @task
    def refresh():
        # Instead of dbt build (which rewrites the incremental model),
        # trigger the MV refresh via the engine
        import trino
        conn = trino.dbapi.connect(host="trino", user="mv_refresher")
        with conn.cursor() as cur:
            cur.execute("REFRESH MATERIALIZED VIEW analytics.daily_revenue")

    refresh()

Step-by-step explanation.

1. The "before" dbt model uses is_incremental() to guard the watermark filter — a standard dbt pattern. On first run, no filter is applied (full table scan). On subsequent runs, the filter reads the current MAX from the target table and only pulls rows newer than that. This is dbt's manual incremental-refresh implementation.
2. The "after" MV drops the is_incremental() guard entirely. The Iceberg MV spec drives incremental refresh via snapshot diffs on sales.orders; no user-side watermark bookkeeping is needed. The SQL becomes ~5 lines shorter and impossible to mis-configure (e.g. wrong unique_key, wrong incremental_strategy).
3. The dbt-iceberg adapter (a real, in-development adapter as of 2026) translates materialized='iceberg_mv' into a CREATE MATERIALIZED VIEW statement against the configured Iceberg catalog. The freshness_max and refresh_mode become TBLPROPERTIES on the MV.
4. The Airflow DAG replaces dbt build --select daily_revenue with a direct REFRESH MATERIALIZED VIEW call to Trino. The refresh no longer runs through dbt's model-execution engine — dbt still owns the DDL definition, but the runtime refresh is engine-managed. This removes ~30 seconds of dbt overhead per refresh.
5. Trade-off: dbt's incremental-model conveniences (test hooks, snapshots, seeds) don't translate 1:1 to MV land. Some teams keep dbt for the modelling layer and use MVs only for the leaf aggregates that need multi-engine reach.

Output.

Aspect	Before (dbt incremental)	After (Iceberg MV)
SQL lines	~15 (with is_incremental)	~5 (no guard)
Refresh trigger	dbt build (~30 s overhead)	REFRESH MV (~2 s overhead)
Multi-engine access	no (single warehouse)	yes (any Iceberg reader)
Watermark logic	manual (unique_key + strategy)	spec-managed (snapshot diff)
Rollback	drop + rerun dbt	ALTER MV STOP; parallel path

Rule of thumb. For any dbt incremental model whose sole purpose is a pure aggregate over an Iceberg source, migrate to an Iceberg MV. Keep dbt for the modelling layer, feature engineering, and tests. Use MVs for the leaf caching layer. Don't force one tool to do both jobs.

Worked example — the four interview signals in a mock loop

Detailed explanation. The senior lakehouse hiring loop for a Staff+ role probes four signals across every Iceberg MV question. A mock loop with a hypothetical candidate walks through each probe and shows the "weak" vs "senior" answer. Use this as a rehearsal script before your next lakehouse-focused interview.

• Probe 1. "Explain freshness for an Iceberg MV."
• Probe 2. "How do you schedule refresh to meet a 5-minute freshness SLA?"
• Probe 3. "How does query rewrite decide when to use the MV?"
• Probe 4. "When is an Iceberg MV not worth it?"

Question. Draft the senior-level answer to each probe with the interview-signal cue in mind.

Input.

Probe	Weak answer	Senior answer
Freshness	"it's fresh when we refresh it"	"snapshot-consistent; pinned source snapshot IDs; is_stale = ANY(pinned ≠ current)"
Refresh cadence	"we cron it every 5 minutes"	"cron at MAX_STALENESS ÷ 2.5; max_active_runs=1; alert on p95 > 70% of contract"
Query rewrite	"we SELECT from the MV"	"optimiser matches logical plan + freshness gate + cost estimate; exact/filter/rollup/join match"
When not worth it	"always use MVs"	"read-hotness × source-scan-cost < refresh-freq × refresh-cost; single-consumer + light-read = MV overhead"

Code.

Mock loop answer template
=========================

Probe 1 — freshness
  "Iceberg MVs are snapshot-consistent, not strictly-consistent. The
   view-metadata pins source snapshot IDs at last refresh; is_stale is
   computed as ANY(current source snapshot != pinned). Users get a
   MAX_STALENESS contract — the engine won't rewrite to the MV if it's
   older than the contract."

Probe 2 — refresh cadence
  "Schedule the refresh at MAX_STALENESS ÷ 2.5 (every 2 minutes for a
   5-minute contract). Enforce max_active_runs=1 to prevent overlaps.
   Emit refresh duration to Prometheus; alert on p95 > 70% of the
   contract. Use incremental-change-scan for upsert sources, plain
   incremental-append-scan for append-only sources."

Probe 3 — query rewrite
  "The optimiser parses the incoming query, enumerates candidate MVs
   whose logical plan is a superset or exact match of the query, gates
   each by freshness against the user's tolerance (per-query, session,
   or MV-default), and picks the cheapest surviving plan. Match kinds
   are exact, filter-pushdown, rollup (like daily → weekly), and join."

Probe 4 — when not worth it
  "MV cost = refresh cost + storage. MV benefit = (query freq) ×
   (source scan cost saved). Break-even is roughly refresh × refresh_cost
   < query × source_cost. If the query fires once a day and refresh
   fires every 5 minutes, you're paying 288× refresh for one lookup —
   pure overhead. MVs shine on read-heavy hot paths."

Step-by-step explanation.

1. Probe 1 tests whether you understand that MV freshness is not real-time — it's a bounded staleness contract. The senior signal is "snapshot-consistent" plus the pinned-snapshot mechanism; weak candidates say "it's fresh when we refresh" without naming the contract.
2. Probe 2 tests operational maturity. Senior candidates cite the MAX_STALENESS ÷ 2.5 scheduler rule and max_active_runs=1 from real production experience; weak candidates say "we cron it" without the safety margin math.
3. Probe 3 tests optimiser-mental-model depth. Senior candidates name the four match kinds and the freshness gate; weak candidates treat the MV as a table you explicitly SELECT from — which misses the whole point of rewrite.
4. Probe 4 is the "senior gate" question — do you know when the pattern doesn't apply? MVs are pure overhead for cold queries; the break-even calculation is the mental model that separates people who apply the pattern reflexively from people who apply it deliberately.
5. The four signals compound. A candidate who nails freshness (Signal 1) but flubs the break-even (Signal 4) gets a "senior-ish" rating — they know the mechanism but not the trade-off. Nailing all four consistently is a Staff-level signal.

Output.

Signal	Rating	Notes
Freshness semantics	Senior	Named "snapshot-consistent" + pinning
Refresh cadence	Senior	Cited MAX_STALENESS ÷ 2.5 rule + overlap prevention
Query rewrite	Senior	Named all four match kinds + freshness gate
When not worth it	Staff	Cited break-even formula + cold-query overhead

Rule of thumb. Rehearse all four signals as a monologue before your next interview. Interviewers ask them in different orders and with different framings, but the underlying signals are stable. Ace all four consistently → Staff-level Iceberg MV rating.

Senior interview question on migration + signals

A senior interviewer might close with: "Walk me through a real migration you've done from a proprietary MV to an Iceberg MV — the DDL translation, the parallel-run reconcile, the cutover, and the lessons learned. Then explain which of the four MV worlds you'd pick for a new greenfield lakehouse and why."

Solution Using Snowflake dynamic table → Iceberg MV migration with 30-day parallel run + Airflow refresh + Trino query rewrite

-- 1. Old Snowflake dynamic table (still running during parallel run)
-- CREATE DYNAMIC TABLE analytics.daily_revenue_dt
--   TARGET_LAG = '5 minutes' WAREHOUSE = compute_wh
-- AS SELECT ... FROM sales.orders GROUP BY 1;

-- 2. New Iceberg MV — canonical target
CREATE MATERIALIZED VIEW analytics.daily_revenue
  TBLPROPERTIES (
    'iceberg.mv.refresh.mode'    = 'incremental',
    'iceberg.mv.freshness.max'   = '5 minutes',
    'iceberg.mv.storage.location'= 's3://lake/mvs/daily_revenue/'
  )
AS
SELECT order_date, SUM(total_cents) AS revenue_cents, COUNT(*) AS order_count
FROM   sales.orders
GROUP  BY order_date;

-- 3. Snowflake external Iceberg table so Snowflake dashboards read
--    the new MV storage without config changes
CREATE OR REPLACE ICEBERG TABLE analytics.daily_revenue_new
  EXTERNAL_VOLUME = 'lake_vol'
  CATALOG         = 'glue_catalog'
  BASE_LOCATION   = 'mvs/daily_revenue/';

# 4. Airflow DAG — refresh at 2-min cadence for 5-min freshness contract
from airflow.decorators import dag, task
from datetime import datetime

@dag(schedule="*/2 * * * *", start_date=datetime(2026, 7, 19),
     catchup=False, max_active_runs=1, dagrun_timeout=300)
def refresh_daily_revenue_mv():

    @task
    def refresh():
        import trino
        conn = trino.dbapi.connect(host="trino", user="mv_refresher")
        with conn.cursor() as cur:
            cur.execute("REFRESH MATERIALIZED VIEW analytics.daily_revenue")

    @task
    def reconcile():
        # Nightly reconcile query — compares old DT to new MV
        import trino
        conn = trino.dbapi.connect(host="trino", user="mv_reconciler")
        with conn.cursor() as cur:
            cur.execute("""
                SELECT SUM(ABS(dt.revenue_cents - ic.revenue_cents))
                FROM   snowflake.analytics.daily_revenue_dt      dt
                FULL OUTER JOIN iceberg.analytics.daily_revenue  ic
                       ON dt.order_date = ic.order_date
            """)
            drift = cur.fetchone()[0] or 0
            print(f"total_drift_cents = {drift}")
            if drift > 100:
                raise RuntimeError("Drift exceeds tolerance; block cutover")

    refresh() >> reconcile()

dag = refresh_daily_revenue_mv()

-- 5. Query rewrite verification (Trino)
EXPLAIN
SELECT order_date, SUM(total_cents) FROM sales.orders
WHERE  order_date >= current_date - 30 GROUP BY 1;
-- Expected: MaterializedViewSubstitution against analytics.daily_revenue

Step-by-step trace.

Migration phase	Days	Actions	Rollback story
Design	3	DDL translation, freshness contract, refresh strategy	reversible; no data written
Deploy Iceberg MV	1	CREATE MV, first refresh	reversible; DROP MV
Parallel run	30	Airflow refresh + nightly reconcile	reversible; keep DT running
BI cutover	1	switch dashboard connections to Iceberg MV	reversible; flip connection back
DT suspend	7	SUSPEND dynamic table; monitor for issues	reversible; RESUME DT
DT drop	1	DROP DYNAMIC TABLE	one-way after this point

After the migration, the aggregate lives in one place (S3 Iceberg MV), is refreshed by one scheduler (Airflow), is read by three engines (Snowflake, Trino, Spark), and honours the same 5-minute freshness contract as the old dynamic table. The dbt DAG loses one incremental model; the Snowflake bill drops by the dynamic table's refresh cost.

Output:

Metric	Before	After
MV storage	Snowflake internal (1×)	S3 Iceberg (1×)
Refresh compute	Snowflake credits (~50/day)	Trino cluster (~1 vCPU-hour/day)
Reader engines	Snowflake only	Snowflake + Trino + Spark
Freshness contract	TARGET_LAG=5m	MAX_STALENESS=5m
Query rewrite	Snowflake DT (automatic)	Trino MV substitution (automatic)
Reconcile drift (30 days)	n/a	< 0.001%
BI dashboard latency	300 ms	300 ms (unchanged)
Ad-hoc Trino query	40 s (full scan)	200 ms (MV)

Why this works — concept by concept:

• Parallel-run gate on drift — the reconcile task blocks cutover until the aggregate drift between the old and new paths is provably below tolerance. This catches aggregation-semantics mismatches (e.g. Snowflake's SUM handling of NULL vs Iceberg's) before they hit dashboards.
• Snowflake external Iceberg table — the migration bridge. Snowflake dashboards keep working during and after cutover because they read the same MV storage via Snowflake's Iceberg-table feature. The BI tool needs zero config changes; just a connection swap at cutover.
• 2-minute Airflow cadence — honours the 5-minute freshness contract with 2.5× safety margin, exactly as prescribed. max_active_runs=1 prevents overlap; dagrun_timeout=300 matches the contract for loud failure.
• Multi-engine reader story — the entire justification. Post-cutover, adding Spark for backfills is a zero-cost read (it just points at the Iceberg MV); adding Athena for ad-hoc SQL is a zero-cost read. Every future engine is free.
• Cost — one 30-day parallel-run window (double refresh cost temporarily), one Airflow DAG, one nightly reconcile query, one Iceberg storage table. Post-migration: 1× storage, 1× refresh, 3× (or more) readers. Compared to running per-engine MVs (N× everything), the payback period is under 3 months for any lakehouse with more than one engine.

SQL
Topic — sql
SQL migration and reconciliation problems

Practice →

Streaming
Topic — streaming
Streaming and lakehouse aggregate problems

Practice →

---

Cheat sheet — Iceberg MV recipes

• Which MV world when. Iceberg MV is the 2026 default when more than one query engine (Spark, Trino, Snowflake, Athena, Databricks) needs to read the same aggregate; the single storage table serves every reader and refresh cost is paid once. Snowflake dynamic tables win inside a Snowflake-only stack because the vendor optimiser and refresh integration are tighter. Cloud-warehouse MVs (BigQuery, Redshift, Databricks) win for their single-engine workloads. dbt incremental models win for batch-only, warehouse-native aggregates that also need dbt's tests, snapshots, and seeds. Streaming MV engines (Materialize, RisingWave, dynamic tables with tight lag) win for sub-second freshness that Iceberg's snapshot cadence cannot serve.

• Iceberg MV DDL template. CREATE MATERIALIZED VIEW <name> TBLPROPERTIES ('iceberg.mv.refresh.mode' = 'incremental', 'iceberg.mv.freshness.max' = '<duration>', 'iceberg.mv.storage.location' = 's3://.../<name>/') AS <select-statement> — includes source tables, group-bys, filters, and only functions that are portable across the dialects you plan to read from. Add multi-dialect representations after creation with ALTER MATERIALIZED VIEW ... ADD REPRESENTATION for cross-engine execution. Set write.format.default = 'parquet' and write.parquet.compression-codec = 'zstd' for optimal storage footprint on hot-path aggregates.

• view-metadata.json inspection recipe. Load the view via PyIceberg (catalog.load_view(name)), read metadata.materialization.source_snapshot_ids (the source pin), metadata.materialization.last_refreshed_timestamp_ms (the freshness timestamp), and metadata.materialization.storage_table_uuid (the pointer to the storage table). Compute is_stale as any(source.metadata.current_snapshot_id != pinned for source, pinned in source_snapshot_ids.items()). Wrap this in a Prometheus exporter emitting iceberg_mv_seconds_since_refresh{mv=...} and iceberg_mv_is_stale{mv=...} for cross-engine freshness observability.

• Incremental refresh recipe — append-only source. Use the Iceberg incremental-append-scan read mode with start-snapshot-id = pinned_id and end-snapshot-id = current_id. The scan reads only data files added between the two snapshots. Aggregate the delta with the same SQL as the MV definition and append to the MV storage table — no MERGE needed because there are no UPDATEs or DELETEs to reconcile. After the append, atomically advance the view-metadata's source-snapshot pin to current_id. Complexity: O(delta size), typically 100-1000× cheaper than full refresh.

• Incremental refresh recipe — upsert / delete source. Use incremental-change-scan which exposes the per-row _iceberg_change_type marker (INSERT, UPDATE_BEFORE, UPDATE_AFTER, DELETE). Compute signed contributions per aggregate key: +1 for INSERT and UPDATE_AFTER, -1 for DELETE and UPDATE_BEFORE. Group by aggregate key and sum. Filter out no-op deltas (WHERE delta != 0). MERGE into MV storage table; MATCHED = add signed delta to existing MV row; NOT MATCHED = insert with the delta. This is the only correct incremental refresh path for a source with mutations.

• SLA-bound refresh cadence. For a MAX_STALENESS = X contract, schedule refresh at X ÷ 2.5 cadence (2 minutes for a 5-minute contract). Enforce max_active_runs=1 in the scheduler to prevent overlapping refreshes that would double-count deltas. Set dagrun_timeout = X so refresh fails loudly rather than silently blowing the contract. Emit refresh duration to Prometheus; alert on p95 > X × 0.7 (approaching the contract) and on mv_refresh_success == 0 for 5m (sustained failure).

• Freshness contract + query rewrite. Set MV-level freshness_max to the strictest consumer's tolerance (e.g. dashboard = 5 min). Allow looser consumers (analysts, offline reports) to widen via per-query OPTIONS (MAX_STALENESS = '1 hour') or session-level SET SESSION iceberg.mv_max_staleness = '1 hour'. Force-fresh consumers (auditors, compliance) use OPTIONS (MAX_STALENESS = '0 seconds') which disables rewrite for that query and forces a source scan. Precedence: query > session > MV default.

• Multi-dialect representation trick. For any MV read by more than one engine, add a per-dialect representation at creation time using ANSI-standard functions only where possible. Non-portable functions (APPROX_PERCENTILE vs approx_percentile, LATERAL FLATTEN vs UNNEST) need each dialect written with matching semantics. Engines resolve their own dialect first; foreign-dialect execution is a warning-emitting fallback. Store the representations under versions[current].representations[].

• Rollup match for finer-grained MV. A daily MV can serve daily, weekly, monthly, quarterly, and yearly queries via rollup match — the optimiser reads MV storage, applies date_trunc('week', day) on top, then re-aggregates with SUM(SUM) GROUP BY week. Materialise only the finest grain worth storing; let the optimiser handle coarser rollups. Works for associative aggregates (SUM, COUNT, MIN, MAX); AVG requires storing both SUM and COUNT; DISTINCT COUNT requires HyperLogLog sketches; percentiles require T-Digest sketches.

• Break-even for MV vs source scan. MV cost = (refresh_freq × refresh_cost) + storage_cost. MV benefit = (query_freq × source_scan_cost_saved) - mv_scan_cost. Break-even: refresh_freq × refresh_cost ≈ query_freq × source_scan_cost_saved. If refresh fires 288 times/day but the query fires once/day, the MV is pure overhead. If the query fires 10 000 times/day, the MV is a 10 000× read-cost saving. Design MVs for read-heavy hot paths only.

• Migration parallel-run pattern. For any proprietary-MV → Iceberg-MV migration, run both refresh paths in parallel for 30 days. Nightly reconcile query full-outer-joins the two aggregates on the group-by key; alert if aggregate drift > tolerance (typically 0.01% or a small absolute cent value). Cut over BI dashboards to the Iceberg MV via a Snowflake external Iceberg table (or equivalent) so no dashboard SQL changes. Suspend the old MV for 7 days as a safety net; drop after successful post-cutover monitoring.

• dbt incremental → Iceberg MV translation. Replace {{ config(materialized='incremental', unique_key=..., incremental_strategy='merge') }} with {{ config(materialized='iceberg_mv', freshness_max=..., refresh_mode='incremental') }} (dbt-iceberg adapter). Remove the {% if is_incremental() %} WHERE ... {% endif %} guard — the MV spec drives incremental refresh via snapshot diff, no manual watermark logic needed. Rewire the dbt DAG to trigger REFRESH MATERIALIZED VIEW from Airflow instead of running dbt build --select <model> for the aggregate.

• Rollback plan for every MV deployment. Keep the source table intact; MVs are always a caching layer, never a source of truth. If the MV produces wrong numbers, set MAX_STALENESS = 0 on the consumer session to force source scans while you investigate. If refresh is broken, disable the Airflow DAG; the MV will become stale but reads still work (they just fall back to source scans automatically). Never delete the storage table without at least one confirmed reader migration to source scans — otherwise you break every rewrite consumer instantly.

• Failure semantics reminder. MV storage table corruption: rebuild from source via full REFRESH MATERIALIZED VIEW. Source-snapshot pin lost: full refresh (bootstrap) and re-pin. Refresh job crashes mid-run: pin was not advanced, next run replays the delta safely (idempotent by design in signed-reconciliation MERGE). Reader engine doesn't support MV rewrite: user explicitly SELECTs from <name>$storage (the storage table) — degraded but functional. Every failure has a bounded recovery.

• Storage retention + snapshot expiry. MV storage tables accumulate snapshots on every refresh — one per refresh cycle. At a 2-minute cadence, that's 720 snapshots/day. Configure Iceberg snapshot expiry: ALTER TABLE ... EXECUTE expire_snapshots(retention_threshold => '7d', min_snapshots_to_keep => 100) — keeps 7 days for time-travel + rollback, drops older. Compaction (OPTIMIZE) runs weekly to rewrite small files from frequent refreshes into larger Parquet files. Both jobs are separate Airflow DAGs, not part of the refresh path.

• When to add a new MV vs extend an existing one. If the new query pattern shares the source table and can be served by a rollup match against an existing MV, do not create a new MV. If the query needs columns the existing MV doesn't hold, first try adding those columns to the existing MV (schema evolution + full refresh). Only create a new MV when the aggregate shape is genuinely different (different group-by keys, different source tables). Every new MV is a new refresh cost + storage footprint + observability surface.

Frequently asked questions

What is an Iceberg materialized view?

An iceberg materialized view is a first-class object in the Apache Iceberg spec that combines a view-metadata.json (schema, versions, per-dialect SQL representations, and a materialization pointer) with a backing Iceberg storage table that holds the cached rows produced by executing the view's defining query. The MV is snapshot-consistent — it pins the source table snapshot IDs at last refresh in the materialization metadata, and any engine that reads the MV can compute is_stale by comparing those pinned IDs against each source's current snapshot ID. This differs sharply from proprietary MVs (Snowflake dynamic tables, BigQuery MVs, Databricks MVs) that store data inside the vendor's engine and expose freshness only through the vendor's own APIs.

The design goal is engine portability: the same MV can be refreshed by Spark, queried by Trino, read by Snowflake (via external Iceberg tables), and scanned by Athena, all reading the same S3 storage table with the same freshness contract. Refresh moves rows into the storage table using either full refresh (rebuild from scratch) or one of the incremental modes (incremental-append-scan for append-only sources, incremental-change-scan for upsert/delete sources); the view-metadata is updated atomically to point at the new pinned source snapshots. In 2026, Spark 4.0, Trino 450+, and Snowflake's Iceberg integration all ship reference implementations; other engines are following.

Iceberg MV vs Snowflake dynamic table vs Databricks materialized view — when do I pick each?

Pick Iceberg MV when more than one query engine needs to read the same cached aggregate — a lakehouse with Snowflake dashboards, Trino ad-hoc analysts, and Spark backfill jobs all pointed at the same MV storage table on S3. Iceberg MV pays refresh cost once and serves any Iceberg-aware reader; the trade-off is that refresh scheduling is your responsibility (Airflow, event-driven, or continuous streaming), the spec's newer features aren't uniformly implemented across engines yet, and vendor-specific optimiser tricks (Snowflake's smart-materialization, Databricks Photon acceleration) don't apply.

Pick Snowflake dynamic tables for pure Snowflake-only workloads where the aggregate never needs to leave Snowflake — the vendor manages incremental refresh, handles TARGET_LAG scheduling automatically, and integrates rewrite deeply with Snowflake's cost-based optimiser. The trade-off is complete lock-in: the dynamic table's rows are invisible to Trino, Spark, or any external reader; adding a second engine forces you to duplicate the aggregate.

Pick Databricks materialized views or BigQuery MVs for the same single-engine reason inside Databricks or BigQuery respectively. Each vendor's MV feature is excellent inside its silo and increasingly weak outside it. If you know you're single-engine forever, the vendor MV is usually the higher-performance choice. If you're multi-engine or planning to be, Iceberg MV wins by default and the migration cost only grows the longer you wait.

How does incremental refresh work on an Iceberg MV?

Incremental refresh reads only the changes to the source tables since the MV's last refresh, computes the incremental contribution to the aggregate, and applies it to the MV storage table — instead of re-executing the MV's defining query end-to-end (which would be full refresh, O(source size)). For append-only sources (event streams, immutable fact tables), the mechanism is incremental-append-scan: Iceberg exposes the data files added between the pinned snapshot ID and the current snapshot ID; the refresh reads only those files, aggregates the delta, and appends the aggregated delta to the MV storage table. Because the aggregate is a SUM/COUNT/etc. over an append-only source, the union of past MV rows plus the new delta rows produces the same result as re-aggregating the full source — mathematically exact and typically 100-1000× cheaper.

For upsert / delete sources (CDC-fed tables, dimension tables), the mechanism is incremental-change-scan, which exposes per-row change-type markers (INSERT, UPDATE_BEFORE, UPDATE_AFTER, DELETE) via Iceberg v2 row-level deletes. The refresh computes signed contributions per aggregate key (+1 for INSERT and UPDATE_AFTER, -1 for DELETE and UPDATE_BEFORE), sums per key, filters no-op deltas, and MERGEs into the MV storage table. This correctly reconciles updates — the pre-image contribution is subtracted, the post-image is added, and the aggregate remains consistent. The complexity is O(delta rows + affected aggregate keys) — still typically 10-1000× cheaper than full refresh, though not as cheap as the append-only path.

What is query rewrite for materialized views and how does it work?

Query rewrite (also called automatic query substitution or MV substitution) is the mechanism where the engine's cost-based query optimiser transparently replaces a scan of the source table with a scan of a matching MV — the user writes SELECT ... FROM sales.orders GROUP BY order_date and the optimiser silently rewrites it to SELECT ... FROM analytics.daily_revenue$storage without the user knowing the MV exists. The optimiser parses the user query into a logical plan, enumerates candidate MVs whose defining plan is a superset or exact match of the user's plan, gates each candidate by freshness (comparing last_refreshed_at against the tolerance the user or MV declared), estimates the cost of each surviving plan, and picks the cheapest.

Iceberg MV rewrite recognises four match kinds: exact match (user query is byte-identical to MV definition after normalisation), filter-pushdown match (user adds a WHERE clause on a column the MV has; optimiser reads MV then filters), rollup match (MV holds a finer aggregate — daily; user asks for coarser — weekly; optimiser reads MV and re-aggregates on top), and join match (MV is a denormalised join; user query joins the same tables; optimiser reads MV instead of re-joining sources). The MAX_STALENESS contract can be set at MV level (default), session level (SET SESSION iceberg.mv_max_staleness), or per-query (OPTIONS (MAX_STALENESS = ...)). Precedence is query > session > MV default; tightening to 0 seconds forces a source scan for freshness-critical queries.

How do I migrate a dbt incremental model to an Iceberg materialized view?

The migration replaces the dbt materialized='incremental' config with the dbt-iceberg adapter's materialized='iceberg_mv' config, removes the {% if is_incremental() %} watermark-filter guard entirely (the MV spec drives incremental refresh via snapshot diffs, not user-side watermark bookkeeping), and rewires the DAG to trigger REFRESH MATERIALIZED VIEW from Airflow instead of running dbt build --select <model> for the aggregate. The aggregate SQL itself stays identical — only the config header and the incremental-guard branch change. Concretely, a 15-line incremental model with unique_key, incremental_strategy='merge', and an is_incremental() WHERE clause becomes a 5-line MV definition with freshness_max and refresh_mode properties.

The parallel-run recipe is the same as any MV migration: keep the dbt model running while the Iceberg MV is populated by Airflow refreshes; nightly full-outer-join reconcile between the dbt-materialised table and the Iceberg MV; cut over consumers when drift is provably below tolerance for 30 consecutive days; suspend the dbt model for 7 days as a safety net; drop the model file when confidence is established. Watch out for dbt-specific conveniences that don't translate — model tests, snapshots, seeds, and freshness checks remain in dbt land; only the materialization moves to the MV. Some teams keep dbt for the modelling layer and use MVs only for leaf aggregates that need multi-engine reach.

Which engines support Iceberg materialized views in 2026?

Spark 4.0 ships full support for CREATE MATERIALIZED VIEW, REFRESH MATERIALIZED VIEW, and query rewrite against Iceberg catalogs (Glue, REST, Hive, Nessie). Spark is the reference implementation for the spec — new spec features usually land in Spark first. Trino 450+ supports MV read + refresh with automatic query rewrite; the Trino coordinator's cost-based optimiser handles exact, filter-pushdown, rollup, and (with tuning) join match kinds. Snowflake exposes Iceberg tables and views via its external Iceberg catalog integration, so a Snowflake session can read an Iceberg MV's storage table as if it were a normal Snowflake table; Snowflake's own rewrite engine does not yet fully understand external MV freshness metadata, so rewrite for external Iceberg MVs is limited compared to native dynamic tables.

AWS Athena supports reading Iceberg MV storage tables via its Iceberg connector but does not yet support triggering refresh or performing MV substitution — you must SELECT from the storage table explicitly. Databricks supports Iceberg tables through Unity Catalog's Iceberg feature, with growing MV support in 2026 that mirrors the Databricks-native MV feature. DuckDB and ClickHouse are following, with early support in nightly builds. PyIceberg (0.9+) supports metadata inspection and manual refresh orchestration from Python for any engine that lacks native DDL. Feature-detect before committing — the spec is stable enough for production use, but individual engine coverage varies feature by feature.

Practice on PipeCode

• Drill the SQL practice library → for the materialized view, aggregate rewrite, incremental-load, and snapshot-diff problems senior lakehouse interviewers love to probe on Iceberg MVs.
• Rehearse aggregation depth on the aggregation practice library → for the SUM / COUNT / rollup / signed-reconciliation patterns that make incremental refresh actually correct on upsert sources.
• Level up window-function fluency on the window functions practice library → for the rolling-window aggregates and time-series rollups that often need MV pre-computation to hit sub-second dashboard latency.
• Sharpen the join axis with the joins practice library → for the denormalised-join MV patterns where the MV pre-computes an expensive join and query rewrite substitutes it for the user's join.
• Layer in the streaming perspective on the streaming practice library → for the CDC-fed sources, change-data-feed reads, and event-driven refresh triggers that push Iceberg MV freshness contracts into sub-minute territory.
• Stack the prerequisites against PipeCode's broader 450+ data-engineering catalogue to anchor the four-lever MV decision framework (freshness, cadence, rewrite, cost/latency) against real graded inputs.

Lock in iceberg materialized views muscle memory

Docs explain the Iceberg MV spec. PipeCode drills explain the decision — when snapshot-consistent freshness is enough, when the append-only fast path applies vs the row-level-delete reconciliation, when query rewrite silently stops firing and hit-rate observability is your only warning, when a dbt incremental model is worth migrating and when it isn't. Pipecode.ai is Leetcode for Data Engineering — pattern-first practice tuned for the production trade-offs senior data engineers actually face on the lakehouse.

Practice SQL problems →
Practice aggregation problems →