Parquet, Arrow, Polars streaming and out-of-core DuckDB.

When the table does not fit in memory

Lesson 6.6.1 put numbers on the word “big”: the ticks of a stock universe, L2 and L3 books or option chains quickly reach hundreds of gigabytes. The usual reaction is to think about a cluster. It is almost always premature: a laptop with a good on-disk format and an engine that does not load what it does not need goes much further than it seems. This lesson walks through that single-machine stack, from the bottom up:

flowchart TB
  P[Parquet on disk] --> A[Arrow in memory]
  A --> PL[Polars lazy]
  A --> D[DuckDB]
  A --> PD[pandas and NumPy]
  PL --> R[Small result]
  D --> R

• Parquet decides how much has to be read: columns, partitions and row groups with statistics (lesson 1.4.7 introduced the format; here we measure what each piece saves).
• Arrow is the in-memory format shared by all the tools in the figure: passing data from one to another does not have to cost a copy.
• Polars lazy and its streaming engine process the query in batches, without loading the whole table (the lazy API and explain() are in lesson 0.5.4).
• DuckDB runs SQL with a memory limit and spills to disk whatever does not fit (basic SQL is in lesson 0.5.5).

Measuring this needs a large dataset. The first cell generates synthetic ticks for 200 stocks from the level 5 panel (the tickers in panel5_activos.csv; the prices have no relation to those of the panel) over the 60 NYSE sessions of the first quarter of 2025. Each stock has its own average activity, from a few hundred to several thousand ticks per session, and its prices follow a random walk with a daily volatility of 2%. One Parquet file is written per session, with the rows sorted by symbol and time, in folders with the Hive layout (fecha=2025-03-14/).

from pathlib import Path

import exchange_calendars as xcals
import numpy as np
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq

tickers = np.array(
    sorted(pd.read_csv("panel5_activos.csv")["ticker"])[:200])
cal = xcals.get_calendar("XNYS")
sesiones = cal.sessions_in_range("2025-01-02", "2025-03-31")
rng = np.random.default_rng(662)
actividad = np.exp(rng.normal(np.log(1250), 0.8, 200))
ultimo = np.exp(rng.uniform(np.log(10), np.log(400), 200))
CENTROS = pa.array(["XNYS", "ARCX", "BATS", "EDGX", "OSCURO"])

def ticks_de_una_sesion(sesion):
    global ultimo
    abre = cal.session_open(sesion).value // 1000   # µs, UTC
    cierra = cal.session_close(sesion).value // 1000
    n = rng.poisson(actividad)            # ticks per symbol
    sim = np.repeat(np.arange(200), n)
    N = len(sim)
    inicio = np.repeat(np.cumsum(n) - n, n)
    ts = rng.integers(abre, cierra - 100_000, N)
    orden = np.lexsort((ts, sim))         # symbol and time
    ts = ts[orden] + (np.arange(N) - inicio)   # no ties
    paso = rng.normal(0, 1, N) * np.repeat(
        0.02 / np.sqrt(np.maximum(n, 1)), n)
    c = np.concatenate([[0.0], np.cumsum(paso)])
    camino = c[1:] - c[np.cumsum(n) - n].repeat(n)
    precio = np.round(np.repeat(ultimo, n) * np.exp(camino), 2)
    ultimo = np.where(n > 0, precio[np.cumsum(n) - 1], ultimo)
    return pa.table({
        "ts": pa.array(ts, pa.timestamp("us", tz="UTC")),
        "simbolo": pa.array(tickers).take(sim),
        "precio": precio,
        "tamano": rng.geometric(0.02, N).astype(np.int32),
        "lado": rng.choice(np.array([-1, 1], np.int8), N),
        "centro": CENTROS.take(
            rng.choice(5, N, p=[.3, .2, .15, .15, .2])),
    })

filas = memoria = 0
for s in sesiones:
    t = ticks_de_una_sesion(s)
    ruta = Path(f"ticks/fecha={s.date()}/parte-0.parquet")
    ruta.parent.mkdir(parents=True)
    pq.write_table(t, ruta, compression="zstd",
                   row_group_size=65_536)
    filas += t.num_rows
    memoria += t.nbytes
ficheros = sorted(Path("ticks").rglob("*.parquet"))
disco = sum(p.stat().st_size for p in ficheros)
print(f"{filas:,} ticks for {len(tickers)} symbols"
      f" in {len(sesiones)} sessions")
