Columnar Engines

The previous lesson cut the memory peak to one thirty-second by reading the dataset chunk by chunk, but in every chunk it parsed all the fields of the record. The region breakdown looked at two columns; five columns were read, two were used, three were thrown away. Chunking lowers the peak, it does not lower the work done.

The question of this lesson is that one: how many values must be touched to produce the same answer. The unit of measure is the number of values read, and next to it stands a second number — access, that is, how many times a separate contiguous region was visited. The layout that keeps the fields of a record adjacent and the layout that keeps a column adjacent work in opposite ways on these two numbers; which one is cheap depends on what the question asks.

• SP11. The dataset is fictional and is the dataset of the previous lessons: 3,199 rows, six columns, 1,329 subscribers, three periods. It is produced with the same seed and the output is identical from run to run. It is a model.
• SP12. Both store layouts are modeled in memory; what is measured is not the storage medium but the number of values touched to produce the answer. It is a model.
• SP13. In the row-based layout, touching a row means reading all six fields of that row; since the fields are written adjacent, a single field cannot be fetched separately.
• SP14. In the columnar layout each column is a separate array, and for every block of 512 values a summary holding the smallest and the largest value is stored alongside the column.
• SP15. Access is the number of visits to a separate contiguous region: a scan from start to finish is one access, and reading the individual positions of a column is as many accesses as there are positions.

Two Layouts

# store.py — the same fictional dataset is held in two layouts and every read is counted.
# MODEL: two layouts were modeled in place of a real store; the measure is how many
# values are touched to produce the answer of a query. 3199 rows, 1329 subscribers.
SEED = 1246
REGION = ("north", "east", "center", "south", "west")
PERIODS = (("2024-01", 1329), ("2024-02", 1035), ("2024-03", 835))
DENSITY = (0, 1, 2, 2, 2, 3, 4, 2, 2, 1)   # center-weighted measurement network
HEADER = ("subscriber_id", "meter_id", "period", "region", "reading_m3", "tier")


def rng(seed: int):
    s = seed & 0xFFFFFFFF

    def next_value() -> float:
        nonlocal s
        s = (1103515245 * s + 12345) & 0xFFFFFFFF
        return s / 4294967296
    return next_value


def tier(o: float) -> str:
    return "0-10" if o <= 10 else "10-25" if o <= 25 else "25-40"


def records() -> list[tuple]:
    r, out = rng(SEED), []
    for period, size in PERIODS:
        for i in range(size):
            o = round(5 + r() * 32.0, 1)
            out.append((10000 + i, f"S-{4000 + i}", period,
                        REGION[DENSITY[i % 10]], o, tier(o)))
    return out


class RowStore:
    """All fields of a record are adjacent: touching a row means reading all
    the fields of that row."""

    def __init__(self, rows):
        self.rows, self.values_read, self.accesses = [tuple(map(str, x)) for x in rows], 0, 0

    def scan(self):
        self.accesses += 1                   # one contiguous pass from start to finish
        for s in self.rows:
            self.values_read += len(HEADER)
            yield s

    def record(self, i: int) -> tuple:
        self.values_read, self.accesses = self.values_read + len(HEADER), self.accesses + 1
        return self.rows[i]


class ColumnStore:
    """Each column is a separate array: only the requested column, even only the
    requested positions of that column, can be read. The block summary is stored
    with the column."""
    BLOCK = 512

    def __init__(self, rows):
        self.cols = {name: [str(x[j]) for x in rows] for j, name in enumerate(HEADER)}
        self.values_read, self.accesses, self.n = 0, 0, len(rows)
        self.summary = {name: [(min(v[i:i + self.BLOCK]), max(v[i:i + self.BLOCK]))
                               for i in range(0, len(v), self.BLOCK)]
                        for name, v in self.cols.items()}

    def read_all(self, name: str) -> list:
        self.values_read, self.accesses = self.values_read + self.n, self.accesses + 1
        return self.cols[name]

    def read_at(self, name: str, positions: list) -> list:
        self.values_read += len(positions)   # scattered access: every position separate
        self.accesses += len(positions)
        return [self.cols[name][i] for i in positions]

    def record(self, i: int) -> tuple:
        self.values_read += len(HEADER)      # six separate visits to six columns
        self.accesses += len(HEADER)
        return tuple(self.cols[name][i] for name in HEADER)


