Columnar Formats and the Lakehouse

"They asked for two columns. I, a humble row, handed over all twenty and apologized for the wait."

A Record That Reads Everything to Answer Anything

A training pipeline and an analytics pipeline ask the same physical question of storage thousands of times a day: read these few attributes from these many records. A churn-feature job wants two columns out of forty; a label-distribution check wants one column out of two hundred. How the bytes are arranged on the object store decides whether that query reads what it asked for or drags the entire dataset off disk to discard most of it. Section 8.2 settled where the data lives (object storage, distributed filesystems). This section settles how it is encoded there, and that single choice moves analytical and feature-read cost by an order of magnitude.

1. Row Layout Versus Columnar Layout Beginner

A table is a logical grid, but disk is one-dimensional, so the grid must be flattened into a byte sequence, and there are two natural ways to do it. A row store writes record 1 in full (all its columns), then record 2 in full, and so on; the fields of one row sit physically adjacent. This is ideal when you fetch whole records by key, which is why transactional databases default to it. A columnar store does the opposite: it writes all values of column 1 together, then all values of column 2, and so on; the values of one column sit physically adjacent. Parquet and ORC are the dominant columnar formats on object storage, and the layout difference is the entire story of why analytical scans are cheap on them.

Two consequences follow directly from putting a column's values next to each other. The first is column pruning: a query that needs two of twenty columns reads only those two contiguous regions and skips the rest, so bytes scanned fall in proportion to columns touched, not rows kept. The second is compression: values within one column share a type and a distribution, and once similar values are adjacent, dictionary encoding, run-length encoding, and bit packing all become dramatically effective, far more than they ever are on the mixed-type byte soup of a row. Figure 8.3.1 shows both layouts side by side and marks the bytes a two-column query must read in each.

2. Measuring the Two Wins: Pruning and Compression Intermediate

The two benefits are easy to state and easy to measure, so we measure them. The code below builds a 200,000-row table of twenty columns: nineteen noisy float columns plus one low-cardinality categorical column (a region code with five distinct values), arranged in clustered runs the way a real table sorted on a key would be. It then computes, from an explicit byte model, how much a row store must read for a full scan versus how much a columnar store reads when a query asks for just two of the twenty columns. It measures the columnar compression ratio on the clustered categorical column. Finally, when pyarrow is installed, it writes a real Parquet file and reads back the compressed size of only the two requested column chunks, so the from-scratch model can be checked against a production encoder.

import io, random

random.seed(7)
N_ROWS, N_COLS, READ_COLS = 200_000, 20, 2     # a query touching 2 of 20 columns

cols = {f"f{c}": [random.gauss(0, 1) for _ in range(N_ROWS)] for c in range(N_COLS - 1)}
categories = ["US", "EU", "APAC", "LATAM", "MEA"]
region = []                                     # low-cardinality key in clustered runs
while len(region) < N_ROWS:
    region.extend([categories[random.randint(0, 4)]] * random.randint(200, 2000))
cols["region"] = region[:N_ROWS]

BF, BC = 8, 8                                   # bytes per float, per category slot
row_bytes = (N_COLS - 1) * BF + BC              # a full row record
row_store_full_scan = row_bytes * N_ROWS       # row store: a query must read every row

want = ["f0", "region"]                         # columnar store reads only these two
columnar_pruned_scan = BF * N_ROWS + BC * N_ROWS
print("row-store full read (MB)   :", f"{row_store_full_scan / 1e6:.1f}")
print("columnar pruned read (MB)  :", f"{columnar_pruned_scan / 1e6:.1f}")
print("bytes-scanned reduction    :", f"{row_store_full_scan / columnar_pruned_scan:.1f}x")

# Compression: similar values adjacent => dictionary + run-length collapses the column.
raw = BC * N_ROWS
codes = [categories.index(v) for v in cols["region"]]
runs = 1 + sum(codes[i] != codes[i - 1] for i in range(1, len(codes)))
rle = sum(len(s) for s in categories) + 5 + runs * 5   # tiny dict + (code, run-length)
print("region raw (MB)            :", f"{raw / 1e6:.2f}")
print("region dict+RLE (MB)       :", f"{rle / 1e6:.2f}")
print("columnar compression ratio :", f"{raw / rle:.1f}x")

