Home › Enterprise Intelligence Architecture

Lakehouse Storage & Incremental Views

ACID over object storage, the Delta Lake and Iceberg table formats, file layout and compaction, change data capture, and incremental view maintenance instead of full recompute.

Last reviewed: 2026-09-04 · Українською

In this area

Object storage is cheap, unbounded — and not a database. It cannot atomically rename a directory, it has no locks, and it does not know which files constitute your table right now. This module walks the mechanism that turns a bucket into a transactional table: the commit log and optimistic concurrency control, the physical file layout that dictates the cost of every query, and the decision that shapes an entire analytical estate — refresh a mart with a delta, or honestly recompute it in full.

ACID over object storage: how a transaction log turns a bucket into a table

E2.1

Object storage is not a filesystem. S3, GCS and MinIO offer no atomic directory rename, no locks, and prefix listings that historically lagged reality. The classic 'a table is a prefix in a bucket' arrangement breaks precisely here: a reader that starts mid-write sees half the new files; a crashed worker leaves orphans behind; two concurrent writers silently clobber each other.

A table format is a metadata layer that makes a set of Parquet files transactional. It answers exactly one question: which files constitute the table at version N. Everything else follows from that answer.

• Delta Lake: an ordered log at _delta_log/000…N.json. A commit is the creation of the file bearing the next version number, and whoever creates it first wins (put-if-absent semantics). Every 10 commits a Parquet checkpoint is written, so a reader reconstructs state from one checkpoint plus at most ten JSON files rather than replaying 40,000 log entries.
• Apache Iceberg: a metadata tree of metadata.json → manifest list → manifest. A commit is an atomic swap of a catalog pointer (REST, Glue, Nessie, HMS) via compare-and-swap.
• The shared mechanism is optimistic concurrency control. A writer pins version N, stages its data files, then attempts to commit N+1. If it loses the race it re-reads the new commits, checks for a logical conflict (did we touch the same files or partitions?) and retries — without rewriting the data it already produced.
• What the architect gets as a side effect: snapshot isolation (a reader sees one consistent snapshot for the whole query), time travel (VERSION AS OF), an atomic MERGE including deletes, and above all the ability to undo a bad load with one statement instead of restoring from backup.

What actually counts as a conflict. Two appends writing different files do not conflict: both merely add new files, and whoever loses the race simply shifts its version number. What conflicts are the operations that read state and depend on what they read — MERGE, UPDATE, DELETE and compaction. Hence the cheapest way to eliminate most conflicts: narrow an operation's scope with a partition predicate. If two writers are guaranteed not to overlap by partition, the conflict check lets both through and neither waits on the other.

Where this breaks in practice. A commit is atomic only to the extent that an atomic primitive exists in the storage layer. Delta on S3 with concurrent multi-cluster writers needs an external coordinator (the DynamoDB log store) or S3 conditional writes; Iceberg needs a catalog that genuinely supports compare-and-swap. A catalog that is 'just a file in the bucket' returns you to exactly what you were escaping: silent version clobbering.

# What a commit actually is: one conditional PUT decides the winner.
# _delta_log/00000000000000000042.json either exists or it does not.

from deltalake import DeltaTable
from deltalake.exceptions import CommitFailedError

def merge_batch(table_uri, batch, attempts=5):
    for attempt in range(attempts):
        dt = DeltaTable(table_uri)
        pinned = dt.version()                 # optimistic: pin version N
        try:
            (dt.merge(source=batch,
                      predicate="t.order_id = s.order_id",
                      source_alias="s", target_alias="t")
               .when_matched_update_all()
               .when_not_matched_insert_all()
               .execute())                    # tries to create version N+1
            return dt.version()
        except CommitFailedError:
            # Someone else won N+1. Re-read the log, re-check the conflict
            # window, retry. Data files already staged are reused, not rewritten.
            print("lost race at v%d, retry %d" % (pinned, attempt + 1))
    raise RuntimeError("commit contention: %d failed attempts" % attempts)

# The audit trail is a table, not a log file:
#   DESCRIBE HISTORY sales.orders        -> version, operation, operationMetrics
#   RESTORE TABLE sales.orders TO VERSION AS OF 41   -- undo a bad load in seconds