print(f"in Arrow (memory): {memoria / 1e6:,.0f} MB")
print(f"in Parquet (disk): {disco / 1e6:,.0f} MB")

20,783,757 ticks for 200 symbols in 60 sessions
in Arrow (memory): 783 MB
in Parquet (disk): 220 MB

Almost 21 million ticks: 783 MB in memory in the Arrow format and 220 MB on disk with Parquet and zstd, 3.6 times less. It is a size chosen so that the lesson runs in a minute on a laptop (the cell takes about ten seconds on the machine it was written on); the mechanisms are the same with a hundred times more data, and the proportions measured below do not depend on the size.

Inside the file: row groups and statistics

A Parquet file is not an opaque block. It is divided into row groups; inside each group, each column is stored separately (column chunk) in compressed pages, and at the end of the file comes the footer with the schema, the position of each chunk and its statistics: the minimum, the maximum and the null count of each column in each group11Note 1 · Apache Parquet, “Concepts”, “File Format” and “Configurations” (which recommends row groups of 512 MB to 1 GB): https://parquet.apache.org/docs/file-format/configurations/, accessed October 23, 2026..

flowchart TB
  F[Parquet file] --> G0[Row group 0]
  F --> G1[Row group 1]
  F --> PIE[Footer: schema and min/max]
  G0 --> C1[Column ts]
  G0 --> C2[Column simbolo]
  C2 --> PG[zstd pages]

The footer is the first thing any reader reads: with it, it decides which chunks it needs without opening the rest. The one for the March 14 session:

pf = pq.ParquetFile("ticks/fecha=2025-03-14/parte-0.parquet")
md = pf.metadata
print(md.num_rows, "rows,", md.num_row_groups, "row groups,",
      f"footer of {md.serialized_size / 1e3:.1f} KB")
filas_rg = []
for i in range(md.num_row_groups):
    rg = md.row_group(i)
    s_sim, s_ts = rg.column(1).statistics, rg.column(0).statistics
    filas_rg.append({
        "grupo": i, "filas": rg.num_rows,
        "simbolo_min": s_sim.min, "simbolo_max": s_sim.max,
        "ts_min": s_ts.min.strftime("%H:%M:%S"),
        "ts_max": s_ts.max.strftime("%H:%M:%S"),
        "MB": round(rg.total_byte_size / 1e6, 1)})
pd.DataFrame(filas_rg)

346529 rows, 6 row groups, footer of 4.5 KB
   grupo  filas simbolo_min simbolo_max    ts_min    ts_max   MB
0      0  65536        BAGO        CELE  13:30:00  19:59:59  0.9
1      1  65536        CELE        DORU  13:30:00  19:59:59  1.0
2      2  65536        DORU        GUBO  13:30:00  19:59:58  0.9
3      3  65536        GUBO        LAVI  13:30:00  19:59:59  0.9
4      4  65536        LAVI        MENO  13:30:00  19:59:59  0.9
5      5  18849        MENO        MONA  13:30:01  19:59:58  0.3

Six groups of 65,536 rows (the last one with the remainder) and a 4.5 KB footer that describes them. Since the file is sorted by symbol, each group covers a narrow alphabetical range (BAGO to CELE, CELE to DORU…) and the whole session in the ts column: from 13:30 to 20:00 UTC, the New York session on daylight saving time, which began on March 9. Those statistics are what enable predicate pushdown: a filter simbolo = 'BAGO' can only be satisfied in group 0, because in the others the minimum is already later in the alphabet. The MB column is the uncompressed size of each group according to the footer.

Predicate pushdown, measured in bytes

That a reader “can” skip data is something you read everywhere. Here it is measured. The next cell wraps the local disk in an fsspec file system that adds up the bytes actually read and passes it to pyarrow.dataset, PyArrow's reader for sets of files. Before measuring, it reads the 60 footers once (a reader keeps them in memory after the first query): that way each query counts only the data.

import datetime as dt

import pyarrow.compute as pc
import pyarrow.dataset as ds
from fsspec.implementations.local import LocalFileSystem
from pyarrow.fs import FSSpecHandler, PyFileSystem

class Contador(LocalFileSystem):
    """Local disk that records how many bytes are read."""
    bytes_leidos = 0
    lecturas = 0

    def _open(self, path, mode="rb", **kw):
        f = super()._open(path, mode, **kw)
        leer = f.read
        def read(n=-1):
            datos = leer(n)
            self.bytes_leidos += len(datos)
            self.lecturas += 1
            return datos
        f.read = read
        return f

