feat(sql): support multiple right-hand-side tables in HORIZON JOIN

[puzpuzpuz]

Documentation PR: questdb/documentation#397

Summary

• Extends HORIZON JOIN to accept multiple right-hand-side (slave) tables in a single query, enabling users to aggregate columns from several time-series sources against a common master table and offset grid in one statement
• Adds both single-threaded and parallel (page-frame-based) execution paths for multi-slave HORIZON JOIN, supporting keyed (ON symbol) and non-keyed (timestamp-only ASOF) variants, including mixed keyed/non-keyed slaves within the same query
• Refactors existing single-slave HORIZON JOIN internals: replaces LongList offsets with long[] to avoid on-heap allocations on the hot path, simplifies BaseAsyncHorizonJoinAtom constructor by removing redundant @Transient parameters already baked into the compiled RecordSink classes, and replaces the always-forward scan strategy in processHorizonTimestamps with an adaptive backward/forward approach controlled by configurable thresholds (cairo.sql.horizon.join.bwd.scan.* properties) — the new heuristic starts with backward-only scans and switches to forward scans when the backward scan cost exceeds a gap-relative or absolute threshold

SQL syntax

The last HORIZON JOIN in the chain carries the RANGE/LIST and AS clauses; preceding HORIZON JOINs omit them:

SELECT avg(b.bid) AS avg_bid, avg(a.ask) AS avg_ask
FROM trades AS t
HORIZON JOIN bids AS b ON (t.sym = b.sym)
HORIZON JOIN asks AS a ON (t.sym = a.sym)
    LIST (-2s, 0, 2s) AS h
GROUP BY h.offset

TODOs

• async factories
• tests: SQL parsing and validation, data correctness, fuzz, parallel
• compare Multi* factory performance vs. single-RHS table HORIZON JOIN
• document the feature

[coderabbitai]

Important

Review skipped

Auto reviews are disabled on this repository. Please check the settings in the CodeRabbit UI or the .coderabbit.yaml file in this repository. To trigger a single review, invoke the @coderabbitai review command.

⚙️ Run configuration

Configuration used: Path: .coderabbit.yaml

Review profile: CHILL

Plan: Pro

Run ID: 00f22146-bdf2-40e7-a32d-cb2896fd5c6d

You can disable this status message by setting the reviews.review_status to false in the CodeRabbit configuration file.

Use the checkbox below for a quick retry:

• 🔍 Trigger review

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch puzpuzpuz_multi_horizon_join

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[bluestreak01]

@puzpuzpuz

PR Review

Critical

1. perWorkerFilters always passed as null to all async factory constructors — resource leak + thread-safety bug

Files: SqlCodeGenerator.java lines 3880/3912/3951 (single-slave) and 6127/6153, 6191/6222 (multi-slave)

Commit c14dc63f53 ("Null perWorkerFilters on ownership transfer") added perWorkerFilters = null for catch-block safety but forgot to save the value to a local variable first. Every other transferred resource follows the correct foo0 = foo; foo = null; pass foo0 pattern. perWorkerFilters does not.

Code path trace (multi-slave, not-keyed):

line 6100:  perWorkerFilters = compileWorkerFiltersConditionally(...)  // non-null list
line 6127:  perWorkerFilters = null;                                  // nulled
line 6153:  ..., perWorkerFilters, ...                                // passes null!

Compare to the correct pattern for filter on the same path:

line 6122:  final Function filter0 = filter;   // saved
line 6126:  filter = null;                     // nulled
line 6151:  ..., filter0, ...                  // correct value passed

On master, perWorkerFilters was a final local (line 3733) that was never nulled and was always passed correctly. This is a regression.

Impact:

1. Resource leak: Per-worker filter Function objects are allocated by compileWorkerFiltersConditionally() but never freed or passed to anyone.
2. Thread-safety violation: In AsyncFilterContext.getFilter(slotId), when perWorkerFilters is null, all workers share the single ownerFilter. Thread-unsafe filters like concat() (which uses a mutable StringSink) will produce corrupted results or crashes under concurrent access. The test testParallelMultiHorizonJoinFilteredThreadUnsafe exercises this case but may pass non-deterministically.
3. Affects 4 call sites: both keyed and not-keyed constructors for both single-slave and multi-slave async factories.

Fix: Add final ObjList<Function> perWorkerFilters0 = perWorkerFilters; before each null assignment, then pass perWorkerFilters0 to the constructor. Apply at all 3 locations where perWorkerFilters = null was added (the single-slave path at 3880 feeds two constructors at 3912 and 3951).

---

Moderate

2. Adaptive scan heuristic changes existing single-slave behavior without mention in PR description

The processHorizonTimestamps method in AsyncHorizonJoinNotKeyedRecordCursorFactory (and the keyed variant) was rewritten with a new adaptive backward/forward scan strategy. The old approach always forward-scanned between consecutive positions. The new approach starts in backward-only mode and switches to forward scan heuristically based on cost thresholds (bwdScanAbsoluteThreshold, bwdScanMinGap, bwdScanSwitchFactor).

This changes runtime behavior for existing single-slave HORIZON JOIN queries, not just the new multi-slave path. The PR description mentions "Refactors existing single-slave HORIZON JOIN internals" for the LongList → long[] change but does not call out this algorithmic change. The performance comparison TODO is checked off, but results are not shared.

Risk: The new heuristic could regress workloads with many distinct keys and small gaps, where the old always-forward approach was cheaper. The configurable thresholds mitigate this, but the default values need benchmarking.

3. No deterministic test coverage for both parallel and sequential multi-horizon paths

