From 330652a844f637238b37bce2c97de412430812c8 Mon Sep 17 00:00:00 2001 From: Melanie Plageman Date: Wed, 26 Feb 2020 09:18:29 -0800 Subject: [PATCH v6] Implement Adaptive Hashjoin If the inner side tuples of a hashjoin will not fit in memory, the hashjoin can be executed in multiple batches. If the statistics on the inner side relation are accurate, planner chooses a multi-batch strategy and sets the number of batches. The query executor measures the real size of the hashtable and increases the number of batches if the hashtable grows too large. The number of batches is always a power of two, so an increase in the number of batches doubles it. Serial hashjoin measures batch size lazily -- waiting until it is loading a batch to determine if it will fit in memory. Parallel hashjoin, on the other hand, completes all changes to the number of batches during the build phase. If it doubles the number of batches, it dumps all the tuples out, reassigns them to batches, measures each batch, and checks that it will fit in the space allowed. In both cases, the executor currently makes a best effort. If a particular batch won't fit in memory, and, upon changing the number of batches none of the tuples move to a new batch, the executor disables growth in the number of batches globally. After growth is disabled, all batches that would have previously triggered an increase in the number of batches instead exceed the space allowed. There is no mechanism to perform a hashjoin within memory constraints if a run of tuples hash to the same batch. Also, hashjoin will continue to double the number of batches if *some* tuples move each time -- even if the batch will never fit in memory -- resulting in an explosion in the number of batches (affecting performance negatively for multiple reasons). Adaptive hashjoin is a mechanism to process a run of inner side tuples with join keys which hash to the same batch in a manner that is efficient and respects the space allowed. When an offending batch causes the number of batches to be doubled and some percentage of the tuples would not move to a new batch, that batch can be marked to "fall back". This mechanism replaces serial hashjoin's "grow_enabled" flag and replaces part of the functionality of parallel hashjoin's "growth = PHJ_GROWTH_DISABLED" flag. However, instead of disabling growth in the number of batches for all batches, it only prevents this batch from causing another increase in the number of batches. When the inner side of this batch is loaded into memory, stripes of arbitrary tuples totaling work_mem in size are loaded into the hashtable. After probing this stripe, the outer side batch is rewound and the next stripe is loaded. Each stripe of inner is probed until all tuples have been processed. Tuples that match are emitted (depending on the join semantics of the particular join type) during probing of a stripe. In order to make left outer join work, unmatched tuples cannot be emitted NULL-extended until all stripes have been probed. To address this, a bitmap is created with a bit for each tuple of the outer side. If a tuple on the outer side matches a tuple from the inner, the corresponding bit is set. At the end of probing all stripes, the executor scans the bitmap and emits unmatched outer tuples. TODOs: - Batch 0 falling back - Implement stripe_barrier deadlock fix - Fix semi-join - Stripe instrumentation for parallel adaptive hashjoin - Do benchmarking and experiment with different fallback threshholds (currently hardcoded to 80% but more parameterizable than before) - Assorted TODOs in the code Co-authored-by: Jesse Zhang Co-authored-by: David Kimura --- src/backend/commands/explain.c | 43 +- src/backend/executor/nodeHash.c | 306 +++++-- src/backend/executor/nodeHashjoin.c | 652 ++++++++++++--- src/backend/postmaster/pgstat.c | 13 +- src/backend/utils/sort/Makefile | 1 + src/backend/utils/sort/sharedbits.c | 285 +++++++ src/backend/utils/sort/sharedtuplestore.c | 112 ++- src/include/commands/explain.h | 1 + src/include/executor/hashjoin.h | 47 +- src/include/executor/instrument.h | 7 + src/include/executor/nodeHash.h | 1 + src/include/executor/tuptable.h | 2 + src/include/nodes/execnodes.h | 5 + src/include/pgstat.h | 5 +- src/include/utils/sharedbits.h | 39 + src/include/utils/sharedtuplestore.h | 19 + src/test/regress/expected/join_hash.out | 945 +++++++++++++++++++++- src/test/regress/sql/join_hash.sql | 127 +++ 18 files changed, 2444 insertions(+), 166 deletions(-) create mode 100644 src/backend/utils/sort/sharedbits.c create mode 100644 src/include/utils/sharedbits.h diff --git a/src/backend/commands/explain.c b/src/backend/commands/explain.c index 7ae6131676..fc26341244 100644 --- a/src/backend/commands/explain.c +++ b/src/backend/commands/explain.c @@ -184,6 +184,8 @@ ExplainQuery(ParseState *pstate, ExplainStmt *stmt, es->wal = defGetBoolean(opt); else if (strcmp(opt->defname, "settings") == 0) es->settings = defGetBoolean(opt); + else if (strcmp(opt->defname, "usage") == 0) + es->usage = defGetBoolean(opt); else if (strcmp(opt->defname, "timing") == 0) { timing_set = true; @@ -312,6 +314,7 @@ NewExplainState(void) /* Set default options (most fields can be left as zeroes). */ es->costs = true; + es->usage = true; /* Prepare output buffer. */ es->str = makeStringInfo(); @@ -3026,22 +3029,50 @@ show_hash_info(HashState *hashstate, ExplainState *es) else if (hinstrument.nbatch_original != hinstrument.nbatch || hinstrument.nbuckets_original != hinstrument.nbuckets) { + ListCell *lc; + ExplainIndentText(es); appendStringInfo(es->str, - "Buckets: %d (originally %d) Batches: %d (originally %d) Memory Usage: %ldkB\n", + "Buckets: %d (originally %d) Batches: %d (originally %d)", hinstrument.nbuckets, hinstrument.nbuckets_original, hinstrument.nbatch, - hinstrument.nbatch_original, - spacePeakKb); + hinstrument.nbatch_original); + if (es->usage) + appendStringInfo(es->str, " Memory Usage: %ldkB\n", spacePeakKb); + else + appendStringInfo(es->str, "\n"); + + foreach(lc, hinstrument.fallback_batches_stats) + { + FallbackBatchStats *fbs = lfirst(lc); + + ExplainIndentText(es); + appendStringInfo(es->str, "Batch: %d Stripes: %d\n", fbs->batchno, fbs->numstripes); + } } else { + ListCell *lc; + ExplainIndentText(es); appendStringInfo(es->str, - "Buckets: %d Batches: %d Memory Usage: %ldkB\n", - hinstrument.nbuckets, hinstrument.nbatch, - spacePeakKb); + "Buckets: %d Batches: %d", + hinstrument.nbuckets, hinstrument.nbatch); + if (es->usage) + appendStringInfo(es->str, " Memory Usage: %ldkB\n", spacePeakKb); + else + appendStringInfo(es->str, "\n"); + foreach(lc, hinstrument.fallback_batches_stats) + { + FallbackBatchStats *fbs = lfirst(lc); + + ExplainIndentText(es); + appendStringInfo(es->str, + "Batch: %d Stripes: %d\n", + fbs->batchno, + fbs->numstripes); + } } } } diff --git a/src/backend/executor/nodeHash.c b/src/backend/executor/nodeHash.c index 5da13ada72..6ecbc76ab5 100644 --- a/src/backend/executor/nodeHash.c +++ b/src/backend/executor/nodeHash.c @@ -80,7 +80,6 @@ static bool ExecParallelHashTuplePrealloc(HashJoinTable hashtable, static void ExecParallelHashMergeCounters(HashJoinTable hashtable); static void ExecParallelHashCloseBatchAccessors(HashJoinTable hashtable); - /* ---------------------------------------------------------------- * ExecHash * @@ -321,6 +320,27 @@ MultiExecParallelHash(HashState *node) * skew). */ pstate->growth = PHJ_GROWTH_DISABLED; + + /* + * In the current design, batch 0 cannot fall back. That + * behavior is an artifact of the existing design where batch + * 0 fills the initial hash table and as an optimization it + * doesn't need a batch file. But, there is no real reason + * that batch 0 shouldn't be allowed to spill. + * + * Consider a hash table where majority of tuples with + * hashvalue 0. These tuples will never relocate no matter how + * many batches exist. If you cannot exceed work_mem, then you + * will be stuck infinitely trying to double the number of + * batches in order to accommodate the tuples that can only + * ever be in batch 0. So, we allow it to be set to fall back + * during the build phase to avoid excessive batch increases + * but we don't check it when loading the actual tuples, so we + * may exceed space_allowed. We set it back to false here so + * that it isn't true during any of the checks that may happen + * during probing. + */ + hashtable->batches[0].shared->hashloop_fallback = false; } } @@ -495,12 +515,14 @@ ExecHashTableCreate(HashState *state, List *hashOperators, List *hashCollations, hashtable->curbatch = 0; hashtable->nbatch_original = nbatch; hashtable->nbatch_outstart = nbatch; - hashtable->growEnabled = true; hashtable->totalTuples = 0; hashtable->partialTuples = 0; hashtable->skewTuples = 0; hashtable->innerBatchFile = NULL; hashtable->outerBatchFile = NULL; + hashtable->hashloop_fallback = NULL; + hashtable->fallback_batches_stats = NULL; + hashtable->curstripe = -1; hashtable->spaceUsed = 0; hashtable->spacePeak = 0; hashtable->spaceAllowed = space_allowed; @@ -572,6 +594,8 @@ ExecHashTableCreate(HashState *state, List *hashOperators, List *hashCollations, palloc0(nbatch * sizeof(BufFile *)); hashtable->outerBatchFile = (BufFile **) palloc0(nbatch * sizeof(BufFile *)); + hashtable->hashloop_fallback = (BufFile **) + palloc0(nbatch * sizeof(BufFile *)); /* The files will not be opened until needed... */ /* ... but make sure we have temp tablespaces established for them */ PrepareTempTablespaces(); @@ -866,6 +890,8 @@ ExecHashTableDestroy(HashJoinTable hashtable) BufFileClose(hashtable->innerBatchFile[i]); if (hashtable->outerBatchFile[i]) BufFileClose(hashtable->outerBatchFile[i]); + if (hashtable->hashloop_fallback[i]) + BufFileClose(hashtable->hashloop_fallback[i]); } } @@ -876,6 +902,9 @@ ExecHashTableDestroy(HashJoinTable hashtable) pfree(hashtable); } +/* Threshhold for tuple relocation during batch split for parallel and serial */ +#define MAX_RELOCATION 0.8 + /* * ExecHashIncreaseNumBatches * increase the original number of batches in order to reduce @@ -886,14 +915,18 @@ ExecHashIncreaseNumBatches(HashJoinTable hashtable) { int oldnbatch = hashtable->nbatch; int curbatch = hashtable->curbatch; + int childbatch; int nbatch; MemoryContext oldcxt; long ninmemory; long nfreed; HashMemoryChunk oldchunks; + int curbatch_outgoing_tuples; + int childbatch_outgoing_tuples; + int target_batch; + FallbackBatchStats *fallback_batch_stats; - /* do nothing if we've decided to shut off growth */ - if (!hashtable->growEnabled) + if (hashtable->hashloop_fallback && hashtable->hashloop_fallback[curbatch]) return; /* safety check to avoid overflow */ @@ -917,6 +950,8 @@ ExecHashIncreaseNumBatches(HashJoinTable hashtable) palloc0(nbatch * sizeof(BufFile *)); hashtable->outerBatchFile = (BufFile **) palloc0(nbatch * sizeof(BufFile *)); + hashtable->hashloop_fallback = (BufFile **) + palloc0(nbatch * sizeof(BufFile *)); /* time to establish the temp tablespaces, too */ PrepareTempTablespaces(); } @@ -927,10 +962,14 @@ ExecHashIncreaseNumBatches(HashJoinTable hashtable) repalloc(hashtable->innerBatchFile, nbatch * sizeof(BufFile *)); hashtable->outerBatchFile = (BufFile **) repalloc(hashtable->outerBatchFile, nbatch * sizeof(BufFile *)); + hashtable->hashloop_fallback = (BufFile **) + repalloc(hashtable->hashloop_fallback, nbatch * sizeof(BufFile *)); MemSet(hashtable->innerBatchFile + oldnbatch, 0, (nbatch - oldnbatch) * sizeof(BufFile *)); MemSet(hashtable->outerBatchFile + oldnbatch, 0, (nbatch - oldnbatch) * sizeof(BufFile *)); + MemSet(hashtable->hashloop_fallback + oldnbatch, 0, + (nbatch - oldnbatch) * sizeof(BufFile *)); } MemoryContextSwitchTo(oldcxt); @@ -942,6 +981,8 @@ ExecHashIncreaseNumBatches(HashJoinTable hashtable) * no longer of the current batch. */ ninmemory = nfreed = 0; + curbatch_outgoing_tuples = childbatch_outgoing_tuples = 0; + childbatch = (1U << (my_log2(hashtable->nbatch) - 1)) | hashtable->curbatch; /* If know we need to resize nbuckets, we can do it while rebatching. */ if (hashtable->nbuckets_optimal != hashtable->nbuckets) @@ -999,6 +1040,7 @@ ExecHashIncreaseNumBatches(HashJoinTable hashtable) /* and add it back to the appropriate bucket */ copyTuple->next.unshared = hashtable->buckets.unshared[bucketno]; hashtable->buckets.unshared[bucketno] = copyTuple; + curbatch_outgoing_tuples++; } else { @@ -1010,6 +1052,16 @@ ExecHashIncreaseNumBatches(HashJoinTable hashtable) hashtable->spaceUsed -= hashTupleSize; nfreed++; + + /* + * TODO: what to do about tuples that don't go to the child + * batch or stay in the current batch? (this is why we are + * counting tuples to child and curbatch with two diff + * variables in case the tuples go to a batch that isn't the + * child) + */ + if (batchno == childbatch) + childbatch_outgoing_tuples++; } /* next tuple in this chunk */ @@ -1030,21 +1082,33 @@ ExecHashIncreaseNumBatches(HashJoinTable hashtable) #endif /* - * If we dumped out either all or none of the tuples in the table, disable - * further expansion of nbatch. This situation implies that we have - * enough tuples of identical hashvalues to overflow spaceAllowed. - * Increasing nbatch will not fix it since there's no way to subdivide the - * group any more finely. We have to just gut it out and hope the server - * has enough RAM. + * For now we do not support fallback in batch 0 as it is a special case + * and assumed to fit in hashtable. + */ + if (curbatch == 0) + return; + + /* + * The same batch should not be marked to fall back more than once */ - if (nfreed == 0 || nfreed == ninmemory) - { - hashtable->growEnabled = false; #ifdef HJDEBUG - printf("Hashjoin %p: disabling further increase of nbatch\n", - hashtable); + if ((childbatch_outgoing_tuples / (float) ninmemory) >= 0.8) + printf("childbatch %i targeted to fallback.", childbatch); + if ((curbatch_outgoing_tuples / (float) ninmemory) >= 0.8) + printf("curbatch %i targeted to fallback.", curbatch); #endif - } + if ((childbatch_outgoing_tuples / (float) ninmemory) >= MAX_RELOCATION && childbatch > 0) + target_batch = childbatch; + else if ((curbatch_outgoing_tuples / (float) ninmemory) >= MAX_RELOCATION && curbatch > 0) + target_batch = curbatch; + else + return; + hashtable->hashloop_fallback[target_batch] = BufFileCreateTemp(false); + + fallback_batch_stats = palloc0(sizeof(FallbackBatchStats)); + fallback_batch_stats->batchno = target_batch; + fallback_batch_stats->numstripes = 0; + hashtable->fallback_batches_stats = lappend(hashtable->fallback_batches_stats, fallback_batch_stats); } /* @@ -1213,7 +1277,6 @@ ExecParallelHashIncreaseNumBatches(HashJoinTable hashtable) WAIT_EVENT_HASH_GROW_BATCHES_DECIDING)) { bool space_exhausted = false; - bool extreme_skew_detected = false; /* Make sure that we have the current dimensions and buckets. */ ExecParallelHashEnsureBatchAccessors(hashtable); @@ -1224,27 +1287,50 @@ ExecParallelHashIncreaseNumBatches(HashJoinTable hashtable) { ParallelHashJoinBatch *batch = hashtable->batches[i].shared; + /* + * All batches were just created anew during + * repartitioning + */ + Assert(!batch->hashloop_fallback); + + /* + * At the time of repartitioning, each batch updates its + * estimated_size to reflect the size of the batch file on + * disk. It is also updated when increasing preallocated + * space in ExecParallelHashTuplePrealloc(). However, + * batch 0 does not store anything on disk so it has no + * estimated_size. + * + * We still want to allow batch 0 to trigger batch growth. + * In order to do that, for batch 0 check whether the + * actual size exceeds space_allowed. It is a little + * backwards at this point as we would have already + * exceeded inserted the allowed space. + */ if (batch->space_exhausted || - batch->estimated_size > pstate->space_allowed) + batch->estimated_size > pstate->space_allowed || + batch->size > pstate->space_allowed) { int parent; + float frac_moved; space_exhausted = true; - /* - * Did this batch receive ALL of the tuples from its - * parent batch? That would indicate that further - * repartitioning isn't going to help (the hash values - * are probably all the same). - */ parent = i % pstate->old_nbatch; - if (batch->ntuples == hashtable->batches[parent].shared->old_ntuples) - extreme_skew_detected = true; + frac_moved = batch->ntuples / (float) hashtable->batches[parent].shared->old_ntuples; + + if (frac_moved >= MAX_RELOCATION) + { + batch->hashloop_fallback = true; + space_exhausted = false; + } } + if (space_exhausted) + break; } - /* Don't keep growing if it's not helping or we'd overflow. */ - if (extreme_skew_detected || hashtable->nbatch >= INT_MAX / 2) + /* Don't keep growing if we'd overflow. */ + if (hashtable->nbatch >= INT_MAX / 2) pstate->growth = PHJ_GROWTH_DISABLED; else if (space_exhausted) pstate->growth = PHJ_GROWTH_NEED_MORE_BATCHES; @@ -1311,11 +1397,28 @@ ExecParallelHashRepartitionFirst(HashJoinTable hashtable) { size_t tuple_size = MAXALIGN(HJTUPLE_OVERHEAD + tuple->t_len); + tupleMetadata metadata; /* It belongs in a later batch. */ + ParallelHashJoinBatch *batch = hashtable->batches[batchno].shared; + + LWLockAcquire(&batch->lock, LW_EXCLUSIVE); + + if (batch->estimated_stripe_size + tuple_size > hashtable->parallel_state->space_allowed) + { + batch->maximum_stripe_number++; + batch->estimated_stripe_size = 0; + } + + batch->estimated_stripe_size += tuple_size; + + metadata.hashvalue = hashTuple->hashvalue; + metadata.stripe = batch->maximum_stripe_number; + LWLockRelease(&batch->lock); + hashtable->batches[batchno].estimated_size += tuple_size; - sts_puttuple(hashtable->batches[batchno].inner_tuples, - &hashTuple->hashvalue, tuple); + + sts_puttuple(hashtable->batches[batchno].inner_tuples, &metadata, tuple); } /* Count this tuple. */ @@ -1363,27 +1466,41 @@ ExecParallelHashRepartitionRest(HashJoinTable hashtable) for (i = 1; i < old_nbatch; ++i) { MinimalTuple tuple; - uint32 hashvalue; + tupleMetadata metadata; /* Scan one partition from the previous generation. */ sts_begin_parallel_scan(old_inner_tuples[i]); - while ((tuple = sts_parallel_scan_next(old_inner_tuples[i], &hashvalue))) + + while ((tuple = sts_parallel_scan_next(old_inner_tuples[i], &metadata.hashvalue))) { size_t tuple_size = MAXALIGN(HJTUPLE_OVERHEAD + tuple->t_len); int bucketno; int batchno; + ParallelHashJoinBatch *batch; /* Decide which partition it goes to in the new generation. */ - ExecHashGetBucketAndBatch(hashtable, hashvalue, &bucketno, + ExecHashGetBucketAndBatch(hashtable, metadata.hashvalue, &bucketno, &batchno); hashtable->batches[batchno].estimated_size += tuple_size; ++hashtable->batches[batchno].ntuples; ++hashtable->batches[i].old_ntuples; + batch = hashtable->batches[batchno].shared; + /* Store the tuple its new batch. */ - sts_puttuple(hashtable->batches[batchno].inner_tuples, - &hashvalue, tuple); + LWLockAcquire(&batch->lock, LW_EXCLUSIVE); + + if (batch->estimated_stripe_size + tuple_size > pstate->space_allowed) + { + batch->maximum_stripe_number++; + batch->estimated_stripe_size = 0; + } + batch->estimated_stripe_size += tuple_size; + metadata.stripe = batch->maximum_stripe_number; + LWLockRelease(&batch->lock); + /* Store the tuple its new batch. */ + sts_puttuple(hashtable->batches[batchno].inner_tuples, &metadata, tuple); CHECK_FOR_INTERRUPTS(); } @@ -1693,6 +1810,12 @@ retry: if (batchno == 0) { + /* + * TODO: if spilling is enabled for batch 0 so that it can fall back, + * we will need to stop loading batch 0 into the hashtable somewhere-- + * maybe here-- and switch to saving tuples to a file. Currently, this + * will simply exceed the space allowed + */ HashJoinTuple hashTuple; /* Try to load it into memory. */ @@ -1715,10 +1838,17 @@ retry: else { size_t tuple_size = MAXALIGN(HJTUPLE_OVERHEAD + tuple->t_len); + ParallelHashJoinBatch *batch; + tupleMetadata metadata; Assert(batchno > 0); /* Try to preallocate space in the batch if necessary. */ + + /* + * TODO: is it okay to only count the tuple when it doesn't fit in the + * preallocated memory? + */ if (hashtable->batches[batchno].preallocated < tuple_size) { if (!ExecParallelHashTuplePrealloc(hashtable, batchno, tuple_size)) @@ -1727,8 +1857,14 @@ retry: Assert(hashtable->batches[batchno].preallocated >= tuple_size); hashtable->batches[batchno].preallocated -= tuple_size; - sts_puttuple(hashtable->batches[batchno].inner_tuples, &hashvalue, - tuple); + batch = hashtable->batches[batchno].shared; + + metadata.hashvalue = hashvalue; + LWLockAcquire(&batch->lock, LW_SHARED); + metadata.stripe = batch->maximum_stripe_number; + LWLockRelease(&batch->lock); + + sts_puttuple(hashtable->batches[batchno].inner_tuples, &metadata, tuple); } ++hashtable->batches[batchno].ntuples; @@ -2697,6 +2833,7 @@ ExecHashAccumInstrumentation(HashInstrumentation *instrument, hashtable->nbatch_original); instrument->space_peak = Max(instrument->space_peak, hashtable->spacePeak); + instrument->fallback_batches_stats = hashtable->fallback_batches_stats; } /* @@ -2850,6 +2987,8 @@ ExecParallelHashTupleAlloc(HashJoinTable hashtable, size_t size, /* Check if it's time to grow batches or buckets. */ if (pstate->growth != PHJ_GROWTH_DISABLED) { + ParallelHashJoinBatchAccessor batch = hashtable->batches[0]; + Assert(curbatch == 0); Assert(BarrierPhase(&pstate->build_barrier) == PHJ_BUILD_HASHING_INNER); @@ -2858,8 +2997,13 @@ ExecParallelHashTupleAlloc(HashJoinTable hashtable, size_t size, * very large tuples or very low work_mem setting, we'll always allow * each backend to allocate at least one chunk. */ - if (hashtable->batches[0].at_least_one_chunk && - hashtable->batches[0].shared->size + + + /* + * TODO: get rid of this check for batch 0 and make it so that + * batch 0 always has to keep trying to increase the number of batches + */ + if (!batch.shared->hashloop_fallback && batch.at_least_one_chunk && + batch.shared->size + chunk_size > pstate->space_allowed) { pstate->growth = PHJ_GROWTH_NEED_MORE_BATCHES; @@ -2891,6 +3035,11 @@ ExecParallelHashTupleAlloc(HashJoinTable hashtable, size_t size, /* We are cleared to allocate a new chunk. */ chunk_shared = dsa_allocate(hashtable->area, chunk_size); + + /* + * TODO: if batch 0 will have stripes, need to account for this memory + * there + */ hashtable->batches[curbatch].shared->size += chunk_size; hashtable->batches[curbatch].at_least_one_chunk = true; @@ -2960,20 +3109,35 @@ ExecParallelHashJoinSetUpBatches(HashJoinTable hashtable, int nbatch) { ParallelHashJoinBatchAccessor *accessor = &hashtable->batches[i]; ParallelHashJoinBatch *shared = NthParallelHashJoinBatch(batches, i); + SharedBits *sbits = ParallelHashJoinBatchOuterBits(shared, pstate->nparticipants); char name[MAXPGPATH]; + char sbname[MAXPGPATH]; + + shared->hashloop_fallback = false; + /* TODO: is it okay to use the same tranche for this lock? */ + LWLockInitialize(&shared->lock, LWTRANCHE_PARALLEL_HASH_JOIN); + shared->maximum_stripe_number = 0; + shared->estimated_stripe_size = 0; /* * All members of shared were zero-initialized. We just need to set * up the Barrier. */ BarrierInit(&shared->batch_barrier, 0); + BarrierInit(&shared->stripe_barrier, 0); + + /* Batch 0 doesn't need to be loaded. */ if (i == 0) { - /* Batch 0 doesn't need to be loaded. */ BarrierAttach(&shared->batch_barrier); - while (BarrierPhase(&shared->batch_barrier) < PHJ_BATCH_PROBING) + while (BarrierPhase(&shared->batch_barrier) < PHJ_BATCH_STRIPING) BarrierArriveAndWait(&shared->batch_barrier, 0); BarrierDetach(&shared->batch_barrier); + + BarrierAttach(&shared->stripe_barrier); + while (BarrierPhase(&shared->stripe_barrier) < PHJ_STRIPE_PROBING) + BarrierArriveAndWait(&shared->stripe_barrier, 0); + BarrierDetach(&shared->stripe_barrier); } /* Initialize accessor state. All members were zero-initialized. */ @@ -2985,7 +3149,7 @@ ExecParallelHashJoinSetUpBatches(HashJoinTable hashtable, int nbatch) sts_initialize(ParallelHashJoinBatchInner(shared), pstate->nparticipants, ParallelWorkerNumber + 1, - sizeof(uint32), + sizeof(tupleMetadata), SHARED_TUPLESTORE_SINGLE_PASS, &pstate->fileset, name); @@ -2995,10 +3159,13 @@ ExecParallelHashJoinSetUpBatches(HashJoinTable hashtable, int nbatch) pstate->nparticipants), pstate->nparticipants, ParallelWorkerNumber + 1, - sizeof(uint32), + sizeof(tupleMetadata), SHARED_TUPLESTORE_SINGLE_PASS, &pstate->fileset, name); + snprintf(sbname, MAXPGPATH, "%s.bitmaps", name); + accessor->sba = sb_initialize(sbits, pstate->nparticipants, + ParallelWorkerNumber + 1, &pstate->sbfileset, sbname); } MemoryContextSwitchTo(oldcxt); @@ -3047,8 +3214,8 @@ ExecParallelHashEnsureBatchAccessors(HashJoinTable hashtable) * It's possible for a backend to start up very late so that the whole * join is finished and the shm state for tracking batches has already * been freed by ExecHashTableDetach(). In that case we'll just leave - * hashtable->batches as NULL so that ExecParallelHashJoinNewBatch() gives - * up early. + * hashtable->batches as NULL so that ExecParallelHashJoinAdvanceBatch() + * gives up early. */ if (!DsaPointerIsValid(pstate->batches)) return; @@ -3070,6 +3237,7 @@ ExecParallelHashEnsureBatchAccessors(HashJoinTable hashtable) { ParallelHashJoinBatchAccessor *accessor = &hashtable->batches[i]; ParallelHashJoinBatch *shared = NthParallelHashJoinBatch(batches, i); + SharedBits *sbits = ParallelHashJoinBatchOuterBits(shared, pstate->nparticipants); accessor->shared = shared; accessor->preallocated = 0; @@ -3083,6 +3251,7 @@ ExecParallelHashEnsureBatchAccessors(HashJoinTable hashtable) pstate->nparticipants), ParallelWorkerNumber + 1, &pstate->fileset); + accessor->sba = sb_attach(sbits, ParallelWorkerNumber + 1, &pstate->sbfileset); } MemoryContextSwitchTo(oldcxt); @@ -3149,6 +3318,7 @@ ExecHashTableDetachBatch(HashJoinTable hashtable) dsa_free(hashtable->area, batch->buckets); batch->buckets = InvalidDsaPointer; } + sb_end_read(hashtable->batches[curbatch].sba); } /* @@ -3165,6 +3335,18 @@ ExecHashTableDetachBatch(HashJoinTable hashtable) } } +bool +ExecHashTableDetachStripe(HashJoinTable hashtable) +{ + int curbatch = hashtable->curbatch; + ParallelHashJoinBatch *batch = hashtable->batches[curbatch].shared; + Barrier *stripe_barrier = &batch->stripe_barrier; + + BarrierDetach(stripe_barrier); + hashtable->curstripe = -1; + return false; +} + /* * Detach from all shared resources. If we are last to detach, clean up. */ @@ -3350,13 +3532,35 @@ ExecParallelHashTuplePrealloc(HashJoinTable hashtable, int batchno, size_t size) { /* * We have determined that this batch would exceed the space budget if - * loaded into memory. Command all participants to help repartition. + * loaded into memory. */ - batch->shared->space_exhausted = true; - pstate->growth = PHJ_GROWTH_NEED_MORE_BATCHES; - LWLockRelease(&pstate->lock); - - return false; + /* TODO: the nested lock is a deadlock waiting to happen. */ + LWLockAcquire(&batch->shared->lock, LW_EXCLUSIVE); + if (!batch->shared->hashloop_fallback) + { + /* + * This batch is not marked to fall back so command all + * participants to help repartition. + */ + batch->shared->space_exhausted = true; + pstate->growth = PHJ_GROWTH_NEED_MORE_BATCHES; + LWLockRelease(&batch->shared->lock); + LWLockRelease(&pstate->lock); + return false; + } + else if (batch->shared->estimated_stripe_size + want + + HASH_CHUNK_HEADER_SIZE > pstate->space_allowed) + { + /* + * This batch is marked to fall back and the current (last) stripe + * does not have enough space to handle the request so we must + * increment the number of stripes in the batch and reset the size + * of its new last stripe. + */ + batch->shared->maximum_stripe_number++; + batch->shared->estimated_stripe_size = 0; + } + LWLockRelease(&batch->shared->lock); } batch->at_least_one_chunk = true; diff --git a/src/backend/executor/nodeHashjoin.c b/src/backend/executor/nodeHashjoin.c index cc8edacdd0..516067f176 100644 --- a/src/backend/executor/nodeHashjoin.c +++ b/src/backend/executor/nodeHashjoin.c @@ -126,7 +126,7 @@ #define HJ_SCAN_BUCKET 3 #define HJ_FILL_OUTER_TUPLE 4 #define HJ_FILL_INNER_TUPLES 5 -#define HJ_NEED_NEW_BATCH 6 +#define HJ_NEED_NEW_STRIPE 6 /* Returns true if doing null-fill on outer relation */ #define HJ_FILL_OUTER(hjstate) ((hjstate)->hj_NullInnerTupleSlot != NULL) @@ -143,10 +143,91 @@ static TupleTableSlot *ExecHashJoinGetSavedTuple(HashJoinState *hjstate, BufFile *file, uint32 *hashvalue, TupleTableSlot *tupleSlot); +static int ExecHashJoinLoadStripe(HashJoinState *hjstate); static bool ExecHashJoinNewBatch(HashJoinState *hjstate); static bool ExecParallelHashJoinNewBatch(HashJoinState *hjstate); +static bool ExecParallelHashJoinLoadStripe(HashJoinState *hjstate); static void ExecParallelHashJoinPartitionOuter(HashJoinState *node); +static bool checkbit(HashJoinState *hjstate); +static void set_match_bit(HashJoinState *hjstate); +static pg_attribute_always_inline bool + IsHashloopFallback(HashJoinTable hashtable); + +#define UINT_BITS (sizeof(unsigned int) * CHAR_BIT) + +static void +set_match_bit(HashJoinState *hjstate) +{ + HashJoinTable hashtable = hjstate->hj_HashTable; + BufFile *statusFile = hashtable->hashloop_fallback[hashtable->curbatch]; + int tupindex = hjstate->hj_CurNumOuterTuples - 1; + size_t unit_size = sizeof(hjstate->hj_CurOuterMatchStatus); + off_t offset = tupindex / UINT_BITS * unit_size; + + int fileno; + off_t cursor; + + BufFileTell(statusFile, &fileno, &cursor); + + /* Extend the statusFile if this is stripe zero. */ + if (hashtable->curstripe == 0) + { + for (; cursor < offset + unit_size; cursor += unit_size) + { + hjstate->hj_CurOuterMatchStatus = 0; + BufFileWrite(statusFile, &hjstate->hj_CurOuterMatchStatus, unit_size); + } + } + + if (cursor != offset) + BufFileSeek(statusFile, 0, offset, SEEK_SET); + + BufFileRead(statusFile, &hjstate->hj_CurOuterMatchStatus, unit_size); + BufFileSeek(statusFile, 0, -unit_size, SEEK_CUR); + + hjstate->hj_CurOuterMatchStatus |= 1U << tupindex % UINT_BITS; + BufFileWrite(statusFile, &hjstate->hj_CurOuterMatchStatus, unit_size); +} + +/* return true if bit is set and false if not */ +static bool +checkbit(HashJoinState *hjstate) +{ + HashJoinTable hashtable = hjstate->hj_HashTable; + int curbatch = hashtable->curbatch; + BufFile *outer_match_statuses; + + int bitno = hjstate->hj_EmitOuterTupleId % UINT_BITS; + + hjstate->hj_EmitOuterTupleId++; + outer_match_statuses = hjstate->hj_HashTable->hashloop_fallback[curbatch]; + + /* + * if current chunk of bitmap is exhausted, read next chunk of bitmap from + * outer_match_status_file + */ + if (bitno == 0) + BufFileRead(outer_match_statuses, &hjstate->hj_CurOuterMatchStatus, + sizeof(hjstate->hj_CurOuterMatchStatus)); + + /* + * check if current tuple's match bit is set in outer match status file + */ + return hjstate->hj_CurOuterMatchStatus & (1U << bitno); +} + +static bool +IsHashloopFallback(HashJoinTable hashtable) +{ + if (hashtable->parallel_state) + return hashtable->batches[hashtable->curbatch].shared->hashloop_fallback; + + if (!hashtable->hashloop_fallback) + return false; + + return hashtable->hashloop_fallback[hashtable->curbatch]; +} /* ---------------------------------------------------------------- * ExecHashJoinImpl @@ -290,6 +371,12 @@ ExecHashJoinImpl(PlanState *pstate, bool parallel) hashNode->hashtable = hashtable; (void) MultiExecProcNode((PlanState *) hashNode); + /* + * After building the hashtable, stripe 0 of batch 0 will have + * been loaded. + */ + hashtable->curstripe = 0; + /* * If the inner relation is completely empty, and we're not * doing a left outer join, we can quit without scanning the @@ -333,12 +420,11 @@ ExecHashJoinImpl(PlanState *pstate, bool parallel) /* Each backend should now select a batch to work on. */ hashtable->curbatch = -1; - node->hj_JoinState = HJ_NEED_NEW_BATCH; - continue; + if (!ExecParallelHashJoinNewBatch(node)) + return NULL; } - else - node->hj_JoinState = HJ_NEED_NEW_OUTER; + node->hj_JoinState = HJ_NEED_NEW_OUTER; /* FALL THRU */ @@ -365,12 +451,18 @@ ExecHashJoinImpl(PlanState *pstate, bool parallel) node->hj_JoinState = HJ_FILL_INNER_TUPLES; } else - node->hj_JoinState = HJ_NEED_NEW_BATCH; + node->hj_JoinState = HJ_NEED_NEW_STRIPE; continue; } econtext->ecxt_outertuple = outerTupleSlot; - node->hj_MatchedOuter = false; + + /* + * Don't reset hj_MatchedOuter after the first stripe as it + * would cancel out whatever we found before + */ + if (node->hj_HashTable->curstripe == 0) + node->hj_MatchedOuter = false; /* * Find the corresponding bucket for this tuple in the main @@ -386,9 +478,15 @@ ExecHashJoinImpl(PlanState *pstate, bool parallel) /* * The tuple might not belong to the current batch (where * "current batch" includes the skew buckets if any). + * + * This should only be done once per tuple per batch. If a + * batch "falls back", its inner side will be split into + * stripes. Any displaced outer tuples should only be + * relocated while probing the first stripe of the inner side. */ if (batchno != hashtable->curbatch && - node->hj_CurSkewBucketNo == INVALID_SKEW_BUCKET_NO) + node->hj_CurSkewBucketNo == INVALID_SKEW_BUCKET_NO && + node->hj_HashTable->curstripe == 0) { bool shouldFree; MinimalTuple mintuple = ExecFetchSlotMinimalTuple(outerTupleSlot, @@ -410,6 +508,13 @@ ExecHashJoinImpl(PlanState *pstate, bool parallel) continue; } + /* + * While probing the phantom stripe, don't increment + * hj_CurNumOuterTuples or extend the bitmap + */ + if (!parallel && hashtable->curstripe != -2) + node->hj_CurNumOuterTuples++; + /* OK, let's scan the bucket for matches */ node->hj_JoinState = HJ_SCAN_BUCKET; @@ -455,6 +560,14 @@ ExecHashJoinImpl(PlanState *pstate, bool parallel) { node->hj_MatchedOuter = true; + if (HJ_FILL_OUTER(node) && IsHashloopFallback(hashtable)) + { + if (parallel) + sb_setbit(hashtable->batches[hashtable->curbatch].sba, econtext->ecxt_outertuple->tts_tuplenum); + else + set_match_bit(node); + } + if (parallel) { /* @@ -508,6 +621,22 @@ ExecHashJoinImpl(PlanState *pstate, bool parallel) */ node->hj_JoinState = HJ_NEED_NEW_OUTER; + if (IsHashloopFallback(hashtable) && HJ_FILL_OUTER(node)) + { + if (hashtable->curstripe != -2) + continue; + + if (parallel) + { + ParallelHashJoinBatchAccessor *accessor = + &node->hj_HashTable->batches[node->hj_HashTable->curbatch]; + + node->hj_MatchedOuter = sb_checkbit(accessor->sba, econtext->ecxt_outertuple->tts_tuplenum); + } + else + node->hj_MatchedOuter = checkbit(node); + } + if (!node->hj_MatchedOuter && HJ_FILL_OUTER(node)) { @@ -534,7 +663,7 @@ ExecHashJoinImpl(PlanState *pstate, bool parallel) if (!ExecScanHashTableForUnmatched(node, econtext)) { /* no more unmatched tuples */ - node->hj_JoinState = HJ_NEED_NEW_BATCH; + node->hj_JoinState = HJ_NEED_NEW_STRIPE; continue; } @@ -550,19 +679,23 @@ ExecHashJoinImpl(PlanState *pstate, bool parallel) InstrCountFiltered2(node, 1); break; - case HJ_NEED_NEW_BATCH: + case HJ_NEED_NEW_STRIPE: /* - * Try to advance to next batch. Done if there are no more. + * Try to advance to next stripe. Then try to advance to the + * next batch if there are no more stripes in this batch. Done + * if there are no more batches. */ if (parallel) { - if (!ExecParallelHashJoinNewBatch(node)) + if (!ExecParallelHashJoinLoadStripe(node) && + !ExecParallelHashJoinNewBatch(node)) return NULL; /* end of parallel-aware join */ } else { - if (!ExecHashJoinNewBatch(node)) + if (!ExecHashJoinLoadStripe(node) && + !ExecHashJoinNewBatch(node)) return NULL; /* end of parallel-oblivious join */ } node->hj_JoinState = HJ_NEED_NEW_OUTER; @@ -751,6 +884,8 @@ ExecInitHashJoin(HashJoin *node, EState *estate, int eflags) hjstate->hj_JoinState = HJ_BUILD_HASHTABLE; hjstate->hj_MatchedOuter = false; hjstate->hj_OuterNotEmpty = false; + hjstate->hj_CurNumOuterTuples = 0; + hjstate->hj_CurOuterMatchStatus = 0; return hjstate; } @@ -917,15 +1052,24 @@ ExecParallelHashJoinOuterGetTuple(PlanState *outerNode, } else if (curbatch < hashtable->nbatch) { + tupleMetadata metadata; MinimalTuple tuple; tuple = sts_parallel_scan_next(hashtable->batches[curbatch].outer_tuples, - hashvalue); + &metadata); + *hashvalue = metadata.hashvalue; + if (tuple != NULL) { ExecForceStoreMinimalTuple(tuple, hjstate->hj_OuterTupleSlot, false); + + /* + * TODO: should we use tupleid instead of position in the serial + * case too? + */ + hjstate->hj_OuterTupleSlot->tts_tuplenum = metadata.tupleid; slot = hjstate->hj_OuterTupleSlot; return slot; } @@ -949,24 +1093,37 @@ ExecHashJoinNewBatch(HashJoinState *hjstate) HashJoinTable hashtable = hjstate->hj_HashTable; int nbatch; int curbatch; - BufFile *innerFile; - TupleTableSlot *slot; - uint32 hashvalue; + BufFile *innerFile = NULL; + BufFile *outerFile = NULL; nbatch = hashtable->nbatch; curbatch = hashtable->curbatch; - if (curbatch > 0) + /* + * We no longer need the previous outer batch file; close it right away to + * free disk space. + */ + if (hashtable->outerBatchFile && hashtable->outerBatchFile[curbatch]) { - /* - * We no longer need the previous outer batch file; close it right - * away to free disk space. - */ - if (hashtable->outerBatchFile[curbatch]) - BufFileClose(hashtable->outerBatchFile[curbatch]); + BufFileClose(hashtable->outerBatchFile[curbatch]); hashtable->outerBatchFile[curbatch] = NULL; } - else /* we just finished the first batch */ + if (IsHashloopFallback(hashtable)) + { + BufFileClose(hashtable->hashloop_fallback[curbatch]); + hashtable->hashloop_fallback[curbatch] = NULL; + } + + /* + * We are surely done with the inner batch file now + */ + if (hashtable->innerBatchFile && hashtable->innerBatchFile[curbatch]) + { + BufFileClose(hashtable->innerBatchFile[curbatch]); + hashtable->innerBatchFile[curbatch] = NULL; + } + + if (curbatch == 0) /* we just finished the first batch */ { /* * Reset some of the skew optimization state variables, since we no @@ -1030,45 +1187,68 @@ ExecHashJoinNewBatch(HashJoinState *hjstate) return false; /* no more batches */ hashtable->curbatch = curbatch; + hashtable->curstripe = -1; + hjstate->hj_CurNumOuterTuples = 0; - /* - * Reload the hash table with the new inner batch (which could be empty) - */ - ExecHashTableReset(hashtable); + if (hashtable->innerBatchFile && hashtable->innerBatchFile[curbatch]) + innerFile = hashtable->innerBatchFile[curbatch]; + + if (innerFile && BufFileSeek(innerFile, 0, 0L, SEEK_SET)) + ereport(ERROR, + (errcode_for_file_access(), + errmsg("could not rewind hash-join temporary file: %m"))); + + /* Need to rewind outer when this is the first stripe of a new batch */ + if (hashtable->outerBatchFile && hashtable->outerBatchFile[curbatch]) + outerFile = hashtable->outerBatchFile[curbatch]; + + if (outerFile && BufFileSeek(outerFile, 0, 0L, SEEK_SET)) + ereport(ERROR, + (errcode_for_file_access(), + errmsg("could not rewind hash-join temporary file: %m"))); + + ExecHashJoinLoadStripe(hjstate); + return true; +} - innerFile = hashtable->innerBatchFile[curbatch]; +static inline void +InstrIncrBatchStripes(List *fallback_batches_stats, int curbatch) +{ + ListCell *lc; - if (innerFile != NULL) + foreach(lc, fallback_batches_stats) { - if (BufFileSeek(innerFile, 0, 0L, SEEK_SET)) - ereport(ERROR, - (errcode_for_file_access(), - errmsg("could not rewind hash-join temporary file: %m"))); + FallbackBatchStats *fallback_batch_stats = lfirst(lc); - while ((slot = ExecHashJoinGetSavedTuple(hjstate, - innerFile, - &hashvalue, - hjstate->hj_HashTupleSlot))) + if (fallback_batch_stats->batchno == curbatch) { - /* - * NOTE: some tuples may be sent to future batches. Also, it is - * possible for hashtable->nbatch to be increased here! - */ - ExecHashTableInsert(hashtable, slot, hashvalue); + fallback_batch_stats->numstripes++; + break; } - - /* - * after we build the hash table, the inner batch file is no longer - * needed - */ - BufFileClose(innerFile); - hashtable->innerBatchFile[curbatch] = NULL; } +} + +/* + * Returns false when the inner batch file is exhausted + */ +static int +ExecHashJoinLoadStripe(HashJoinState *hjstate) +{ + HashJoinTable hashtable = hjstate->hj_HashTable; + int curbatch = hashtable->curbatch; + TupleTableSlot *slot; + uint32 hashvalue; + bool loaded_inner = false; + + if (hashtable->curstripe == -2) + return false; /* * Rewind outer batch file (if present), so that we can start reading it. + * TODO: This is only necessary if this is not the first stripe of the + * batch */ - if (hashtable->outerBatchFile[curbatch] != NULL) + if (hashtable->outerBatchFile && hashtable->outerBatchFile[curbatch]) { if (BufFileSeek(hashtable->outerBatchFile[curbatch], 0, 0L, SEEK_SET)) ereport(ERROR, @@ -1076,9 +1256,78 @@ ExecHashJoinNewBatch(HashJoinState *hjstate) errmsg("could not rewind hash-join temporary file: %m"))); } - return true; + hashtable->curstripe++; + + if (!hashtable->innerBatchFile || !hashtable->innerBatchFile[curbatch]) + return false; + + /* + * Reload the hash table with the new inner stripe + */ + ExecHashTableReset(hashtable); + + while ((slot = ExecHashJoinGetSavedTuple(hjstate, + hashtable->innerBatchFile[curbatch], + &hashvalue, + hjstate->hj_HashTupleSlot))) + { + /* + * NOTE: some tuples may be sent to future batches. Also, it is + * possible for hashtable->nbatch to be increased here! + */ + uint32 hashTupleSize; + /* + * TODO: wouldn't it be cool if this returned the size of the tuple + * inserted + */ + ExecHashTableInsert(hashtable, slot, hashvalue); + loaded_inner = true; + + if (!IsHashloopFallback(hashtable)) + continue; + + hashTupleSize = slot->tts_ops->get_minimal_tuple(slot)->t_len + HJTUPLE_OVERHEAD; + + if (hashtable->spaceUsed + hashTupleSize + + hashtable->nbuckets_optimal * sizeof(HashJoinTuple) + > hashtable->spaceAllowed) + break; + } + + /* + * if we didn't load anything and it is a FOJ/LOJ fallback batch, we will + * transition to emit unmatched outer tuples next. we want to know how + * many tuples were in the batch in that case, so don't zero it out then + */ + + /* + * if we loaded anything into the hashtable or it is the phantom stripe, + * must proceed to probing + */ + if (loaded_inner) + { + hjstate->hj_CurNumOuterTuples = 0; + InstrIncrBatchStripes(hashtable->fallback_batches_stats, curbatch); + return true; + } + + if (IsHashloopFallback(hashtable) && HJ_FILL_OUTER(hjstate)) + { + /* + * if we didn't load anything and it is a fallback batch, we will + * prepare to emit outer tuples during the phantom stripe probing + */ + hashtable->curstripe = -2; + hjstate->hj_EmitOuterTupleId = 0; + hjstate->hj_CurOuterMatchStatus = 0; + BufFileSeek(hashtable->hashloop_fallback[curbatch], 0, 0, SEEK_SET); + BufFileSeek(hashtable->outerBatchFile[curbatch], 0, 0L, SEEK_SET); + return true; + } + return false; } + /* * Choose a batch to work on, and attach to it. Returns true if successful, * false if there are no more batches. @@ -1101,10 +1350,18 @@ ExecParallelHashJoinNewBatch(HashJoinState *hjstate) /* * If we were already attached to a batch, remember not to bother checking * it again, and detach from it (possibly freeing the hash table if we are - * last to detach). + * last to detach). curbatch is set when the batch_barrier phase is either + * PHJ_BATCH_LOADING or PHJ_BATCH_STRIPING (note that the + * PHJ_BATCH_LOADING case will fall through to the PHJ_BATCH_STRIPING + * case). The PHJ_BATCH_STRIPING case returns to the caller. So when this + * function is reentered with a curbatch >= 0 then we must be done + * probing. */ + if (hashtable->curbatch >= 0) { + if (IsHashloopFallback(hashtable)) + sb_end_write(hashtable->batches[hashtable->curbatch].sba); hashtable->batches[hashtable->curbatch].done = true; ExecHashTableDetachBatch(hashtable); } @@ -1119,13 +1376,8 @@ ExecParallelHashJoinNewBatch(HashJoinState *hjstate) hashtable->nbatch; do { - uint32 hashvalue; - MinimalTuple tuple; - TupleTableSlot *slot; - if (!hashtable->batches[batchno].done) { - SharedTuplestoreAccessor *inner_tuples; Barrier *batch_barrier = &hashtable->batches[batchno].shared->batch_barrier; @@ -1136,7 +1388,15 @@ ExecParallelHashJoinNewBatch(HashJoinState *hjstate) /* One backend allocates the hash table. */ if (BarrierArriveAndWait(batch_barrier, WAIT_EVENT_HASH_BATCH_ELECTING)) + { ExecParallelHashTableAlloc(hashtable, batchno); + + /* + * one worker needs to 0 out the read_pages of all the + * participants in the new batch + */ + sts_reinitialize(hashtable->batches[batchno].inner_tuples); + } /* Fall through. */ case PHJ_BATCH_ALLOCATING: @@ -1145,40 +1405,15 @@ ExecParallelHashJoinNewBatch(HashJoinState *hjstate) WAIT_EVENT_HASH_BATCH_ALLOCATING); /* Fall through. */ - case PHJ_BATCH_LOADING: - /* Start (or join in) loading tuples. */ - ExecParallelHashTableSetCurrentBatch(hashtable, batchno); - inner_tuples = hashtable->batches[batchno].inner_tuples; - sts_begin_parallel_scan(inner_tuples); - while ((tuple = sts_parallel_scan_next(inner_tuples, - &hashvalue))) - { - ExecForceStoreMinimalTuple(tuple, - hjstate->hj_HashTupleSlot, - false); - slot = hjstate->hj_HashTupleSlot; - ExecParallelHashTableInsertCurrentBatch(hashtable, slot, - hashvalue); - } - sts_end_parallel_scan(inner_tuples); - BarrierArriveAndWait(batch_barrier, - WAIT_EVENT_HASH_BATCH_LOADING); - /* Fall through. */ - - case PHJ_BATCH_PROBING: + case PHJ_BATCH_STRIPING: - /* - * This batch is ready to probe. Return control to - * caller. We stay attached to batch_barrier so that the - * hash table stays alive until everyone's finished - * probing it, but no participant is allowed to wait at - * this barrier again (or else a deadlock could occur). - * All attached participants must eventually call - * BarrierArriveAndDetach() so that the final phase - * PHJ_BATCH_DONE can be reached. - */ ExecParallelHashTableSetCurrentBatch(hashtable, batchno); - sts_begin_parallel_scan(hashtable->batches[batchno].outer_tuples); + sts_begin_parallel_scan(hashtable->batches[batchno].inner_tuples); + if (hashtable->batches[batchno].shared->hashloop_fallback) + sb_initialize_accessor(hashtable->batches[hashtable->curbatch].sba, + sts_get_tuplenum(hashtable->batches[hashtable->curbatch].outer_tuples)); + hashtable->curstripe = -1; + ExecParallelHashJoinLoadStripe(hjstate); return true; case PHJ_BATCH_DONE: @@ -1203,6 +1438,220 @@ ExecParallelHashJoinNewBatch(HashJoinState *hjstate) return false; } + + +/* + * Returns true if ready to probe and false if the inner is exhausted + * (there are no more stripes) + */ +bool +ExecParallelHashJoinLoadStripe(HashJoinState *hjstate) +{ + HashJoinTable hashtable = hjstate->hj_HashTable; + int batchno = hashtable->curbatch; + ParallelHashJoinBatch *batch = hashtable->batches[batchno].shared; + Barrier *stripe_barrier = &batch->stripe_barrier; + SharedTuplestoreAccessor *outer_tuples; + SharedTuplestoreAccessor *inner_tuples; + ParallelHashJoinBatchAccessor *accessor; + dsa_pointer_atomic *buckets; + + outer_tuples = hashtable->batches[batchno].outer_tuples; + inner_tuples = hashtable->batches[batchno].inner_tuples; + + if (hashtable->curstripe >= 0) + { + BarrierArriveAndWait(stripe_barrier, WAIT_EVENT_HASH_STRIPE_PROBING); + } + else if (hashtable->curstripe == -1) + { + int phase = BarrierAttach(stripe_barrier); + + /* + * If a worker enters this phase machine on a stripe number greater + * than the batch's maximum stripe number, then: 1) The batch is done, + * or 2) The batch is on the phantom stripe that's used for hashloop + * fallback Either way the worker can't contribute so just detach and + * move on. + */ + if (PHJ_STRIPE_NUMBER(phase) > batch->maximum_stripe_number) + return ExecHashTableDetachStripe(hashtable); + + hashtable->curstripe = PHJ_STRIPE_NUMBER(phase); + } + else if (hashtable->curstripe == -2) + { + sts_end_parallel_scan(outer_tuples); + sb_end_read(hashtable->batches[batchno].sba); + return ExecHashTableDetachStripe(hashtable); + } + + /* + * The outer side is exhausted and either 1) the current stripe of the + * inner side is exhausted and it is time to advance the stripe 2) the + * last stripe of the inner side is exhausted and it is time to advance + * the batch + */ + for (;;) + { + int phase = BarrierPhase(stripe_barrier); + + switch (PHJ_STRIPE_PHASE(phase)) + { + case PHJ_STRIPE_ELECTING: + if (BarrierArriveAndWait(stripe_barrier, WAIT_EVENT_HASH_STRIPE_ELECTING)) + { + sts_reinitialize(outer_tuples); + + /* + * set the rewound flag back to false to prepare for the + * next stripe + */ + sts_reset_rewound(inner_tuples); + } + + /* Fall through. */ + + case PHJ_STRIPE_RESETTING: + /* TODO: not needed for phantom stripe */ + BarrierArriveAndWait(stripe_barrier, WAIT_EVENT_HASH_STRIPE_RESETTING); + + case PHJ_STRIPE_LOADING: + { + MinimalTuple tuple; + tupleMetadata metadata; + + /* + * Start (or join in) loading the next stripe of inner + * tuples. + */ + + /* + * I'm afraid there potential issue if a worker joins in + * this phase and doesn't do the actions and resetting of + * variables in sts_resume_parallel_scan. that is, if it + * doesn't reset start_page and read_next_page in between + * stripes. For now, call it. However, I think it might be + * able to be removed. + */ + + /* + * TODO: sts_resume_parallel_scan() is overkill for stripe + * 0 of each batch + */ + sts_resume_parallel_scan(inner_tuples); + + while ((tuple = sts_parallel_scan_next(inner_tuples, &metadata))) + { + /* The tuple is from a previous stripe. Skip it */ + if (metadata.stripe < PHJ_STRIPE_NUMBER(phase)) + continue; + + /* + * tuple from future. time to back out read_page. end + * of stripe + */ + if (metadata.stripe > PHJ_STRIPE_NUMBER(phase)) + { + sts_parallel_scan_rewind(inner_tuples); + continue; + } + + ExecForceStoreMinimalTuple(tuple, hjstate->hj_HashTupleSlot, false); + ExecParallelHashTableInsertCurrentBatch( + hashtable, + hjstate->hj_HashTupleSlot, + metadata.hashvalue); + } + BarrierArriveAndWait(stripe_barrier, WAIT_EVENT_HASH_STRIPE_LOADING); + /* Fall through. */ + } + + case PHJ_STRIPE_PROBING: + + /* + * do this again here in case a worker began the scan and then + * entered after loading before probing + */ + sts_end_parallel_scan(inner_tuples); + sts_begin_parallel_scan(outer_tuples); + return true; + + case PHJ_STRIPE_DONE: + + if (PHJ_STRIPE_NUMBER(phase) >= batch->maximum_stripe_number) + { + /* + * Handle the phantom stripe case. + */ + if (batch->hashloop_fallback && HJ_FILL_OUTER(hjstate)) + goto fallback_stripe; + + /* Return if this is the last stripe */ + return ExecHashTableDetachStripe(hashtable); + } + + /* this, effectively, increments the stripe number */ + if (BarrierArriveAndWait(stripe_barrier, WAIT_EVENT_HASH_STRIPE_LOADING)) + { + /* + * reset inner's hashtable and recycle the existing bucket array. + */ + buckets = (dsa_pointer_atomic *) + dsa_get_address(hashtable->area, batch->buckets); + + for (size_t i = 0; i < hashtable->nbuckets; ++i) + dsa_pointer_atomic_write(&buckets[i], InvalidDsaPointer); + } + + hashtable->curstripe++; + continue; + + default: + elog(ERROR, "unexpected stripe phase %d. pid %i. batch %i.", BarrierPhase(stripe_barrier), MyProcPid, batchno); + } + } + +fallback_stripe: + accessor = &hashtable->batches[hashtable->curbatch]; + sb_end_write(accessor->sba); + + /* Ensure that only a single worker is attached to the barrier */ + if (!BarrierArriveAndWait(stripe_barrier, WAIT_EVENT_HASH_STRIPE_LOADING)) + return ExecHashTableDetachStripe(hashtable); + + + /* No one except the last worker will run this code */ + hashtable->curstripe = -2; + + /* + * reset inner's hashtable and recycle the existing bucket array. + */ + buckets = (dsa_pointer_atomic *) + dsa_get_address(hashtable->area, batch->buckets); + + for (size_t i = 0; i < hashtable->nbuckets; ++i) + dsa_pointer_atomic_write(&buckets[i], InvalidDsaPointer); + + /* + * If all workers (including this one) have finished probing the batch, + * one worker is elected to Loop through the outer match status files from + * all workers that were attached to this batch Combine them into one + * bitmap Use the bitmap, loop through the outer batch file again, and + * emit unmatched tuples All workers will detach from the batch barrier + * and the last worker will clean up the hashtable. All workers except the + * last worker will end their scans of the outer and inner side. The last + * worker will end its scan of the inner side + */ + + sb_combine(accessor->sba); + sts_reinitialize(outer_tuples); + + sts_begin_parallel_scan(outer_tuples); + + return true; +} + /* * ExecHashJoinSaveTuple * save a tuple to a batch file. @@ -1372,6 +1821,9 @@ ExecReScanHashJoin(HashJoinState *node) node->hj_MatchedOuter = false; node->hj_FirstOuterTupleSlot = NULL; + node->hj_CurNumOuterTuples = 0; + node->hj_CurOuterMatchStatus = 0; + /* * if chgParam of subnode is not null then plan will be re-scanned by * first ExecProcNode. @@ -1402,7 +1854,6 @@ ExecParallelHashJoinPartitionOuter(HashJoinState *hjstate) ExprContext *econtext = hjstate->js.ps.ps_ExprContext; HashJoinTable hashtable = hjstate->hj_HashTable; TupleTableSlot *slot; - uint32 hashvalue; int i; Assert(hjstate->hj_FirstOuterTupleSlot == NULL); @@ -1410,6 +1861,8 @@ ExecParallelHashJoinPartitionOuter(HashJoinState *hjstate) /* Execute outer plan, writing all tuples to shared tuplestores. */ for (;;) { + tupleMetadata metadata; + slot = ExecProcNode(outerState); if (TupIsNull(slot)) break; @@ -1418,17 +1871,23 @@ ExecParallelHashJoinPartitionOuter(HashJoinState *hjstate) hjstate->hj_OuterHashKeys, true, /* outer tuple */ HJ_FILL_OUTER(hjstate), - &hashvalue)) + &metadata.hashvalue)) { int batchno; int bucketno; bool shouldFree; + SharedTuplestoreAccessor *accessor; + MinimalTuple mintup = ExecFetchSlotMinimalTuple(slot, &shouldFree); - ExecHashGetBucketAndBatch(hashtable, hashvalue, &bucketno, + ExecHashGetBucketAndBatch(hashtable, metadata.hashvalue, &bucketno, &batchno); - sts_puttuple(hashtable->batches[batchno].outer_tuples, - &hashvalue, mintup); + accessor = hashtable->batches[batchno].outer_tuples; + + /* cannot count on deterministic order of tupleids */ + metadata.tupleid = sts_increment_ntuples(accessor); + + sts_puttuple(hashtable->batches[batchno].outer_tuples, &metadata.hashvalue, mintup); if (shouldFree) heap_free_minimal_tuple(mintup); @@ -1494,6 +1953,7 @@ ExecHashJoinInitializeDSM(HashJoinState *state, ParallelContext *pcxt) /* Set up the space we'll use for shared temporary files. */ SharedFileSetInit(&pstate->fileset, pcxt->seg); + SharedFileSetInit(&pstate->sbfileset, pcxt->seg); /* Initialize the shared state in the hash node. */ hashNode = (HashState *) innerPlanState(state); diff --git a/src/backend/postmaster/pgstat.c b/src/backend/postmaster/pgstat.c index 50eea2e8a8..02ca9654ec 100644 --- a/src/backend/postmaster/pgstat.c +++ b/src/backend/postmaster/pgstat.c @@ -3780,8 +3780,17 @@ pgstat_get_wait_ipc(WaitEventIPC w) case WAIT_EVENT_HASH_BATCH_ELECTING: event_name = "Hash/Batch/Electing"; break; - case WAIT_EVENT_HASH_BATCH_LOADING: - event_name = "Hash/Batch/Loading"; + case WAIT_EVENT_HASH_STRIPE_ELECTING: + event_name = "Hash/Stripe/Electing"; + break; + case WAIT_EVENT_HASH_STRIPE_RESETTING: + event_name = "Hash/Stripe/RESETTING"; + break; + case WAIT_EVENT_HASH_STRIPE_LOADING: + event_name = "Hash/Stripe/Loading"; + break; + case WAIT_EVENT_HASH_STRIPE_PROBING: + event_name = "Hash/Stripe/Probing"; break; case WAIT_EVENT_HASH_BUILD_ALLOCATING: event_name = "Hash/Build/Allocating"; diff --git a/src/backend/utils/sort/Makefile b/src/backend/utils/sort/Makefile index 7ac3659261..f11fe85aeb 100644 --- a/src/backend/utils/sort/Makefile +++ b/src/backend/utils/sort/Makefile @@ -16,6 +16,7 @@ override CPPFLAGS := -I. -I$(srcdir) $(CPPFLAGS) OBJS = \ logtape.o \ + sharedbits.o \ sharedtuplestore.o \ sortsupport.o \ tuplesort.o \ diff --git a/src/backend/utils/sort/sharedbits.c b/src/backend/utils/sort/sharedbits.c new file mode 100644 index 0000000000..37df04844e --- /dev/null +++ b/src/backend/utils/sort/sharedbits.c @@ -0,0 +1,285 @@ +#include "postgres.h" +#include "storage/buffile.h" +#include "utils/sharedbits.h" + +/* + * TODO: put a comment about not currently supporting parallel scan of the SharedBits + * To support parallel scan, need to introduce many more mechanisms + */ + +/* Per-participant shared state */ +struct SharedBitsParticipant +{ + bool present; + bool writing; +}; + +/* Shared control object */ +struct SharedBits +{ + int nparticipants; /* Number of participants that can write. */ + int64 nbits; + char name[NAMEDATALEN]; /* A name for this bitstore. */ + + SharedBitsParticipant participants[FLEXIBLE_ARRAY_MEMBER]; +}; + +/* backend-local state */ +struct SharedBitsAccessor +{ + int participant; + SharedBits *bits; + SharedFileSet *fileset; + BufFile *write_file; + BufFile *combined; +}; + +SharedBitsAccessor * +sb_attach(SharedBits *sbits, int my_participant_number, SharedFileSet *fileset) +{ + SharedBitsAccessor *accessor = palloc0(sizeof(SharedBitsAccessor)); + + accessor->participant = my_participant_number; + accessor->bits = sbits; + accessor->fileset = fileset; + accessor->write_file = NULL; + accessor->combined = NULL; + return accessor; +} + +SharedBitsAccessor * +sb_initialize(SharedBits *sbits, + int participants, + int my_participant_number, + SharedFileSet *fileset, + char *name) +{ + SharedBitsAccessor *accessor; + + sbits->nparticipants = participants; + strcpy(sbits->name, name); + sbits->nbits = 0; /* TODO: maybe delete this */ + + accessor = palloc0(sizeof(SharedBitsAccessor)); + accessor->participant = my_participant_number; + accessor->bits = sbits; + accessor->fileset = fileset; + accessor->write_file = NULL; + accessor->combined = NULL; + return accessor; +} + +/* TODO: is "initialize_accessor" a clear enough API for this? (making the file)? */ +void +sb_initialize_accessor(SharedBitsAccessor *accessor, uint32 nbits) +{ + char name[MAXPGPATH]; + uint32 num_to_write; + + snprintf(name, MAXPGPATH, "%s.p%d.bitmap", accessor->bits->name, accessor->participant); + + accessor->write_file = + BufFileCreateShared(accessor->fileset, name); + + accessor->bits->participants[accessor->participant].present = true; + /* TODO: check this math. tuplenumber will be too high? */ + num_to_write = nbits / 8 + 1; + + /* + * TODO: add tests that could exercise a problem with junk being written + * to bitmap + */ + + /* + * TODO: is there a better way to write the bytes to the file without + * calling BufFileWrite() like this? palloc()ing an undetermined number of + * bytes feels like it is against the spirit of this patch to begin with, + * but the many function calls seem expensive + */ + for (int i = 0; i < num_to_write; i++) + { + unsigned char byteToWrite = 0; + + BufFileWrite(accessor->write_file, &byteToWrite, 1); + } + + if (BufFileSeek(accessor->write_file, 0, 0L, SEEK_SET)) + ereport(ERROR, + (errcode_for_file_access(), + errmsg("could not rewind hash-join temporary file: %m"))); +} + +size_t +sb_estimate(int participants) +{ + return offsetof(SharedBits, participants) + participants * sizeof(SharedBitsParticipant); +} + + +void +sb_setbit(SharedBitsAccessor *accessor, uint64 bit) +{ + SharedBitsParticipant *const participant = + &accessor->bits->participants[accessor->participant]; + + /* TODO: use an unsigned int instead of a byte */ + unsigned char current_outer_byte; + + Assert(accessor->write_file); + + if (!participant->writing) + { + participant->writing = true; + } + + BufFileSeek(accessor->write_file, 0, bit / 8, SEEK_SET); + BufFileRead(accessor->write_file, ¤t_outer_byte, 1); + + current_outer_byte |= 1U << (bit % 8); + + BufFileSeek(accessor->write_file, 0, -1, SEEK_CUR); + BufFileWrite(accessor->write_file, ¤t_outer_byte, 1); +} + +bool +sb_checkbit(SharedBitsAccessor *accessor, uint32 n) +{ + bool match; + uint32 bytenum = n / 8; + unsigned char bit = n % 8; + unsigned char byte_to_check = 0; + + Assert(accessor->combined); + + /* seek to byte to check */ + if (BufFileSeek(accessor->combined, + 0, + bytenum, + SEEK_SET)) + ereport(ERROR, + (errcode_for_file_access(), + errmsg( + "could not rewind shared outer temporary file: %m"))); + /* read byte containing ntuple bit */ + if (BufFileRead(accessor->combined, &byte_to_check, 1) == 0) + ereport(ERROR, + (errcode_for_file_access(), + errmsg( + "could not read byte in outer match status bitmap: %m."))); + /* if bit is set */ + match = ((byte_to_check) >> bit) & 1; + + return match; +} + +BufFile * +sb_combine(SharedBitsAccessor *accessor) +{ + /* + * TODO: this tries to close an outer match status file for each + * participant in the tuplestore. technically, only participants in the + * barrier could have outer match status files, however, all but one + * participant continue on and detach from the barrier so we won't have a + * reliable way to close only files for those attached to the barrier + */ + BufFile **statuses; + BufFile *combined_bitmap_file; + int statuses_length; + + int nbparticipants = 0; + + for (int l = 0; l < accessor->bits->nparticipants; l++) + { + SharedBitsParticipant participant = accessor->bits->participants[l]; + + if (participant.present) + { + Assert(!participant.writing); + nbparticipants++; + } + } + statuses = palloc(sizeof(BufFile *) * nbparticipants); + + /* + * Open the bitmap shared BufFile from each participant. TODO: explain why + * file can be NULLs + */ + statuses_length = 0; + + for (int i = 0; i < accessor->bits->nparticipants; i++) + { + char bitmap_filename[MAXPGPATH]; + BufFile *file; + + /* TODO: make a function that will do this */ + snprintf(bitmap_filename, MAXPGPATH, "%s.p%d.bitmap", accessor->bits->name, i); + + if (!accessor->bits->participants[i].present) + continue; + file = BufFileOpenShared(accessor->fileset, bitmap_filename); + + Assert(file); + + statuses[statuses_length++] = file; + } + + combined_bitmap_file = BufFileCreateTemp(false); + + for (int64 cur = 0; cur < BufFileSize(statuses[0]); cur++) /* make it while not EOF */ + { + /* + * TODO: make this use an unsigned int instead of a byte so it isn't + * so slow + */ + unsigned char combined_byte = 0; + + for (int i = 0; i < statuses_length; i++) + { + unsigned char read_byte; + + BufFileRead(statuses[i], &read_byte, 1); + combined_byte |= read_byte; + } + + BufFileWrite(combined_bitmap_file, &combined_byte, 1); + } + + if (BufFileSeek(combined_bitmap_file, 0, 0L, SEEK_SET)) + ereport(ERROR, + (errcode_for_file_access(), + errmsg("could not rewind hash-join temporary file: %m"))); + + for (int i = 0; i < statuses_length; i++) + BufFileClose(statuses[i]); + pfree(statuses); + + accessor->combined = combined_bitmap_file; + return combined_bitmap_file; +} + +void +sb_end_write(SharedBitsAccessor *sba) +{ + SharedBitsParticipant + *const participant = &sba->bits->participants[sba->participant]; + + participant->writing = false; + + /* + * TODO: this should not be needed if flow is correct. need to fix that + * and get rid of this check + */ + if (sba->write_file) + BufFileClose(sba->write_file); + sba->write_file = NULL; +} + +void +sb_end_read(SharedBitsAccessor *accessor) +{ + if (accessor->combined == NULL) + return; + + BufFileClose(accessor->combined); + accessor->combined = NULL; +} diff --git a/src/backend/utils/sort/sharedtuplestore.c b/src/backend/utils/sort/sharedtuplestore.c index c3ab494a45..0e3b3de2b6 100644 --- a/src/backend/utils/sort/sharedtuplestore.c +++ b/src/backend/utils/sort/sharedtuplestore.c @@ -52,6 +52,7 @@ typedef struct SharedTuplestoreParticipant { LWLock lock; BlockNumber read_page; /* Page number for next read. */ + bool rewound; BlockNumber npages; /* Number of pages written. */ bool writing; /* Used only for assertions. */ } SharedTuplestoreParticipant; @@ -60,6 +61,7 @@ typedef struct SharedTuplestoreParticipant struct SharedTuplestore { int nparticipants; /* Number of participants that can write. */ + pg_atomic_uint32 ntuples; /* Number of tuples in this tuplestore. */ int flags; /* Flag bits from SHARED_TUPLESTORE_XXX */ size_t meta_data_size; /* Size of per-tuple header. */ char name[NAMEDATALEN]; /* A name for this tuplestore. */ @@ -85,6 +87,8 @@ struct SharedTuplestoreAccessor char *read_buffer; /* A buffer for loading tuples. */ size_t read_buffer_size; BlockNumber read_next_page; /* Lowest block we'll consider reading. */ + BlockNumber start_page; /* page to reset p->read_page to if back out + * required */ /* State for writing. */ SharedTuplestoreChunk *write_chunk; /* Buffer for writing. */ @@ -137,6 +141,7 @@ sts_initialize(SharedTuplestore *sts, int participants, Assert(my_participant_number < participants); sts->nparticipants = participants; + pg_atomic_init_u32(&sts->ntuples, 1); sts->meta_data_size = meta_data_size; sts->flags = flags; @@ -158,6 +163,7 @@ sts_initialize(SharedTuplestore *sts, int participants, LWLockInitialize(&sts->participants[i].lock, LWTRANCHE_SHARED_TUPLESTORE); sts->participants[i].read_page = 0; + sts->participants[i].rewound = false; sts->participants[i].writing = false; } @@ -277,6 +283,45 @@ sts_begin_parallel_scan(SharedTuplestoreAccessor *accessor) accessor->read_participant = accessor->participant; accessor->read_file = NULL; accessor->read_next_page = 0; + accessor->start_page = 0; +} + +void +sts_resume_parallel_scan(SharedTuplestoreAccessor *accessor) +{ + int i PG_USED_FOR_ASSERTS_ONLY; + SharedTuplestoreParticipant *p; + + /* End any existing scan that was in progress. */ + sts_end_parallel_scan(accessor); + + /* + * Any backend that might have written into this shared tuplestore must + * have called sts_end_write(), so that all buffers are flushed and the + * files have stopped growing. + */ + for (i = 0; i < accessor->sts->nparticipants; ++i) + Assert(!accessor->sts->participants[i].writing); + + /* + * We will start out reading the file that THIS backend wrote. There may + * be some caching locality advantage to that. + */ + + /* + * TODO: does this still apply in the multi-stripe case? It seems like if + * a participant file is exhausted for the current stripe it might be + * better to remember that + */ + accessor->read_participant = accessor->participant; + accessor->read_file = NULL; + p = &accessor->sts->participants[accessor->read_participant]; + + /* TODO: find a better solution than this for resuming the parallel scan */ + LWLockAcquire(&p->lock, LW_SHARED); + accessor->start_page = p->read_page; + LWLockRelease(&p->lock); + accessor->read_next_page = 0; } /* @@ -295,6 +340,7 @@ sts_end_parallel_scan(SharedTuplestoreAccessor *accessor) BufFileClose(accessor->read_file); accessor->read_file = NULL; } + accessor->start_page = 0; } /* @@ -531,7 +577,13 @@ sts_parallel_scan_next(SharedTuplestoreAccessor *accessor, void *meta_data) for (;;) { /* Can we read more tuples from the current chunk? */ - if (accessor->read_ntuples < accessor->read_ntuples_available) + /* + * Added a check for accessor->read_file being present here, as it + * became relevant for adaptive hashjoin. Not sure if this has other + * consequences for correctness + */ + + if (accessor->read_ntuples < accessor->read_ntuples_available && accessor->read_file) return sts_read_tuple(accessor, meta_data); /* Find the location of a new chunk to read. */ @@ -541,7 +593,7 @@ sts_parallel_scan_next(SharedTuplestoreAccessor *accessor, void *meta_data) /* We can skip directly past overflow pages we know about. */ if (p->read_page < accessor->read_next_page) p->read_page = accessor->read_next_page; - eof = p->read_page >= p->npages; + eof = p->read_page >= p->npages || p->rewound; if (!eof) { /* Claim the next chunk. */ @@ -549,9 +601,22 @@ sts_parallel_scan_next(SharedTuplestoreAccessor *accessor, void *meta_data) /* Advance the read head for the next reader. */ p->read_page += STS_CHUNK_PAGES; accessor->read_next_page = p->read_page; + + /* + * initialize start_page to the read_page this participant will + * start reading from + */ + accessor->start_page = read_page; } LWLockRelease(&p->lock); + if (!eof) + { + char name[MAXPGPATH]; + + sts_filename(name, accessor, accessor->read_participant); + } + if (!eof) { SharedTuplestoreChunk chunk_header; @@ -613,6 +678,7 @@ sts_parallel_scan_next(SharedTuplestoreAccessor *accessor, void *meta_data) if (accessor->read_participant == accessor->participant) break; accessor->read_next_page = 0; + accessor->start_page = 0; /* Go around again, so we can get a chunk from this file. */ } @@ -621,6 +687,48 @@ sts_parallel_scan_next(SharedTuplestoreAccessor *accessor, void *meta_data) return NULL; } +void +sts_parallel_scan_rewind(SharedTuplestoreAccessor *accessor) +{ + SharedTuplestoreParticipant *p = + &accessor->sts->participants[accessor->read_participant]; + + /* + * Only set the read_page back to the start of the sts_chunk this worker + * was reading if some other worker has not already done so. It could be + * the case that this worker saw a tuple from a future stripe and another + * worker did also in its sts_chunk and it already set read_page to its + * start_page If so, we want to set read_page to the lowest value to + * ensure that we read all tuples from the stripe (don't miss tuples) + */ + LWLockAcquire(&p->lock, LW_EXCLUSIVE); + p->read_page = Min(p->read_page, accessor->start_page); + p->rewound = true; + LWLockRelease(&p->lock); + + accessor->read_ntuples_available = 0; + accessor->read_next_page = 0; +} + +void +sts_reset_rewound(SharedTuplestoreAccessor *accessor) +{ + for (int i = 0; i < accessor->sts->nparticipants; ++i) + accessor->sts->participants[i].rewound = false; +} + +uint32 +sts_increment_ntuples(SharedTuplestoreAccessor *accessor) +{ + return pg_atomic_fetch_add_u32(&accessor->sts->ntuples, 1); +} + +uint32 +sts_get_tuplenum(SharedTuplestoreAccessor *accessor) +{ + return pg_atomic_read_u32(&accessor->sts->ntuples); +} + /* * Create the name used for the BufFile that a given participant will write. */ diff --git a/src/include/commands/explain.h b/src/include/commands/explain.h index ba661d32a6..0ba9d856c8 100644 --- a/src/include/commands/explain.h +++ b/src/include/commands/explain.h @@ -46,6 +46,7 @@ typedef struct ExplainState bool timing; /* print detailed node timing */ bool summary; /* print total planning and execution timing */ bool settings; /* print modified settings */ + bool usage; /* print memory usage */ ExplainFormat format; /* output format */ /* state for output formatting --- not reset for each new plan tree */ int indent; /* current indentation level */ diff --git a/src/include/executor/hashjoin.h b/src/include/executor/hashjoin.h index 79b634e8ed..9ffcd84806 100644 --- a/src/include/executor/hashjoin.h +++ b/src/include/executor/hashjoin.h @@ -19,6 +19,7 @@ #include "storage/barrier.h" #include "storage/buffile.h" #include "storage/lwlock.h" +#include "utils/sharedbits.h" /* ---------------------------------------------------------------- * hash-join hash table structures @@ -152,6 +153,7 @@ typedef struct ParallelHashJoinBatch { dsa_pointer buckets; /* array of hash table buckets */ Barrier batch_barrier; /* synchronization for joining this batch */ + Barrier stripe_barrier; /* synchronization for stripes */ dsa_pointer chunks; /* chunks of tuples loaded */ size_t size; /* size of buckets + chunks in memory */ @@ -160,6 +162,17 @@ typedef struct ParallelHashJoinBatch size_t old_ntuples; /* number of tuples before repartitioning */ bool space_exhausted; + /* Adaptive HashJoin */ + + /* + * after finishing build phase, hashloop_fallback cannot change, and does + * not require a lock to read + */ + bool hashloop_fallback; + int maximum_stripe_number; + size_t estimated_stripe_size; /* size of last stripe in batch */ + LWLock lock; + /* * Variable-sized SharedTuplestore objects follow this struct in memory. * See the accessor macros below. @@ -177,10 +190,17 @@ typedef struct ParallelHashJoinBatch ((char *) ParallelHashJoinBatchInner(batch) + \ MAXALIGN(sts_estimate(nparticipants)))) +/* Accessor for sharedbits following a ParallelHashJoinBatch. */ +#define ParallelHashJoinBatchOuterBits(batch, nparticipants) \ + ((SharedBits *) \ + ((char *) ParallelHashJoinBatchOuter(batch, nparticipants) + \ + MAXALIGN(sts_estimate(nparticipants)))) + /* Total size of a ParallelHashJoinBatch and tuplestores. */ #define EstimateParallelHashJoinBatch(hashtable) \ (MAXALIGN(sizeof(ParallelHashJoinBatch)) + \ - MAXALIGN(sts_estimate((hashtable)->parallel_state->nparticipants)) * 2) + MAXALIGN(sts_estimate((hashtable)->parallel_state->nparticipants)) * 2 + \ + MAXALIGN(sb_estimate((hashtable)->parallel_state->nparticipants))) /* Accessor for the nth ParallelHashJoinBatch given the base. */ #define NthParallelHashJoinBatch(base, n) \ @@ -207,6 +227,7 @@ typedef struct ParallelHashJoinBatchAccessor bool done; /* flag to remember that a batch is done */ SharedTuplestoreAccessor *inner_tuples; SharedTuplestoreAccessor *outer_tuples; + SharedBitsAccessor *sba; } ParallelHashJoinBatchAccessor; /* @@ -251,6 +272,7 @@ typedef struct ParallelHashJoinState pg_atomic_uint32 distributor; /* counter for load balancing */ SharedFileSet fileset; /* space for shared temporary files */ + SharedFileSet sbfileset; } ParallelHashJoinState; /* The phases for building batches, used by build_barrier. */ @@ -263,9 +285,17 @@ typedef struct ParallelHashJoinState /* The phases for probing each batch, used by for batch_barrier. */ #define PHJ_BATCH_ELECTING 0 #define PHJ_BATCH_ALLOCATING 1 -#define PHJ_BATCH_LOADING 2 -#define PHJ_BATCH_PROBING 3 -#define PHJ_BATCH_DONE 4 +#define PHJ_BATCH_STRIPING 2 +#define PHJ_BATCH_DONE 3 + +/* The phases for probing each stripe of each batch used with stripe barriers */ +#define PHJ_STRIPE_ELECTING 0 +#define PHJ_STRIPE_RESETTING 1 +#define PHJ_STRIPE_LOADING 2 +#define PHJ_STRIPE_PROBING 3 +#define PHJ_STRIPE_DONE 4 +#define PHJ_STRIPE_NUMBER(n) ((n) / 5) +#define PHJ_STRIPE_PHASE(n) ((n) % 5) /* The phases of batch growth while hashing, for grow_batches_barrier. */ #define PHJ_GROW_BATCHES_ELECTING 0 @@ -313,8 +343,6 @@ typedef struct HashJoinTableData int nbatch_original; /* nbatch when we started inner scan */ int nbatch_outstart; /* nbatch when we started outer scan */ - bool growEnabled; /* flag to shut off nbatch increases */ - double totalTuples; /* # tuples obtained from inner plan */ double partialTuples; /* # tuples obtained from inner plan by me */ double skewTuples; /* # tuples inserted into skew tuples */ @@ -329,6 +357,13 @@ typedef struct HashJoinTableData BufFile **innerBatchFile; /* buffered virtual temp file per batch */ BufFile **outerBatchFile; /* buffered virtual temp file per batch */ + /* + * Adaptive hashjoin variables + */ + BufFile **hashloop_fallback; /* outer match status files if fall back */ + List *fallback_batches_stats; /* per hashjoin batch statistics */ + int curstripe; /* current stripe #; 0 on 1st pass, -2 on phantom stripe */ + /* * Info about the datatype-specific hash functions for the datatypes being * hashed. These are arrays of the same length as the number of hash join diff --git a/src/include/executor/instrument.h b/src/include/executor/instrument.h index 50d672b270..bcac88f7f3 100644 --- a/src/include/executor/instrument.h +++ b/src/include/executor/instrument.h @@ -14,6 +14,7 @@ #define INSTRUMENT_H #include "portability/instr_time.h" +#include "nodes/pg_list.h" typedef struct BufferUsage @@ -39,6 +40,12 @@ typedef struct WalUsage uint64 wal_bytes; /* size of WAL records produced */ } WalUsage; +typedef struct FallbackBatchStats +{ + int batchno; + int numstripes; +} FallbackBatchStats; + /* Flag bits included in InstrAlloc's instrument_options bitmask */ typedef enum InstrumentOption { diff --git a/src/include/executor/nodeHash.h b/src/include/executor/nodeHash.h index 64d2ce693c..f85308738b 100644 --- a/src/include/executor/nodeHash.h +++ b/src/include/executor/nodeHash.h @@ -31,6 +31,7 @@ extern void ExecParallelHashTableAlloc(HashJoinTable hashtable, extern void ExecHashTableDestroy(HashJoinTable hashtable); extern void ExecHashTableDetach(HashJoinTable hashtable); extern void ExecHashTableDetachBatch(HashJoinTable hashtable); +extern bool ExecHashTableDetachStripe(HashJoinTable hashtable); extern void ExecParallelHashTableSetCurrentBatch(HashJoinTable hashtable, int batchno); diff --git a/src/include/executor/tuptable.h b/src/include/executor/tuptable.h index f7df70b5ab..0c0d87d1d3 100644 --- a/src/include/executor/tuptable.h +++ b/src/include/executor/tuptable.h @@ -129,6 +129,7 @@ typedef struct TupleTableSlot MemoryContext tts_mcxt; /* slot itself is in this context */ ItemPointerData tts_tid; /* stored tuple's tid */ Oid tts_tableOid; /* table oid of tuple */ + uint32 tts_tuplenum; /* a tuple id for use when ctid cannot be used */ } TupleTableSlot; /* routines for a TupleTableSlot implementation */ @@ -425,6 +426,7 @@ static inline TupleTableSlot * ExecClearTuple(TupleTableSlot *slot) { slot->tts_ops->clear(slot); + slot->tts_tuplenum = 0; /* TODO: should this be done elsewhere? */ return slot; } diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h index 4fee043bb2..41a4133c3a 100644 --- a/src/include/nodes/execnodes.h +++ b/src/include/nodes/execnodes.h @@ -1957,6 +1957,10 @@ typedef struct HashJoinState int hj_JoinState; bool hj_MatchedOuter; bool hj_OuterNotEmpty; + /* Adaptive Hashjoin variables */ + int hj_CurNumOuterTuples; /* number of outer tuples in a batch */ + unsigned int hj_CurOuterMatchStatus; + int hj_EmitOuterTupleId; } HashJoinState; @@ -2359,6 +2363,7 @@ typedef struct HashInstrumentation int nbatch; /* number of batches at end of execution */ int nbatch_original; /* planned number of batches */ Size space_peak; /* peak memory usage in bytes */ + List *fallback_batches_stats; /* per hashjoin batch stats */ } HashInstrumentation; /* ---------------- diff --git a/src/include/pgstat.h b/src/include/pgstat.h index b8041d9988..9ebdeeeb8a 100644 --- a/src/include/pgstat.h +++ b/src/include/pgstat.h @@ -857,7 +857,10 @@ typedef enum WAIT_EVENT_EXECUTE_GATHER, WAIT_EVENT_HASH_BATCH_ALLOCATING, WAIT_EVENT_HASH_BATCH_ELECTING, - WAIT_EVENT_HASH_BATCH_LOADING, + WAIT_EVENT_HASH_STRIPE_ELECTING, + WAIT_EVENT_HASH_STRIPE_RESETTING, + WAIT_EVENT_HASH_STRIPE_LOADING, + WAIT_EVENT_HASH_STRIPE_PROBING, WAIT_EVENT_HASH_BUILD_ALLOCATING, WAIT_EVENT_HASH_BUILD_ELECTING, WAIT_EVENT_HASH_BUILD_HASHING_INNER, diff --git a/src/include/utils/sharedbits.h b/src/include/utils/sharedbits.h new file mode 100644 index 0000000000..de43279de8 --- /dev/null +++ b/src/include/utils/sharedbits.h @@ -0,0 +1,39 @@ +/*------------------------------------------------------------------------- + * + * sharedbits.h + * Simple mechanism for sharing bits between backends. + * + * Portions Copyright (c) 1996-2020, PostgreSQL Global Development Group + * Portions Copyright (c) 1994, Regents of the University of California + * + * src/include/utils/sharedbits.h + * + *------------------------------------------------------------------------- + */ +#ifndef SHAREDBITS_H +#define SHAREDBITS_H + +#include "storage/sharedfileset.h" + +struct SharedBits; +typedef struct SharedBits SharedBits; + +struct SharedBitsParticipant; +typedef struct SharedBitsParticipant SharedBitsParticipant; + +struct SharedBitsAccessor; +typedef struct SharedBitsAccessor SharedBitsAccessor; + +extern SharedBitsAccessor *sb_attach(SharedBits *sbits, int my_participant_number, SharedFileSet *fileset); +extern SharedBitsAccessor *sb_initialize(SharedBits *sbits, int participants, int my_participant_number, SharedFileSet *fileset, char *name); +extern void sb_initialize_accessor(SharedBitsAccessor *accessor, uint32 nbits); +extern size_t sb_estimate(int participants); + +extern void sb_setbit(SharedBitsAccessor *accessor, uint64 bit); +extern bool sb_checkbit(SharedBitsAccessor *accessor, uint32 n); +extern BufFile *sb_combine(SharedBitsAccessor *accessor); + +extern void sb_end_write(SharedBitsAccessor *sba); +extern void sb_end_read(SharedBitsAccessor *accessor); + +#endif /* SHAREDBITS_H */ diff --git a/src/include/utils/sharedtuplestore.h b/src/include/utils/sharedtuplestore.h index 9754504cc5..99aead8a4a 100644 --- a/src/include/utils/sharedtuplestore.h +++ b/src/include/utils/sharedtuplestore.h @@ -22,6 +22,17 @@ typedef struct SharedTuplestore SharedTuplestore; struct SharedTuplestoreAccessor; typedef struct SharedTuplestoreAccessor SharedTuplestoreAccessor; +struct tupleMetadata; +typedef struct tupleMetadata tupleMetadata; +struct tupleMetadata +{ + uint32 hashvalue; + union + { + uint32 tupleid; /* tuple number or id on the outer side */ + int stripe; /* stripe number for inner side */ + }; +}; /* * A flag indicating that the tuplestore will only be scanned once, so backing @@ -49,6 +60,8 @@ extern void sts_reinitialize(SharedTuplestoreAccessor *accessor); extern void sts_begin_parallel_scan(SharedTuplestoreAccessor *accessor); +extern void sts_resume_parallel_scan(SharedTuplestoreAccessor *accessor); + extern void sts_end_parallel_scan(SharedTuplestoreAccessor *accessor); extern void sts_puttuple(SharedTuplestoreAccessor *accessor, @@ -58,4 +71,10 @@ extern void sts_puttuple(SharedTuplestoreAccessor *accessor, extern MinimalTuple sts_parallel_scan_next(SharedTuplestoreAccessor *accessor, void *meta_data); +extern void sts_parallel_scan_rewind(SharedTuplestoreAccessor *accessor); + +extern void sts_reset_rewound(SharedTuplestoreAccessor *accessor); +extern uint32 sts_increment_ntuples(SharedTuplestoreAccessor *accessor); +extern uint32 sts_get_tuplenum(SharedTuplestoreAccessor *accessor); + #endif /* SHAREDTUPLESTORE_H */ diff --git a/src/test/regress/expected/join_hash.out b/src/test/regress/expected/join_hash.out index 3a91c144a2..98a90a85e4 100644 --- a/src/test/regress/expected/join_hash.out +++ b/src/test/regress/expected/join_hash.out @@ -443,7 +443,7 @@ $$ $$); original | final ----------+------- - 1 | 2 + 1 | 4 (1 row) rollback to settings; @@ -478,7 +478,7 @@ $$ $$); original | final ----------+------- - 1 | 2 + 1 | 4 (1 row) rollback to settings; @@ -1013,3 +1013,944 @@ WHERE (1 row) ROLLBACK; +-- Serial Adaptive Hash Join +CREATE TYPE stub AS (hash INTEGER, value CHAR(8098)); +CREATE FUNCTION stub_hash(item stub) +RETURNS INTEGER AS $$ +DECLARE + batch_size INTEGER; +BEGIN + batch_size := 4; + RETURN item.hash << (batch_size - 1); +END; $$ LANGUAGE plpgsql IMMUTABLE LEAKPROOF STRICT PARALLEL SAFE; +CREATE FUNCTION stub_eq(item1 stub, item2 stub) +RETURNS BOOLEAN AS $$ +BEGIN + RETURN item1.hash = item2.hash AND item1.value = item2.value; +END; $$ LANGUAGE plpgsql IMMUTABLE LEAKPROOF STRICT PARALLEL SAFE; +CREATE OPERATOR = ( + FUNCTION = stub_eq, + LEFTARG = stub, + RIGHTARG = stub, + COMMUTATOR = =, + HASHES, MERGES +); +CREATE OPERATOR CLASS stub_hash_ops +DEFAULT FOR TYPE stub USING hash AS + OPERATOR 1 =(stub, stub), + FUNCTION 1 stub_hash(stub); +CREATE TABLE probeside(a stub); +ALTER TABLE probeside ALTER COLUMN a SET STORAGE PLAIN; +-- non-fallback batch with unmatched outer tuple +INSERT INTO probeside SELECT '(2, "")' FROM generate_series(1, 1); +-- fallback batch unmatched outer tuple (in first stripe maybe) +INSERT INTO probeside SELECT '(1, "unmatched outer tuple")' FROM generate_series(1, 1); +-- fallback batch matched outer tuple +INSERT INTO probeside SELECT '(1, "")' FROM generate_series(1, 5); +-- fallback batch unmatched outer tuple (in last stripe maybe) +-- When numbatches=4, hash 5 maps to batch 1, but after numbatches doubles to +-- 8 batches hash 5 maps to batch 5. +INSERT INTO probeside SELECT '(5, "")' FROM generate_series(1, 1); +-- non-fallback batch matched outer tuple +INSERT INTO probeside SELECT '(3, "")' FROM generate_series(1, 1); +-- batch with 3 stripes where non-first/non-last stripe contains unmatched outer tuple +INSERT INTO probeside SELECT '(6, "")' FROM generate_series(1, 5); +INSERT INTO probeside SELECT '(6, "unmatched outer tuple")' FROM generate_series(1, 1); +INSERT INTO probeside SELECT '(6, "")' FROM generate_series(1, 1); +CREATE TABLE hashside_wide(a stub, id int); +ALTER TABLE hashside_wide ALTER COLUMN a SET STORAGE PLAIN; +-- falls back with an unmatched inner tuple that is in fist, middle, and last +-- stripe +INSERT INTO hashside_wide SELECT '(1, "unmatched inner tuple in first stripe")', 1 FROM generate_series(1, 1); +INSERT INTO hashside_wide SELECT '(1, "")', 1 FROM generate_series(1, 9); +INSERT INTO hashside_wide SELECT '(1, "unmatched inner tuple in middle stripe")', 1 FROM generate_series(1, 1); +INSERT INTO hashside_wide SELECT '(1, "")', 1 FROM generate_series(1, 9); +INSERT INTO hashside_wide SELECT '(1, "unmatched inner tuple in last stripe")', 1 FROM generate_series(1, 1); +-- doesn't fall back -- matched tuple +INSERT INTO hashside_wide SELECT '(3, "")', 3 FROM generate_series(1, 1); +INSERT INTO hashside_wide SELECT '(6, "")', 6 FROM generate_series(1, 20); +ANALYZE probeside, hashside_wide; +SET enable_nestloop TO off; +SET enable_mergejoin TO off; +SET work_mem = 64; +SELECT (probeside.a).hash, TRIM((probeside.a).value), hashside_wide.id, (hashside_wide.a).hash, TRIM((hashside_wide.a).value) +FROM probeside +LEFT OUTER JOIN hashside_wide USING (a) +ORDER BY 1, 2, 3, 4, 5; + hash | btrim | id | hash | btrim +------+-----------------------+----+------+------- + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | unmatched outer tuple | | | + 2 | | | | + 3 | | 3 | 3 | + 5 | | | | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | unmatched outer tuple | | | +(215 rows) + +EXPLAIN (ANALYZE, summary off, timing off, costs off, usage off) SELECT * FROM probeside +LEFT OUTER JOIN hashside_wide USING (a); + QUERY PLAN +---------------------------------------------------------------- + Hash Left Join (actual rows=215 loops=1) + Hash Cond: (probeside.a = hashside_wide.a) + -> Seq Scan on probeside (actual rows=16 loops=1) + -> Hash (actual rows=42 loops=1) + Buckets: 8 (originally 8) Batches: 32 (originally 8) + Batch: 1 Stripes: 3 + Batch: 6 Stripes: 3 + -> Seq Scan on hashside_wide (actual rows=42 loops=1) +(8 rows) + +SELECT (probeside.a).hash, TRIM((probeside.a).value), hashside_wide.id, (hashside_wide.a).hash, TRIM((hashside_wide.a).value) +FROM probeside +RIGHT OUTER JOIN hashside_wide USING (a) +ORDER BY 1, 2, 3, 4, 5; + hash | btrim | id | hash | btrim +------+-------+----+------+---------------------------------------- + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 3 | | 3 | 3 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + | | 1 | 1 | unmatched inner tuple in first stripe + | | 1 | 1 | unmatched inner tuple in last stripe + | | 1 | 1 | unmatched inner tuple in middle stripe +(214 rows) + +EXPLAIN (ANALYZE, summary off, timing off, costs off, usage off) SELECT * FROM probeside +RIGHT OUTER JOIN hashside_wide USING (a); + QUERY PLAN +---------------------------------------------------------------- + Hash Right Join (actual rows=214 loops=1) + Hash Cond: (probeside.a = hashside_wide.a) + -> Seq Scan on probeside (actual rows=16 loops=1) + -> Hash (actual rows=42 loops=1) + Buckets: 8 (originally 8) Batches: 32 (originally 8) + Batch: 1 Stripes: 3 + Batch: 6 Stripes: 3 + -> Seq Scan on hashside_wide (actual rows=42 loops=1) +(8 rows) + +SELECT (probeside.a).hash, TRIM((probeside.a).value), hashside_wide.id, (hashside_wide.a).hash, TRIM((hashside_wide.a).value) +FROM probeside +FULL OUTER JOIN hashside_wide USING (a) +ORDER BY 1, 2, 3, 4, 5; + hash | btrim | id | hash | btrim +------+-----------------------+----+------+---------------------------------------- + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | | 1 | 1 | + 1 | unmatched outer tuple | | | + 2 | | | | + 3 | | 3 | 3 | + 5 | | | | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | | 6 | 6 | + 6 | unmatched outer tuple | | | + | | 1 | 1 | unmatched inner tuple in first stripe + | | 1 | 1 | unmatched inner tuple in last stripe + | | 1 | 1 | unmatched inner tuple in middle stripe +(218 rows) + +EXPLAIN (ANALYZE, summary off, timing off, costs off, usage off) SELECT * FROM probeside +FULL OUTER JOIN hashside_wide USING (a); + QUERY PLAN +---------------------------------------------------------------- + Hash Full Join (actual rows=218 loops=1) + Hash Cond: (probeside.a = hashside_wide.a) + -> Seq Scan on probeside (actual rows=16 loops=1) + -> Hash (actual rows=42 loops=1) + Buckets: 8 (originally 8) Batches: 32 (originally 8) + Batch: 1 Stripes: 3 + Batch: 6 Stripes: 3 + -> Seq Scan on hashside_wide (actual rows=42 loops=1) +(8 rows) + +/* +-- semi-join testcase +EXPLAIN (ANALYZE, summary off, timing off, costs off, usage off) +SELECT probeside.* FROM probeside WHERE EXISTS (SELECT * FROM hashside_wide WHERE probeside.a=a); + +SELECT (probeside.a).hash, TRIM((probeside.a).value) +FROM probeside WHERE EXISTS (SELECT * FROM hashside_wide WHERE probeside.a=a) ORDER BY 1, 2; +*/ +-- anti-join testcase +EXPLAIN (ANALYZE, summary off, timing off, costs off, usage off) +SELECT probeside.* FROM probeside WHERE NOT EXISTS (SELECT * FROM hashside_wide WHERE probeside.a=a); + QUERY PLAN +---------------------------------------------------------------- + Hash Anti Join (actual rows=4 loops=1) + Hash Cond: (probeside.a = hashside_wide.a) + -> Seq Scan on probeside (actual rows=16 loops=1) + -> Hash (actual rows=42 loops=1) + Buckets: 8 (originally 8) Batches: 32 (originally 8) + Batch: 1 Stripes: 3 + Batch: 6 Stripes: 3 + -> Seq Scan on hashside_wide (actual rows=42 loops=1) +(8 rows) + +SELECT (probeside.a).hash, TRIM((probeside.a).value) +FROM probeside WHERE NOT EXISTS (SELECT * FROM hashside_wide WHERE probeside.a=a) ORDER BY 1, 2; + hash | btrim +------+----------------------- + 1 | unmatched outer tuple + 2 | + 5 | + 6 | unmatched outer tuple +(4 rows) + +-- Test spill of batch 0 gives correct results. +CREATE TABLE probeside_batch0(a stub); +ALTER TABLE probeside_batch0 ALTER COLUMN a SET STORAGE PLAIN; +INSERT INTO probeside_batch0 SELECT '(0, "")' FROM generate_series(1, 13); +INSERT INTO probeside_batch0 SELECT '(0, "unmatched outer")' FROM generate_series(1, 1); +CREATE TABLE hashside_wide_batch0(a stub, id int); +ALTER TABLE hashside_wide_batch0 ALTER COLUMN a SET STORAGE PLAIN; +INSERT INTO hashside_wide_batch0 SELECT '(0, "")', 1 FROM generate_series(1, 9); +ANALYZE probeside_batch0, hashside_wide_batch0; +SELECT (probeside_batch0.a).hash, ((((probeside_batch0.a).hash << 7) >> 3) & 31) AS batchno, TRIM((probeside_batch0.a).value), hashside_wide_batch0.id, hashside_wide_batch0.ctid, (hashside_wide_batch0.a).hash, TRIM((hashside_wide_batch0.a).value) +FROM probeside_batch0 +LEFT OUTER JOIN hashside_wide_batch0 USING (a) +ORDER BY 1, 2, 3, 4, 5; + hash | batchno | btrim | id | ctid | hash | btrim +------+---------+-----------------+----+-------+------+------- + 0 | 0 | | 1 | (0,1) | 0 | + 0 | 0 | | 1 | (0,1) | 0 | + 0 | 0 | | 1 | (0,1) | 0 | + 0 | 0 | | 1 | (0,1) | 0 | + 0 | 0 | | 1 | (0,1) | 0 | + 0 | 0 | | 1 | (0,1) | 0 | + 0 | 0 | | 1 | (0,1) | 0 | + 0 | 0 | | 1 | (0,1) | 0 | + 0 | 0 | | 1 | (0,1) | 0 | + 0 | 0 | | 1 | (0,1) | 0 | + 0 | 0 | | 1 | (0,1) | 0 | + 0 | 0 | | 1 | (0,1) | 0 | + 0 | 0 | | 1 | (0,1) | 0 | + 0 | 0 | | 1 | (1,1) | 0 | + 0 | 0 | | 1 | (1,1) | 0 | + 0 | 0 | | 1 | (1,1) | 0 | + 0 | 0 | | 1 | (1,1) | 0 | + 0 | 0 | | 1 | (1,1) | 0 | + 0 | 0 | | 1 | (1,1) | 0 | + 0 | 0 | | 1 | (1,1) | 0 | + 0 | 0 | | 1 | (1,1) | 0 | + 0 | 0 | | 1 | (1,1) | 0 | + 0 | 0 | | 1 | (1,1) | 0 | + 0 | 0 | | 1 | (1,1) | 0 | + 0 | 0 | | 1 | (1,1) | 0 | + 0 | 0 | | 1 | (1,1) | 0 | + 0 | 0 | | 1 | (2,1) | 0 | + 0 | 0 | | 1 | (2,1) | 0 | + 0 | 0 | | 1 | (2,1) | 0 | + 0 | 0 | | 1 | (2,1) | 0 | + 0 | 0 | | 1 | (2,1) | 0 | + 0 | 0 | | 1 | (2,1) | 0 | + 0 | 0 | | 1 | (2,1) | 0 | + 0 | 0 | | 1 | (2,1) | 0 | + 0 | 0 | | 1 | (2,1) | 0 | + 0 | 0 | | 1 | (2,1) | 0 | + 0 | 0 | | 1 | (2,1) | 0 | + 0 | 0 | | 1 | (2,1) | 0 | + 0 | 0 | | 1 | (2,1) | 0 | + 0 | 0 | | 1 | (3,1) | 0 | + 0 | 0 | | 1 | (3,1) | 0 | + 0 | 0 | | 1 | (3,1) | 0 | + 0 | 0 | | 1 | (3,1) | 0 | + 0 | 0 | | 1 | (3,1) | 0 | + 0 | 0 | | 1 | (3,1) | 0 | + 0 | 0 | | 1 | (3,1) | 0 | + 0 | 0 | | 1 | (3,1) | 0 | + 0 | 0 | | 1 | (3,1) | 0 | + 0 | 0 | | 1 | (3,1) | 0 | + 0 | 0 | | 1 | (3,1) | 0 | + 0 | 0 | | 1 | (3,1) | 0 | + 0 | 0 | | 1 | (3,1) | 0 | + 0 | 0 | | 1 | (4,1) | 0 | + 0 | 0 | | 1 | (4,1) | 0 | + 0 | 0 | | 1 | (4,1) | 0 | + 0 | 0 | | 1 | (4,1) | 0 | + 0 | 0 | | 1 | (4,1) | 0 | + 0 | 0 | | 1 | (4,1) | 0 | + 0 | 0 | | 1 | (4,1) | 0 | + 0 | 0 | | 1 | (4,1) | 0 | + 0 | 0 | | 1 | (4,1) | 0 | + 0 | 0 | | 1 | (4,1) | 0 | + 0 | 0 | | 1 | (4,1) | 0 | + 0 | 0 | | 1 | (4,1) | 0 | + 0 | 0 | | 1 | (4,1) | 0 | + 0 | 0 | | 1 | (5,1) | 0 | + 0 | 0 | | 1 | (5,1) | 0 | + 0 | 0 | | 1 | (5,1) | 0 | + 0 | 0 | | 1 | (5,1) | 0 | + 0 | 0 | | 1 | (5,1) | 0 | + 0 | 0 | | 1 | (5,1) | 0 | + 0 | 0 | | 1 | (5,1) | 0 | + 0 | 0 | | 1 | (5,1) | 0 | + 0 | 0 | | 1 | (5,1) | 0 | + 0 | 0 | | 1 | (5,1) | 0 | + 0 | 0 | | 1 | (5,1) | 0 | + 0 | 0 | | 1 | (5,1) | 0 | + 0 | 0 | | 1 | (5,1) | 0 | + 0 | 0 | | 1 | (6,1) | 0 | + 0 | 0 | | 1 | (6,1) | 0 | + 0 | 0 | | 1 | (6,1) | 0 | + 0 | 0 | | 1 | (6,1) | 0 | + 0 | 0 | | 1 | (6,1) | 0 | + 0 | 0 | | 1 | (6,1) | 0 | + 0 | 0 | | 1 | (6,1) | 0 | + 0 | 0 | | 1 | (6,1) | 0 | + 0 | 0 | | 1 | (6,1) | 0 | + 0 | 0 | | 1 | (6,1) | 0 | + 0 | 0 | | 1 | (6,1) | 0 | + 0 | 0 | | 1 | (6,1) | 0 | + 0 | 0 | | 1 | (6,1) | 0 | + 0 | 0 | | 1 | (7,1) | 0 | + 0 | 0 | | 1 | (7,1) | 0 | + 0 | 0 | | 1 | (7,1) | 0 | + 0 | 0 | | 1 | (7,1) | 0 | + 0 | 0 | | 1 | (7,1) | 0 | + 0 | 0 | | 1 | (7,1) | 0 | + 0 | 0 | | 1 | (7,1) | 0 | + 0 | 0 | | 1 | (7,1) | 0 | + 0 | 0 | | 1 | (7,1) | 0 | + 0 | 0 | | 1 | (7,1) | 0 | + 0 | 0 | | 1 | (7,1) | 0 | + 0 | 0 | | 1 | (7,1) | 0 | + 0 | 0 | | 1 | (7,1) | 0 | + 0 | 0 | | 1 | (8,1) | 0 | + 0 | 0 | | 1 | (8,1) | 0 | + 0 | 0 | | 1 | (8,1) | 0 | + 0 | 0 | | 1 | (8,1) | 0 | + 0 | 0 | | 1 | (8,1) | 0 | + 0 | 0 | | 1 | (8,1) | 0 | + 0 | 0 | | 1 | (8,1) | 0 | + 0 | 0 | | 1 | (8,1) | 0 | + 0 | 0 | | 1 | (8,1) | 0 | + 0 | 0 | | 1 | (8,1) | 0 | + 0 | 0 | | 1 | (8,1) | 0 | + 0 | 0 | | 1 | (8,1) | 0 | + 0 | 0 | | 1 | (8,1) | 0 | + 0 | 0 | unmatched outer | | | | +(118 rows) + diff --git a/src/test/regress/sql/join_hash.sql b/src/test/regress/sql/join_hash.sql index 68c1a8c7b6..1f70300d02 100644 --- a/src/test/regress/sql/join_hash.sql +++ b/src/test/regress/sql/join_hash.sql @@ -538,3 +538,130 @@ WHERE AND hjtest_1.a <> hjtest_2.b; ROLLBACK; + +-- Serial Adaptive Hash Join + +CREATE TYPE stub AS (hash INTEGER, value CHAR(8098)); + +CREATE FUNCTION stub_hash(item stub) +RETURNS INTEGER AS $$ +DECLARE + batch_size INTEGER; +BEGIN + batch_size := 4; + RETURN item.hash << (batch_size - 1); +END; $$ LANGUAGE plpgsql IMMUTABLE LEAKPROOF STRICT PARALLEL SAFE; + +CREATE FUNCTION stub_eq(item1 stub, item2 stub) +RETURNS BOOLEAN AS $$ +BEGIN + RETURN item1.hash = item2.hash AND item1.value = item2.value; +END; $$ LANGUAGE plpgsql IMMUTABLE LEAKPROOF STRICT PARALLEL SAFE; + +CREATE OPERATOR = ( + FUNCTION = stub_eq, + LEFTARG = stub, + RIGHTARG = stub, + COMMUTATOR = =, + HASHES, MERGES +); + +CREATE OPERATOR CLASS stub_hash_ops +DEFAULT FOR TYPE stub USING hash AS + OPERATOR 1 =(stub, stub), + FUNCTION 1 stub_hash(stub); + +CREATE TABLE probeside(a stub); +ALTER TABLE probeside ALTER COLUMN a SET STORAGE PLAIN; +-- non-fallback batch with unmatched outer tuple +INSERT INTO probeside SELECT '(2, "")' FROM generate_series(1, 1); +-- fallback batch unmatched outer tuple (in first stripe maybe) +INSERT INTO probeside SELECT '(1, "unmatched outer tuple")' FROM generate_series(1, 1); +-- fallback batch matched outer tuple +INSERT INTO probeside SELECT '(1, "")' FROM generate_series(1, 5); +-- fallback batch unmatched outer tuple (in last stripe maybe) +-- When numbatches=4, hash 5 maps to batch 1, but after numbatches doubles to +-- 8 batches hash 5 maps to batch 5. +INSERT INTO probeside SELECT '(5, "")' FROM generate_series(1, 1); +-- non-fallback batch matched outer tuple +INSERT INTO probeside SELECT '(3, "")' FROM generate_series(1, 1); +-- batch with 3 stripes where non-first/non-last stripe contains unmatched outer tuple +INSERT INTO probeside SELECT '(6, "")' FROM generate_series(1, 5); +INSERT INTO probeside SELECT '(6, "unmatched outer tuple")' FROM generate_series(1, 1); +INSERT INTO probeside SELECT '(6, "")' FROM generate_series(1, 1); + +CREATE TABLE hashside_wide(a stub, id int); +ALTER TABLE hashside_wide ALTER COLUMN a SET STORAGE PLAIN; +-- falls back with an unmatched inner tuple that is in fist, middle, and last +-- stripe +INSERT INTO hashside_wide SELECT '(1, "unmatched inner tuple in first stripe")', 1 FROM generate_series(1, 1); +INSERT INTO hashside_wide SELECT '(1, "")', 1 FROM generate_series(1, 9); +INSERT INTO hashside_wide SELECT '(1, "unmatched inner tuple in middle stripe")', 1 FROM generate_series(1, 1); +INSERT INTO hashside_wide SELECT '(1, "")', 1 FROM generate_series(1, 9); +INSERT INTO hashside_wide SELECT '(1, "unmatched inner tuple in last stripe")', 1 FROM generate_series(1, 1); + +-- doesn't fall back -- matched tuple +INSERT INTO hashside_wide SELECT '(3, "")', 3 FROM generate_series(1, 1); +INSERT INTO hashside_wide SELECT '(6, "")', 6 FROM generate_series(1, 20); + +ANALYZE probeside, hashside_wide; + +SET enable_nestloop TO off; +SET enable_mergejoin TO off; +SET work_mem = 64; + +SELECT (probeside.a).hash, TRIM((probeside.a).value), hashside_wide.id, (hashside_wide.a).hash, TRIM((hashside_wide.a).value) +FROM probeside +LEFT OUTER JOIN hashside_wide USING (a) +ORDER BY 1, 2, 3, 4, 5; + +EXPLAIN (ANALYZE, summary off, timing off, costs off, usage off) SELECT * FROM probeside +LEFT OUTER JOIN hashside_wide USING (a); + +SELECT (probeside.a).hash, TRIM((probeside.a).value), hashside_wide.id, (hashside_wide.a).hash, TRIM((hashside_wide.a).value) +FROM probeside +RIGHT OUTER JOIN hashside_wide USING (a) +ORDER BY 1, 2, 3, 4, 5; + +EXPLAIN (ANALYZE, summary off, timing off, costs off, usage off) SELECT * FROM probeside +RIGHT OUTER JOIN hashside_wide USING (a); + +SELECT (probeside.a).hash, TRIM((probeside.a).value), hashside_wide.id, (hashside_wide.a).hash, TRIM((hashside_wide.a).value) +FROM probeside +FULL OUTER JOIN hashside_wide USING (a) +ORDER BY 1, 2, 3, 4, 5; + +EXPLAIN (ANALYZE, summary off, timing off, costs off, usage off) SELECT * FROM probeside +FULL OUTER JOIN hashside_wide USING (a); + +/* +-- semi-join testcase +EXPLAIN (ANALYZE, summary off, timing off, costs off, usage off) +SELECT probeside.* FROM probeside WHERE EXISTS (SELECT * FROM hashside_wide WHERE probeside.a=a); + +SELECT (probeside.a).hash, TRIM((probeside.a).value) +FROM probeside WHERE EXISTS (SELECT * FROM hashside_wide WHERE probeside.a=a) ORDER BY 1, 2; +*/ + +-- anti-join testcase +EXPLAIN (ANALYZE, summary off, timing off, costs off, usage off) +SELECT probeside.* FROM probeside WHERE NOT EXISTS (SELECT * FROM hashside_wide WHERE probeside.a=a); + +SELECT (probeside.a).hash, TRIM((probeside.a).value) +FROM probeside WHERE NOT EXISTS (SELECT * FROM hashside_wide WHERE probeside.a=a) ORDER BY 1, 2; + +-- Test spill of batch 0 gives correct results. +CREATE TABLE probeside_batch0(a stub); +ALTER TABLE probeside_batch0 ALTER COLUMN a SET STORAGE PLAIN; +INSERT INTO probeside_batch0 SELECT '(0, "")' FROM generate_series(1, 13); +INSERT INTO probeside_batch0 SELECT '(0, "unmatched outer")' FROM generate_series(1, 1); + +CREATE TABLE hashside_wide_batch0(a stub, id int); +ALTER TABLE hashside_wide_batch0 ALTER COLUMN a SET STORAGE PLAIN; +INSERT INTO hashside_wide_batch0 SELECT '(0, "")', 1 FROM generate_series(1, 9); +ANALYZE probeside_batch0, hashside_wide_batch0; + +SELECT (probeside_batch0.a).hash, ((((probeside_batch0.a).hash << 7) >> 3) & 31) AS batchno, TRIM((probeside_batch0.a).value), hashside_wide_batch0.id, hashside_wide_batch0.ctid, (hashside_wide_batch0.a).hash, TRIM((hashside_wide_batch0.a).value) +FROM probeside_batch0 +LEFT OUTER JOIN hashside_wide_batch0 USING (a) +ORDER BY 1, 2, 3, 4, 5; -- 2.20.1