disco_contado = Contador()
lago = ds.dataset(
    str(Path("ticks").resolve()), format="parquet",
    filesystem=PyFileSystem(FSSpecHandler(disco_contado)),
    partitioning=ds.partitioning(
        pa.schema([("fecha", pa.date32())]), flavor="hive"))
for frag in lago.get_fragments():   # read the 60 footers once
    frag.ensure_complete_metadata()
print(f"footers: {disco_contado.lecturas} reads,"
      f" {disco_contado.bytes_leidos / 1e6:.1f} MB")

def medir(nombre, columnas=None, filtro=None):
    disco_contado.bytes_leidos = disco_contado.lecturas = 0
    t = lago.to_table(columns=columnas, filter=filtro)
    return {"consulta": nombre, "filas": t.num_rows,
            "MB_leidos": round(disco_contado.bytes_leidos / 1e6, 2),
            "pct_lago": round(100 * disco_contado.bytes_leidos
                              / disco, 2)}

f14 = pc.field("fecha") == dt.date(2025, 3, 14)
bago = pc.field("simbolo") == "BAGO"
consultas = pd.DataFrame([
    medir("everything"),
    medir("price only", ["precio"]),
    medir("one session", None, f14),
    medir("BAGO in one session", ["ts", "precio"], f14 & bago),
    medir("BAGO in the whole lake", ["ts", "precio"], bago),
    medir("precio > 1000", ["precio"], pc.field("precio") > 1000),
])
consultas

footers: 61 reads, 4.0 MB
                 consulta     filas  MB_leidos  pct_lago
0              everything  20783757     219.23     99.88
1              price only  20783757      35.20     16.04
2             one session    346529       3.62      1.65
3     BAGO in one session      1568       0.59      0.27
4  BAGO in the whole lake     90225      35.57     16.21
5           precio > 1000         0       0.00      0.00

The figure in the header plots the last column. The footers are read with 61 reads (one per file plus one more, for the first file, which ds.dataset opens to infer the schema) and add up to 4.0 MB. After that, each query:

• Everything: 219.23 MB, 99.88% of the lake. What is missing up to 100% is the footers, already read.
• Price only: 16%. This is projection: the other columns are not even touched. The price weighs more than its proportional share because it is the least repetitive column; symbols and venues, with few distinct values, are dictionary-encoded and take up almost nothing.
• One session: 1.65%, one file out of 60. This is partition pruning: the value of fecha is in the path, and the reader discards 59 files without opening them.
• BAGO in one session: 0.59 MB, 0.27% of the lake. Partition plus statistics: inside the file, only group 0 can contain BAGO, and from that group only three columns are read (ts and precio, which are requested, and simbolo, which is needed to filter row by row inside the group).
• BAGO in the whole lake: 16.2%. Without a date filter all 60 files are opened, but in each one only the first group. It reads almost as much as “price only” to return 230 times fewer rows.
• precio > 1000: zero bytes. The maximum of precio in the footer of each group is below 1,000, so the reader knows that no row satisfies the filter without reading a single page.

Statistics only help if the data is sorted by the filter column. The next cell checks this: it rewrites the March 14 session sorted by time and counts, with split_by_row_group, how many groups survive two different filters in each version.

un_dia = pq.read_table("ticks/fecha=2025-03-14/parte-0.parquet")
pq.write_table(un_dia.sort_by("ts"), "por_hora.parquet",
               compression="zstd", row_group_size=65_536)

def grupos_leidos(ruta, filtro):
    frag = next(ds.dataset(ruta).get_fragments())
    vivos = frag.split_by_row_group(filtro)
    return sum(len(f.row_groups) for f in vivos)

franja = ((pc.field("ts") >= pd.Timestamp("2025-03-14 15:00Z"))
          & (pc.field("ts") < pd.Timestamp("2025-03-14 15:05Z")))
orden = pd.DataFrame({
    "simbolo = BAGO": [
        grupos_leidos("ticks/fecha=2025-03-14", bago),
        grupos_leidos("por_hora.parquet", bago)],
    "5 minutes": [
        grupos_leidos("ticks/fecha=2025-03-14", franja),
        grupos_leidos("por_hora.parquet", franja)],
}, index=["sorted by symbol", "sorted by time"])
orden.columns.name = "row groups read"
orden