File layout, clustering and compaction: why your query opens 200,000 objects

E2.2

Query cost in a lakehouse is set by the number of objects you must open, not by the volume of data. Every object-store GET costs 20–80 ms to first byte plus a separate metadata round trip. An 8 GB table spread across 200,000 files of 40 KB reads an order of magnitude slower than the same table in 60 files of 128 MB, even though the byte count is identical.

Three levers, and they must not be conflated:

• Partitioning — physical directory splitting on a low-cardinality key (date, region). Working rule: a partition should weigh hundreds of megabytes at minimum. Partitioning by customer_id or by a second-precision timestamp guarantees a million directories holding one file each. Iceberg removes part of the pain with hidden partitioning (days(ts), bucket(16, customer_id)): the query filters on the natural column and the engine derives the partition.
• Data skipping — metadata carries per-file min/max and null counts for each column. The planner discards a file without opening it. This works exactly as well as the min/max ranges are narrow, and they are narrow only when the data is physically ordered on the predicate column.
• Clustering (Z-order, liquid clustering) — rewriting files so correlated values sit together. Z-ordering on the two or three columns people actually filter by tightens min/max ranges; Z-ordering on a column absent from every predicate buys nothing but the compute bill.

Compaction is not housekeeping, it is part of the SLA. A streaming job with a one-minute trigger produces at least 1,440 files per partition per day. OPTIMIZE (Delta) or rewrite_data_files (Iceberg) bin-packs them into a 128 MB – 1 GB target. But the rewritten files do not vanish immediately: old snapshots are needed for time travel and for queries already in flight. VACUUM and expire_snapshots reclaim space only after the retention window — and this is exactly where architects destroy their own recoverability by setting retention to zero hours to save on storage.

Copy-on-write versus merge-on-read. COW rewrites an entire file for one updated row: reads are fast, writes are expensive, and write amplification on point updates reaches hundreds of times. MOR writes separate delete and delta files: writes are cheap, reads pay a merge cost until compaction catches up. Deletion vectors in Delta are the managed compromise — the row is marked positionally and the physical rewrite is deferred to scheduled compaction.

-- Delta: bin-pack, then cluster on the columns the predicates actually use.
OPTIMIZE sales.orders
  WHERE order_date >= current_date() - INTERVAL 7 DAYS
  ZORDER BY (customer_id, status);

-- Retention is a recovery decision, not a cleanup chore.
-- 168h keeps a week of time travel; RETAIN 0 HOURS destroys it.
VACUUM sales.orders RETAIN 168 HOURS;

-- Iceberg: the same two jobs, as explicit procedures.
CALL catalog.system.rewrite_data_files(
  table      => 'sales.orders',
  strategy   => 'sort',
  sort_order => 'customer_id, status',
  options    => map('target-file-size-bytes', '536870912',  -- 512 MB
                    'min-input-files',        '32')
);
CALL catalog.system.expire_snapshots('sales.orders', TIMESTAMP '2026-08-25 00:00:00');

-- Hidden partitioning: the query filters on ts, the engine prunes on days(ts).
-- No derived date_str column for analysts to remember (or forget).
ALTER TABLE sales.orders ADD PARTITION FIELD days(ts);
ALTER TABLE sales.orders ADD PARTITION FIELD bucket(16, customer_id);

CDC and incremental views: when a delta beats a full recompute

E2.3

Step 1 — how you learn a change happened at all. High-watermark polling (WHERE updated_at > :watermark) is the cheapest and most treacherous option: it cannot see physical deletes, it loses a row updated twice inside one interval, and it breaks silently whenever the source updates a record around the trigger that maintains the column. Log-based CDC (WAL or binlog via Debezium) yields an ordered INSERT/UPDATE/DELETE stream with LSNs — the only way to propagate deletes correctly beyond the OLTP boundary.

Step 2 — landing the stream. Raw changes are written append-only. Current state is derived by a MERGE with mandatory de-duplication: for each key take the record with the highest LSN in the micro-batch. Without it, two updates to one row inside a batch produce a non-deterministic result, and the engine rightly refuses to run the MERGE.