r0 = records()
rs, cs = RowStore(r0), ColumnStore(r0)
print(f"seed {SEED}; {len(r0)} rows, {len(HEADER)} columns, "
      f"{len(r0) * len(HEADER)} values")
print(f"block size {ColumnStore.BLOCK}, {len(cs.summary['period'])} block summaries "
      f"in the period column")
print("first record:", rs.record(0))

seed 1246; 3199 rows, 6 columns, 19194 values
block size 512, 7 block summaries in the period column
first record: ('10000', 'S-4000', '2024-01', 'north', '9.4', '0-10')

The number draws the limit of the dataset: six columns and 3,199 rows, 19,194 values in total. The row-based layout keeps these values adjacent record by record; the six fields of the first record sit side by side. The columnar layout spreads the same values over six separate arrays and keeps one summary for each array’s seven blocks. The content of the dataset is the same; its placement is different.

One Question, Four Read Paths

# query.py — a single question is answered in four ways: "the total reading per region in the given
# period". The answer is the same on all four paths; what changes is the number of values read and
# accesses, the number of visits to a separate contiguous region.
def row_path(d: RowStore, target: str) -> dict:
    t = {}
    for s in d.scan():                       # all six fields of the record come along
        if s[2] == target:
            t[s[3]] = round(t.get(s[3], 0.0) + float(s[4]), 1)
    return t


def column_early(d: ColumnStore, target: str) -> dict:   # three columns end to end
    period, region, reading = d.read_all("period"), d.read_all("region"), d.read_all("reading_m3")
    return aggregate([(region[i], reading[i]) for i, v in enumerate(period) if v == target])


def column_late(d: ColumnStore, target: str) -> dict:    # filter first, then positions
    positions = [i for i, v in enumerate(d.read_all("period")) if v == target]
    return aggregate(list(zip(d.read_at("region", positions),
                              d.read_at("reading_m3", positions))))


def column_block(d: ColumnStore, target: str) -> tuple[dict, int]:
    summary, positions, skipped = d.summary["period"], [], 0
    d.values_read, d.accesses = d.values_read + 2 * len(summary), d.accesses + 1
    for bi, (lo, hi) in enumerate(summary):
        if not lo <= target <= hi:           # the block summary does not cover the target
            skipped += 1
            continue
        start = bi * d.BLOCK
        piece = d.cols["period"][start:start + d.BLOCK]
        d.values_read, d.accesses = d.values_read + len(piece), d.accesses + 1
        positions += [start + i for i, v in enumerate(piece) if v == target]
    return aggregate(list(zip(d.read_at("region", positions),
                              d.read_at("reading_m3", positions)))), skipped


def aggregate(pairs: list) -> dict:
    t = {}
    for b, o in pairs:
        t[b] = round(t.get(b, 0.0) + float(o), 1)
    return t


def run(path: str, dataset: list, target: str) -> tuple:
    if path == "row-based scan":
        d = RowStore(dataset)
        t = row_path(d, target)
    else:
        d = ColumnStore(dataset)
        t = (column_early(d, target) if "early" in path else
             column_late(d, target) if "late" in path else column_block(d, target)[0])
    return d, t


PATHS = ("row-based scan", "columnar, early materialization",
         "columnar, late materialization", "columnar, block skipping")
G = (34, 14, 9, 9, 11, 12)
fmt = lambda h: "".join(str(v).rjust(G[j]) if j else str(v).ljust(G[0])
                        for j, v in enumerate(h))
print(fmt(["read path", "values read", "ratio", "accesses", "center m3", "total m3"]))
base = 0
for path in PATHS:
    d, t = run(path, r0, "2024-03")
    base = base or d.values_read
    print(fmt([path, d.values_read, f"{base / d.values_read:.1f}x", d.accesses, t["center"],
               round(sum(t.values()), 1)]))