row groups read   simbolo = BAGO  5 minutes
sorted by symbol               1          6
sorted by time                 6          1

Symmetric: sorted by symbol, the symbol filter reads 1 group out of 6 and a 5-minute window reads all of them (each group has ticks from the whole session); sorted by time, the reverse. No single order serves everything, and the choice depends on the usual queries. For a backtest that goes through one asset at a time, symbol and time; to rebuild the whole market at one instant (a consolidated book, a cross-section), time. Some lakes keep both copies. Another option is to partition by one column and sort by the other, as this lesson does: partition by date and sort by symbol, so that a filter by date and symbol reads a single group from a single file.

Arrow: the same bytes for everyone

Reading little from disk is useless if afterwards each tool copies the data into its own format. Apache Arrow defines how columns are laid out in memory (contiguous value buffers, a validity bitmap for nulls and offsets for strings) in a language-agnostic way22Note 2 · Apache Arrow, “Arrow Columnar Format”: https://arrow.apache.org/docs/format/Columnar.html, accessed October 23, 2026.. Polars, DuckDB, pandas with ArrowDtype and PyArrow understand that format and can work on the same buffers without copying them (zero-copy). The cell checks this with the March 14 session: it measures how much new memory the Arrow allocator reserves in each conversion and whether the price buffer is the same.

import duckdb
import polars as pl

t = un_dia.combine_chunks()          # one chunk per column
direccion = t.column("precio").chunk(0).buffers()[1].address

def mb_nuevos(funcion):
    antes = pa.total_allocated_bytes()
    resultado = funcion()
    return resultado, (pa.total_allocated_bytes() - antes) / 1e6

df_pl, mb_pl = mb_nuevos(lambda: pl.from_arrow(t))
df_ar, mb_ar = mb_nuevos(
    lambda: t.to_pandas(types_mapper=pd.ArrowDtype))
df_np, mb_np = mb_nuevos(lambda: t.to_pandas())
v = t.column("precio").chunk(0).to_numpy(zero_copy_only=True)

print(f"Polars:            {mb_pl:5.1f} MB new,",
      "same buffer:",
      df_pl["precio"].to_arrow().buffers()[1].address == direccion)
print(f"pandas ArrowDtype: {mb_ar:5.1f} MB new")
print(f"pandas NumPy:      {mb_np:5.1f} MB new")
print("NumPy zero-copy:", v.ctypes.data == direccion,
      "| writable:", v.flags.writeable)
duckdb.sql("""
    SELECT count(*) AS ticks,
           round(sum(precio * tamano) / sum(tamano), 4) AS vwap
    FROM t WHERE simbolo = 'BAGO'
""")

Polars:              0.0 MB new, same buffer: True
pandas ArrowDtype:   0.0 MB new
pandas NumPy:       12.8 MB new
NumPy zero-copy: True | writable: False
┌───────┬─────────┐
│ ticks │  vwap   │
│ int64 │ double  │
├───────┼─────────┤
│  1568 │ 23.9505 │
└───────┴─────────┘

What each line says:

• Polars reserves nothing: its DataFrame points to the same price buffer as the Arrow table (the address comparison gives True; the addresses are not printed because they change on every run).
• pandas with ArrowDtype does not either: each column is a wrapper around the Arrow array.
• pandas with NumPy (what to_pandas() returns by default) reserves 12.8 MB in the Arrow allocator: the copy of the columns into the format NumPy expects.
• NumPy can view the price buffer without copying it, but read-only (writable: False): writing to it would change data that the other tools believe is immutable.
• DuckDB queries the Python variable t directly (the replacement scan from lesson 0.5.5) and reads its buffers in place: 1,568 BAGO ticks, the same ones medir read.

Zero-copy has conditions. The most common one in market data is nulls. Arrow marks them in a separate bitmap and leaves an arbitrary value in the slot; NumPy has no nulls for float64, so to hand them over it has to copy and write NaN in those slots. If no copy is demanded, PyArrow refuses:

con_nulos = pa.array([101.5, None, 101.7])
con_nulos.to_numpy(zero_copy_only=True)

pyarrow.lib.ArrowInvalid: Needed to copy 1 chunks with 1 nulls, but zero_copy_only was True

The other conditions: that the column is in a single chunk (that is why the cell uses combine_chunks(); a table read from several files has several) and that the type has the same representation in both tools. Strings, for example, are a byte buffer with offsets in Arrow and Python objects in a pandas object column: going from one to the other always copies.

