Improvement: Rewrite hash join hash table for better scalability

[sim1984]

Replace hash join table with a flat bucket-array structure

Summary

The old HashJoin implementation used a fixed-size hash table of 1009 buckets, with a per-bucket CollisionList object holding a SortedArray of {hash, position} entries. On inner streams beyond a few hundred thousand records, average chain lengths reached thousands and lookups degenerated into scattered binary searches over heap-allocated buffers. Attempts to make the table grow dynamically did not help: cardinality estimates for table functions and stored procedures use a fixed default of 1000, so rehashing either never fired or fired when it was already too late, and each rehash added allocation and copying overhead on top of the lookup cost.

This change replaces the table with a flat per-stream bucket array. During the build phase, put() only appends raw {hash, position} pairs into a single contiguous Array<Entry>. Once all inner streams have been read, build() picks the final table size from the exact number of collected entries and performs a counting sort over the entries array: one pass to count bucket sizes, one prefix-sum pass to compute offsets, and one redistribution pass into a temporary buffer. After build(), records sharing a bucket occupy a contiguous range in a single flat array, sorted by hash. Lookup is a binary search inside that range.

No CollisionList objects, no per-bucket allocations, no rehashing at any point.

What changed

• HashJoin::HashTable is rewritten around a Stream inner class holding entries, bucketStart, and an iterator cursor.
• put() is a plain append; bucket assignment is deferred until build().
• build() chooses the table size from the actual number of entries (HASH_SIZES up to 1000003) and performs the counting sort described above.
• bucketStart has tableSize + 1 elements so that a bucket can be expressed as a half-open range without a special case for the last one.
• setup() checks that every stream has a non-empty bucket for the target hash, then primes each stream's cursor via reset().
• HashJoin::maxCapacity() now returns MAX_HASH_SIZE * HASH_LOAD_FACTOR, which corresponds to the actual capacity of the flat layout.

The lookup API (setup / reset / iterate) and its usage from fetchRecord() are unchanged, so no changes were needed in the rest of HashJoin or in RecordSource.

Test results

A test suite covering pure GENERATE_SERIES joins, integer keys with uniform / skewed / low-cardinality / NULL distributions, string keys, composite keys, and indexed variants is attached (hash-join-tests.sql). Both builds were run on the same machine, with the same database configuration (DefaultDbCachePages = 32K, TempCacheLimit = 1G) and freshly created data. All COUNT(*) results are identical between the two builds and match the expected values.

Elapsed times, in seconds:

Test	Old (1009)	New (flat)	Δ
GS 1M × 1k	0.083	0.075	−10%
GS 1M × 100k	0.123	0.101	−18%
GS 1M × 1M	0.448	0.326	−27%
GS 1M × 10M	2.799	1.690	−40%
GS 10M × 10M	8.728	4.875	−44%
GS mod-join (duplicates)	0.559	0.508	−9%
GS 3-way chain	0.088	0.077	−12%
1k × 1k	0.000	0.001	~0
100k × 100k	0.054	0.049	−9%
1M × 1M	0.627	0.506	−19%
1M × 1k	0.159	0.154	−3%
1M × 100k	0.222	0.191	−14%
Skewed (10k-row hot key)	0.225	0.191	−15%
Low cardinality (100 ids)	0.186	0.166	−11%
NULL in join key	0.049	0.045	−8%
Composite (int, int)	0.021	0.021	~0
Two hashed streams	0.161	0.153	−5%
SEMI (EXISTS)	0.016	0.016	~0
VARCHAR(36) join	0.083	0.073	−12%
VARCHAR(36) SEMI	0.079	0.070	−11%
(int, varchar) key	0.090	0.079	−12%
Indexed 100k	0.054	0.050	−7%
Indexed 1M	0.649	0.507	−22%
Indexed VARCHAR	0.078	0.072	−8%

Summary: parity on inputs below ~10k records, 9–18% faster on 100k, 19–27% faster on 1M, 40–44% faster on 10M. No regressions in any test, including the duplicate-heavy and skewed cases that used to give the old layout trouble.

Query plans are also unchanged between the two builds, including the indexed variants where the optimizer may pick an index-based plan instead of a hash join.

Memory

Structural memory usage is identical between the two builds — Delta memory in the test output matches to the byte for every test. The only difference is the transient peak during build(): the flat layout allocates a temporary buffer of the same size as the entries array for the redistribution pass, so Max memory on the 10M test grows from ~749 MB to ~839 MB. The temporary buffer is released immediately after build() returns; the additional allocation does not persist across queries.

For deployments where a single query runs multiple hash joins concurrently, this peak is worth keeping in mind. A possible future improvement is an in-place redistribution using cycle-following and a bitmask of visited positions, which would eliminate the temporary buffer at the cost of more random memory access during build.

hash-join-tests.sql

[aafemt]

IMHO, there is no need to round up to the next prime. The main purpose of prime's usage is to flatten distribution of keys among array. Hash functions already aimed to even distribution, so usage of primes hardly can improve situation.