Memory and spilling¶

A query in ursus runs under a memory budget. When the operators that hold data across batches, a sort, a group-by, a join's build side, reach it, they write what they hold to disk and carry on, and the answer is the one they give in memory. An operator that cannot spill fails at the budget with an error naming itself, instead of growing until the operating system kills the process.

The default budget¶

A query with no limit of its own is budgeted at half the memory the process can have: the smaller of its container's limit and the machine's memory on Linux, and the machine's memory on macOS and Windows. Half, because ursus's buffers live on the Go heap, which grows to about twice what is live before it collects.

Under that default, ursus also sets the Go runtime's soft memory limit near the same ceiling, so the heap's slack stays under it. GOMEMLIMIT, or calling ursus.LeaveProcessMemoryLimit() before the first query, keeps a program's own choice.

A limit of your own¶

WithMemoryLimit sets the budget for one query, WithSpillDir where its files go, and WithMemoryStats reports what it held and how many files it wrote. A group-by over 25,000 keys under 256 KiB:

keys := make([]int64, 100_000)
for i := range keys {
    keys[i] = int64(i * 7919 % 25_000)
}
var stats ursus.MemoryStats
df, err := ursus.Frame(ursus.Values("k", keys)).
    GroupBy(ursus.Col("k")).Agg(ursus.Len().Alias("n")).
    Select(ursus.Len().Alias("groups"), ursus.Col("n").Max().Alias("largest")).
    Collect(context.Background(),
        ursus.WithMemoryLimit(256<<10),
        ursus.WithSpillDir(os.TempDir()),
        ursus.WithMemoryStats(&stats))
if err != nil {
    log.Fatal(err)
}
fmt.Println(df)
fmt.Println("spilled:", stats.Spills > 0)

Output

shape: (1, 2)
┌────────┬─────────┐
│ groups │ largest │
│ Uint64 │ Uint64  │
├────────┼─────────┤
│ 25000  │ 4       │
└────────┴─────────┘

spilled: true

The spill files are removed when the query ends, on success, on an error, and when a loop over CollectBatches breaks early. WithMemoryLimit(0) turns the budget off.

What spills¶

sort, and a top-k with any k;
group-by, including the values a median, a quantile or an NUnique holds, the pairs of a CorrSpearman and the counts of a Mode;
join, both sides partitioned;
unique;
window functions, over many partitions or one huge one;
keyed Rolling and GroupByDynamic;
tail and reverse.

JoinAsOf, MergeSorted and HStack stream their input, as Rolling and GroupByDynamic do, so they hold little to begin with. Changed in 0.6.7: a CorrSpearman over a group larger than the limit spills its pairs, in a group-by, a window or a temporal group, and a Mode its counts in a group-by or a temporal group, where each failed.

What does not¶

A few answers are as large as their input, and cannot be divided across files: Implode holds every value of a group as its answer, for one. Each such operator fails at the budget, and its error says what it holds, why it cannot spill, and what the alternatives are:

keys := make([]int64, 100_000)
_, err := ursus.Frame(ursus.Values("k", keys)).
    Select(ursus.Col("k").Implode()).
    Collect(context.Background(),
        ursus.WithMemoryLimit(64<<10),
        ursus.WithSpillDir(os.TempDir()))
fmt.Println(err)

Output

ursus: group_by: memory limit of 64.0KiB exceeded: the query is holding 105.2KiB, 105.2KiB of it in this operator
  group_by holds one entry per distinct key, and implode holds every VALUE as its answer — which partitioning to disk cannot divide, because it divides across keys (quantile, median, n_unique, mode and corr_spearman spill their values); raise the limit with WithMemoryLimit
  sort, group_by, join, unique, over, reverse and tail spill to disk; join_asof, merge_sorted, hstack, rolling and group_by_dynamic stream their input; an over computing a rolling window by time one of whose periods holds more than the limit fails rather than exceed it, and so does implode over one group larger than it

Collect itself holds its whole answer, by definition. An answer larger than memory goes through CollectBatches, or WriteParquet and WriteCSV, which stream it.

A larger-than-memory query¶

A sort of more Parquet than fits in memory, written to a file as it comes out of the sort, holding at most 512 MiB:

ctx := context.Background()
err := ursus.ScanParquet("events/*.parquet").
    Sort(ursus.Desc(ursus.Col("ts"))).
    SinkParquet(ctx, "sorted.parquet",
        ursus.WithMemoryLimit(512<<20),
        ursus.WithSpillDir("/var/tmp/ursus"))
if err != nil {
    log.Fatal(err)
}

Each path that spills is tested under a limit far below its input, against its answer without one.