Polars lazy: the plan before the data

Lesson 0.5.4 introduced scan_parquet and explain() with one file. With a partitioned lake, the plan also shows which files are opened. The query asks for BAGO's number of ticks and volume per trading venue in the week of March 3 to 7. First the plan as written (optimized=False) and then the one Polars runs:

from datetime import date

ticks = pl.scan_parquet("ticks/**/*.parquet", hive_partitioning=True)
semana = (
    ticks
    .filter(pl.col("fecha").is_between(date(2025, 3, 3),
                                       date(2025, 3, 7))
            & (pl.col("simbolo") == "BAGO"))
    .group_by("centro")
    .agg(ticks=pl.len(),
         volumen=pl.col("tamano").cast(pl.Int64).sum())
    .sort("centro")
)
print(semana.explain(optimized=False))
print()
print(semana.explain())

SORT BY [col("centro")]
  AGGREGATE[maintain_order: false]
    [len().alias("ticks"), col("tamano").cast(Int64).sum().alias("volumen")] BY [col("centro")]
    FROM
    FILTER col("simbolo") == "BAGO"
    FROM
      FILTER col("fecha").is_between([2025-03-03, 2025-03-07])
      FROM
        Parquet SCAN [ticks/fecha=2025-01-02/parte-0.parquet, ... 59 other sources]
        PROJECT */7 COLUMNS
        ESTIMATED ROWS: 20780871

SORT BY [col("centro")]
  AGGREGATE[maintain_order: false]
    [len().alias("ticks"), col("tamano").cast(Int64).sum().alias("volumen")] BY [col("centro")]
    FROM
    simple π 2/2 ["centro", "tamano"]
      Parquet SCAN [ticks/fecha=2025-03-03/parte-0.parquet, ... 4 other sources]
      PROJECT 4/7 COLUMNS
      SELECTION: (col("fecha").is_between([2025-03-03, 2025-03-07]) & (col("simbolo") == "BAGO"))
      ESTIMATED ROWS: 1731740

Plans are read from the bottom up. Unoptimized, Polars would read all 60 files (Parquet SCAN [… 59 other sources]) with all the columns (PROJECT */7 COLUMNS: the six in the file plus fecha, which comes from the path) and filter afterwards, in two steps. The optimized plan changes three things:

• Partition pruning: Parquet SCAN [ticks/fecha=2025-03-03/…, … 4 other sources]. The date filter is resolved with the folder names: five files out of 60.
• Projection: PROJECT 4/7 COLUMNS. It reads simbolo and fecha to filter, and centro and tamano to aggregate.
• Filter pushed down to the reader: SELECTION: … inside the Parquet SCAN. The reader applies the filter while reading, and can use the group statistics to skip those that do not contain BAGO.

ESTIMATED ROWS is an estimate from the file sizes, not a count. The result:

semana.collect()

shape: (5, 3)
┌────────┬───────┬─────────┐
│ centro ┆ ticks ┆ volumen │
│ ---    ┆ ---   ┆ ---     │
│ str    ┆ u32   ┆ i64     │
╞════════╪═══════╪═════════╡
│ ARCX   ┆ 1562  ┆ 77312   │
│ BATS   ┆ 1106  ┆ 56094   │
│ EDGX   ┆ 1147  ┆ 57924   │
│ OSCURO ┆ 1560  ┆ 79308   │
│ XNYS   ┆ 2224  ┆ 114732  │
└────────┴───────┴─────────┘

BAGO had 7,599 ticks that week, spread across the five venues in proportions similar to those of the generator (30% on XNYS, 20% on ARCX and on the dark pool, OSCURO, 15% on BATS and on EDGX): the query reads what it has to read and nothing more.

The streaming engine

collect() runs the plan with the default engine. With engine="streaming", Polars runs the same plan in batches of rows: it reads a chunk of each file, filters it and aggregates it, and frees the memory before reading the next one. The full table is never in memory, only the state of the aggregation33Note 3 · Polars, user guide, “Streaming”: https://docs.pola.rs/user-guide/concepts/streaming/, accessed October 23, 2026.. The cell computes daily bars for the 200 symbols (open, high, low, close, volume, VWAP and number of ticks) from the 21 million ticks:

