{"id":8131,"date":"2026-02-18T20:15:07","date_gmt":"2026-02-18T19:15:07","guid":{"rendered":"https:\/\/launix.de\/launix\/?p=8131"},"modified":"2026-03-10T11:17:18","modified_gmt":"2026-03-10T10:17:18","slug":"36-faster-without-changing-a-single-query-inside-memcps-new-scan-engine","status":"publish","type":"post","link":"https:\/\/launix.de\/launix\/en\/36-faster-without-changing-a-single-query-inside-memcps-new-scan-engine\/","title":{"rendered":"36% Faster Without Changing a Single Query: Inside MemCP’s New Scan Engine"},"content":{"rendered":"\n

MemCP is an in-memory columnar database written in Go with a Scheme scripting layer for query compilation, planning and optimization. Its storage engine organizes data into shards, each containing a compressed main storage and a small append-only delta buffer for recent writes. Until now, the two primary scan paths \u2014 unordered aggregation (scan<\/code>) and ordered retrieval (scan_order<\/code>) \u2014 each contained their own inline loops for reading columns, dispatching between main and delta storage, calling the map function, and folding the reduce. This worked, but the duplication made it harder to optimize either path and \u2014 more critically \u2014 impossible to swap in a network-transparent implementation for distributed query execution.<\/p>\n\n\n\n\n\n\n\n

This post describes the refactoring we just landed: ShardMapReducer<\/strong>, a universal batch map+reduce interface that unifies both scan paths, eliminates per-row heap allocations, and lays the groundwork for clustering.<\/p>\n\n\n\n

The Problem: Two Scan Paths, Duplicated Logic<\/h2>\n\n\n\n

Before this change, scan.go<\/code> and scan_order.go<\/code> each independently:<\/p>\n\n\n\n

    \n
  1. Opened column readers for the requested columns<\/li>\n\n\n\n
  2. Iterated over matching record IDs<\/li>\n\n\n\n
  3. For each ID, branched on whether it lived in main storage or the delta buffer<\/li>\n\n\n\n
  4. Called ColumnStorage.GetValue(id)<\/code> or getDelta(id - mainCount, col)<\/code><\/li>\n\n\n\n
  5. Invoked the map callback<\/li>\n\n\n\n
  6. Folded the result into an accumulator via reduce<\/li>\n<\/ol>\n\n\n\n

    The two paths differed only in how<\/em> record IDs were produced: scan<\/code> iterates an index and filters inline; scan_order<\/code> sorts filtered IDs per shard, then merges them globally via a priority queue. But the map+reduce hot loop was copy-pasted between them, and any optimization (JIT compilation, batch column reads, SIMD) would have to be applied twice.<\/p>\n\n\n\n

    Worse, both paths interleaved filter evaluation with map+reduce execution in a single loop. For each row, the code would read the filter columns, evaluate the condition, then \u2014 if the row passed \u2014 immediately read the map columns and call map+reduce. This meant the CPU cache had to juggle both filter-column data and map-column data simultaneously, evicting one to load the other on every row.<\/p>\n\n\n\n

    The Solution: ShardMapReducer<\/h2>\n\n\n\n

    We introduced a single struct that encapsulates the map+reduce pipeline:<\/p>\n\n\n\n

    type ShardMapReducer struct {\n    shard     *storageShard\n    mainCols  []ColumnStorage   \/\/ direct main storage references\n    colNames  []string          \/\/ for delta access\n    isUpdate  []bool            \/\/ $update columns get UpdateFunction\n    args      []scm.Scmer       \/\/ pre-allocated argument buffer\n    mapFn     func(...scm.Scmer) scm.Scmer\n    reduceFn  func(...scm.Scmer) scm.Scmer\n    mainCount uint\n}<\/code><\/pre>\n\n\n\n
    The key method is Stream(acc, recids) -> acc<\/code>: given an accumulator and a batch of record IDs, it returns the new accumulator after applying map+reduce to each record. Column readers are opened once at construction (OpenMapReducer<\/code>), and the args buffer is pre-allocated \u2014 no allocations in the hot path.<\/p>\n\n\n\n
    Run Detection: Main vs Delta<\/h3>\n\n\n\n
    The Stream<\/code> method doesn’t just iterate blindly. It partitions the incoming ID batch into contiguous runs<\/strong> of the same storage type:<\/p>\n\n\n\n
    recids: [3, 7, 12, 15, 1001, 1003, 18, 22, 1005]\n         \\_____________\/  \\________\/  \\_____\/ \\__\/\n          MainBlock(4)  DeltaBlock(2) Main(2) Delta(1)<\/code><\/pre>\n\n\n\n
    Each run is dispatched to either processMainBlock<\/code> or processDeltaBlock<\/code>. The main block is a tight loop with direct ColumnStorage.GetValue<\/code> calls \u2014 no branching on storage type per row. In the common case (99%+ of data in compressed main storage, only a handful of recent delta rows), this means a single processMainBlock<\/code> call covers almost the entire batch.<\/p>\n\n\n\n
    These two methods are the future JIT compilation targets \u2014 more on that below.<\/p>\n\n\n\n
    Stack Buffer in scan.go<\/h3>\n\n\n\n
    The unordered scan path now collects filtered record IDs into a stack-allocated buffer before flushing them to the MapReducer:<\/p>\n\n\n\n
    var buf [1024]uint\nbufN := 0\n\n\/\/ ... inside index iteration:\nbuf[bufN] = idx\nbufN++\nif bufN == 1024 {\n    acc = mapper.Stream(acc, buf[:bufN])\n    bufN = 0\n}<\/code><\/pre>\n\n\n\n
    1024 items x 8 bytes = 8 KB \u2014 fits comfortably in L1 cache, requires zero heap allocation. The old code called map+reduce inline for every passing row; the new code batches them, letting the MapReducer process a dense block of IDs with optimal memory access patterns.<\/p>\n\n\n\n
    Batched Merge in scan_order<\/h3>\n\n\n\n
    The ordered scan path merges sorted per-shard results via a global priority queue. The merge pops items one at a time from the heap, accumulating consecutive same-shard items into a buffer. When the next heap pop yields a different shard, the buffer is flushed to the previous shard’s MapReducer:<\/p>\n\n\n\n
    var buf [1024]uint\nbufN := 0\nvar bufShard *shardqueue\n\nfor len(q.q) > 0 {\n    qx := q.q[0]                    \/\/ peek at heap root\n    item := qx.items[0]             \/\/ take its front item\n    qx.items = qx.items[1:]         \/\/ consume it\n\n    if bufShard != nil && bufShard != qx {\n        \/\/ shard changed \u2014 flush buffer to previous shard's mapper\n        acc = bufShard.mapper.Stream(acc, buf[:bufN])\n        bufN = 0\n    }\n    bufShard = qx\n    buf[bufN] = item\n    bufN++\n    if bufN == len(buf) {            \/\/ buffer full \u2014 flush\n        acc = bufShard.mapper.Stream(acc, buf[:bufN])\n        bufN = 0\n    }\n\n    \/\/ re-heapify or remove exhausted shard\n    if len(qx.items) > 0 { heap.Fix(&q, 0) } else { heap.Pop(&q) }\n}\n\/\/ flush remaining\nif bufN > 0 { acc = bufShard.mapper.Stream(acc, buf[:bufN]) }<\/code><\/pre>\n\n\n\n
    This uses the same 8 KB stack buffer as the unordered path. When a single shard dominates a region of the sort order, the buffer fills with consecutive same-shard items and flushes in one Stream()<\/code> call. For cluster mode, each Stream()<\/code> call becomes a network RPC \u2014 batching N consecutive same-shard items into one call reduces round-trips from N to ~N\/batchsize.<\/p>\n\n\n\n
    Benchmark Results<\/h2>\n\n\n\n
    We benchmarked the new code against the previous version on a table with 2 million rows, 4 columns (INT, INT, INT, TEXT). Five runs per query, median reported:<\/p>\n\n\n\n
    Query<\/th>Baseline<\/th>ShardMapReducer<\/th>Speedup<\/th><\/tr><\/thead>
    SELECT COUNT(*)<\/code><\/td>1273 ms<\/td>946 ms<\/td>1.35x (-26%)<\/strong><\/td><\/tr>
    SELECT SUM(value)<\/code><\/td>1394 ms<\/td>1078 ms<\/td>1.29x (-23%)<\/strong><\/td><\/tr>
    ORDER BY value DESC LIMIT 100<\/code><\/td>1607 ms<\/td>1096 ms<\/td>1.47x (-32%)<\/strong><\/td><\/tr>
    WHERE value > 500000<\/code> (filter+count)<\/td>1614 ms<\/td>1034 ms<\/td>1.56x (-36%)<\/strong><\/td><\/tr>
    GROUP BY category % 100<\/code><\/td>\u2014<\/td>\u2014<\/td>\u2014<\/td><\/tr><\/tbody><\/table><\/figure>\n\n\n\n
    The variance also dropped significantly (e.g., SCAN+FILTER: +\/-20 ms vs +\/-278 ms on the baseline), confirming more predictable memory access patterns.<\/p>\n\n\n\n
    Where Does the Speedup Come From?<\/h2>\n\n\n\n
    A 23-36% improvement from what is essentially a refactoring \u2014 no algorithmic changes, same data structures, same query plans \u2014 deserves a closer look. There is no single silver bullet; the speedup is the cumulative effect of removing many small inefficiencies that, at 2 million iterations, add up to hundreds of milliseconds.<\/p>\n\n\n\n
    Eliminated Main\/Delta Branching Per Row<\/h3>\n\n\n\n
    The old code checked for every column of every row whether the record lived in main storage or the delta buffer:<\/p>\n\n\n\n
    \/\/ Old: inside the per-row loop, for EACH column\nif idx < t.main_count {\n    mdataset[i] = mcols[i].GetValue(idx)\n} else {\n    mdataset[i] = t.getDelta(int(idx-t.main_count), col)\n}<\/code><\/pre>\n\n\n\n
    With 2M rows and 4 columns, that is 8 million conditional branches<\/strong>. The branch predictor learns “almost always main”, but each misprediction costs roughly 15 CPU cycles on modern hardware.<\/p>\n\n\n\n
    The new code partitions the batch once<\/strong> via run detection. processMainBlock<\/code> contains no branch<\/strong> \u2014 just a tight loop calling col.GetValue(id)<\/code>. With 99.9% of data in main storage, the entire 2M-row scan typically executes as a single processMainBlock<\/code> call.<\/p>\n\n\n\n
    Phase Separation: Filter vs Map+Reduce<\/h3>\n\n\n\n
    This is likely the largest contributor, especially for filtered scans.<\/p>\n\n\n\n
    The old code ran everything in one interleaved loop:<\/p>\n\n\n\n
    for each row:\n  read condition columns   -> load cache lines for filter columns\n  evaluate filter\n  if passes:\n    read callback columns  -> load cache lines for map columns (evicts filter data!)\n    call map + reduce<\/code><\/pre>\n\n\n\n
    The new code separates these into distinct phases:<\/p>\n\n\n\n
    Phase 1 (filter): for each row:\n  read condition columns only   -> L1 stays hot for filter columns\n  if passes: buf[n++] = idx\n\nPhase 2 (map+reduce): mapper.Stream(buf[:n])\n  read callback columns only    -> L1 now fully dedicated to map columns\n  call map + reduce<\/code><\/pre>\n\n\n\n
    In Phase 1, only the filter columns compete for cache space. In Phase 2, only the map columns are accessed. The old code forced both column sets to share the same cache lines, causing thrashing on every row that passed the filter.<\/p>\n\n\n\n
    Why SCAN+FILTER Benefits Most (1.56x)<\/h3>\n\n\n\n
    WHERE value > 500000<\/code> filters roughly half the rows. The old code still interleaved filter and map column reads for every passing row \u2014 the cache pollution hit it on every match. The new code collects all passing IDs first (Phase 1 touches only the filter column), then processes them in a clean sequential sweep through the map columns (Phase 2). The more selective the filter, the denser and more cache-friendly the resulting ID batches become.<\/p>\n\n\n\n
    Why COUNT Benefits Least (1.35x)<\/h3>\n\n\n\n
    COUNT has no real map phase \u2014 map is essentially identity, reduce is a counter increment. There are no “map columns” to read, so the phase-separation advantage barely applies. The 35% speedup for COUNT comes almost entirely from the eliminated per-row branching and the pre-allocated args buffer. It represents the “floor” of what the structural improvements buy you even without cache-locality gains.<\/p>\n\n\n\n
    Pre-Allocated Args Buffer<\/h3>\n\n\n\n
    ShardMapReducer.args<\/code> is allocated once at construction and reused for every row. The old code reused a mdataset<\/code> slice too, but allocated it fresh per scan call. More importantly, the MapReducer is a longer-lived object that the Go runtime can place more efficiently, reducing GC tracing overhead across millions of iterations.<\/p>\n\n\n\n
    Stack Buffer Eliminates GC Pressure<\/h3>\n\n\n\n
    var buf [1024]uint<\/code> lives on the Go stack \u2014 no heap allocation, no GC tracking. The old code had no equivalent buffer and invoked map+reduce inline per row. With 2M rows, the cumulative GC scheduling overhead was measurable \u2014 the reduced variance (+\/-20 ms vs +\/-278 ms for SCAN+FILTER) is a direct indicator of less GC interference.<\/p>\n\n\n\n
    The Compound Effect<\/h3>\n\n\n\n
    None of these improvements alone would explain 36%. But they multiply:<\/p>\n\n\n\n