print(f"on the block-skipping path {column_block(ColumnStore(r0), '2024-03')[1]} of 7 blocks "
      f"were never opened; 2024-03 has 835 rows")

read path                            values read    ratio accesses  center m3    total m3
row-based scan                             19194     1.0x        1     8752.6     17307.2
columnar, early materialization             9597     2.0x        3     8752.6     17307.2
columnar, late materialization              4869     3.9x     1671     8752.6     17307.2
columnar, block skipping                    2835     6.8x     1674     8752.6     17307.2
on the block-skipping path 4 of 7 blocks were never opened; 2024-03 has 835 rows

All four paths give the same answer: 8,752.6 m³ in the center region and 17,307.2 m³ in total. The values read fall from 19,194 to 2,835 — 6.8 times. The gain does not come from a single item; it comes from the stacking of three separate decisions, and each row of the table adds one decision.

That the measure is values read is not a convenience but the choice that makes the comparison portable. The time of the same four paths changes from run to run with the machine worked on, the load at that moment and the state of the cache; the number of values read derives from the layout and the question and comes out the same on every run. How many times cheaper one path is than another is a defensible sentence in this unit.

The first decision is column selection. The question asks about three columns; the row-based layout brings all six, because the fields of the record are adjacent. The columnar layout reads three, and the values read are halved. This is a difference of layout, not of product: every structure that keeps a column adjacent gives the same gain.

The second decision is order. Early materialization reads all three columns from start to finish; late materialization first reads the filter column, finds the 835 matching positions and takes only those positions from the other two columns. The values read fall from 9,597 to 4,869. The cost is in the accesses column: 1,671 accesses instead of three, because scattered positions are now being read instead of a contiguous array.

The third decision is the block summary. The period column had been divided into seven blocks of 512, and for each block the smallest and the largest value had been stored. If the target period does not fall within a block’s range, that block is never opened; four of the seven blocks were not opened and the values read dropped from 4,869 to 2,835. The block summary added fourteen values to the reading and in return saved reading 2,048 values.

What the Gain Depends On

# clustering.py — block skipping depends on the data being clustered on that column.
# When the same dataset is written in subscriber order, every block carries all three
# periods, no summary stays outside, and the skipping gain drops to zero.
by_subscriber = sorted(r0, key=lambda x: (x[0], x[2]))
r = rng(99)                                     # our own generator with a visible seed
shuffled = [x for _, x in sorted((r(), x) for x in r0)]
G2 = (20, 14, 14, 10, 12)
fmt2 = lambda h: "".join(str(v).rjust(G2[j]) if j else str(v).ljust(G2[0])
                         for j, v in enumerate(h))
print(fmt2(["dataset layout", "skipped blocks", "values read", "accesses", "total m3"]))
for label, dataset in [("period order", r0), ("subscriber order", by_subscriber),
                       ("shuffled (seed 99)", shuffled)]:
    d = ColumnStore(dataset)
    t, skipped = column_block(d, "2024-03")
    print(fmt2([label, f"{skipped}/{len(d.summary['period'])}", d.values_read, d.accesses,
                round(sum(t.values()), 1)]))

# The selectivity of the filter decides which path is cheap. The formulas are read from the
# measured paths themselves: the early path reads three columns end to end, the late path
# reads the filter column and collects the remaining m positions one by one from two columns.
n = len(r0)
G3 = (22, 14, 13, 15, 12)
fmt3 = lambda h: "".join(str(v).rjust(G3[j]) if j else str(v).ljust(G3[0])
                         for j, v in enumerate(h))
print(fmt3(["matching rows", "read early", "read late", "late accesses", "late gain"]))
for m in (835, 1329, 2691):
    print(fmt3([f"{m} ({100 * m // n} percent)", 3 * n, n + 2 * m, 1 + 2 * m,
                f"{(3 * n) / (n + 2 * m):.2f}x"]))