diarias = (
    ticks
    .with_columns(importe=pl.col("precio") * pl.col("tamano"))
    .group_by("simbolo", "fecha")
    .agg(apertura=pl.col("precio").first(),
         maximo=pl.col("precio").max(),
         minimo=pl.col("precio").min(),
         cierre=pl.col("precio").last(),
         volumen=pl.col("tamano").cast(pl.Int64).sum(),
         importe=pl.col("importe").sum(),
         ticks=pl.len())
    .with_columns(vwap=(pl.col("importe") / pl.col("volumen"))
                  .round(4))
    .drop("importe")
    .sort("simbolo", "fecha")
    .collect(engine="streaming")
)
print(diarias.shape, diarias["ticks"].sum())
diarias.filter(pl.col("simbolo") == "BAGO").select(
    "fecha", "apertura", "cierre", "volumen", "ticks", "vwap"
).head(3)

(12000, 9) 20783757
shape: (3, 6)
┌────────────┬──────────┬────────┬─────────┬───────┬─────────┐
│ fecha      ┆ apertura ┆ cierre ┆ volumen ┆ ticks ┆ vwap    │
│ ---        ┆ ---      ┆ ---    ┆ ---     ┆ ---   ┆ ---     │
│ date       ┆ f64      ┆ f64    ┆ i64     ┆ u32   ┆ f64     │
╞════════════╪══════════╪════════╪═════════╪═══════╪═════════╡
│ 2025-01-02 ┆ 27.23    ┆ 27.54  ┆ 80546   ┆ 1546  ┆ 27.5508 │
│ 2025-01-03 ┆ 27.53    ┆ 27.39  ┆ 72071   ┆ 1481  ┆ 27.254  │
│ 2025-01-06 ┆ 27.38    ┆ 26.85  ┆ 75824   ┆ 1497  ┆ 27.0718 │
└────────────┴──────────┴────────┴─────────┴───────┴─────────┘

first() and last() respect the read order, which here is that of the files (by symbol and time); in a lake without that order you would have to use sort_by("ts") inside the aggregation. The 12,000 bars (200 symbols × 60 sessions) add up to the 20,783,757 ticks: none has been lost along the way.

What matters does not show in the output: memory. Measured on the machine the lesson was written on (16 cores, each variant in a separate process, peak resident memory of the process), the same daily bars:

The full read pays for the 783 MB of Arrow plus the copies and the intermediate structures. The in-memory lazy query already saves (it reads only the columns it needs), and streaming drops to a quarter of the full read. With a lake ten times larger, the first three would grow in a similar way; the streaming one, much less, because it only keeps 12,000 groups.

Out-of-core DuckDB

DuckDB tackles the same problem from the other side: a memory manager with a limit (memory_limit) and a temporary folder (temp_directory) to which the operators that support it spill whatever does not fit (spilling): aggregations, joins, sorts and window functions44Note 4 · DuckDB 1.5, “Tuning Workloads”: https://duckdb.org/docs/1.5/guides/performance/how_to_tune_workloads, accessed October 23, 2026.. The test: a tick-by-tick log return with lag() over the 21 million rows, which forces each symbol to be sorted by time, with a 300 MB limit.

con = duckdb.connect()
con.sql("SET memory_limit = '300MB'")
con.sql("SET temp_directory = 'duckdb_tmp'")
con.sql("SET threads = 4")
print(con.sql("SELECT current_setting('memory_limit')").fetchone())

RV = """
    WITH r AS (
        SELECT simbolo,
               ln(precio / lag(precio) OVER (
                   PARTITION BY simbolo ORDER BY ts)) AS ret
        FROM read_parquet('ticks/**/*.parquet',
                          hive_partitioning = true)
    )
    SELECT count(*) AS ticks, count(ret) AS rendimientos,
           round(sum(ret * ret), 3) AS suma_r2
    FROM r
"""
con.sql(RV)

('286.1 MiB',)
┌──────────┬──────────────┬─────────┐
│  ticks   │ rendimientos │ suma_r2 │
│  int64   │    int64     │ double  │
├──────────┼──────────────┼─────────┤
│ 20783757 │     20783557 │   5.427 │
└──────────┴──────────────┴─────────┘

300MB is stored as 286.1 MiB (DuckDB shows mebibytes, of 1,048,576 bytes). The query finishes: 20,783,757 ticks and 20,783,557 returns (rendimientos), 200 fewer, because the first tick of each symbol has no predecessor and lag() gives null. suma_r2 is the sum of squared returns of the 200 symbols over the 60 sessions, including the jumps between one day's close and the next day's open: 5.427, about 0.027 per symbol. Spread over 60 sessions, √(0.027 / 60) ≈ 2.1% daily, the volatility the prices were generated with (2%) plus the noise from rounding to the cent, which in cheap stocks adds tick-by-tick variance. It is the known answer, and it serves to check that the window has paired each tick with the previous one of its symbol.