try:                                            # cross-check against a real Parquet writer
    import pyarrow as pa, pyarrow.parquet as pq
    table = pa.table(cols); buf = io.BytesIO()
    pq.write_table(table, buf, compression="snappy")
    md = pq.ParquetFile(io.BytesIO(buf.getvalue())).metadata
    idx = {table.schema.field(i).name: i for i in range(table.num_columns)}
    pruned = sum(md.row_group(g).column(idx[n]).total_compressed_size
                 for g in range(md.num_row_groups) for n in want)
    print("parquet total (MB)         :", f"{buf.getbuffer().nbytes / 1e6:.2f}")
    print("parquet 2-col chunks (MB)  :", f"{pruned / 1e6:.2f}")
    print("parquet column-prune ratio :", f"{buf.getbuffer().nbytes / pruned:.1f}x")
except Exception as e:
    print("pyarrow not available:", type(e).__name__)

row-store full read (MB)   : 32.0
columnar pruned read (MB)  : 3.2
bytes-scanned reduction    : 10.0x
region raw (MB)            : 1.60
region dict+RLE (MB)       : 0.00
columnar compression ratio : 2075.2x
parquet total (MB)         : 35.57
parquet 2-col chunks (MB)  : 1.87
parquet column-prune ratio : 19.0x

The numbers separate the two effects cleanly. Column pruning alone gives the $10\times$ scan reduction from the byte model: two columns instead of twenty, with no compression assumed. The real Parquet file reaches $19\times$ because pruning and per-column compression stack, and the clustered categorical column shows why compression is a columnar privilege, collapsing by more than $2000\times$ once its repeated values lie next to one another. A row store cannot reach that ratio on the same column, because in a row layout each region value is wedged between nineteen unrelated floats and never forms a run. Real workloads sit between these extremes, but the direction is invariant: the wider the table and the more selective the query, the more a columnar layout wins.

Column pruning is not a trick the storage layer pulls off alone; the query engine has to know which columns the query actually needs and push that knowledge down to the reader. This is exactly the projection pushdown that Spark's Catalyst optimizer performs: it inspects the logical plan, determines the minimal set of columns required, and instructs the Parquet reader to skip the rest before a single irrelevant byte leaves the object store. The columnar format and the optimizer are partners; we built the optimizer side of this in Section 7.3, and Parquet is the storage side that makes its pruning pay off.

import pyarrow.parquet as pq

# Read ONLY two columns from a wide Parquet file. The reader fetches just those
# column chunks; the other eighteen columns are never decoded or transferred.
table = pq.read_table("features.parquet", columns=["f0", "region"])

# Push a predicate down too: rows are skipped using per-row-group min/max stats,
# so whole chunks are never read when they cannot match the filter.
table = pq.read_table("features.parquet", columns=["f0", "region"],
                      filters=[("region", "=", "EU")])

3. Apache Arrow: The Columnar Layout, In Memory Intermediate

Parquet is a columnar layout on disk, optimized for compact storage and fast scans. Apache Arrow is the same columnar idea in memory, optimized for fast processing and, above all, for sharing. Arrow defines a standard in-memory layout for columnar data: a column is a contiguous buffer of values plus a validity bitmap, with a byte specification precise enough that any system implementing it lays the bytes out identically. That shared specification is what makes Arrow valuable. When two systems both speak Arrow, one can hand a table to the other by passing a pointer to the buffers, with no serialization, no copy, and no per-row translation. This is the zero-copy exchange that lets a Parquet reader, a query engine, a Python dataframe, and a GPU library move data between them at memory bandwidth rather than parsing speed.

For AI pipelines the consequence is direct. A feature-engineering step in Spark can produce Arrow batches that a training data loader consumes without a serialize-then-deserialize round trip, and a vectorized user-defined function can run Python code over Arrow batches at near-native speed because the data arrives in the layout NumPy and pandas already use. That vectorized-UDF path, where Spark ships Arrow batches to Python workers instead of pickling row by row, is exactly the mechanism we examined in Section 7.8; Arrow is the wire format and the in-memory format that makes it fast. The throughline is that columnar is not only a disk decision: it is the shape data should keep as it crosses every boundary in the pipeline.

