From 2f195950733112bc84e12b33556d070ee8823cc6 Mon Sep 17 00:00:00 2001 From: Tomas Vondra Date: Mon, 1 Jun 2026 14:22:36 +0200 Subject: [PATCH v2.1 3/6] Use Bloom filters in Parallel Hash joins Extend the Bloom filter support to parallel-aware hash joins. The rules for when a filter is built and probed are the same as for serial joins, but the filter lives in the DSA area of the join and is shared by all participants. It is built in one of two ways: * A filter created while the hash table is being built (the planner expects multiple batches, or a single-batch table has to add batches) has to cover tuples inserted by all participants. Each participant adds the hashes it inserts to a private filter of the same dimensions, and merges it into the shared filter before leaving the build phase. * A filter built adaptively during probing is built from the complete hash table, under the ParallelHashJoinState lock, by the first participant whose lookups show a low enough match rate. The others start probing it once their own lookups show a low enough match rate, or when they attach to the hash table later. A batched join probes the filter while partitioning the outer relation, so rejected tuples are never written to the batch files. Tuples read back from a batch file have already been checked and are not probed again. EXPLAIN combines the Bloom filter statistics of all participants, and with VERBOSE also shows the counts of each worker. Notes: * The adaptive build scans the whole shared hash table while holding the ParallelHashJoinState lock exclusively, blocking the participants that reach a window boundary in the meantime. This is with interrupts held off, which is a bit annoying. We could add an intermediate state PHJ_BLOOM_BUILDING, so that other participants can continue probing. * As in the serial case, a filter created when a single-batch table starts batching may be far too small. * Neither the private filters nor the shared one are part of the memory accounting. * PHJ_BLOOM_DISABLED is defined but never used. The idea is to maybe use it to permanently disable filters with too bad false positive rate. --- src/backend/commands/explain.c | 117 ++++- src/backend/executor/nodeHash.c | 441 ++++++++++++++++-- src/backend/executor/nodeHashjoin.c | 77 ++- src/backend/lib/bloomfilter.c | 79 +++- src/include/executor/hashjoin.h | 48 ++ src/include/executor/instrument_node.h | 2 +- src/include/executor/nodeHash.h | 3 + src/include/lib/bloomfilter.h | 6 + src/include/nodes/execnodes.h | 3 +- src/test/regress/expected/join_hash_bloom.out | 145 ++++++ src/test/regress/sql/join_hash_bloom.sql | 45 ++ 11 files changed, 926 insertions(+), 40 deletions(-) diff --git a/src/backend/commands/explain.c b/src/backend/commands/explain.c index 8f2707fbcc2..8571b86f85f 100644 --- a/src/backend/commands/explain.c +++ b/src/backend/commands/explain.c @@ -3436,6 +3436,117 @@ show_hash_info(HashState *hashstate, ExplainState *es) worker_hi->nbatch_original); hinstrument.space_peak = Max(hinstrument.space_peak, worker_hi->space_peak); + + /* + * Each participant (in a parallel-aware hash join) probes its own + * outer tuples, so the probe and match counts are summed. + */ + hinstrument.bloom_nprobes += worker_hi->bloom_nprobes; + hinstrument.bloom_nmatches += worker_hi->bloom_nmatches; + hinstrument.hash_nlookups += worker_hi->hash_nlookups; + hinstrument.hash_nmatches += worker_hi->hash_nmatches; + + /* + * It's possible only some participants used a Bloom filter. For + * example the leader might not have used the filter, but we still + * want to show the info collected from workers. + * + * Parallel-aware and parallel-oblivious filters behave in very + * different ways. In parallel-aware joins the filter is built in + * a synchronized way, but then some participants may choose not + * to use it. The filter parameters are taken from the shared + * filter. + * + * In parallel-oblivious joins each participant builds its own + * private filter, independently of other participants. The filter + * parameters are taken from the private filters, and participants + * that never built a filter report zeroes. We take the maxima, + * same as we do for hash tables. + */ + hinstrument.bloom_used |= worker_hi->bloom_used; + hinstrument.bloom_nbytes = Max(hinstrument.bloom_nbytes, + worker_hi->bloom_nbytes); + hinstrument.bloom_nhashfuncs = Max(hinstrument.bloom_nhashfuncs, + worker_hi->bloom_nhashfuncs); + hinstrument.bloom_false_positive_rate = Max(hinstrument.bloom_false_positive_rate, + worker_hi->bloom_false_positive_rate); + } + + /* + * In VERBOSE mode, also show the probe counts of each worker (the + * merged totals are shown below). Skip this when per-worker details + * are hidden, i.e. when es->workers_state is NULL; ExplainOpenWorker + * must not be called in that case. + */ + if (es->workers_state && es->verbose) + { + for (i = 0; i < shared_info->num_workers; ++i) + { + HashInstrumentation *worker_hi = &shared_info->hinstrument[i]; + + /* ignore workers that did not probe anything */ + if (worker_hi->hash_nlookups == 0 && worker_hi->bloom_nprobes == 0) + continue; + + ExplainOpenWorker(i, es); + + if (es->format == EXPLAIN_FORMAT_TEXT) + { + if (worker_hi->hash_nlookups > 0) + { + ExplainIndentText(es); + appendStringInfo(es->str, + "Hash Lookups: " INT64_FORMAT " Matches: " INT64_FORMAT " Match Rate: %.3f%%\n", + worker_hi->hash_nlookups, + worker_hi->hash_nmatches, + (100.0 * worker_hi->hash_nmatches / worker_hi->hash_nlookups)); + } + + if (worker_hi->bloom_nprobes > 0) + { + ExplainIndentText(es); + appendStringInfo(es->str, + "Bloom Filter Probes: " INT64_FORMAT " Matches: " INT64_FORMAT " Match Rate: %.3f%%\n", + worker_hi->bloom_nprobes, + worker_hi->bloom_nmatches, + (100.0 * worker_hi->bloom_nmatches / worker_hi->bloom_nprobes)); + } + } + else + { + double match_rate; + + if (worker_hi->hash_nlookups > 0) + { + match_rate = (double) worker_hi->hash_nmatches / + worker_hi->hash_nlookups; + + ExplainPropertyInteger("Hash Lookups", NULL, + worker_hi->hash_nlookups, es); + ExplainPropertyInteger("Hash Matches", NULL, + worker_hi->hash_nmatches, es); + ExplainPropertyFloat("Hash Match Rate", NULL, + (100.0 * match_rate), 3, es); + } + + if (worker_hi->bloom_nprobes > 0) + { + match_rate = (double) worker_hi->bloom_nmatches / + worker_hi->bloom_nprobes; + + ExplainOpenGroup("Bloom Filter", "Bloom Filter", true, es); + ExplainPropertyInteger("Probes", NULL, + worker_hi->bloom_nprobes, es); + ExplainPropertyInteger("Matches", NULL, + worker_hi->bloom_nmatches, es); + ExplainPropertyFloat("Match Rate", NULL, + (100.0 * match_rate), 3, es); + ExplainCloseGroup("Bloom Filter", "Bloom Filter", true, es); + } + } + + ExplainCloseWorker(i, es); + } } } @@ -3501,7 +3612,7 @@ show_hash_info(HashState *hashstate, ExplainState *es) ExplainPropertyFloat("Hash Match Rate", NULL, (100.0 * match_rate), 3, es); } - else + else if (hinstrument.hash_nlookups > 0) { ExplainIndentText(es); appendStringInfo(es->str, @@ -3547,7 +3658,7 @@ show_hash_info(HashState *hashstate, ExplainState *es) ExplainPropertyFloat("False Positive Rate", NULL, 100.0 * hinstrument.bloom_false_positive_rate, 3, es); - if (es->analyze) + if (es->verbose) { ExplainPropertyInteger("Probes", NULL, hinstrument.bloom_nprobes, es); @@ -3568,7 +3679,7 @@ show_hash_info(HashState *hashstate, ExplainState *es) hinstrument.bloom_nhashfuncs, 100.0 * hinstrument.bloom_false_positive_rate); - if (es->analyze) + if (es->verbose) { ExplainIndentText(es); appendStringInfo(es->str, diff --git a/src/backend/executor/nodeHash.c b/src/backend/executor/nodeHash.c index a84b17d9113..2e4c44fe801 100644 --- a/src/backend/executor/nodeHash.c +++ b/src/backend/executor/nodeHash.c @@ -83,6 +83,9 @@ static bool ExecParallelHashTuplePrealloc(HashJoinTable hashtable, static void ExecParallelHashMergeCounters(HashJoinTable hashtable); static void ExecParallelHashCloseBatchAccessors(HashJoinTable hashtable); +static void ExecParallelHashBloomCreateFilter(HashJoinTable hashtable, int64 nelems); +static void ExecParallelHashBloomBuildFilter(HashJoinTable hashtable); + /* FIXME would it make sense to mark some of these inline? */ static void ExecHashBloomCreateFilter(HashJoinTable hashtable, bool complete); static void ExecHashBloomBuildFilter(HashJoinTable hashtable, bool complete); @@ -399,6 +402,18 @@ MultiExecParallelHash(HashState *node) ExecParallelHashIncreaseNumBuckets(hashtable); ExecParallelHashEnsureBatchAccessors(hashtable); ExecParallelHashTableSetCurrentBatch(hashtable, 0); + + /* + * When in a multi-batch case, we want to build a shared filter + * from the very beginning. So create per-worker filters with + * matching parameters, and we'll merge them at the end of the + * build. (We might have already created the private filter while + * helping to repartition, in which case we must not create + * another one.) + */ + if ((pstate->bloom_nelems > 0) && (hashtable->bloomFilter == NULL)) + ExecParallelHashBloomCreateFilter(hashtable, pstate->bloom_nelems); + for (;;) { bool isnull; @@ -420,6 +435,11 @@ MultiExecParallelHash(HashState *node) /* normal case with a non-null join key */ ExecParallelHashTableInsert(hashtable, slot, hashvalue); hashtable->reportTuples++; + + /* add the hash to the private Bloom filter */ + if (hashtable->bloomFilter != NULL) + bloom_add_element(hashtable->bloomFilter, + (unsigned char *) &hashvalue, sizeof(uint32)); } else if (node->keep_null_tuples) { @@ -445,6 +465,35 @@ MultiExecParallelHash(HashState *node) */ ExecParallelHashMergeCounters(hashtable); + /* + * If we built a private Bloom filter, merge it into the shared + * filter now, while still holding the build barrier, so that the + * shared filter is complete by the time anyone starts probing. + * + * This covers both the case when we know about batching from the + * beginning of the build (with the shared filter allocated in + * ExecHashTableCreate), and when we start batching only during + * the build (in ExecParallelHashIncreaseNumBatches). + */ + if (hashtable->bloomFilter != NULL && + DsaPointerIsValid(pstate->bloom_filter)) + { + bloom_filter *shared; + + LWLockAcquire(&pstate->lock, LW_EXCLUSIVE); + shared = (bloom_filter *) + dsa_get_address(hashtable->area, pstate->bloom_filter); + bloom_merge(shared, hashtable->bloomFilter); + LWLockRelease(&pstate->lock); + + /* + * Free the private filter, we won't need it anymore, and we + * will set the pointer to the shared one in a bit. + */ + bloom_free(hashtable->bloomFilter); + hashtable->bloomFilter = NULL; + } + BarrierDetach(&pstate->grow_buckets_barrier); BarrierDetach(&pstate->grow_batches_barrier); @@ -477,6 +526,29 @@ MultiExecParallelHash(HashState *node) hashtable->log2_nbuckets = pg_ceil_log2_32(hashtable->nbuckets); hashtable->totalTuples = pstate->total_tuples; + /* + * In the multi-batch case, set the pointer at the filter in shared + * memory, so that the participants can probe it directly. In the + * single-batch case the filter may get built adaptively later, during + * probing (and we'll set the pointer then). + * + * If this participant attached late, another participant may have built + * the filter already, or may be building it right now. The build barrier + * does not order us with that build, so hold the lock while looking at + * the filter. Otherwise we might see it as built before all its bits are + * set, and then reject outer tuples that do have a match. + * + * XXX We could also add memory barriers somewhere, but this seems nicer. + */ + LWLockAcquire(&pstate->lock, LW_SHARED); + if (pstate->bloom_state == PHJ_BLOOM_BUILT && + DsaPointerIsValid(pstate->bloom_filter)) + { + hashtable->bloomFilter = (bloom_filter *) + dsa_get_address(hashtable->area, pstate->bloom_filter); + } + LWLockRelease(&pstate->lock); + /* * Unless we're completely done and the batch state has been freed, make * sure we have accessors. @@ -649,6 +721,10 @@ ExecHashTableCreate(HashState *state) hashtable->bloomSampleProbes = 0; hashtable->bloomProbes = 0; hashtable->bloomMatches = 0; + hashtable->bloomUsed = false; + hashtable->bloomNHashfuncs = 0; + hashtable->bloomNBytes = 0; + hashtable->bloomFPR = 0; hashtable->hashLookups = 0; hashtable->hashMatches = 0; hashtable->hashLookupsTotal = 0; @@ -753,6 +829,43 @@ ExecHashTableCreate(HashState *state) */ pstate->nbuckets = nbuckets; ExecParallelHashTableAlloc(hashtable, 0); + + /* + * If we already know we'll need more than one batch, set up a + * shared Bloom filter right away so that it includes all inner + * tuples. Each worker builds a private filter while hashing and + * merges it into this one at the end of the build (see + * MultiExecParallelHash). + * + * We size it from the planner's estimate so that all the filters + * have matching dimensions (which is required for merging). The + * estimate may be off, so expect at least 1000 tuples and double + * the result, similar to the hedge the serial code applies when + * sizing a filter before the build is complete. We don't want to + * go much further than that - Bloom filters degrade smoothly when + * undersized, but once the filter exceeds the CPU caches it gets + * much slower. In the worst case the filter gets adaptively + * disabled once too many probes find a match. + */ + if (nbatch > 1 && hashtable->bloomEnabled) + { + pstate->bloom_nelems = Max((int64) rows, 1000); + pstate->bloom_nelems *= 2.0; /* FIXME can overflow in int64 */ + + /* same minimum as elsewhere */ + pstate->bloom_nelems = Max(pstate->bloom_nelems, 1000); + + pstate->bloom_filter = + dsa_allocate(hashtable->area, + bloom_estimate(pstate->bloom_nelems, + work_mem, + BLOOM_MIN_FILTER_SIZE)); + bloom_init(dsa_get_address(hashtable->area, + pstate->bloom_filter), + pstate->bloom_nelems, work_mem, + BLOOM_MIN_FILTER_SIZE, 0); + pstate->bloom_state = PHJ_BLOOM_BUILT; + } } /* @@ -972,6 +1085,13 @@ ExecHashBloomBuildFilter(HashJoinTable hashtable, bool complete) * When sampling the filter probes, most tuples bypass the filter and the * function returns false without consulting it. * + * The caller is responsible for probing each outer tuple only once. Tuples + * written to a batch file have already been checked (either while processing + * batch 0 of a serial join, or while partitioning the outer relation of a + * Parallel Hash join), so there's no point in checking them again when they + * are read back. Doing so would waste cycles and distort the probe counters + * used to drive the adaptive sampling. + * * XXX We can't mark it inline, it's called from a different translation unit. * Maybe we could optimize it in some other way? */ @@ -980,13 +1100,6 @@ ExecHashBloomReject(HashJoinTable hashtable, uint32 hashvalue) { bool reject = false; - /* - * Ignore the filter after processing the first batch (all tuples spilled - * to temporary files already went through the check). - */ - if (hashtable->curbatch != 0) - return false; - /* If there's no filter, all tuples should pass. */ if (hashtable->bloomFilter == NULL) return false; @@ -1077,6 +1190,8 @@ ExecHashBloomAdaptiveCreate(HashJoinTable hashtable) int64 hashLookups, hashMatches; + ParallelHashJoinState *pstate = hashtable->parallel_state; + /* Bail out if Bloom filters are disabled. */ if (!hashtable->bloomEnabled) return; @@ -1085,14 +1200,10 @@ ExecHashBloomAdaptiveCreate(HashJoinTable hashtable) if (hashtable->bloomFilter != NULL) return; - /* We can't build filters for parallel hash joins. */ - if (hashtable->parallel_state != NULL) - return; - - /* All serial batched runs should have a filter created automatically. */ + /* All batched runs should have a filter created automatically. */ Assert(hashtable->nbatch == 1); - /* */ + /* use the total counts, including the not-yet-accumulated part */ hashLookups = hashtable->hashLookupsTotal + hashtable->hashLookups; hashMatches = hashtable->hashMatchesTotal + hashtable->hashMatches; @@ -1122,7 +1233,63 @@ ExecHashBloomAdaptiveCreate(HashJoinTable hashtable) if (hashMatches > hashLookups * BLOOM_BUILD_THRESHOLD) return; - ExecHashBloomBuildFilter(hashtable, true); + if (!pstate) + { + /* serial join */ + ExecHashBloomBuildFilter(hashtable, true); + } + else + { + /* + * Parallel join: coordinate so that only one backend builds the + * shared filter (by scanning the current hash table). The first + * backend to finish its sample builds the filter; later backends + * simply observe the result. + * + * The workers may hit the adaptive build threshold at different times + * (or maybe some workers may not hit it at all), in which case the + * worker will not know about the filter and won't probe it. But + * that's fine - it's up to the worker to decide whether to probe or + * not (based on its local stats). We're not adding items to the hash + * table, so this can't cause missing data or anything like that. + * + * XXX It's not great this happens while holding an exclusive LWLock. + * That may take a bit of time (proportional to the hash table size, + * so roughly hash_mem * nparticipants), with disabled interrupts + * (which disables query cancel, etc.). We might do this differently - + * build without a lock, then publish the result, maybe using some + * barriers. In fact, we could simply allocate and publish the shared + * filter, so that everyone starts using it when they choose to. So + * we'd switch to PHJ_BLOOM_BUILDING, and release the lock until the + * end of the build. And everyone would treat it as "no rejection" + * filter. And then later once the build completes, we'd flip it to + * "PHJ_BLOOM_BUILT", and backends would start probing it. The + * difference is whether backends can continue working during the + * build, or should pause and wait. + * + * XXX Notice this happens during probing, in single-batch case, after + * all tuples were added to the in-memory hashtable. There are no more + * items added by any participant. + */ + LWLockAcquire(&pstate->lock, LW_EXCLUSIVE); + + if (pstate->bloom_state == PHJ_BLOOM_NONE) + { + ExecParallelHashBloomBuildFilter(hashtable); + } + + /* + * If the shared filter is already built (possibly by this process), + * set the pointer in the hashtable - that's what we use for probing. + */ + if (pstate->bloom_state == PHJ_BLOOM_BUILT) + { + hashtable->bloomFilter = (bloom_filter *) + dsa_get_address(hashtable->area, pstate->bloom_filter); + } + + LWLockRelease(&pstate->lock); + } } /* @@ -1461,6 +1628,78 @@ ExecHashTableDestroy(HashJoinTable hashtable) pfree(hashtable); } +/* + * ExecParallelHashBloomCreateFilter + * Allocate an empty Bloom filter for this hash table. + * + * The filter is allocated in the long-lived hashCxt so that it survives + * per-batch resets. "nelems" is an estimate of the number of inner tuples, + * used to size the filter; it should be computed identically by every + * participant of a Parallel Hash join so that local filters can be merged. + */ +static void +ExecParallelHashBloomCreateFilter(HashJoinTable hashtable, int64 nelems) +{ + MemoryContext oldcxt; + + Assert(hashtable->bloomFilter == NULL); + + oldcxt = MemoryContextSwitchTo(hashtable->hashCxt); + hashtable->bloomFilter = bloom_create_custom(nelems, work_mem, + BLOOM_MIN_FILTER_SIZE, 0); + MemoryContextSwitchTo(oldcxt); +} + +/* + * ExecParallelHashBloomBuildFilter + * Build the shared Bloom filter for a parallel single-batch hash table. + * + * Called with pstate->lock held by a single backend that has decided the join + * is selective enough to benefit. The completed shared hash table is scanned + * to populate the filter, which is then published for all backends to probe. + */ +static void +ExecParallelHashBloomBuildFilter(HashJoinTable hashtable) +{ + ParallelHashJoinState *pstate = hashtable->parallel_state; + bloom_filter *shared; + int i; + + Assert(hashtable->nbatch == 1); + Assert(pstate->bloom_state == PHJ_BLOOM_NONE); + Assert(!DsaPointerIsValid(pstate->bloom_filter)); + + /* + * Size exactly for the expected number of tuples. We only call this from + * the adaptive build during probing, at which point we know the total + * number of tuples. + */ + pstate->bloom_nelems = Max((int64) pstate->total_tuples, 1000); + pstate->bloom_filter = + dsa_allocate(hashtable->area, + bloom_estimate(pstate->bloom_nelems, work_mem, + BLOOM_MIN_FILTER_SIZE)); + shared = bloom_init(dsa_get_address(hashtable->area, + pstate->bloom_filter), + pstate->bloom_nelems, work_mem, + BLOOM_MIN_FILTER_SIZE, 0); + + for (i = 0; i < hashtable->nbuckets; i++) + { + HashJoinTuple tuple = ExecParallelHashFirstTuple(hashtable, i); + + while (tuple != NULL) + { + bloom_add_element(shared, + (unsigned char *) &tuple->hashvalue, + sizeof(uint32)); + tuple = ExecParallelHashNextTuple(hashtable, tuple); + } + } + + pstate->bloom_state = PHJ_BLOOM_BUILT; +} + /* * Consider adjusting the allowed hash table size, depending on the number * of batches, to minimize the overall memory usage (for both the hashtable @@ -1721,6 +1960,7 @@ ExecParallelHashIncreaseNumBatches(HashJoinTable hashtable) ParallelHashJoinBatch *old_batch0; int new_nbatch; int i; + int64 ntuples; /* Move the old batch out of the way. */ old_batch0 = hashtable->batches[0].shared; @@ -1728,6 +1968,17 @@ ExecParallelHashIncreaseNumBatches(HashJoinTable hashtable) pstate->old_nbatch = hashtable->nbatch; pstate->batches = InvalidDsaPointer; + /* + * Number of tuples in batch 0 so far, used to size the Bloom + * filter below. While there's a single batch, each + * participant folds its local count into the shared one + * whenever it allocates a new chunk (see + * ExecParallelHashTupleAlloc), so the shared value covers all + * participants except for their current chunks. Add our own + * local count to get a bit closer. + */ + ntuples = old_batch0->ntuples + hashtable->batches[0].ntuples; + /* Free this backend's old accessors. */ ExecParallelHashCloseBatchAccessors(hashtable); @@ -1807,6 +2058,56 @@ ExecParallelHashIncreaseNumBatches(HashJoinTable hashtable) for (i = 0; i < new_nbuckets; ++i) dsa_pointer_atomic_init(&buckets[i], InvalidDsaPointer); pstate->nbuckets = new_nbuckets; + + /* + * Create the new shared filter (we'll create the new + * private per-worker filters in + * ExecParallelHashRepartitionFirst). + * + * sizing: We assume we've only seen 1/2 the tuples so + * far. We clearly expected fewer tuples (otherwise we'd + * choose batching right away), but maybe we should use a + * bigger factor? We can easily be off by an order of + * magnitude. OTOH we don't want to overdo it, oversized + * filters get somewhat useless, especially once larger + * than CPU cache. + * + * XXX Is there a better way to estimate the number of + * tuples we'll see for the inner relation? + * + * XXX This might be a good fit for scalable Bloom filters + * (or some other type of filter?) + * + * XXX This should probably check pstate->bloom_filter, + * instead of the private hashtable->bloomEnabled. Can + * those two disagree? + */ + if (hashtable->bloomEnabled) + { + /* + * Double the combined number of tuples all + * participants inserted so far (into the only batch + * we have). + */ + pstate->bloom_nelems = Max(pstate->bloom_nelems, ntuples); + pstate->bloom_nelems *= 2.0; /* FIXME can overflow in + * int64 */ + + /* same minimum as elsewhere */ + pstate->bloom_nelems = Max(pstate->bloom_nelems, 1000); + + pstate->bloom_filter = + dsa_allocate(hashtable->area, + bloom_estimate(pstate->bloom_nelems, + work_mem, + BLOOM_MIN_FILTER_SIZE)); + bloom_init(dsa_get_address(hashtable->area, + pstate->bloom_filter), + pstate->bloom_nelems, work_mem, + BLOOM_MIN_FILTER_SIZE, 0); + pstate->bloom_state = PHJ_BLOOM_BUILT; + } + } else { @@ -1930,8 +2231,28 @@ ExecParallelHashRepartitionFirst(HashJoinTable hashtable) dsa_pointer chunk_shared; HashMemoryChunk chunk; + /* + * If starting to batch (from nbatch=1), we need to create a local filter + * and populate it with entries in each worker. + * + * XXX This should probably check pstate->bloom_filter, instead of the + * private hashtable->bloomEnabled. Can those two disagree? + */ + bool build_filter = hashtable->bloomEnabled && + (hashtable->parallel_state->old_nbatch == 1); + Assert(hashtable->nbatch == hashtable->parallel_state->nbatch); + /* + * Build filter with the same parameters as the shared filter created in + * ExecParallelHashIncreaseNumBatches (so that we can merge them later). + */ + if (build_filter) + { + ExecParallelHashBloomCreateFilter(hashtable, + hashtable->parallel_state->bloom_nelems); + } + while ((chunk = ExecParallelHashPopChunkQueue(hashtable, &chunk_shared))) { size_t idx = 0; @@ -1949,6 +2270,12 @@ ExecParallelHashRepartitionFirst(HashJoinTable hashtable) ExecHashGetBucketAndBatch(hashtable, hashTuple->hashvalue, &bucketno, &batchno); + /* insert everything into the filter */ + if (build_filter) + bloom_add_element(hashtable->bloomFilter, + (unsigned char *) &hashTuple->hashvalue, + sizeof(uint32)); + Assert(batchno < hashtable->nbatch); if (batchno == 0) { @@ -2573,6 +2900,11 @@ ExecScanHashBucket(HashJoinState *hjstate, * On success, the inner tuple is stored into hjstate->hj_CurTuple and * econtext->ecxt_innertuple, using hjstate->hj_HashTupleSlot as the slot * for the latter. + * + * XXX This also updates the hashMatches, but only when starting to scan + * the hash bucket (with hj_CurTuple == NULL), so that it agrees with the + * hashLookups counter. This matters for joins with multiple matches on the + * inner side - the hashLookups is incremented only once per outer tuple. */ bool ExecParallelScanHashBucket(HashJoinState *hjstate, @@ -2607,6 +2939,23 @@ ExecParallelScanHashBucket(HashJoinState *hjstate, if (ExecQualAndReset(hjclauses, econtext)) { + /* + * Count the first match found for this outer tuple. This may + * also create the Bloom filter - the decision is made on the + * lookup that completes a window of BLOOM_BUILD_WINDOW + * lookups, and that lookup may well be a match. + * + * If an outer tuple has multiple matching inner tuples, we + * want it to count as a single match, so that it's comparable + * to counters for the Bloom filter (which also counts each + * outer as a single probe). + */ + if (hjstate->hj_CurTuple == NULL) + { + hashtable->hashMatches++; + ExecHashBloomAdaptiveCreate(hashtable); + } + hjstate->hj_CurTuple = hashTuple; return true; } @@ -2615,6 +2964,10 @@ ExecParallelScanHashBucket(HashJoinState *hjstate, hashTuple = ExecParallelHashNextTuple(hashtable, hashTuple); } + /* first call only (no match found) */ + if (hjstate->hj_CurTuple == NULL) + ExecHashBloomAdaptiveCreate(hashtable); + /* * no match */ @@ -3415,6 +3768,31 @@ ExecHashRetrieveInstrumentation(HashState *node) memcpy(node->shared_info, shared_info, size); } +/* + * Remember the parameters of the Bloom filter in the hash table, so that + * ExecHashAccumInstrumentation() can report them after the filter itself is + * gone (see the comment on these fields in hashjoin.h). + * + * Computing the false positive rate counts the bits set in the whole filter, + * so callers should only bother when the node is instrumented. + */ +void +ExecHashBloomRememberFilter(HashJoinTable hashtable, bloom_filter *filter) +{ + /* + * Do this only once, on the first call - bloom_false_positive_rate() + * walks the whole filter, so we don't want to be calling it on every + * shutdown (which for SPI and cursors can be quite often). + */ + if (hashtable->bloomUsed) + return; + + hashtable->bloomUsed = true; + hashtable->bloomNHashfuncs = bloom_num_hash_funcs(filter); + hashtable->bloomNBytes = bloom_total_bits(filter) / BITS_PER_BYTE; + hashtable->bloomFPR = bloom_false_positive_rate(filter); +} + /* * Accumulate instrumentation data from 'hashtable' into an * initially-zeroed HashInstrumentation struct. @@ -3446,28 +3824,33 @@ ExecHashAccumInstrumentation(HashInstrumentation *instrument, hashtable->spacePeak); /* - * Record Bloom filter information, if a filter was built. Do this only - * once, on the first call - bloom_total_bits() walks the whole filter, so - * we don't want to be calling it on every shutdown (which for SPI and - * cursors can be quite often). + * Record Bloom filter information, if a filter was built. Unlike for the + * hash table fields above, the last instance that uses a filter simply + * wins. The parameters of different instances (e.g. before and after a + * rescan) can't be combined in a meaningful way - taking the maxima of + * the size, number of hash functions and false positive rate would + * describe a filter that never existed. * - * XXX Shouldn't this use Max(), just like the block above? + * Normally the filter is still alive and we can inspect it directly. On a + * parallel rescan, however, ExecHashJoinReInitializeDSM() frees the + * shared filter before the leader gets here, and only the copy of the + * parameters it made in the hash table is left - so report that. */ - if (hashtable->bloomFilter != NULL && !instrument->bloom_used) + if (hashtable->bloomFilter != NULL) + ExecHashBloomRememberFilter(hashtable, hashtable->bloomFilter); + + if (hashtable->bloomUsed) { instrument->bloom_used = true; - instrument->bloom_nhashfuncs - = bloom_num_hash_funcs(hashtable->bloomFilter); - instrument->bloom_nbytes - = bloom_total_bits(hashtable->bloomFilter) / BITS_PER_BYTE; - instrument->bloom_false_positive_rate - = bloom_false_positive_rate(hashtable->bloomFilter); + instrument->bloom_nhashfuncs = hashtable->bloomNHashfuncs; + instrument->bloom_nbytes = hashtable->bloomNBytes; + instrument->bloom_false_positive_rate = hashtable->bloomFPR; } /* - * Record hash-table probe statistics. - * - * XXX Shouldn't this use Max(), just like the earlier block? + * Bloom filter probe and match counts are cumulative, so sum them across + * successive hash table instances (e.g. rescans) rather than taking the + * maximum. */ instrument->bloom_nprobes += hashtable->bloomProbes; instrument->bloom_nmatches += hashtable->bloomMatches; diff --git a/src/backend/executor/nodeHashjoin.c b/src/backend/executor/nodeHashjoin.c index 2cd7fceb0bc..fbf1132058b 100644 --- a/src/backend/executor/nodeHashjoin.c +++ b/src/backend/executor/nodeHashjoin.c @@ -523,8 +523,16 @@ ExecHashJoinImpl(PlanState *pstate, bool parallel) * hash-table lookup and any spilling to a later batch. * Jumping to HJ_FILL_OUTER_TUPLE emits a null-extended row * for outer joins and simply discards the tuple otherwise. + * + * Only tuples coming straight from the outer plan need to be + * checked. Tuples read back from a batch file were already + * checked when they were written: in a serial join while + * processing batch 0, in a Parallel Hash join while + * partitioning the outer relation up front (see + * ExecParallelHashJoinPartitionOuter). */ - if (!parallel && ExecHashBloomReject(hashtable, hashvalue)) + if ((parallel ? (hashtable->nbatch == 1) : (hashtable->curbatch == 0)) && + ExecHashBloomReject(hashtable, hashvalue)) { node->hj_JoinState = HJ_FILL_OUTER_TUPLE; continue; @@ -1842,9 +1850,18 @@ ExecParallelHashJoinPartitionOuter(HashJoinState *hjstate) econtext, &isnull)); - if (!isnull) + /* + * Consult the Bloom filter before writing a tuple to a batch file. + * The shared filter is complete at this point (all participants + * merged their private filters into it before leaving the build + * phase), so a negative answer means the tuple cannot match in any + * batch and there's no point in spilling it. We treat it just like a + * tuple with a NULL join key: outer joins have to emit it + * null-extended, other join types can simply discard it. + */ + if (!isnull && !ExecHashBloomReject(hashtable, hashvalue)) { - /* normal case with a non-null join key */ + /* non-null join key that the filter did not reject */ int batchno; int bucketno; bool shouldFree; @@ -1860,7 +1877,10 @@ ExecParallelHashJoinPartitionOuter(HashJoinState *hjstate) } else if (hjstate->hj_KeepNullTuples) { - /* null join key, but we must save tuple to be emitted later */ + /* + * null join key, or rejected by the Bloom filter, but we must + * save tuple to be emitted later + */ if (hjstate->hj_NullOuterTupleStore == NULL) hjstate->hj_NullOuterTupleStore = ExecHashBuildNullTupleStore(hashtable); tuplestore_puttupleslot(hjstate->hj_NullOuterTupleStore, slot); @@ -1920,6 +1940,9 @@ ExecHashJoinInitializeDSM(HashJoinState *state, ParallelContext *pcxt) pg_atomic_init_u32(&pstate->distributor, 0); pstate->nparticipants = pcxt->nworkers + 1; pstate->total_tuples = 0; + pstate->bloom_filter = InvalidDsaPointer; + pstate->bloom_state = PHJ_BLOOM_NONE; + pstate->bloom_nelems = 0; LWLockInitialize(&pstate->lock, LWTRANCHE_PARALLEL_HASH_JOIN); BarrierInit(&pstate->build_barrier, 0); @@ -1991,6 +2014,52 @@ ExecHashJoinReInitializeDSM(HashJoinState *state, ParallelContext *pcxt) /* Reset build_barrier to PHJ_BUILD_ELECT so we can go around again. */ BarrierInit(&pstate->build_barrier, 0); + + /* + * Remember the parameters of the shared Bloom filter in the leader's hash + * table before we free the filter below, so that EXPLAIN can still report + * them. (ExecHashTableDetach above does not touch the filter, it only + * releases the batch descriptors.) The hash table itself is destroyed + * only later, by ExecReScanHashJoin, which also accumulates the + * instrumentation from it - and by then the filter is gone, so + * ExecHashAccumInstrumentation falls back to this copy. Look at the + * shared filter rather than hj_HashTable->bloomFilter, which stays NULL + * if another participant built the filter adaptively and the leader never + * got around to adopting it. Without instrumentation nothing reads the + * copy, so skip it (the false positive rate is not free to compute). + * + * The hash table may be NULL here while a shared filter exists: with + * parallel_leader_participation = off the leader never builds one, and + * then there's nothing to accumulate from in the leader either. + */ + if (hashNode->ps.instrument && + (state->hj_HashTable != NULL) && + DsaPointerIsValid(pstate->bloom_filter)) + { + bloom_filter *filter; + + filter = (bloom_filter *) + dsa_get_address(state->js.ps.state->es_query_dsa, + pstate->bloom_filter); + + ExecHashBloomRememberFilter(state->hj_HashTable, filter); + } + + /* Free any shared Bloom filter from the previous scan and reset state. */ + if (DsaPointerIsValid(pstate->bloom_filter)) + dsa_free(state->js.ps.state->es_query_dsa, pstate->bloom_filter); + pstate->bloom_filter = InvalidDsaPointer; + pstate->bloom_state = PHJ_BLOOM_NONE; + pstate->bloom_nelems = 0; + + /* + * Reset the "private" pointer, pointing at the DSA we just released. It + * can't point to a private per-participant filter: those are merged into + * the shared filter and freed before any participant leaves the build + * phase (see MultiExecParallelHash), so there's nothing to free here. + */ + if (state->hj_HashTable) + state->hj_HashTable->bloomFilter = NULL; } void diff --git a/src/backend/lib/bloomfilter.c b/src/backend/lib/bloomfilter.c index 6f977718fb6..d483dd302d7 100644 --- a/src/backend/lib/bloomfilter.c +++ b/src/backend/lib/bloomfilter.c @@ -103,6 +103,81 @@ bloom_bitset_bytes(int64 total_elems, int bloom_work_mem, size_t min_filter_size return bitset_bytes; } +/* + * Amount of memory (in bytes) that a Bloom filter sized for the given + * parameters occupies, including the fixed-size header. This lets callers + * place a Bloom filter in caller-managed storage (for example shared memory) + * with bloom_init(). + * + * The minimum size of the bitset (in bytes) is provided by the caller. + * + * See bloom_create_custom(). + */ +size_t +bloom_estimate(int64 total_elems, int bloom_work_mem, size_t min_filter_size) +{ + return offsetof(bloom_filter, bitset) + + sizeof(unsigned char) * bloom_bitset_bytes(total_elems, bloom_work_mem, + min_filter_size); +} + +/* + * Initialize a Bloom filter in caller-provided memory. + * + * "ptr" must point to a sufficiently large buffer, of at least + * bloom_estimate(total_elems, bloom_work_mem, min_filter_size) + * bytes. This is useful when the filter must live in memory that the caller + * manages itself, such as a DSA allocation shared between parallel workers. + * + * Two filters initialized with identical total_elems, bloom_work_mem, + * min_filter_size and seed values share the same dimensions and may be + * combined with bloom_merge(). + * + * The minimum size of the bitset (in bytes) is provided by the caller. See + * bloom_create_custom(). + */ +bloom_filter * +bloom_init(void *ptr, int64 total_elems, int bloom_work_mem, + size_t min_filter_size, uint64 seed) +{ + bloom_filter *filter = (bloom_filter *) ptr; + uint64 bitset_bytes = bloom_bitset_bytes(total_elems, bloom_work_mem, + min_filter_size); + uint64 bitset_bits = bitset_bytes * BITS_PER_BYTE; + + filter->k_hash_funcs = optimal_k(bitset_bits, total_elems); + filter->seed = seed; + filter->m = bitset_bits; + memset(filter->bitset, 0, bitset_bytes); + + return filter; +} + +/* + * Merge the bits set in "src" into "dst". + * + * Both filters must have been created with identical dimensions (that is, the + * same total_elems, bloom_work_mem, minimum bitset size and seed values). + * After this call "dst" reports an element as possibly-present if it was + * possibly-present in either of the input filters, which is exactly the + * filter that would have resulted from adding every element of both filters + * to a single Bloom filter. + */ +void +bloom_merge(bloom_filter *dst, const bloom_filter *src) +{ + uint64 bitset_bytes; + uint64 i; + + Assert(dst->m == src->m); + Assert(dst->k_hash_funcs == src->k_hash_funcs); + Assert(dst->seed == src->seed); + + bitset_bytes = dst->m / BITS_PER_BYTE; + for (i = 0; i < bitset_bytes; i++) + dst->bitset[i] |= src->bitset[i]; +} + /* * Create Bloom filter in caller's memory context. We aim for a false positive * rate of between 1% and 2% when bitset size is not constrained by memory @@ -120,7 +195,7 @@ bloom_bitset_bytes(int64 total_elems, int bloom_work_mem, size_t min_filter_size * bits, and the largest possible bitset is 512MB (2^32 bits). The * implementation allocates only enough memory to target its standard false * positive rate, using a simple formula with caller's total_elems estimate as - * an input. The bitset might be as small as min_bitset_bytes, even when + * an input. The bitset might be as small as min_filter_size, even when * bloom_work_mem is much higher. * * The Bloom filter is seeded using a value provided by the caller. Using a @@ -129,7 +204,7 @@ bloom_bitset_bytes(int64 total_elems, int bloom_work_mem, size_t min_filter_size * Callers that don't care about this pass a constant as their seed, typically * 0. Callers can also use a pseudo-random seed, eg from pg_prng_uint64(). * - * min_bitset_bytes is the minimum bitset size. This is useful for callers + * min_filter_size is the minimum bitset size. This is useful for callers * that want to allow filters smaller than the default DEFAULT_MIN_BITSET_BYTES * (1MB), for example when fingerprinting small sets where the 1MB minimum * would waste memory and would not fit into CPU caches. The bitset is still diff --git a/src/include/executor/hashjoin.h b/src/include/executor/hashjoin.h index 315bcea2db6..12e231c401f 100644 --- a/src/include/executor/hashjoin.h +++ b/src/include/executor/hashjoin.h @@ -276,8 +276,42 @@ typedef struct ParallelHashJoinState pg_atomic_uint32 distributor; /* counter for load balancing */ SharedFileSet fileset; /* space for shared temporary files */ + + /* + * Shared Bloom filter state. When a Parallel Hash join uses a Bloom + * filter, the filter lives in the DSA area pointed to by "bloom_filter". + * "bloom_state" coordinates building it (see PHJ_BLOOM_* constants), and + * "bloom_nelems" is the element count estimate used to size it so that + * other workers build mergeable local filters of identical dimensions. + * + * During the hash build, these fields are only set by the participant + * elected to allocate the hash table or to increase the number of + * batches, and the barriers make them visible to everyone else. + * + * During an adaptive build, a participant may build the filter for a + * single-batch join adaptively, concurrently with participants probing + * the hash table or have only just attached. Any participant who may run + * concurrently with such a build must hold "lock" while reading these + * fields; otherwise it might see the filter as built before all of its + * bits are set. + * + * XXX We could also do this with memory barriers, but LWLock is a more + * traditional way to do this. + * + * XXX The "lock" comment says it's protecting "above" fields, but now + * it's also protecting the bloom fields. + */ + dsa_pointer bloom_filter; /* shared bloom_filter, or InvalidDsaPointer */ + int bloom_state; /* PHJ_BLOOM_* */ + int64 bloom_nelems; /* element estimate used to size the filter */ } ParallelHashJoinState; +/* Values for ParallelHashJoinState.bloom_state. */ +#define PHJ_BLOOM_NONE 0 /* no decision made yet */ +#define PHJ_BLOOM_BUILT 1 /* shared filter is built and usable */ +/* FIXME The DISABLED macro is not used anywhere, maybe remove? */ +#define PHJ_BLOOM_DISABLED 2 /* decided not to use a bloom filter */ + /* The phases for building batches, used by build_barrier. */ #define PHJ_BUILD_ELECT 0 #define PHJ_BUILD_ALLOCATE 1 @@ -411,6 +445,20 @@ typedef struct HashJoinTableData int64 hashLookupsTotal; int64 hashMatchesTotal; + /* + * Parameters of the Bloom filter, for EXPLAIN. + * ExecHashAccumInstrumentation() copies them from bloomFilter while the + * filter is alive, and reports the copy when it is not: in parallel hash + * joins the shared filter is freed by ExecHashJoinReInitializeDSM() on + * rescan, before the leader accumulates the instrumentation from the old + * hash table, so that function saves the parameters here first. Only + * maintained when the node is instrumented. + */ + bool bloomUsed; /* a filter was built for this hash table */ + int bloomNHashfuncs; /* number of hash functions */ + uint64 bloomNBytes; /* size of the filter, in bytes */ + double bloomFPR; /* false positive rate */ + /* * totalTuples is the running total of tuples inserted into either the * main or skew hash tables. reportTuples is the number of tuples that we diff --git a/src/include/executor/instrument_node.h b/src/include/executor/instrument_node.h index d196dd3a97c..1a783d33e3b 100644 --- a/src/include/executor/instrument_node.h +++ b/src/include/executor/instrument_node.h @@ -252,7 +252,7 @@ typedef struct HashInstrumentation int nbatch_original; /* planned number of batches */ Size space_peak; /* peak memory usage in bytes */ - /* Bloom filter statistics (only the non-parallel hash join builds one) */ + /* Bloom filter statistics */ bool bloom_used; /* was a Bloom filter built? */ int bloom_nhashfuncs; /* number of hash functions used */ uint64 bloom_nbytes; /* size of the filter's bitset, in bytes */ diff --git a/src/include/executor/nodeHash.h b/src/include/executor/nodeHash.h index 4559f85bdc6..f236a02e5ae 100644 --- a/src/include/executor/nodeHash.h +++ b/src/include/executor/nodeHash.h @@ -15,6 +15,7 @@ #define NODEHASH_H #include "access/parallel.h" +#include "lib/bloomfilter.h" #include "nodes/execnodes.h" struct SharedHashJoinBatch; @@ -71,6 +72,8 @@ extern void ExecHashInitializeDSM(HashState *node, ParallelContext *pcxt); extern void ExecHashInitializeWorker(HashState *node, ParallelWorkerContext *pwcxt); extern void ExecHashRetrieveInstrumentation(HashState *node); extern void ExecShutdownHash(HashState *node); +extern void ExecHashBloomRememberFilter(HashJoinTable hashtable, + bloom_filter *filter); extern void ExecHashAccumInstrumentation(HashInstrumentation *instrument, HashJoinTable hashtable); diff --git a/src/include/lib/bloomfilter.h b/src/include/lib/bloomfilter.h index 923ef9f3186..eeedbeb73e3 100644 --- a/src/include/lib/bloomfilter.h +++ b/src/include/lib/bloomfilter.h @@ -19,6 +19,12 @@ extern bloom_filter *bloom_create(int64 total_elems, int bloom_work_mem, uint64 seed); extern bloom_filter *bloom_create_custom(int64 total_elems, int bloom_work_mem, size_t min_filter_size, uint64 seed); +extern size_t bloom_estimate(int64 total_elems, int bloom_work_mem, + size_t min_filter_size); +extern bloom_filter *bloom_init(void *ptr, int64 total_elems, + int bloom_work_mem, size_t min_filter_size, + uint64 seed); +extern void bloom_merge(bloom_filter *dst, const bloom_filter *src); extern void bloom_free(bloom_filter *filter); extern void bloom_add_element(bloom_filter *filter, unsigned char *elem, size_t len); diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h index fe8681ba642..038978a24e6 100644 --- a/src/include/nodes/execnodes.h +++ b/src/include/nodes/execnodes.h @@ -2203,7 +2203,8 @@ typedef struct MergeJoinState * outer joins * hj_NullInnerTupleSlot prepared null tuple for left/full outer joins * hj_NullOuterTupleStore tuplestore holding outer tuples that have - * null join keys (but must be emitted anyway); + * null join keys, or were rejected by the + * Bloom filter (but must be emitted anyway); * used only in multi-batch parallel joins * hj_FirstOuterTupleSlot first tuple retrieved from outer plan * hj_JoinState current state of ExecHashJoin state machine diff --git a/src/test/regress/expected/join_hash_bloom.out b/src/test/regress/expected/join_hash_bloom.out index b0b75996483..986ed953884 100644 --- a/src/test/regress/expected/join_hash_bloom.out +++ b/src/test/regress/expected/join_hash_bloom.out @@ -642,6 +642,151 @@ $$); t | f | t | t (1 row) +-- test parallel hash joins +SET work_mem = '512kB'; +SET max_parallel_workers_per_gather = 2; +SET parallel_setup_cost = 0; +SET parallel_tuple_cost = 0; +ALTER TABLE hash_bloom_fact SET (parallel_workers = 2); +-- non-selective in-memory hash join does not use Bloom filters +SET enable_hashjoin_bloom = off; +SELECT * FROM hash_bloom_explain($$SELECT * FROM hash_bloom_fact f JOIN hash_bloom_dimension d ON (f.did = d.id)$$); + hash_bloom_explain +------------------------------------------------------------------------------ + Gather (actual rows=N loops=N) + Workers Planned: 2 + Workers Launched: N + -> Hash Join (actual rows=N loops=N) + Hash Cond: (f.did = d.id) + -> Parallel Seq Scan on hash_bloom_fact f (actual rows=N loops=N) + -> Hash (actual rows=N loops=N) + Buckets: N Batches: 1 Memory Usage: N + -> Seq Scan on hash_bloom_dimension d (actual rows=N loops=N) +(9 rows) + +SET enable_hashjoin_bloom = on; +SELECT * FROM hash_bloom_explain($$SELECT * FROM hash_bloom_fact f JOIN hash_bloom_dimension d ON (f.did = d.id)$$); + hash_bloom_explain +------------------------------------------------------------------------------ + Gather (actual rows=N loops=N) + Workers Planned: 2 + Workers Launched: N + -> Hash Join (actual rows=N loops=N) + Hash Cond: (f.did = d.id) + -> Parallel Seq Scan on hash_bloom_fact f (actual rows=N loops=N) + -> Hash (actual rows=N loops=N) + Buckets: N Batches: 1 Memory Usage: N + -> Seq Scan on hash_bloom_dimension d (actual rows=N loops=N) +(9 rows) + +-- a selective in-memory join uses a filter (after 1000 lookups) +SET enable_hashjoin_bloom = off; +SELECT * FROM hash_bloom_explain($$SELECT * FROM hash_bloom_fact f JOIN hash_bloom_dimension d ON (f.did = d.id) WHERE d.r < 0.5$$); + hash_bloom_explain +------------------------------------------------------------------------------ + Gather (actual rows=N loops=N) + Workers Planned: 2 + Workers Launched: N + -> Hash Join (actual rows=N loops=N) + Hash Cond: (f.did = d.id) + -> Parallel Seq Scan on hash_bloom_fact f (actual rows=N loops=N) + -> Hash (actual rows=N loops=N) + Buckets: N Batches: 1 Memory Usage: N + -> Seq Scan on hash_bloom_dimension d (actual rows=N loops=N) + Filter: (r < '0.5'::double precision) + Rows Removed by Filter: 4982 +(11 rows) + +SET enable_hashjoin_bloom = on; +SELECT * FROM hash_bloom_explain($$SELECT * FROM hash_bloom_fact f JOIN hash_bloom_dimension d ON (f.did = d.id) WHERE d.r < 0.5$$); + hash_bloom_explain +--------------------------------------------------------------------------------- + Gather (actual rows=N loops=N) + Workers Planned: 2 + Workers Launched: N + -> Hash Join (actual rows=N loops=N) + Hash Cond: (f.did = d.id) + -> Parallel Seq Scan on hash_bloom_fact f (actual rows=N loops=N) + -> Hash (actual rows=N loops=N) + Buckets: N Batches: 1 Memory Usage: N + Bloom Filter: Size: N Hash Functions: N False Positive Rate: N% + -> Seq Scan on hash_bloom_dimension d (actual rows=N loops=N) + Filter: (r < '0.5'::double precision) + Rows Removed by Filter: 4982 +(12 rows) + +-- force batching +SET work_mem = '128kB'; +-- batched join always creates a Bloom filter, but then disables it if +-- not selective enough +SET enable_hashjoin_bloom = off; +SELECT * FROM hash_bloom_explain($$SELECT * FROM hash_bloom_fact f JOIN hash_bloom_dimension d ON (f.did = d.id)$$); + hash_bloom_explain +------------------------------------------------------------------------------ + Gather (actual rows=N loops=N) + Workers Planned: 2 + Workers Launched: N + -> Hash Join (actual rows=N loops=N) + Hash Cond: (f.did = d.id) + -> Parallel Seq Scan on hash_bloom_fact f (actual rows=N loops=N) + -> Hash (actual rows=N loops=N) + Buckets: N Batches: N Memory Usage: N + -> Seq Scan on hash_bloom_dimension d (actual rows=N loops=N) +(9 rows) + +SET enable_hashjoin_bloom = on; +SELECT * FROM hash_bloom_explain($$SELECT * FROM hash_bloom_fact f JOIN hash_bloom_dimension d ON (f.did = d.id)$$); + hash_bloom_explain +--------------------------------------------------------------------------------- + Gather (actual rows=N loops=N) + Workers Planned: 2 + Workers Launched: N + -> Hash Join (actual rows=N loops=N) + Hash Cond: (f.did = d.id) + -> Parallel Seq Scan on hash_bloom_fact f (actual rows=N loops=N) + -> Hash (actual rows=N loops=N) + Buckets: N Batches: N Memory Usage: N + Bloom Filter: Size: N Hash Functions: N False Positive Rate: N% + -> Seq Scan on hash_bloom_dimension d (actual rows=N loops=N) +(10 rows) + +-- batched join always creates a Bloom filter, and keeps using it if +-- selective enough +SET enable_hashjoin_bloom = off; +SELECT * FROM hash_bloom_explain($$SELECT * FROM hash_bloom_fact f JOIN hash_bloom_dimension d ON (f.did = d.id) WHERE d.r < 0.5$$); + hash_bloom_explain +------------------------------------------------------------------------------ + Gather (actual rows=N loops=N) + Workers Planned: 2 + Workers Launched: N + -> Hash Join (actual rows=N loops=N) + Hash Cond: (f.did = d.id) + -> Parallel Seq Scan on hash_bloom_fact f (actual rows=N loops=N) + -> Hash (actual rows=N loops=N) + Buckets: N Batches: N Memory Usage: N + -> Seq Scan on hash_bloom_dimension d (actual rows=N loops=N) + Filter: (r < '0.5'::double precision) + Rows Removed by Filter: 4982 +(11 rows) + +SET enable_hashjoin_bloom = on; +SELECT * FROM hash_bloom_explain($$SELECT * FROM hash_bloom_fact f JOIN hash_bloom_dimension d ON (f.did = d.id) WHERE d.r < 0.5$$); + hash_bloom_explain +--------------------------------------------------------------------------------- + Gather (actual rows=N loops=N) + Workers Planned: 2 + Workers Launched: N + -> Hash Join (actual rows=N loops=N) + Hash Cond: (f.did = d.id) + -> Parallel Seq Scan on hash_bloom_fact f (actual rows=N loops=N) + -> Hash (actual rows=N loops=N) + Buckets: N Batches: N Memory Usage: N + Bloom Filter: Size: N Hash Functions: N False Positive Rate: N% + -> Seq Scan on hash_bloom_dimension d (actual rows=N loops=N) + Filter: (r < '0.5'::double precision) + Rows Removed by Filter: 4982 +(12 rows) + DROP TABLE hash_bloom_fact; DROP TABLE hash_bloom_dimension; DROP TABLE hash_bloom_probe; diff --git a/src/test/regress/sql/join_hash_bloom.sql b/src/test/regress/sql/join_hash_bloom.sql index ba7018e5382..f37136ea538 100644 --- a/src/test/regress/sql/join_hash_bloom.sql +++ b/src/test/regress/sql/join_hash_bloom.sql @@ -340,6 +340,51 @@ SELECT * FROM hash_bloom_stats($$ FROM (VALUES (0), (50000), (99990)) v(x) $$); +-- test parallel hash joins +SET work_mem = '512kB'; +SET max_parallel_workers_per_gather = 2; +SET parallel_setup_cost = 0; +SET parallel_tuple_cost = 0; + +ALTER TABLE hash_bloom_fact SET (parallel_workers = 2); + +-- non-selective in-memory hash join does not use Bloom filters + +SET enable_hashjoin_bloom = off; +SELECT * FROM hash_bloom_explain($$SELECT * FROM hash_bloom_fact f JOIN hash_bloom_dimension d ON (f.did = d.id)$$); + +SET enable_hashjoin_bloom = on; +SELECT * FROM hash_bloom_explain($$SELECT * FROM hash_bloom_fact f JOIN hash_bloom_dimension d ON (f.did = d.id)$$); + +-- a selective in-memory join uses a filter (after 1000 lookups) + +SET enable_hashjoin_bloom = off; +SELECT * FROM hash_bloom_explain($$SELECT * FROM hash_bloom_fact f JOIN hash_bloom_dimension d ON (f.did = d.id) WHERE d.r < 0.5$$); + +SET enable_hashjoin_bloom = on; +SELECT * FROM hash_bloom_explain($$SELECT * FROM hash_bloom_fact f JOIN hash_bloom_dimension d ON (f.did = d.id) WHERE d.r < 0.5$$); + +-- force batching +SET work_mem = '128kB'; + +-- batched join always creates a Bloom filter, but then disables it if +-- not selective enough + +SET enable_hashjoin_bloom = off; +SELECT * FROM hash_bloom_explain($$SELECT * FROM hash_bloom_fact f JOIN hash_bloom_dimension d ON (f.did = d.id)$$); + +SET enable_hashjoin_bloom = on; +SELECT * FROM hash_bloom_explain($$SELECT * FROM hash_bloom_fact f JOIN hash_bloom_dimension d ON (f.did = d.id)$$); + +-- batched join always creates a Bloom filter, and keeps using it if +-- selective enough + +SET enable_hashjoin_bloom = off; +SELECT * FROM hash_bloom_explain($$SELECT * FROM hash_bloom_fact f JOIN hash_bloom_dimension d ON (f.did = d.id) WHERE d.r < 0.5$$); + +SET enable_hashjoin_bloom = on; +SELECT * FROM hash_bloom_explain($$SELECT * FROM hash_bloom_fact f JOIN hash_bloom_dimension d ON (f.did = d.id) WHERE d.r < 0.5$$); + DROP TABLE hash_bloom_fact; DROP TABLE hash_bloom_dimension; DROP TABLE hash_bloom_probe; -- 2.55.0