diff --git a/src/jrd/recsrc/HashJoin.cpp b/src/jrd/recsrc/HashJoin.cpp index 8b416c15742..46ee8a8682e 100644 --- a/src/jrd/recsrc/HashJoin.cpp +++ b/src/jrd/recsrc/HashJoin.cpp @@ -32,6 +32,9 @@ #include "../jrd/intl_proto.h" #include "../jrd/optimizer/Optimizer.h" +#include +#include + #include "RecordSource.h" using namespace Firebird; @@ -43,210 +46,330 @@ using namespace Jrd; // Data access: hash join // ---------------------- -// NS: FIXME - Why use static hash table here??? Hash table shall support dynamic resizing -static constexpr ULONG HASH_SIZE = 1009; -static constexpr ULONG BUCKET_PREALLOCATE_SIZE = 32; // 256 bytes per bucket +namespace +{ + // Predefined table sizes (all prime). Table sizing is done once, in build(), + // based on the actual number of collected entries. + constexpr ULONG HASH_SIZES[] = + { + 1009, 2003, 4001, 8009, 16001, 32003, + 64007, 128021, 256019, 512009, 1000003 + }; + + constexpr ULONG MIN_HASH_SIZE = HASH_SIZES[0]; + constexpr ULONG MAX_HASH_SIZE = HASH_SIZES[std::size(HASH_SIZES) - 1]; + + // Desired average number of entries per bucket. We sort each bucket + // by hash and use binary search on lookup, so even a moderately long + // bucket is cheap. A small factor here means more buckets and a larger + // bucketStart array; a large factor means longer buckets. + constexpr ULONG HASH_LOAD_FACTOR = 4; + + constexpr ULONG nextHashSize(ULONG current) + { + for (const ULONG size : HASH_SIZES) + { + if (size > current) + return size; + } + return MAX_HASH_SIZE; + } + +} unsigned HashJoin::maxCapacity() noexcept { - // Binary search across 1000 collisions is computationally similar to - // linear search across 10 collisions. We use this number as a rough - // estimation of whether the lookup performance is likely to be acceptable. - return HASH_SIZE * 1000; + // Upper bound on the number of records the optimizer may reasonably + // feed into a single hash join. Above this the bucket scan starts + // to dominate and the optimizer should pick a different join method. + return MAX_HASH_SIZE * HASH_LOAD_FACTOR; } class HashJoin::HashTable final : public PermanentStorage { - class CollisionList + // A single {hash, position} pair. The position refers to a record + // inside the corresponding BufferedStream. + struct Entry { - static constexpr FB_SIZE_T INVALID_ITERATOR = FB_SIZE_T(~0); - - struct Entry - { - Entry() noexcept - : hash(0), position(0) - {} - - Entry(ULONG h, ULONG pos) noexcept - : hash(h), position(pos) - {} - - static const ULONG generate(const Entry& item) noexcept - { - return item.hash; - } - - ULONG hash; - ULONG position; - }; + ULONG hash; + ULONG position; + }; - public: - CollisionList(MemoryPool& pool) - : m_collisions(pool, BUCKET_PREALLOCATE_SIZE), - m_iterator(INVALID_ITERATOR) + // Per-stream state and the operations on it. Entries are collected + // unsorted in put(), then rearranged in build() so that records + // sharing the same bucket occupy a contiguous range and are sorted + // by hash within it. After build() every lookup is a binary search + // inside the bucket's range. + struct Stream : public PermanentStorage + { + Array entries; // all records, grouped by bucket after build() + Array bucketStart; // size = tableSize + 1; see build() for layout + ULONG iterator; // current cursor for iterate() + + explicit Stream(MemoryPool& pool) + : PermanentStorage(pool) + , entries(pool) + , bucketStart(pool) + , iterator(0) + {} + + // Collect one raw entry. No bucket logic here, so this can be + // called freely while reading the inner stream. + void put(ULONG hash, ULONG position) { - m_collisions.setSortMode(FB_ARRAY_SORT_MANUAL); + entries.add({ hash, position }); } - void sort() + // Group entries by bucket and sort each bucket by hash. + // + // The table size is chosen by HashTable::build() from the total + // number of entries across all streams, not from any cardinality + // estimate, so this runs exactly once and never re-runs. + // + // Each bucket is expressed as a half-open range + // [bucketStart[b], bucketStart[b + 1]). The extra (tableSize + 1)-th + // element is needed so that the range of the last bucket can be + // written uniformly, without a special case. + void build(ULONG tableSize); + + bool bucketEmpty(ULONG slot) const noexcept { - m_collisions.sort(); + return bucketStart[slot] == bucketStart[slot + 1]; } - ULONG getCount() const noexcept - { - return (ULONG) m_collisions.getCount(); - } - void add(ULONG hash, ULONG position) + // Position the cursor at the first entry with hash >= the target. + // If the target is absent, the cursor ends up at the first entry + // greater than it, and iterate() immediately returns false. + void reset(ULONG slot, ULONG hash) { - m_collisions.add(Entry(hash, position)); - } + ULONG lo = bucketStart[slot]; + ULONG hi = bucketStart[slot + 1]; - bool locate(ULONG hash) - { - if (m_collisions.find(hash, m_iterator)) - return true; + while (lo < hi) + { + const ULONG mid = lo + (hi - lo) / 2; + if (entries[mid].hash < hash) + lo = mid + 1; + else + hi = mid; + } - m_iterator = INVALID_ITERATOR; - return false; + iterator = lo; } - bool iterate(ULONG hash, ULONG& position) noexcept + bool iterate(ULONG slot, ULONG hash, ULONG& position) noexcept { - if (m_iterator >= m_collisions.getCount()) - return false; - - const Entry& collision = m_collisions[m_iterator++]; + const ULONG end = bucketStart[slot + 1]; - if (hash != collision.hash) - { - m_iterator = INVALID_ITERATOR; + if (iterator >= end || entries[iterator].hash != hash) return false; - } - position = collision.position; + position = entries[iterator].position; + iterator++; return true; } - - private: - SortedArray, ULONG, Entry> m_collisions; - FB_SIZE_T m_iterator; }; public: - HashTable(MemoryPool& pool, ULONG streamCount, ULONG tableSize = HASH_SIZE) - : PermanentStorage(pool), m_streamCount(streamCount), - m_tableSize(tableSize), m_slot(0) + HashTable(MemoryPool& pool, ULONG streamCount) + : PermanentStorage(pool), m_streams(pool, streamCount) { - m_collisions = FB_NEW_POOL(pool) CollisionList*[streamCount * tableSize]; - memset(m_collisions, 0, streamCount * tableSize * sizeof(CollisionList*)); + for (ULONG i = 0; i < streamCount; i++) + m_streams.add(FB_NEW_POOL(pool) Stream(pool)); } ~HashTable() { - for (ULONG i = 0; i < m_streamCount * m_tableSize; i++) - delete m_collisions[i]; - - delete[] m_collisions; + for (auto* s : m_streams) + delete s; } + // Phase 1: collect raw (hash, position) pairs into the appropriate + // stream. Called by HashJoin::internalGetRecord() while the inner + // stream is being read. void put(ULONG stream, ULONG hash, ULONG position) { - const ULONG slot = hash % m_tableSize; - - fb_assert(stream < m_streamCount); - fb_assert(slot < m_tableSize); - - CollisionList* collisions = m_collisions[stream * m_tableSize + slot]; - - if (!collisions) - { - collisions = FB_NEW_POOL(getPool()) CollisionList(getPool()); - m_collisions[stream * m_tableSize + slot] = collisions; - } + fb_assert(stream < m_streams.getCount()); - collisions->add(hash, position); + m_streams[stream]->put(hash, position); } - bool setup(ULONG hash) - { - const ULONG slot = hash % m_tableSize; - - for (ULONG i = 0; i < m_streamCount; i++) - { - CollisionList* const collisions = m_collisions[i * m_tableSize + slot]; - - if (!collisions) - return false; + // Phase 2: once all inner streams are fully read, group their entries + // by bucket. Called exactly once, never re-run. + void build(); - if (!collisions->locate(hash)) - return false; - } + // Lookup API. The typical usage pattern is: + // + // if (table->setup(hash)) + // while (table->iterate(stream, hash, position)) { ... } + // + // setup() primes all streams to the first matching record in the + // current bucket, iterate() advances the cursor of a single stream. + // If iterate() advances past the end of the bucket, the caller is + // expected to advance the previous stream (see HashJoin::fetchRecord) + // and call reset() for the current stream to restart the scan of + // matching records. - m_slot = slot; - return true; - } + bool setup(ULONG hash); void reset(ULONG stream, ULONG hash) { - fb_assert(stream < m_streamCount); + fb_assert(stream < m_streams.getCount()); - CollisionList* const collisions = m_collisions[stream * m_tableSize + m_slot]; - collisions->locate(hash); + m_streams[stream]->reset(m_slot, hash); } bool iterate(ULONG stream, ULONG hash, ULONG& position) noexcept { - fb_assert(stream < m_streamCount); + fb_assert(stream < m_streams.getCount()); - CollisionList* const collisions = m_collisions[stream * m_tableSize + m_slot]; - return collisions->iterate(hash, position); + return m_streams[stream]->iterate(m_slot, hash, position); } - void sort() +private: + Array m_streams; + ULONG m_tableSize = 0; + ULONG m_slot = 0; +}; + +void HashJoin::HashTable::Stream::build(ULONG tableSize) +{ + const ULONG n = entries.getCount(); + + bucketStart.grow(tableSize + 1); + memset(bucketStart.begin(), 0, (tableSize + 1) * sizeof(ULONG)); + + // Pass 1: count how many entries fall into each bucket. + // bucketStart[b + 1] is used as a per-bucket counter, so after + // this pass bucketStart[b + 1] holds the size of bucket b. + // The +1 offset lets us compute prefix sums below without + // touching bucketStart[0], which stays zero. + for (ULONG k = 0; k < n; k++) + bucketStart[entries[k].hash % tableSize + 1]++; + + // Pass 2: turn per-bucket counts into start offsets. + // After this pass bucketStart[b] is the index of the first + // entry of bucket b, and bucketStart[b + 1] is one past the last. + for (ULONG b = 0; b < tableSize; b++) + bucketStart[b + 1] += bucketStart[b]; + + // Pass 3: physically move entries into their buckets. We cannot + // write into entries while still reading from it without + // corrupting data, so we allocate a temporary buffer and use a + // separate cursor array 'next' that starts as a copy of bucketStart + // and is incremented as entries are placed. + // + // This is the only place where O(n) temporary memory is needed. + // Peak footprint during build is roughly twice the size of the + // entries array; the temporary buffer is released right after. + if (n > 0) { - for (ULONG i = 0; i < m_streamCount * m_tableSize; i++) + Array temp(getPool(), n); + temp.grow(n); + + Array next(getPool(), tableSize); + next.grow(tableSize); + memcpy(next.begin(), bucketStart.begin(), tableSize * sizeof(ULONG)); + + for (ULONG k = 0; k < n; k++) { - if (const auto collisions = m_collisions[i]) - collisions->sort(); + const Entry& e = entries[k]; + temp[next[e.hash % tableSize]++] = e; } + entries.assign(temp); + } + + // Pass 4: sort each bucket by hash. Singletons are skipped. + // Records with equal hashes may end up in any order relative to + // each other, which is fine for the join: they all compare equal + // on the join key anyway. + auto compareHash = [](const Entry& a, const Entry& b) { return a.hash < b.hash; }; + + for (ULONG b = 0; b < tableSize; b++) + { + const ULONG start = bucketStart[b]; + const ULONG end = bucketStart[b + 1]; + if (end > start + 1) + std::sort(entries.begin() + start, entries.begin() + end, compareHash); + } + #ifdef PRINT_HASH_TABLE + // Per-stream bucket statistics. Useful when tuning HASH_LOAD_FACTOR + // or debugging hash distribution for a particular key type. + { FB_UINT64 total = 0; - ULONG min = MAX_ULONG, max = 0, count = 0; + ULONG min = MAX_ULONG, max = 0, occupied = 0; - for (ULONG i = 0; i < m_streamCount * m_tableSize; i++) + for (ULONG b = 0; b < tableSize; b++) { - CollisionList* const collisions = m_collisions[i]; - if (!collisions) + const ULONG cnt = bucketStart[b + 1] - bucketStart[b]; + if (cnt == 0) continue; - const auto cnt = collisions->getCount(); - if (cnt < min) min = cnt; if (cnt > max) max = cnt; total += cnt; - count++; + occupied++; } - if (count) + if (occupied) { - printf("Hash table size %u, count %u, buckets %u, min %u, max %u, avg %u\n", - m_tableSize, (ULONG) total, count, min, max, (ULONG) (total / count)); + printf("Hash table size %u, entries %u, buckets %u, min %u, max %u, avg %u\n", + tableSize, (ULONG)total, occupied, min, max, + (ULONG)(total / occupied)); } + } #endif +} + +void HashJoin::HashTable::build() +{ + // The table size is chosen from the *actual* number of collected + // entries, not from any cardinality estimate. Estimates for table + // functions and stored procedures use a fixed default and may be + // off by orders of magnitude; getting the size wrong is exactly + // what a dynamic resizing scheme would be paying for later. + // Since all inner streams are fully materialized by the time + // build() runs, we can afford to decide once and for all. + ULONG total = 0; + for (auto* s : m_streams) + total += s->entries.getCount(); + + const ULONG desired = total / (m_streams.getCount() * HASH_LOAD_FACTOR); + m_tableSize = nextHashSize(desired); + + for (auto* s : m_streams) + s->build(m_tableSize); +} + +bool HashJoin::HashTable::setup(ULONG hash) +{ + const ULONG slot = hash % m_tableSize; + + // Every stream must have at least one entry in this bucket, otherwise + // the join cannot produce output for this leader row. This is a cheap + // O(1) check on bucketStart and lets us skip binary searches for the + // common case of a leader row with no counterpart. + for (auto* s : m_streams) + { + if (s->bucketEmpty(slot)) + return false; } -private: - const ULONG m_streamCount; - const ULONG m_tableSize; - CollisionList** m_collisions; - ULONG m_slot; -}; + m_slot = slot; + + // Prime the cursor of each stream to the first entry with the target + // hash (or to its insertion point if the target is absent). + for (ULONG i = 0; i < m_streams.getCount(); i++) + reset(i, hash); + + return true; +} HashJoin::HashJoin(thread_db* tdbb, CompilerScratch* csb, JoinType joinType, @@ -456,7 +579,7 @@ bool HashJoin::internalGetRecord(thread_db* tdbb) const } } - impure->irsb_hash_table->sort(); + impure->irsb_hash_table->build(); } // Compute and hash the comparison keys