How we made WINDOW JOIN parallel and vectorized

QuestDB is the open-source time-series database for demanding workloads—from trading floors to mission control. It delivers ultra-low latency, high ingestion throughput, and a multi-tier storage engine. Native support for Parquet and SQL keeps your data portable, AI-ready—no vendor lock-in.

Consider a workload that comes up constantly on a trading desk: for every executed trade, attach the average bid and ask within a 1-second window around the trade. Without a dedicated operator it takes two joins, an ASOF JOIN for the carry-forward quote at the window start plus a range join for the rows inside the window, stitched with UNION ALL and folded with a GROUP BY: -- QuestDB timestamps are microseconds, so 1_000_000 is 1 second.

The same query, dedicated operator: SELECT t.*, avg(p.bid) avg_bid, avg(p.ask) avg_askFROM trades tWINDOW JOIN prices p ON p.sym = t.symbol RANGE BETWEEN 1 second PRECEDING AND 1 second FOLLOWING; Now the operator knows what it is doing: for every row on the left-hand side of the join (LHS - trades here), find rows on the right-hand side (RHS - prices) whose timestamp falls inside a [lo, hi] window around the LHS timestamp, restrict to matching symbol keys, and reduce them with a batch of aggregate functions. Making that fast comes down to two pieces: data-level parallelism over the LHS, plus a low-cardinality fast path that copies values into contiguous buffers so the SIMD aggregation kernels we already ship for SAMPLE BY run on window slices unchanged. Benchmarked against Timescale, DuckDB, and ClickHouse on a 50M-row trades table joined to a 150M-row prices table, the parallel + SIMD path runs 5.0x faster than QuestDB's own single-threaded fallback and 25x faster than ClickHouse's best rewrite. Data-level parallelism QuestDB stores data in append-only column files, partitioned by time. The query engine reads them as a sequence of page frames: contiguous, columnar slabs of memory that map directly onto file pages. Filtering and aggregation both work at this granularity: a page frame is the unit of dispatch to a worker thread. WINDOW JOIN follows the same model. The LHS table is sliced into page frames; each worker takes a frame and is responsible for producing the aggregate result for every LHS row in that frame. To do that it needs the RHS rows that fall inside the union of all windows the frame covers. Concretely, for a frame whose LHS timestamps run from tLo to tHi with a [-w_lo, +w_hi] window, the worker needs RHS rows in [tLo - w_lo, tHi + w_hi].

Locating that slice cheaply is what makes the parallel plan viable, and the enabler is QuestDB's storage layout: rows in both tables are kept in designated timestamp order on disk, so the RHS slice for any time range collapses to a single binary search per worker rather than a scan per LHS row. Then, for the join keys present in the LHS frame, the worker builds a small in-memory index from the RHS slice: a per-key list of RHS timestamps, plus per-key arrays of the values to aggregate. Once that index is built, the inner loop over LHS rows is just two binary searches per row, one for the window's low bound and one for its high, followed by an aggregate over the resulting contiguous range. Both binary searches walk forward monotonically, so they amortize across rows in the same frame. Roughly: LHS page frames ┌─────┬─────┬─────┬─────┬─────┐ │ F0 │ F1 │ F2 │ F3 │ ... │ └──┬──┴──┬──┴──┬──┴──┬──┴─────┘ │ │ │ │ ┌──────┴┐ ┌──┴───┐ │ │ │worker0│ │worker1│ ... │ workers pulled from a shared pool └───┬───┘ └──┬────┘ │ one frame at a time │ │ │ ┌───────┴───────┐│ ┌───────┴───────┐ │ RHS slice + ││ │ RHS slice + │

