Columnar Storage Formats

The previous lesson counted the coordination overhead item around the computation and wrote how many items a stage adds, but it never opened what the partitions read when a stage starts. A partition reads its share from a file, and the layout of that file determines how many values must be touched to produce the same answer.

The measure of this lesson is the number of values read, and what is measured is not an engine but a file. The distinction matters: the Scale and Performance topic of the Data Analysis with Python course used the same unit of measure to measure the read path — when an engine applies the filter, when it materializes, which column the dataset is sorted by. Here the read path is fixed and the write order is fixed too; the only thing that moves is the layout of the file itself: whether the values are separated by column boundaries, whether they are divided into row groups, whether a row group header exists and how much space that header takes.

• AS19. The dataset is inherited from the Data Collection and Preparation, Data Analysis with Python, and Statistical Analysis and Exploratory Data Analysis courses and is fictional: 3,199 rows, 1,329 subscribers, three periods, five regions, tariff tiers 10/25/40 m³. It is written with seven columns and amounts to 22,393 values in total. It is a model.
• AS20. No file format product is used; the three layouts are modeled in memory and every value read is counted. It is a model.
• AS21. The write order is fixed: the file is written in the subscriber order the source gave. The effect of the sorting decision on skipping was measured in the Scale and Performance topic of the Data Analysis with Python course and is not repeated here.
• AS22. A row group header carries three values per column: the minimum, the maximum, and the row count. The header is written to the file and reading it is counted too.
• AS23. A row group is not opened if the filter range falls entirely outside the minimum and maximum values of that group.
• AS24. The query is fixed: the total consumption in a given subscriber range. It asks for two columns and its answer is 8,869.88 m³.

Three Layouts, One Query

# layout.py — the same FICTIONAL dataset is written in three file layouts and a single query is
# run in all three. MODEL: no file format product is used; the layouts are modeled in memory
# and the measure is the number of values read to produce the answer.
import math

SEED, RAW, M32 = 20260218, 1400, 0xFFFFFFFF
REGION = [("north", 0.28, 21), ("south", 0.22, 17), ("east", 0.18, 26),
          ("west", 0.14, 14), ("center", 0.18, 23)]
COLUMN = ("subscriber_id", "region", "period", "household", "satisfaction", "consumption_m3", "tier")


def rng(t):
    x = ((t ^ (t >> 16)) * 2246822507) & M32
    x = ((x ^ (x >> 13)) * 3266489909) & M32
    s = [(x ^ (x >> 16)) & M32]

    def next_value():
        s[0] = (s[0] * 1664525 + 1013904223) & M32
        return s[0] / 4294967296
    return next_value


def pick(u, w):
    t = 0.0
    for i, x in enumerate(w):
        t += x
        if u < t:
            return i
    return len(w) - 1


def tier(v):
    return "0-10" if v <= 10 else "10-25" if v <= 25 else "25-40" if v <= 40 else "over 40"


DATASET = []
for i in range(RAW):
    r = rng(SEED + i)
    b = REGION[pick(r(), [x[1] for x in REGION])]
    household = 1 + pick(r(), [0.06, 0.24, 0.30, 0.24, 0.11, 0.05])
    satisfied = 1 + pick(r(), [0.08, 0.14, 0.27, 0.34, 0.17])
    if r() < 0.046:
        continue
    subscriber_id, q = 10001 + i, rng(SEED + 7000 + 10001 + i)
    for d in range(pick(q(), [0.05, 0.12, 0.21, 0.62])):
        v = 0.0 if q() < 0.038 else math.floor(
            b[2] * math.exp((q() + q() + q() - 1.5) * 0.62)
            * (1 - d * 0.05) * 100 + 0.5) / 100
        DATASET.append((subscriber_id, b[0], d + 1, household, satisfied, v, tier(v)))
        q()

N, S = len(DATASET), len(COLUMN)
LOW, HIGH = 10601, 10800          # query: total consumption in this subscriber range
selected = [s for s in DATASET if LOW <= s[0] <= HIGH]
print(f"seed {SEED}; {N} rows, {S} columns, {N * S} values")
print(f"file was written in the order the source gave: subscriber {min(s[0] for s in DATASET)} - "
      f"{max(s[0] for s in DATASET)}")
print(f"query range {LOW}-{HIGH}: {len(selected)} rows match, correct answer "
      f"{sum(s[5] for s in selected):.2f} m3")

seed 20260218; 3199 rows, 7 columns, 22393 values
file was written in the order the source gave: subscriber 10001 - 11399
query range 10601-10800: 453 rows match, correct answer 8869.88 m3

The boundary of the dataset is 22,393 values. The query concerns 453 rows of these values and asks for two columns; that is, the values the answer genuinely needs number 906. The ratio between the two is the frame of all the tables of this lesson: the better a layout is, the closer the number of values read gets to that 906, and no layout can go below it.