Step 3 — maintaining the aggregate. This is the module's core decision. Not every aggregate can be updated from a delta:

• Self-maintainable: COUNT, SUM, AVG (kept as a sum/count pair). An inverse operation exists, so 'add the new, subtract the old' stays exact even under deletes.
• Not self-maintainable: MIN, MAX, COUNT(DISTINCT), percentiles. Deleting the current maximum tells you nothing about the new one — you need a recompute of the affected groups or an auxiliary structure (value counters, HLL or t-digest sketches). An architect who 'optimised' MAX incrementally gets a mart that drifts quietly and looks plausible while doing so.

Step 4 — when incrementality is justified at all. A crude but workable rule: the incremental path wins while the fraction of changed partitions stays small — roughly under 10–15% daily churn. Above that threshold, the cost of reading the change feed, running the MERGE, rewriting files and compacting afterwards exceeds the cost of an honest CREATE OR REPLACE. So a mature architecture keeps both branches: an hourly incremental refresh plus a full reconciling recompute daily or weekly that absorbs late and out-of-order arrivals.

And where NoSQL fits. A lakehouse gives you scans, joins and ACID, but 100–300 ms per point lookup is not something you put behind a product API. A key-value store answers in single-digit milliseconds but offers neither joins nor a consistent snapshot across entities. The right answer is not either/or: the lakehouse remains the system of record while the KV store is a derived serving projection rebuilt by the same change stream. The duplication here is latency deliberately bought, not a normalisation mistake.

-- 1. De-duplicate the micro-batch: one row per key, highest LSN wins.
CREATE OR REPLACE TEMP VIEW changes AS
SELECT * FROM (
  SELECT *, ROW_NUMBER() OVER (PARTITION BY order_id ORDER BY lsn DESC) AS rn
  FROM bronze.orders_cdc
  WHERE _commit_version > (SELECT last_version FROM ops.watermarks WHERE job = 'orders_silver')
) WHERE rn = 1;

-- 2. Inserts, updates AND deletes land in one atomic commit.
MERGE INTO silver.orders t
USING changes s ON t.order_id = s.order_id
WHEN MATCHED AND s.op = 'D'       THEN DELETE
WHEN MATCHED AND s.lsn > t.lsn    THEN UPDATE SET *
WHEN NOT MATCHED AND s.op <> 'D'  THEN INSERT *;

-- 3. Self-maintainable aggregate: apply +new / -old straight from the feed.
MERGE INTO gold.customer_totals g
USING (
  SELECT customer_id,
         SUM(CASE WHEN _change_type IN ('insert', 'update_postimage') THEN amount
                  ELSE -amount END) AS delta_amount
  FROM table_changes('silver.orders', 812, 947)   -- versions since the last run
  GROUP BY customer_id
) d ON g.customer_id = d.customer_id
WHEN MATCHED     THEN UPDATE SET g.total = g.total + d.delta_amount
WHEN NOT MATCHED THEN INSERT (customer_id, total) VALUES (d.customer_id, d.delta_amount);

-- 4. NOT self-maintainable: MAX cannot be repaired from a delta.
--    Recompute only the groups the change feed actually touched.
CREATE OR REPLACE TABLE gold.customer_peak AS
SELECT customer_id, MAX(amount) AS max_amount
FROM silver.orders
WHERE customer_id IN (
  SELECT DISTINCT customer_id FROM table_changes('silver.orders', 812, 947)
)
GROUP BY customer_id;

Sources this area derives from

1. DeltaLakeOptimizationTechniquesforScalableLakehouseArchitectures
2. Characterizing and Fixing Silent Data Loss in Spark on AWS
3. LakeVilla A Modular and Non Invasive Toolbox for Lakehouse
4. Proof Gated Publication Verify Before Commit Content Integ
5. Smart Compaction Predicting Compaction Utility from Lakeho
6. Delta Tensor Efficient Vector and Tensor Storage in Delta
7. An Architecture for Real Time Warehousin
8. Big Data Techniques Systems Applications
9. Big Data Analytic Approaches Classificat

Continue in this track