HorizonJoinTest randomizes parallelHorizonJoinEnabled in the constructor, so any single CI run covers only one of the two code paths through generateMultiHorizonJoinFactory. The ParallelHorizonJoinFuzzTest forces the parallel path explicitly, but the sequential path for multi-horizon relies on randomization (or on specific tests that disable page frames, like testMultiHorizonJoinLimitOnMaster).

4. No parser test for RANGE/LIST appearing on a non-last HORIZON JOIN

The parser now makes RANGE/LIST optional on non-last HORIZON JOINs. But what if a user writes RANGE/LIST on the first of two HORIZON JOINs? The parser would consume it and attach it to the first join model. Then findHorizonOffsetContext would find the context on a non-last join, potentially producing incorrect behavior. There should be a test that verifies this case produces a clear error or correct behavior.

---

Minor

5. Commit messages "More tests" (x2) are too vague

Commits 6121cd470d and a283dbbda5 both titled "More tests" don't describe what they test.

6. freeObjListAndClear vs freeObjList inconsistency

MultiHorizonJoinNotKeyedRecordCursorFactory._close() (line 178) uses freeObjListAndClear(groupByFunctions) while AsyncMultiHorizonJoinNotKeyedRecordCursorFactory._close() (line 513) uses freeObjList(groupByFunctions). Worth aligning.

---

Downgraded (false positives)

Original claim	Why dismissed
Ambiguous column name resolution in buildMultiHorizonColumnMappings no-alias fallback silently picks first slave	The no-alias path is a defensive fallback; JoinRecordMetadata stores alias-prefixed names, so the alias path handles real queries.
closeSlaveOnFailure not restored after pending horizon slave accumulation	Line 5059 resets closeSlaveOnFailure = true at top of each iteration. No bug.
MultiHorizonJoinSymbolTableSource mutable masterSource field not volatile	of() called before dispatchAndAwait() creates happens-before edge. Safe.
Bare timestamp/offset names mapped to SOURCE_SEQUENCE could shadow real columns	JoinRecordMetadata stores alias-prefixed names. Not a real risk.
Single HORIZON JOIN without RANGE/LIST now parsed successfully (deferred error)	Post-loop check at line 5286-5292 catches this and produces a clear error.

---

Summary

Verdict: Request changes — the perWorkerFilters null bug (Finding 1) is a confirmed regression that causes resource leaks and thread-safety violations on the parallel execution path with stolen filters. It must be fixed before merge.

12 draft findings verified, 5 downgraded as false positives, 7 confirmed findings reported.

[puzpuzpuz]

@bluestreak01 all items but 3 and 5 are addressed in the following commits: ed88dcb, 802ef32, 76296af

[glasstiger]

[PR Coverage check]

😍 pass : 1345 / 1965 (68.45%)

file detail

	path	covered line	new line	coverage
🔵	io/questdb/griffin/SqlCodeGenerator.java	0	354	00.00%
🔵	io/questdb/griffin/engine/table/MultiHorizonJoinRecord.java	37	115	32.17%
🔵	io/questdb/griffin/engine/table/MultiHorizonJoinSymbolTableSource.java	19	29	65.52%
🔵	io/questdb/griffin/engine/table/AsyncHorizonJoinNotKeyedRecordCursorFactory.java	2	3	66.67%
🔵	io/questdb/griffin/engine/table/AsyncMultiHorizonJoinNotKeyedAtom.java	24	30	80.00%
🔵	io/questdb/griffin/engine/table/AsyncMultiHorizonJoinRecordCursorFactory.java	159	194	81.96%
🔵	io/questdb/griffin/engine/table/AsyncMultiHorizonJoinRecordCursor.java	120	146	82.19%
🔵	io/questdb/griffin/engine/table/MultiHorizonJoinRecordCursorFactory.java	175	205	85.37%
🔵	io/questdb/griffin/engine/table/AsyncMultiHorizonJoinNotKeyedRecordCursor.java	114	130	87.69%
🔵	io/questdb/griffin/engine/table/MultiHorizonJoinNotKeyedRecordCursorFactory.java	166	185	89.73%
🔵	io/questdb/griffin/engine/table/BaseAsyncMultiHorizonJoinAtom.java	222	248	89.52%
🔵	io/questdb/griffin/engine/table/AsyncMultiHorizonJoinAtom.java	58	64	90.62%
🔵	io/questdb/griffin/engine/table/AsyncMultiHorizonJoinNotKeyedRecordCursorFactory.java	151	164	92.07%
🔵	io/questdb/griffin/SqlParser.java	6	6	100.00%
🔵	io/questdb/griffin/engine/table/AsyncHorizonJoinAtom.java	8	8	100.00%
🔵	io/questdb/griffin/engine/table/HorizonTimestampIterator.java	6	6	100.00%
🔵	io/questdb/griffin/engine/table/HorizonJoinNotKeyedRecordCursorFactory.java	9	9	100.00%
🔵	io/questdb/griffin/engine/table/HorizonJoinSlaveState.java	20	20	100.00%
🔵	io/questdb/griffin/engine/table/AsyncGroupByNotKeyedRecordCursorFactory.java	1	1	100.00%
🔵	io/questdb/griffin/engine/table/HorizonJoinTimeFrameHelper.java	26	26	100.00%
🔵	io/questdb/griffin/engine/table/AsyncHorizonJoinRecordCursorFactory.java	2	2	100.00%
🔵	io/questdb/griffin/engine/table/AsyncHorizonTimestampIterator.java	7	7	100.00%
🔵	io/questdb/griffin/engine/table/BaseAsyncHorizonJoinAtom.java	8	8	100.00%
🔵	io/questdb/griffin/engine/table/HorizonJoinRecordCursorFactory.java	5	5	100.00%