# reading.py — the same query is run in three layouts. Measure: number of values read.
# D1 row-ordered flat layout, D2 column-bounded single row group,
# D3 row-grouped and headed columnar layout.
REQUESTED = ("subscriber_id", "consumption_m3")
FILTER, HEADER_VALUE = "subscriber_id", 3      # minimum, maximum, row count


def row_groups(r):
    return [DATASET[i:i + r] for i in range(0, N, r)]


def d1():
    return N * S, N, "all fields of a row contiguous"


def d2():
    return N * len(REQUESTED), N, "only the requested columns"


def d3(r, all_headers=False):
    g = row_groups(r)
    header = HEADER_VALUE * len(g) * (S if all_headers else 1)
    opened = [k for k in g if not (max(x[0] for x in k) < LOW
                                   or min(x[0] for x in k) > HIGH)]
    data = sum(len(k) for k in opened) * len(REQUESTED)
    return header + data, sum(len(k) for k in opened), len(g) - len(opened)


print(f"query asks for {len(REQUESTED)} columns: {', '.join(REQUESTED)}; "
      f"header is {HEADER_VALUE} values per row group")
print(f"{'layout':<40}{'values read':>14}{'rows read':>14}{'ratio':>8}")
t1, s1, _ = d1()
t2, s2, _ = d2()
t3, s3, skipped = d3(512)
print(f"{'D1 row-ordered flat layout':<40}{t1:>14}{s1:>14}{t1 / t1:>7.2f}x")
print(f"{'D2 column-bounded, single row group':<40}{t2:>14}{s2:>14}{t1 / t2:>7.2f}x")
print(f"{'D3 row groups (512) with headers':<40}{t3:>14}{s3:>14}{t1 / t3:>7.2f}x")
print(f"in D3, {skipped} of the {len(row_groups(512))} row groups were never opened")

query asks for 2 columns: subscriber_id, consumption_m3; header is 3 values per row group
layout                                     values read     rows read   ratio
D1 row-ordered flat layout                       22393          3199   1.00x
D2 column-bounded, single row group               6398          3199   3.50x
D3 row groups (512) with headers                  2069          1024  10.82x
in D3, 5 of the 7 row groups were never opened

The three layouts give the same answer and the values read fall from 22,393 to 2,069 — 10.82 times. The gain comes from two separate layout properties and each row of the table adds one of them.

The first property is the column boundary. In the row-ordered flat layout the seven fields of a row are written side by side; to read two fields you have to pass over all seven. In a file drawn with column boundaries each column sits in its own contiguous area and only two of them are read: 22,393 values fall to 6,398. This gain comes not from a query or an engine but from how the file was written.

The second property is the row group and the header. The file is divided into seven groups of 512 rows, and for each group the minimum, maximum, and row count of the subscriber column are written. If the query range does not touch the boundaries of a group, that group is not opened; five of the seven groups were not opened and the rows read fell from 3,199 to 1,024. The cost of reading the headers is 21 values, and what it saved is 4,350 values.

The third column shows the remaining share of skipping. The query wants 453 rows and the rows opened are 1,024; the 571 rows in between are read, and the filter eliminates them one by one. The reason is that skipping works at the grain of the row group: if a group touches the range with even a single row, it is opened in its entirety. The fineness of skipping is determined by the row group size, and that is the subject of the next section.

Row Group Size Is a Write Decision

# row_group.py — the row group size is a WRITE decision. A small row group gives fine skipping
# and many headers, a large row group gives coarse skipping and few headers.
# The write order is fixed: the subscriber order the source gave.
print(f"{'row group':<14}{'groups':>6}{'header':>8}{'skipped':>9}"
      f"{'opened rows':>14}{'data':>7}{'total':>8}{'ratio':>8}{'all headers':>12}")
best = None
for r in (32, 64, 128, 256, 512, 1024, 3199):
    g = row_groups(r)
    total, opened, skipped = d3(r)
    all_hdr = d3(r, True)[0]
    header = HEADER_VALUE * len(g)
    print(f"{r:<14}{len(g):>6}{header:>8}{skipped:>9}{opened:>14}"
          f"{opened * len(REQUESTED):>7}{total:>8}{t1 / total:>7.2f}x{all_hdr:>12}")
    if best is None or total < best[1]:
        best = (r, total)
all_best = min(((r, d3(r, True)[0]) for r in (32, 64, 128, 256, 512, 1024, 3199)),
               key=lambda x: x[1])
print(f"if only the filter column header is read, the best row group size is "
      f"{best[0]}: {best[1]} values, {t1 / best[1]:.2f} times")
print(f"if the headers of all columns are read, the best size is {all_best[0]}: {all_best[1]} values, "
      f"{t1 / all_best[1]:.2f} times; same file, two read decisions, two different bests")

# The gain depends on how many columns the query asks for.
print(f"{'requested cols':<15}{'D1':>8}{'D2':>8}{'D2 ratio':>10}"
      f"{'D3 (128)':>10}{'D3 ratio':>10}")
for k in (1, 2, 3, 5, 7):
    REQUESTED = COLUMN[:k]
    a, _, _ = d2()
    b = d3(128)[0]
    print(f"{k:<15}{t1:>8}{a:>8}{t1 / a:>9.2f}x{b:>10}{t1 / b:>9.2f}x")