4. The Lakehouse: A Transactional Table Over Object Storage Intermediate

Columnar files on object storage are fast to scan but, by themselves, a loose pile of immutable blobs. They have no notion of a table version, no atomic way to add or replace rows, and no protection against a reader seeing a half-written update. The data lake gave us cheap, scalable storage and surrendered the guarantees a warehouse always had. The lakehouse closes that gap: it is a transactional table layer placed over the same columnar files on the same object storage, adding the warehouse's guarantees without giving up the lake's cost and scale. Delta Lake, Apache Iceberg, and Apache Hudi are the three dominant implementations, and although they differ in detail they share one mechanism: a metadata log of commits that defines, at every point in time, exactly which data files constitute the table.

That commit log is what turns a directory of Parquet files into a table with real semantics. ACID transactions mean a writer's changes appear all at once or not at all, so a reader never sees a partial commit even while a write is in flight. Schema evolution means columns can be added, renamed, or retyped through metadata, with old files reinterpreted under the new schema rather than rewritten. Time travel means every commit is a queryable version, so you can read the table exactly as it stood at a past commit or timestamp. Efficient upserts (merge, update, delete) mean you can change a few rows without rewriting the dataset, by writing new files and recording in the log which old files they supersede. Figure 8.3.2 shows the layering: immutable columnar files at the bottom, a metadata commit log defining table versions in the middle, and the engines that read and write through that log on top.

The mechanism behind every one of these features is the same idea this chapter keeps returning to. The data files are immutable shards on object storage; the table is the set of shards a given commit points to. An upsert does not edit a file in place, an operation object stores do not support well anyway; it writes new shards and atomically commits a new version of the file set. Time travel is just reading the file set named by an older commit. This is partitioning and sharding wearing a transactional hat, and it is the same arc that began as data shards in this chapter and returns as sharded parameters and model shards later in the book.

5. Why This Matters for AI: Reproducible, Versioned Training Tables Advanced

A model is only as reproducible as the data it was trained on, and "the data" on a plain data lake is a moving target: files get appended, corrected, and backfilled, so the dataset you trained on last month may no longer exist in the same form. The lakehouse fixes this by making the training set an explicit, immutable version. You pin training to a specific table commit, record that version id alongside the model checkpoint, and months later you can read the table exactly as it stood, re-run the pipeline, and get a dataset identical to the original byte for byte. Time travel turns "which data produced this model?" from an archaeology problem into a single version number, and it does so on cheap object storage rather than an expensive warehouse. This data-versioning story is foundational enough that Section 8.9 develops lineage and reproducible dataset versioning in full.

The efficient-upsert capability matters just as much for the daily reality of feature pipelines. Training and feature tables are rarely static; labels get corrected, late-arriving events backfill yesterday's partition, and a feature definition changes and must be recomputed for a subset of entities. On raw Parquet, any of these forces a full rewrite of the affected partitions. On a lakehouse table, a MERGE updates only the rows that changed by writing new files and retiring superseded ones, so a correction touches kilobytes of metadata and a handful of files rather than the whole dataset. The same transactional table that ingests a stream of corrections is also the one a training job reads from at a pinned version, which is why the lakehouse sits at the seam between batch and streaming ingestion that Chapter 9 takes up.

Columnar encoding decides what a scan costs, Arrow decides what crossing a system boundary costs, and the lakehouse decides what a table version costs, and together they turn cheap object storage into a reproducible, queryable, transactional foundation for training and feature data. What none of them decides yet is how the files are arranged within a table: how rows are grouped into partitions, how small files are compacted into large ones, and how data is sorted so that pruning and predicate pushdown actually skip the bytes they could. That layout question is where the costs in this section are won or lost in practice, and it is the subject of Section 8.4.

6. Lance: ML-Native Columnar Format Intermediate