What you do not see: measured on the lesson's machine, DuckDB wrote about 500 MB to duckdb_tmp during the query and the process did not go above about 530 MB of resident memory. Without a limit, the same query used ≈ 1.3 GB and wrote nothing to disk. The temporary files are deleted when the query finishes.

With less memory, the number of threads matters: each thread needs its own buffers. With 100 MB and four threads, the query fails; with one thread, it finishes (more slowly):

con.sql("SET memory_limit = '100MB'")
try:
    con.sql(RV).fetchall()
except duckdb.OutOfMemoryException as e:
    lineas = str(e).splitlines()
    print(lineas[0].split(":")[0])
    print("\n".join(l for l in lineas if l.startswith("*")))
con.sql("SET threads = 1")
print(con.sql(RV).fetchall())

Out of Memory Error
* Reducing the number of threads (SET threads=X)
* Disabling insertion-order preservation (SET preserve_insertion_order=false)
* Increasing the memory limit (SET memory_limit='...GB')
[(20783757, 20783557, 5.427)]

The error message comes with three remedies, in this order: fewer threads, preserve_insertion_order = false (DuckDB stops guaranteeing that rows come out in the order they went in when the query has no ORDER BY, which saves memory in large imports and exports) and more memory. With one thread, the result is identical to the previous one; measured on the lesson's machine, it took about eight seconds, versus about three with four threads and 300 MB.

The DuckDB documentation spells out the limits of spilling: all blocking operators (grouping, joining, sorting and windows) can work with more data than memory, but some aggregate functions, such as string_agg(), do not spill to disk, and a query with several blocking operators can run out of memory even if each one on its own would not⁴.

What to use and when

The rule that sums up the lesson: before buying memory or setting up a cluster, look at the plan and count the bytes. In the lesson's lake, the difference between the worst and the best way to ask for BAGO's ticks for one day is 370 times (219 MB versus 0.59 MB), without changing anything in the hardware.

Exercises

1. Without filtering by fecha, ask the lake (medir) for the ts and precio columns of the 15:00 to 15:05 UTC window on March 14 (franja). How many bytes does it read and why is it not the 220 MB of the lake?

pd.DataFrame([medir("window without date", ["ts", "precio"], franja),
              medir("window with date", ["ts", "precio"],
                    franja & f14)])

              consulta  filas  MB_leidos  pct_lago
0  window without date   4481       3.08       1.4
1     window with date   4481       3.08       1.4

2. Compute the same daily bars with DuckDB (with arg_min and arg_max for the open and the close) and check that they match the Polars ones.

con.sql("SET threads = 4")
con.sql("SET memory_limit = '300MB'")
d_duck = con.sql("""
    SELECT simbolo, fecha,
           arg_min(precio, ts) AS apertura, max(precio) AS maximo,
           min(precio) AS minimo, arg_max(precio, ts) AS cierre,
           sum(tamano) AS volumen,
           round(sum(precio * tamano) / sum(tamano), 4) AS vwap,
           count(*) AS ticks
    FROM read_parquet('ticks/**/*.parquet', hive_partitioning = true)
    GROUP BY simbolo, fecha ORDER BY simbolo, fecha
""").pl()
cols = ["apertura", "maximo", "minimo", "cierre", "volumen",
        "vwap", "ticks"]
iguales = {c: bool((d_duck[c] == diarias[c].cast(d_duck[c].dtype))
                   .all()) for c in cols}
print(d_duck.shape, iguales)

(12000, 9) {'apertura': True, 'maximo': True, 'minimo': True, 'cierre': True, 'volumen': True, 'vwap': True, 'ticks': True}

import shutil

con.close()
shutil.rmtree("ticks")
shutil.rmtree("duckdb_tmp", ignore_errors=True)
Path("por_hora.parquet").unlink()
print(sorted(p.name for p in Path(".").glob("ticks*")))

[]

3. A colleague wants to store the L2 book updates (lesson 1.1.3) of 500 stocks, about 2 billion rows a month, for two uses: backtests of one asset at a time and reconstructions of the whole market at specific moments. Propose the partitioning, the order and the row group size.