d = ColumnStore(r0)
column_late(d, "2024-03")
print(f"the measured late path at 835 rows: {d.values_read} values and {d.accesses} accesses; "
      f"formula {n + 2 * 835} and {1 + 2 * 835}")

# The opposite direction: fetching all of a record.
SAMPLE = list(range(0, n, 320))
rs2, cs2 = RowStore(r0), ColumnStore(r0)
for i in SAMPLE:
    rs2.record(i)
    cs2.record(i)
print(f"all fields of {len(SAMPLE)} records: row-based {rs2.values_read} values / {rs2.accesses} "
      f"accesses, columnar {cs2.values_read} values / {cs2.accesses} accesses")

dataset layout      skipped blocks   values read  accesses    total m3
period order                   4/7          2835      1674     17307.2
subscriber order               2/7          4244      1676     17307.2
shuffled (seed 99)             0/7          4883      1678     17307.2
matching rows             read early    read late  late accesses   late gain
835 (26 percent)                9597         4869           1671       1.97x
1329 (41 percent)               9597         5857           2659       1.64x
2691 (84 percent)               9597         8581           5383       1.12x
the measured late path at 835 rows: 4869 values and 1671 accesses; formula 4869 and 1671
all fields of 10 records: row-based 60 values / 10 accesses, columnar 60 values / 60 accesses

The first table is the hidden default of the lesson. Block skipping looks like a property of the structure, but it depends on the dataset being clustered on that column. When the same dataset is written in subscriber order, all three periods are mixed into every block, the summaries cover the target, and only two of the seven blocks are skipped; in shuffled order none is skipped. The values read rise from 2,835 to 4,244 and from there to 4,883, and the 6.8-times gain falls to four times. The total comes out as 17,307.2 m³ in all three runs; the number the decision moves is not the answer but the values read to produce it. The question of which column to sort the dataset on is a layout decision, and it is invisible when the query is written.

The second table draws the limit of late materialization. When the filter leaves 835 rows, the late path reads 1.97 times fewer values than the early path; when it leaves 2,691 rows the gain falls to 1.12 times and scattered access rises from 1,671 to 5,383. As selectivity drops the position list grows, scattered reading approaches contiguous reading, and the filter moved forward does not pay for itself. The formulas are read from the measured paths themselves, and for the 835-row case they match the measured value exactly: 4,869 values, 1,671 accesses.

The last line measures the opposite direction. Fetching all of ten records reads 60 values in both layouts, but the row-based layout does it with 10 accesses and the columnar layout with 60. The columnar placement wins on questions that read few columns over many rows and loses on questions that read many columns over few rows. Layout is not a convenience preference; it is the decision of which question will be answered cheaply.

Summary

• The same question was answered by four paths and all four gave 8,752.6 m³ for the center and 17,307.2 m³ in total; the only thing that changed is the number of values read.
• The row-based scan reads 19,194 values; the columnar layout reads 9,597 with the three needed columns, 4,869 by moving the filter forward, and 2,835 by skipping blocks — 6.8 times.
• Late materialization lowers the values read but raises the accesses from three to 1,671; the gain depends on selectivity and falls to 1.12 times when the matching rows rise to 2,691.
• Block skipping depends on clustering: in the period-ordered dataset four of the seven blocks are skipped, in the subscriber-ordered dataset two, in the shuffled dataset none, and the values read rise from 2,835 to 4,883.
• Fetching all of a single record is 10 accesses in the row-based layout and 60 accesses in the columnar layout; the gain of the layout changes sign with the direction of the question.

Next Step

This lesson cut the values read by 6.8 times, but the whole computation was still done in a single place. As the dataset grows, the limit of that single place arrives: even if the values read drop to one sixth, when 3,199 rows become 3.2 billion rows the values a single machine would read still do not fit. The next lesson distributes the work across several partitions and measures the line item that distribution itself adds — how many units partitioning, shuffle and coordination add to the computation, and at what job size that line item turns into gain.