Parquet was designed for analytical queries that scan millions of rows and read a handful of columns. That access pattern, a wide sequential sweep, is not what a training loop performs. A training loop reads sample i by index, jumping to one row out of millions, then sample j, then k, in shuffled order. Parquet's row-group structure makes random access expensive: to retrieve one row you must decompress the entire row group it belongs to. The Lance file format, specified in the VLDB 2025 paper (arXiv:2504.15247), is designed from the start for this ML access pattern. It stores data in page-level random-access chunks sized for single-sample retrieval, so fetching row i decodes one small page rather than a multi-megabyte row group. In benchmarks on image and embedding datasets, Lance delivers approximately 100x faster random access than Parquet for single-sample index reads, while maintaining competitive scan throughput for batch reads.

Two additional features make Lance particularly suited to ML pipelines. First, Lance stores ANN (approximate nearest-neighbor) vector indices inside the same file as the data, so a moderate-scale embedding retrieval query requires no external vector database: the index lives next to the vectors. Second, Lance uses a Git-style versioning model: each write creates a new snapshot, and rolling back to a prior version is a metadata-only operation that does not rewrite the data files. The following callout shows how to connect to a LanceDB table in three lines, and Section 8.9 returns to the versioning discipline these snapshots support.

import lancedb

db = lancedb.connect("./my_db")
table = db.create_table("embeddings", data=df)

Lance connects to two threads that run through this chapter and beyond. Its built-in ANN index is the same distributed vector-search problem that Chapter 25 addresses at larger scale: Lance handles the moderate-scale case in a single file, and Chapter 25's distributed index shards handle the billion-vector case across a cluster. Its snapshot versioning resembles Delta Lake's transaction log (discussed earlier in this section): both treat each write as an immutable new version and make rollback a pointer swap rather than a data rewrite. The design choice that separates them is the access pattern each optimizes for: Delta Lake optimizes for transactional correctness on wide analytical tables; Lance optimizes for random-access speed on embedding and multimodal ML datasets.

7. Table Format Convergence: Iceberg v3, UniForm, and XTable Intermediate

The lakehouse ecosystem that Section 4 introduced fragmented around three competing transaction-log formats. Apache Iceberg (originated at Netflix), Delta Lake (Databricks), and Apache Hudi (Uber) each solve the same core problem, ACID commits over immutable Parquet files on object storage, but with different metadata schemas, different manifest structures, and different APIs. Until recently, a table written in Delta Lake required a full data copy to be read by an Iceberg-native engine such as Trino or Snowflake, and vice versa. Two lines of work in 2024 and 2025 are closing that gap without copying data.

Iceberg v3 (ratified 2025) adds three capabilities to the core spec: multi-level aggregated statistics for finer predicate pushdown across nested partition hierarchies; native variant and semi-structured data types that store JSON-like structures without schema coercion; and encryption at the manifest level, so individual partitions can be access-controlled without re-encrypting data files. The statistics and variant-type additions are particularly relevant to feature stores that mix structured and semi-structured columns in the same table.

The interoperability problem is addressed by two metadata-translation projects. Delta UniForm (shipped in Delta Lake 3.x) generates Iceberg and Hudi metadata alongside every Delta commit, so a table written as Delta Lake is simultaneously readable as an Iceberg table by any Iceberg-native engine. No data is copied: only a second set of manifest files is maintained. Apache XTable (formerly OneTable, donated to Apache in 2024) is an open, engine-agnostic translator that converts the transaction log of any of the three formats into any other on demand, again without touching the Parquet data files. The cost of interoperability is one metadata round-trip, not a data copy.

Above the format layer, two catalog projects aim to register tables once and serve them to any engine. Unity Catalog (open-sourced by Databricks in 2024) provides governance, lineage, and access control for Delta, Iceberg, and other formats through a single catalog API. Apache Polaris (donated to Apache in 2024) is an Iceberg REST catalog implementation that lets Spark, Trino, DuckDB, Flink, and ML frameworks all discover and read the same table through a standard HTTP interface. Together they represent a governance layer that sits above the format choice rather than being tied to it.

These convergence efforts connect back to the thesis thread running through this section. The sharding and immutable-file model is the same across all three lakehouse formats; what differed was the metadata envelope around those shards. UniForm and XTable demonstrate that once the data layer is stable (immutable columnar files on object storage), interoperability can be achieved by translating only the thin metadata layer above it, a further confirmation that the immutable-shard model is the durable substrate and the metadata layer is where policy, versioning, and now cross-format compatibility live.