│ per-key ││ │ per-key │ │ value arrays ││ │ value arrays │ └───────┬───────┘│ └───────┬───────┘ │ │ │ ▼ ▼ ▼ ┌───────────────────────────────────┐ │ for each LHS row in frame: │ │ bsearch lo, bsearch hi, │ │ aggregate(value_array[lo:hi]) │ └───────────────────────────────────┘ The reduce step is per-frame and lock-free; workers do not share any mutable state across frames. That covers the outer parallelism. The inner loop, where each LHS row aggregates its RHS slice, is where SIMD comes in. The low-cardinality-key fast path For low-cardinality equality joins paired with vectorizable aggregates, we copy the RHS values into per-key contiguous buffers so the SIMD kernels we already ship for SAMPLE BY can run on the window slice unchanged. The SIMD kernels are hand-tuned sum, avg, min, max, and a few others that munch eight doubles per loop iteration, but only on contiguous arrays. The RHS rows in a window are not contiguous; they are scattered across page frames, threaded through mixed columns. Our first cut just walked the matching rows one at a time and called computeBatch per row. Correct, but it left those kernels on the table. The query shape that makes the extra pass worth it is the most common one in practice: a low-cardinality equality join (the canonical p.sym = t.symbol case) and an aggregated numeric column. With few distinct keys, the per-key value buffers fit in cache, and the cost of materializing them is more than recouped by handing each window slice to a vectorized aggregate. The fast path is gated on two conditions: the query has to use aggregates we have vectorized implementations for (the sum / avg / min / max / count family over numeric types), and the join condition has to fit the low-cardinality shape that the planner recognizes (today, single-symbol equi-joins).

When both hold, the worker iterates its RHS slice once to build the per-key timestamp index and copy the aggregated columns into per-key value buffers in the same pass. The buffers are typed and packed: all bid values for symbol AAPL land in one contiguous block of doubles, in RHS timestamp order. For queries that don't qualify, we skip the buffer copy entirely and fall back to the scalar computeBatch loop. Once the inner loop reaches an LHS row, the per-key value buffer is already laid out the way SIMD wants it. The two binary searches give us [rowLo, rowHi) into the buffer, and we hand the slice straight to the vectorized aggregate: for (int i = 0; i < groupByFuncCount; i++) { groupByFunctions.getQuick(i).computeBatch( value, // running aggregate state bufferStart(i) + typeSize * rowLo, // pointer into the per-key buffer (int) (rowHi - rowLo), // slice length 0 );} The extra pass over the RHS slice is not free; it reads each RHS column twice (once to put the values into the per-key buffer, once for the aggregate to consume them). For a generic plan that would be worse. But:

The aggregate loop in QuestDB is SIMD-bound on the input if the input is contiguous, and otherwise call-bound on computeBatch per row. On doubles, AVX2 buys us about an order of magnitude here. The RHS columns are read sequentially with prefetching, which is the case the memory subsystem is happiest with. The per-key value buffers are sized to the RHS slice for the current LHS frame, so the working set stays in L2/L3.

It took some staring at flame graphs to convince ourselves the second pass was the right call. Once we did, the fast path beat the scalar path even on small frames, and the gap widened as the LHS frame got bigger, because more LHS rows shared the same indexed buffer. The general-predicate path still exists for queries that do not qualify for the fast path. It runs the scalar computeBatch loop without the buffer-copy step.

EXPLAIN shows you which path your query is on: Async Window Fast Join workers: 23 vectorized: true symbol: symbol=sym window lo: 1000000 preceding (include prevailing) window hi: 1000000 following PageFrame Row forward scan Frame forward scan on: trades PageFrame Row forward scan Frame forward scan on: prices The fast variant is Async Window Fast Join (parallel) or Window Fast Join (serial), with vectorized: true when SIMD kernels are engaged for your aggregates. Adding a non-vectorizable aggregate flips vectorized: false on the same operator; switching the join key off a symbol column falls back to Async Window Join entirely. The SIMD (AVX2) hot loop Once the slice is contiguous, the aggregation is the same C++ kernel that SAMPLE BY uses. Take the canonical query the operator was designed for: SELECT t.timestamp, avg(p.bid)FROM trades tWINDOW JOIN prices p ON p.sym = t.symbol RANGE BETWEEN 1 second PRECEDING AND 1 second FOLLOWING; Once the per-key value buffer is built, the per-LHS-row work is just avg(double) over a contiguous slice of it. avg decomposes into a sum plus a count of non-null values (avg = sum / count, and nulls in QuestDB are encoded as NaN), so what actually runs is a fused sum + NaN-count kernel called sumDoubleAcc. The hot loop processes eight doubles per iteration on AVX2: ; sumDoubleAcc, AVX2 hot loop body (rax = &d[i+8], rcx = i).Lloop: vmovupd ymm3, [rax-0xfa0] ; load d[i+4..i+7] vcmpunordpd ymm1, ymm3, ymm3 ; NaN mask, lane 1 vmovupd ymm2, [rax-0xfc0] ; load d[i..i+3] vcmpunordpd