REQUESTED = ("subscriber_id", "consumption_m3")

row group     groups  header  skipped   opened rows   data   total   ratio all headers
32               100     300       85           480    960    1260  17.77x        3060
64                50     150       42           512   1024    1174  19.07x        2074
128               25      75       20           640   1280    1355  16.53x        1805
256               13      39       10           768   1536    1575  14.22x        1809
512                7      21        5          1024   2048    2069  10.82x        2195
1024               4      12        3          1024   2048    2060  10.87x        2132
3199               1       3        0          3199   6398    6401   3.50x        6419
if only the filter column header is read, the best row group size is 64: 1174 values, 19.07 times
if the headers of all columns are read, the best size is 128: 1805 values, 12.41 times; same file, two read decisions, two different bests
requested cols       D1      D2  D2 ratio  D3 (128)  D3 ratio
1                 22393    3199     7.00x       715    31.32x
2                 22393    6398     3.50x      1355    16.53x
3                 22393    9597     2.33x      1995    11.22x
5                 22393   15995     1.40x      3275     6.84x
7                 22393   22393     1.00x      4555     4.92x

The first table is the hidden decision of the lesson. The row group size is chosen when a file is written, not when a query is written, and it costs in both directions. In a small group skipping gets finer — in a 32-row group 85 of a hundred groups are skipped and only 480 rows are opened — but the headers make you read 300 values, nearly a third of the data share. In a large group the header gets cheap, only three values in a single 3,199-row group, but nothing is left to skip and the read rises to 6,401 values.

The size that reads the fewest values is 64 rows: 1,174 values and 19.07 times against 22,393. The top of the table is not sharp — all the sizes between 32 and 128 stay above 16 times — but the ends are expensive, and the two ends are expensive for two separate reasons. It needs to be written again that what is measured in this table is not an engine behavior: the write order is the same in all seven rows, the query is the same, the read path is the same. The only thing that changes is how many rows apart a header is written to the file.

It also needs to be written where the ceiling is. The values the answer genuinely needed were 906; the best layout reads 1,174 values, that is, it is 1.30 times away from the best attainable. The remaining difference is made of two items: the header of 150 values and the 118 values of the 59 rows that fail to pass the filter. These two items cannot be zeroed together at any row group size, because one grows with the size and the other shrinks with it. The job of a file layout is not to meet the need of the query at zero cost but to find the size that keeps the sum of the two items the smallest.

The last column adds the second decision. In some layouts the reader pulls only the header of the filter column; in others all the column headers of a row group sit as a single block and are read together. In the second case the header is seven times more expensive and the best size moves from 64 to 128: 1,805 values, 12.41 times. The same file, two defensible read decisions, two different best sizes. A layout has no single number called “the best row group size”; that number cannot be produced without writing which header is read.

The second table shows that the gain depends on the query. The gain of the column boundary melts with the number of columns asked for and falls to 1.00 times when all seven columns are asked for: for a query that reads all the columns, the columnar layout is no different from the flat layout. Row group skipping does not melt, because skipping rows is independent of choosing columns; in the same row the ratio stays at 4.92 times. The column boundary lets you skip the column that will not be read, the row group header the row that will not be read; the gains of the two are multiplied, and when one is zeroed the other continues.

This distinction determines which question will be asked when a file layout is chosen. Queries that read few columns and scan many rows gain from the column boundary; queries that read a given range gain from the row group header. A query that does neither — an export asking for all the columns and all the rows — gains from no layout: the column boundary gives 1.00 times, there is no row group to skip, and the read rises above the flat layout by the value of the headers. The layout decision is made before the query, and which query will be answered cheaply is determined by that decision.

Summary

• The 3,199 rows and 7 columns of the fictional dataset amount to 22,393 values; the values the query genuinely needs are two columns of 453 rows, that is, 906.
• The same query reads 22,393 values in the row-ordered flat layout, 6,398 in the column-bounded layout, and 2,069 in the row-grouped layout with headers; five of the seven row groups are never opened and the gain is 10.82 times.
• The row group size costs in both directions: in a 32-row group the headers make you read 300 values, in a single 3,199-row group nothing is skipped; the size that reads the fewest values is 64 and gives 19.07 times with 1,174 values.
• Which headers are read is the second decision: when all the column headers are read, the best size moves to 128 and the gain falls to 12.41 times.
• The gain of the column boundary melts with the number of columns asked for and falls to 1.00 times in a seven-column query; row group skipping keeps giving 4.92 times in the same query.

Next Step

Four lessons counted the items of distribution one by one: records moved, coordination overhead items, values read. All three were items that could be reduced and a decision was found in each. What remains is a serial fraction that cannot be reduced. Whatever percentage of a job is indivisible, however many units are added, the speedup cannot rise above a ceiling. The next lesson presses on this ceiling by increasing the number of units and writes with numbers when distribution is unnecessary; the same lesson closes the course and the Data Analytics curriculum.