granicus.if.org Git - postgresql/blob - src/backend/executor/nodeHash.c

   1 /*-------------------------------------------------------------------------
   2  *
   3  * nodeHash.c
   4  *        Routines to hash relations for hashjoin
   5  *
   6  * Portions Copyright (c) 1996-2017, PostgreSQL Global Development Group
   7  * Portions Copyright (c) 1994, Regents of the University of California
   8  *
   9  *
  10  * IDENTIFICATION
  11  *        src/backend/executor/nodeHash.c
  12  *
  13  *-------------------------------------------------------------------------
  14  */
  15 /*
  16  * INTERFACE ROUTINES
  17  *              MultiExecHash   - generate an in-memory hash table of the relation
  18  *              ExecInitHash    - initialize node and subnodes
  19  *              ExecEndHash             - shutdown node and subnodes
  20  */
  21
  22 #include "postgres.h"
  23
  24 #include <math.h>
  25 #include <limits.h>
  26
  27 #include "access/htup_details.h"
  28 #include "catalog/pg_statistic.h"
  29 #include "commands/tablespace.h"
  30 #include "executor/execdebug.h"
  31 #include "executor/hashjoin.h"
  32 #include "executor/nodeHash.h"
  33 #include "executor/nodeHashjoin.h"
  34 #include "miscadmin.h"
  35 #include "utils/dynahash.h"
  36 #include "utils/memutils.h"
  37 #include "utils/lsyscache.h"
  38 #include "utils/syscache.h"
  39
  40
  41 static void ExecHashIncreaseNumBatches(HashJoinTable hashtable);
  42 static void ExecHashIncreaseNumBuckets(HashJoinTable hashtable);
  43 static void ExecHashBuildSkewHash(HashJoinTable hashtable, Hash *node,
  44                                           int mcvsToUse);
  45 static void ExecHashSkewTableInsert(HashJoinTable hashtable,
  46                                                 TupleTableSlot *slot,
  47                                                 uint32 hashvalue,
  48                                                 int bucketNumber);
  49 static void ExecHashRemoveNextSkewBucket(HashJoinTable hashtable);
  50
  51 static void *dense_alloc(HashJoinTable hashtable, Size size);
  52
  53 /* ----------------------------------------------------------------
  54  *              ExecHash
  55  *
  56  *              stub for pro forma compliance
  57  * ----------------------------------------------------------------
  58  */
  59 TupleTableSlot *
  60 ExecHash(HashState *node)
  61 {
  62         elog(ERROR, "Hash node does not support ExecProcNode call convention");
  63         return NULL;
  64 }
  65
  66 /* ----------------------------------------------------------------
  67  *              MultiExecHash
  68  *
  69  *              build hash table for hashjoin, doing partitioning if more
  70  *              than one batch is required.
  71  * ----------------------------------------------------------------
  72  */
  73 Node *
  74 MultiExecHash(HashState *node)
  75 {
  76         PlanState  *outerNode;
  77         List       *hashkeys;
  78         HashJoinTable hashtable;
  79         TupleTableSlot *slot;
  80         ExprContext *econtext;
  81         uint32          hashvalue;
  82
  83         /* must provide our own instrumentation support */
  84         if (node->ps.instrument)
  85                 InstrStartNode(node->ps.instrument);
  86
  87         /*
  88          * get state info from node
  89          */
  90         outerNode = outerPlanState(node);
  91         hashtable = node->hashtable;
  92
  93         /*
  94          * set expression context
  95          */
  96         hashkeys = node->hashkeys;
  97         econtext = node->ps.ps_ExprContext;
  98
  99         /*
 100          * get all inner tuples and insert into the hash table (or temp files)
 101          */
 102         for (;;)
 103         {
 104                 slot = ExecProcNode(outerNode);
 105                 if (TupIsNull(slot))
 106                         break;
 107                 /* We have to compute the hash value */
 108                 econtext->ecxt_innertuple = slot;
 109                 if (ExecHashGetHashValue(hashtable, econtext, hashkeys,
 110                                                                  false, hashtable->keepNulls,
 111                                                                  &hashvalue))
 112                 {
 113                         int                     bucketNumber;
 114
 115                         bucketNumber = ExecHashGetSkewBucket(hashtable, hashvalue);
 116                         if (bucketNumber != INVALID_SKEW_BUCKET_NO)
 117                         {
 118                                 /* It's a skew tuple, so put it into that hash table */
 119                                 ExecHashSkewTableInsert(hashtable, slot, hashvalue,
 120                                                                                 bucketNumber);
 121                                 hashtable->skewTuples += 1;
 122                         }
 123                         else
 124                         {
 125                                 /* Not subject to skew optimization, so insert normally */
 126                                 ExecHashTableInsert(hashtable, slot, hashvalue);
 127                         }
 128                         hashtable->totalTuples += 1;
 129                 }
 130         }
 131
 132         /* resize the hash table if needed (NTUP_PER_BUCKET exceeded) */
 133         if (hashtable->nbuckets != hashtable->nbuckets_optimal)
 134                 ExecHashIncreaseNumBuckets(hashtable);
 135
 136         /* Account for the buckets in spaceUsed (reported in EXPLAIN ANALYZE) */
 137         hashtable->spaceUsed += hashtable->nbuckets * sizeof(HashJoinTuple);
 138         if (hashtable->spaceUsed > hashtable->spacePeak)
 139                 hashtable->spacePeak = hashtable->spaceUsed;
 140
 141         /* must provide our own instrumentation support */
 142         if (node->ps.instrument)
 143                 InstrStopNode(node->ps.instrument, hashtable->totalTuples);
 144
 145         /*
 146          * We do not return the hash table directly because it's not a subtype of
 147          * Node, and so would violate the MultiExecProcNode API.  Instead, our
 148          * parent Hashjoin node is expected to know how to fish it out of our node
 149          * state.  Ugly but not really worth cleaning up, since Hashjoin knows
 150          * quite a bit more about Hash besides that.
 151          */
 152         return NULL;
 153 }
 154
 155 /* ----------------------------------------------------------------
 156  *              ExecInitHash
 157  *
 158  *              Init routine for Hash node
 159  * ----------------------------------------------------------------
 160  */
 161 HashState *
 162 ExecInitHash(Hash *node, EState *estate, int eflags)
 163 {
 164         HashState  *hashstate;
 165
 166         /* check for unsupported flags */
 167         Assert(!(eflags & (EXEC_FLAG_BACKWARD | EXEC_FLAG_MARK)));
 168
 169         /*
 170          * create state structure
 171          */
 172         hashstate = makeNode(HashState);
 173         hashstate->ps.plan = (Plan *) node;
 174         hashstate->ps.state = estate;
 175         hashstate->hashtable = NULL;
 176         hashstate->hashkeys = NIL;      /* will be set by parent HashJoin */
 177
 178         /*
 179          * Miscellaneous initialization
 180          *
 181          * create expression context for node
 182          */
 183         ExecAssignExprContext(estate, &hashstate->ps);
 184
 185         /*
 186          * initialize our result slot
 187          */
 188         ExecInitResultTupleSlot(estate, &hashstate->ps);
 189
 190         /*
 191          * initialize child expressions
 192          */
 193         hashstate->ps.qual =
 194                 ExecInitQual(node->plan.qual, (PlanState *) hashstate);
 195
 196         /*
 197          * initialize child nodes
 198          */
 199         outerPlanState(hashstate) = ExecInitNode(outerPlan(node), estate, eflags);
 200
 201         /*
 202          * initialize tuple type. no need to initialize projection info because
 203          * this node doesn't do projections
 204          */
 205         ExecAssignResultTypeFromTL(&hashstate->ps);
 206         hashstate->ps.ps_ProjInfo = NULL;
 207
 208         return hashstate;
 209 }
 210
 211 /* ---------------------------------------------------------------
 212  *              ExecEndHash
 213  *
 214  *              clean up routine for Hash node
 215  * ----------------------------------------------------------------
 216  */
 217 void
 218 ExecEndHash(HashState *node)
 219 {
 220         PlanState  *outerPlan;
 221
 222         /*
 223          * free exprcontext
 224          */
 225         ExecFreeExprContext(&node->ps);
 226
 227         /*
 228          * shut down the subplan
 229          */
 230         outerPlan = outerPlanState(node);
 231         ExecEndNode(outerPlan);
 232 }
 233
 234
 235 /* ----------------------------------------------------------------
 236  *              ExecHashTableCreate
 237  *
 238  *              create an empty hashtable data structure for hashjoin.
 239  * ----------------------------------------------------------------
 240  */
 241 HashJoinTable
 242 ExecHashTableCreate(Hash *node, List *hashOperators, bool keepNulls)
 243 {
 244         HashJoinTable hashtable;
 245         Plan       *outerNode;
 246         int                     nbuckets;
 247         int                     nbatch;
 248         int                     num_skew_mcvs;
 249         int                     log2_nbuckets;
 250         int                     nkeys;
 251         int                     i;
 252         ListCell   *ho;
 253         MemoryContext oldcxt;
 254
 255         /*
 256          * Get information about the size of the relation to be hashed (it's the
 257          * "outer" subtree of this node, but the inner relation of the hashjoin).
 258          * Compute the appropriate size of the hash table.
 259          */
 260         outerNode = outerPlan(node);
 261
 262         ExecChooseHashTableSize(outerNode->plan_rows, outerNode->plan_width,
 263                                                         OidIsValid(node->skewTable),
 264                                                         &nbuckets, &nbatch, &num_skew_mcvs);
 265
 266         /* nbuckets must be a power of 2 */
 267         log2_nbuckets = my_log2(nbuckets);
 268         Assert(nbuckets == (1 << log2_nbuckets));
 269
 270         /*
 271          * Initialize the hash table control block.
 272          *
 273          * The hashtable control block is just palloc'd from the executor's
 274          * per-query memory context.
 275          */
 276         hashtable = (HashJoinTable) palloc(sizeof(HashJoinTableData));
 277         hashtable->nbuckets = nbuckets;
 278         hashtable->nbuckets_original = nbuckets;
 279         hashtable->nbuckets_optimal = nbuckets;
 280         hashtable->log2_nbuckets = log2_nbuckets;
 281         hashtable->log2_nbuckets_optimal = log2_nbuckets;
 282         hashtable->buckets = NULL;
 283         hashtable->keepNulls = keepNulls;
 284         hashtable->skewEnabled = false;
 285         hashtable->skewBucket = NULL;
 286         hashtable->skewBucketLen = 0;
 287         hashtable->nSkewBuckets = 0;
 288         hashtable->skewBucketNums = NULL;
 289         hashtable->nbatch = nbatch;
 290         hashtable->curbatch = 0;
 291         hashtable->nbatch_original = nbatch;
 292         hashtable->nbatch_outstart = nbatch;
 293         hashtable->growEnabled = true;
 294         hashtable->totalTuples = 0;
 295         hashtable->skewTuples = 0;
 296         hashtable->innerBatchFile = NULL;
 297         hashtable->outerBatchFile = NULL;
 298         hashtable->spaceUsed = 0;
 299         hashtable->spacePeak = 0;
 300         hashtable->spaceAllowed = work_mem * 1024L;
 301         hashtable->spaceUsedSkew = 0;
 302         hashtable->spaceAllowedSkew =
 303                 hashtable->spaceAllowed * SKEW_WORK_MEM_PERCENT / 100;
 304         hashtable->chunks = NULL;
 305
 306 #ifdef HJDEBUG
 307         printf("Hashjoin %p: initial nbatch = %d, nbuckets = %d\n",
 308                    hashtable, nbatch, nbuckets);
 309 #endif
 310
 311         /*
 312          * Get info about the hash functions to be used for each hash key. Also
 313          * remember whether the join operators are strict.
 314          */
 315         nkeys = list_length(hashOperators);
 316         hashtable->outer_hashfunctions =
 317                 (FmgrInfo *) palloc(nkeys * sizeof(FmgrInfo));
 318         hashtable->inner_hashfunctions =
 319                 (FmgrInfo *) palloc(nkeys * sizeof(FmgrInfo));
 320         hashtable->hashStrict = (bool *) palloc(nkeys * sizeof(bool));
 321         i = 0;
 322         foreach(ho, hashOperators)
 323         {
 324                 Oid                     hashop = lfirst_oid(ho);
 325                 Oid                     left_hashfn;
 326                 Oid                     right_hashfn;
 327
 328                 if (!get_op_hash_functions(hashop, &left_hashfn, &right_hashfn))
 329                         elog(ERROR, "could not find hash function for hash operator %u",
 330                                  hashop);
 331                 fmgr_info(left_hashfn, &hashtable->outer_hashfunctions[i]);
 332                 fmgr_info(right_hashfn, &hashtable->inner_hashfunctions[i]);
 333                 hashtable->hashStrict[i] = op_strict(hashop);
 334                 i++;
 335         }
 336
 337         /*
 338          * Create temporary memory contexts in which to keep the hashtable working
 339          * storage.  See notes in executor/hashjoin.h.
 340          */
 341         hashtable->hashCxt = AllocSetContextCreate(CurrentMemoryContext,
 342                                                                                            "HashTableContext",
 343                                                                                            ALLOCSET_DEFAULT_SIZES);
 344
 345         hashtable->batchCxt = AllocSetContextCreate(hashtable->hashCxt,
 346                                                                                                 "HashBatchContext",
 347                                                                                                 ALLOCSET_DEFAULT_SIZES);
 348
 349         /* Allocate data that will live for the life of the hashjoin */
 350
 351         oldcxt = MemoryContextSwitchTo(hashtable->hashCxt);
 352
 353         if (nbatch > 1)
 354         {
 355                 /*
 356                  * allocate and initialize the file arrays in hashCxt
 357                  */
 358                 hashtable->innerBatchFile = (BufFile **)
 359                         palloc0(nbatch * sizeof(BufFile *));
 360                 hashtable->outerBatchFile = (BufFile **)
 361                         palloc0(nbatch * sizeof(BufFile *));
 362                 /* The files will not be opened until needed... */
 363                 /* ... but make sure we have temp tablespaces established for them */
 364                 PrepareTempTablespaces();
 365         }
 366
 367         /*
 368          * Prepare context for the first-scan space allocations; allocate the
 369          * hashbucket array therein, and set each bucket "empty".
 370          */
 371         MemoryContextSwitchTo(hashtable->batchCxt);
 372
 373         hashtable->buckets = (HashJoinTuple *)
 374                 palloc0(nbuckets * sizeof(HashJoinTuple));
 375
 376         /*
 377          * Set up for skew optimization, if possible and there's a need for more
 378          * than one batch.  (In a one-batch join, there's no point in it.)
 379          */
 380         if (nbatch > 1)
 381                 ExecHashBuildSkewHash(hashtable, node, num_skew_mcvs);
 382
 383         MemoryContextSwitchTo(oldcxt);
 384
 385         return hashtable;
 386 }
 387
 388
 389 /*
 390  * Compute appropriate size for hashtable given the estimated size of the
 391  * relation to be hashed (number of rows and average row width).
 392  *
 393  * This is exported so that the planner's costsize.c can use it.
 394  */
 395
 396 /* Target bucket loading (tuples per bucket) */
 397 #define NTUP_PER_BUCKET                 1
 398
 399 void
 400 ExecChooseHashTableSize(double ntuples, int tupwidth, bool useskew,
 401                                                 int *numbuckets,
 402                                                 int *numbatches,
 403                                                 int *num_skew_mcvs)
 404 {
 405         int                     tupsize;
 406         double          inner_rel_bytes;
 407         long            bucket_bytes;
 408         long            hash_table_bytes;
 409         long            skew_table_bytes;
 410         long            max_pointers;
 411         long            mppow2;
 412         int                     nbatch = 1;
 413         int                     nbuckets;
 414         double          dbuckets;
 415
 416         /* Force a plausible relation size if no info */
 417         if (ntuples <= 0.0)
 418                 ntuples = 1000.0;
 419
 420         /*
 421          * Estimate tupsize based on footprint of tuple in hashtable... note this
 422          * does not allow for any palloc overhead.  The manipulations of spaceUsed
 423          * don't count palloc overhead either.
 424          */
 425         tupsize = HJTUPLE_OVERHEAD +
 426                 MAXALIGN(SizeofMinimalTupleHeader) +
 427                 MAXALIGN(tupwidth);
 428         inner_rel_bytes = ntuples * tupsize;
 429
 430         /*
 431          * Target in-memory hashtable size is work_mem kilobytes.
 432          */
 433         hash_table_bytes = work_mem * 1024L;
 434
 435         /*
 436          * If skew optimization is possible, estimate the number of skew buckets
 437          * that will fit in the memory allowed, and decrement the assumed space
 438          * available for the main hash table accordingly.
 439          *
 440          * We make the optimistic assumption that each skew bucket will contain
 441          * one inner-relation tuple.  If that turns out to be low, we will recover
 442          * at runtime by reducing the number of skew buckets.
 443          *
 444          * hashtable->skewBucket will have up to 8 times as many HashSkewBucket
 445          * pointers as the number of MCVs we allow, since ExecHashBuildSkewHash
 446          * will round up to the next power of 2 and then multiply by 4 to reduce
 447          * collisions.
 448          */
 449         if (useskew)
 450         {
 451                 skew_table_bytes = hash_table_bytes * SKEW_WORK_MEM_PERCENT / 100;
 452
 453                 /*----------
 454                  * Divisor is:
 455                  * size of a hash tuple +
 456                  * worst-case size of skewBucket[] per MCV +
 457                  * size of skewBucketNums[] entry +
 458                  * size of skew bucket struct itself
 459                  *----------
 460                  */
 461                 *num_skew_mcvs = skew_table_bytes / (tupsize +
 462                                                                                          (8 * sizeof(HashSkewBucket *)) +
 463                                                                                          sizeof(int) +
 464                                                                                          SKEW_BUCKET_OVERHEAD);
 465                 if (*num_skew_mcvs > 0)
 466                         hash_table_bytes -= skew_table_bytes;
 467         }
 468         else
 469                 *num_skew_mcvs = 0;
 470
 471         /*
 472          * Set nbuckets to achieve an average bucket load of NTUP_PER_BUCKET when
 473          * memory is filled, assuming a single batch; but limit the value so that
 474          * the pointer arrays we'll try to allocate do not exceed work_mem nor
 475          * MaxAllocSize.
 476          *
 477          * Note that both nbuckets and nbatch must be powers of 2 to make
 478          * ExecHashGetBucketAndBatch fast.
 479          */
 480         max_pointers = (work_mem * 1024L) / sizeof(HashJoinTuple);
 481         max_pointers = Min(max_pointers, MaxAllocSize / sizeof(HashJoinTuple));
 482         /* If max_pointers isn't a power of 2, must round it down to one */
 483         mppow2 = 1L << my_log2(max_pointers);
 484         if (max_pointers != mppow2)
 485                 max_pointers = mppow2 / 2;
 486
 487         /* Also ensure we avoid integer overflow in nbatch and nbuckets */
 488         /* (this step is redundant given the current value of MaxAllocSize) */
 489         max_pointers = Min(max_pointers, INT_MAX / 2);
 490
 491         dbuckets = ceil(ntuples / NTUP_PER_BUCKET);
 492         dbuckets = Min(dbuckets, max_pointers);
 493         nbuckets = (int) dbuckets;
 494         /* don't let nbuckets be really small, though ... */
 495         nbuckets = Max(nbuckets, 1024);
 496         /* ... and force it to be a power of 2. */
 497         nbuckets = 1 << my_log2(nbuckets);
 498
 499         /*
 500          * If there's not enough space to store the projected number of tuples and
 501          * the required bucket headers, we will need multiple batches.
 502          */
 503         bucket_bytes = sizeof(HashJoinTuple) * nbuckets;
 504         if (inner_rel_bytes + bucket_bytes > hash_table_bytes)
 505         {
 506                 /* We'll need multiple batches */
 507                 long            lbuckets;
 508                 double          dbatch;
 509                 int                     minbatch;
 510                 long            bucket_size;
 511
 512                 /*
 513                  * Estimate the number of buckets we'll want to have when work_mem is
 514                  * entirely full.  Each bucket will contain a bucket pointer plus
 515                  * NTUP_PER_BUCKET tuples, whose projected size already includes
 516                  * overhead for the hash code, pointer to the next tuple, etc.
 517                  */
 518                 bucket_size = (tupsize * NTUP_PER_BUCKET + sizeof(HashJoinTuple));
 519                 lbuckets = 1L << my_log2(hash_table_bytes / bucket_size);
 520                 lbuckets = Min(lbuckets, max_pointers);
 521                 nbuckets = (int) lbuckets;
 522                 nbuckets = 1 << my_log2(nbuckets);
 523                 bucket_bytes = nbuckets * sizeof(HashJoinTuple);
 524
 525                 /*
 526                  * Buckets are simple pointers to hashjoin tuples, while tupsize
 527                  * includes the pointer, hash code, and MinimalTupleData.  So buckets
 528                  * should never really exceed 25% of work_mem (even for
 529                  * NTUP_PER_BUCKET=1); except maybe for work_mem values that are not
 530                  * 2^N bytes, where we might get more because of doubling. So let's
 531                  * look for 50% here.
 532                  */
 533                 Assert(bucket_bytes <= hash_table_bytes / 2);
 534
 535                 /* Calculate required number of batches. */
 536                 dbatch = ceil(inner_rel_bytes / (hash_table_bytes - bucket_bytes));
 537                 dbatch = Min(dbatch, max_pointers);
 538                 minbatch = (int) dbatch;
 539                 nbatch = 2;
 540                 while (nbatch < minbatch)
 541                         nbatch <<= 1;
 542         }
 543
 544         Assert(nbuckets > 0);
 545         Assert(nbatch > 0);
 546
 547         *numbuckets = nbuckets;
 548         *numbatches = nbatch;
 549 }
 550
 551
 552 /* ----------------------------------------------------------------
 553  *              ExecHashTableDestroy
 554  *
 555  *              destroy a hash table
 556  * ----------------------------------------------------------------
 557  */
 558 void
 559 ExecHashTableDestroy(HashJoinTable hashtable)
 560 {
 561         int                     i;
 562
 563         /*
 564          * Make sure all the temp files are closed.  We skip batch 0, since it
 565          * can't have any temp files (and the arrays might not even exist if
 566          * nbatch is only 1).
 567          */
 568         for (i = 1; i < hashtable->nbatch; i++)
 569         {
 570                 if (hashtable->innerBatchFile[i])
 571                         BufFileClose(hashtable->innerBatchFile[i]);
 572                 if (hashtable->outerBatchFile[i])
 573                         BufFileClose(hashtable->outerBatchFile[i]);
 574         }
 575
 576         /* Release working memory (batchCxt is a child, so it goes away too) */
 577         MemoryContextDelete(hashtable->hashCxt);
 578
 579         /* And drop the control block */
 580         pfree(hashtable);
 581 }
 582
 583 /*
 584  * ExecHashIncreaseNumBatches
 585  *              increase the original number of batches in order to reduce
 586  *              current memory consumption
 587  */
 588 static void
 589 ExecHashIncreaseNumBatches(HashJoinTable hashtable)
 590 {
 591         int                     oldnbatch = hashtable->nbatch;
 592         int                     curbatch = hashtable->curbatch;
 593         int                     nbatch;
 594         MemoryContext oldcxt;
 595         long            ninmemory;
 596         long            nfreed;
 597         HashMemoryChunk oldchunks;
 598
 599         /* do nothing if we've decided to shut off growth */
 600         if (!hashtable->growEnabled)
 601                 return;
 602
 603         /* safety check to avoid overflow */
 604         if (oldnbatch > Min(INT_MAX / 2, MaxAllocSize / (sizeof(void *) * 2)))
 605                 return;
 606
 607         nbatch = oldnbatch * 2;
 608         Assert(nbatch > 1);
 609
 610 #ifdef HJDEBUG
 611         printf("Hashjoin %p: increasing nbatch to %d because space = %zu\n",
 612                    hashtable, nbatch, hashtable->spaceUsed);
 613 #endif
 614
 615         oldcxt = MemoryContextSwitchTo(hashtable->hashCxt);
 616
 617         if (hashtable->innerBatchFile == NULL)
 618         {
 619                 /* we had no file arrays before */
 620                 hashtable->innerBatchFile = (BufFile **)
 621                         palloc0(nbatch * sizeof(BufFile *));
 622                 hashtable->outerBatchFile = (BufFile **)
 623                         palloc0(nbatch * sizeof(BufFile *));
 624                 /* time to establish the temp tablespaces, too */
 625                 PrepareTempTablespaces();
 626         }
 627         else
 628         {
 629                 /* enlarge arrays and zero out added entries */
 630                 hashtable->innerBatchFile = (BufFile **)
 631                         repalloc(hashtable->innerBatchFile, nbatch * sizeof(BufFile *));
 632                 hashtable->outerBatchFile = (BufFile **)
 633                         repalloc(hashtable->outerBatchFile, nbatch * sizeof(BufFile *));
 634                 MemSet(hashtable->innerBatchFile + oldnbatch, 0,
 635                            (nbatch - oldnbatch) * sizeof(BufFile *));
 636                 MemSet(hashtable->outerBatchFile + oldnbatch, 0,
 637                            (nbatch - oldnbatch) * sizeof(BufFile *));
 638         }
 639
 640         MemoryContextSwitchTo(oldcxt);
 641
 642         hashtable->nbatch = nbatch;
 643
 644         /*
 645          * Scan through the existing hash table entries and dump out any that are
 646          * no longer of the current batch.
 647          */
 648         ninmemory = nfreed = 0;
 649
 650         /* If know we need to resize nbuckets, we can do it while rebatching. */
 651         if (hashtable->nbuckets_optimal != hashtable->nbuckets)
 652         {
 653                 /* we never decrease the number of buckets */
 654                 Assert(hashtable->nbuckets_optimal > hashtable->nbuckets);
 655
 656                 hashtable->nbuckets = hashtable->nbuckets_optimal;
 657                 hashtable->log2_nbuckets = hashtable->log2_nbuckets_optimal;
 658
 659                 hashtable->buckets = repalloc(hashtable->buckets,
 660                                                                 sizeof(HashJoinTuple) * hashtable->nbuckets);
 661         }
 662
 663         /*
 664          * We will scan through the chunks directly, so that we can reset the
 665          * buckets now and not have to keep track which tuples in the buckets have
 666          * already been processed. We will free the old chunks as we go.
 667          */
 668         memset(hashtable->buckets, 0, sizeof(HashJoinTuple) * hashtable->nbuckets);
 669         oldchunks = hashtable->chunks;
 670         hashtable->chunks = NULL;
 671
 672         /* so, let's scan through the old chunks, and all tuples in each chunk */
 673         while (oldchunks != NULL)
 674         {
 675                 HashMemoryChunk nextchunk = oldchunks->next;
 676
 677                 /* position within the buffer (up to oldchunks->used) */
 678                 size_t          idx = 0;
 679
 680                 /* process all tuples stored in this chunk (and then free it) */
 681                 while (idx < oldchunks->used)
 682                 {
 683                         HashJoinTuple hashTuple = (HashJoinTuple) (oldchunks->data + idx);
 684                         MinimalTuple tuple = HJTUPLE_MINTUPLE(hashTuple);
 685                         int                     hashTupleSize = (HJTUPLE_OVERHEAD + tuple->t_len);
 686                         int                     bucketno;
 687                         int                     batchno;
 688
 689                         ninmemory++;
 690                         ExecHashGetBucketAndBatch(hashtable, hashTuple->hashvalue,
 691                                                                           &bucketno, &batchno);
 692
 693                         if (batchno == curbatch)
 694                         {
 695                                 /* keep tuple in memory - copy it into the new chunk */
 696                                 HashJoinTuple copyTuple;
 697
 698                                 copyTuple = (HashJoinTuple) dense_alloc(hashtable, hashTupleSize);
 699                                 memcpy(copyTuple, hashTuple, hashTupleSize);
 700
 701                                 /* and add it back to the appropriate bucket */
 702                                 copyTuple->next = hashtable->buckets[bucketno];
 703                                 hashtable->buckets[bucketno] = copyTuple;
 704                         }
 705                         else
 706                         {
 707                                 /* dump it out */
 708                                 Assert(batchno > curbatch);
 709                                 ExecHashJoinSaveTuple(HJTUPLE_MINTUPLE(hashTuple),
 710                                                                           hashTuple->hashvalue,
 711                                                                           &hashtable->innerBatchFile[batchno]);
 712
 713                                 hashtable->spaceUsed -= hashTupleSize;
 714                                 nfreed++;
 715                         }
 716
 717                         /* next tuple in this chunk */
 718                         idx += MAXALIGN(hashTupleSize);
 719
 720                         /* allow this loop to be cancellable */
 721                         CHECK_FOR_INTERRUPTS();
 722                 }
 723
 724                 /* we're done with this chunk - free it and proceed to the next one */
 725                 pfree(oldchunks);
 726                 oldchunks = nextchunk;
 727         }
 728
 729 #ifdef HJDEBUG
 730         printf("Hashjoin %p: freed %ld of %ld tuples, space now %zu\n",
 731                    hashtable, nfreed, ninmemory, hashtable->spaceUsed);
 732 #endif
 733
 734         /*
 735          * If we dumped out either all or none of the tuples in the table, disable
 736          * further expansion of nbatch.  This situation implies that we have
 737          * enough tuples of identical hashvalues to overflow spaceAllowed.
 738          * Increasing nbatch will not fix it since there's no way to subdivide the
 739          * group any more finely. We have to just gut it out and hope the server
 740          * has enough RAM.
 741          */
 742         if (nfreed == 0 || nfreed == ninmemory)
 743         {
 744                 hashtable->growEnabled = false;
 745 #ifdef HJDEBUG
 746                 printf("Hashjoin %p: disabling further increase of nbatch\n",
 747                            hashtable);
 748 #endif
 749         }
 750 }
 751
 752 /*
 753  * ExecHashIncreaseNumBuckets
 754  *              increase the original number of buckets in order to reduce
 755  *              number of tuples per bucket
 756  */
 757 static void
 758 ExecHashIncreaseNumBuckets(HashJoinTable hashtable)
 759 {
 760         HashMemoryChunk chunk;
 761
 762         /* do nothing if not an increase (it's called increase for a reason) */
 763         if (hashtable->nbuckets >= hashtable->nbuckets_optimal)
 764                 return;
 765
 766 #ifdef HJDEBUG
 767         printf("Hashjoin %p: increasing nbuckets %d => %d\n",
 768                    hashtable, hashtable->nbuckets, hashtable->nbuckets_optimal);
 769 #endif
 770
 771         hashtable->nbuckets = hashtable->nbuckets_optimal;
 772         hashtable->log2_nbuckets = hashtable->log2_nbuckets_optimal;
 773
 774         Assert(hashtable->nbuckets > 1);
 775         Assert(hashtable->nbuckets <= (INT_MAX / 2));
 776         Assert(hashtable->nbuckets == (1 << hashtable->log2_nbuckets));
 777
 778         /*
 779          * Just reallocate the proper number of buckets - we don't need to walk
 780          * through them - we can walk the dense-allocated chunks (just like in
 781          * ExecHashIncreaseNumBatches, but without all the copying into new
 782          * chunks)
 783          */
 784         hashtable->buckets =
 785                 (HashJoinTuple *) repalloc(hashtable->buckets,
 786                                                                 hashtable->nbuckets * sizeof(HashJoinTuple));
 787
 788         memset(hashtable->buckets, 0, hashtable->nbuckets * sizeof(HashJoinTuple));
 789
 790         /* scan through all tuples in all chunks to rebuild the hash table */
 791         for (chunk = hashtable->chunks; chunk != NULL; chunk = chunk->next)
 792         {
 793                 /* process all tuples stored in this chunk */
 794                 size_t          idx = 0;
 795
 796                 while (idx < chunk->used)
 797                 {
 798                         HashJoinTuple hashTuple = (HashJoinTuple) (chunk->data + idx);
 799                         int                     bucketno;
 800                         int                     batchno;
 801
 802                         ExecHashGetBucketAndBatch(hashtable, hashTuple->hashvalue,
 803                                                                           &bucketno, &batchno);
 804
 805                         /* add the tuple to the proper bucket */
 806                         hashTuple->next = hashtable->buckets[bucketno];
 807                         hashtable->buckets[bucketno] = hashTuple;
 808
 809                         /* advance index past the tuple */
 810                         idx += MAXALIGN(HJTUPLE_OVERHEAD +
 811                                                         HJTUPLE_MINTUPLE(hashTuple)->t_len);
 812                 }
 813         }
 814 }
 815
 816
 817 /*
 818  * ExecHashTableInsert
 819  *              insert a tuple into the hash table depending on the hash value
 820  *              it may just go to a temp file for later batches
 821  *
 822  * Note: the passed TupleTableSlot may contain a regular, minimal, or virtual
 823  * tuple; the minimal case in particular is certain to happen while reloading
 824  * tuples from batch files.  We could save some cycles in the regular-tuple
 825  * case by not forcing the slot contents into minimal form; not clear if it's
 826  * worth the messiness required.
 827  */
 828 void
 829 ExecHashTableInsert(HashJoinTable hashtable,
 830                                         TupleTableSlot *slot,
 831                                         uint32 hashvalue)
 832 {
 833         MinimalTuple tuple = ExecFetchSlotMinimalTuple(slot);
 834         int                     bucketno;
 835         int                     batchno;
 836
 837         ExecHashGetBucketAndBatch(hashtable, hashvalue,
 838                                                           &bucketno, &batchno);
 839
 840         /*
 841          * decide whether to put the tuple in the hash table or a temp file
 842          */
 843         if (batchno == hashtable->curbatch)
 844         {
 845                 /*
 846                  * put the tuple in hash table
 847                  */
 848                 HashJoinTuple hashTuple;
 849                 int                     hashTupleSize;
 850                 double          ntuples = (hashtable->totalTuples - hashtable->skewTuples);
 851
 852                 /* Create the HashJoinTuple */
 853                 hashTupleSize = HJTUPLE_OVERHEAD + tuple->t_len;
 854                 hashTuple = (HashJoinTuple) dense_alloc(hashtable, hashTupleSize);
 855
 856                 hashTuple->hashvalue = hashvalue;
 857                 memcpy(HJTUPLE_MINTUPLE(hashTuple), tuple, tuple->t_len);
 858
 859                 /*
 860                  * We always reset the tuple-matched flag on insertion.  This is okay
 861                  * even when reloading a tuple from a batch file, since the tuple
 862                  * could not possibly have been matched to an outer tuple before it
 863                  * went into the batch file.
 864                  */
 865                 HeapTupleHeaderClearMatch(HJTUPLE_MINTUPLE(hashTuple));
 866
 867                 /* Push it onto the front of the bucket's list */
 868                 hashTuple->next = hashtable->buckets[bucketno];
 869                 hashtable->buckets[bucketno] = hashTuple;
 870
 871                 /*
 872                  * Increase the (optimal) number of buckets if we just exceeded the
 873                  * NTUP_PER_BUCKET threshold, but only when there's still a single
 874                  * batch.
 875                  */
 876                 if (hashtable->nbatch == 1 &&
 877                         ntuples > (hashtable->nbuckets_optimal * NTUP_PER_BUCKET))
 878                 {
 879                         /* Guard against integer overflow and alloc size overflow */
 880                         if (hashtable->nbuckets_optimal <= INT_MAX / 2 &&
 881                                 hashtable->nbuckets_optimal * 2 <= MaxAllocSize / sizeof(HashJoinTuple))
 882                         {
 883                                 hashtable->nbuckets_optimal *= 2;
 884                                 hashtable->log2_nbuckets_optimal += 1;
 885                         }
 886                 }
 887
 888                 /* Account for space used, and back off if we've used too much */
 889                 hashtable->spaceUsed += hashTupleSize;
 890                 if (hashtable->spaceUsed > hashtable->spacePeak)
 891                         hashtable->spacePeak = hashtable->spaceUsed;
 892                 if (hashtable->spaceUsed +
 893                         hashtable->nbuckets_optimal * sizeof(HashJoinTuple)
 894                         > hashtable->spaceAllowed)
 895                         ExecHashIncreaseNumBatches(hashtable);
 896         }
 897         else
 898         {
 899                 /*
 900                  * put the tuple into a temp file for later batches
 901                  */
 902                 Assert(batchno > hashtable->curbatch);
 903                 ExecHashJoinSaveTuple(tuple,
 904                                                           hashvalue,
 905                                                           &hashtable->innerBatchFile[batchno]);
 906         }
 907 }
 908
 909 /*
 910  * ExecHashGetHashValue
 911  *              Compute the hash value for a tuple
 912  *
 913  * The tuple to be tested must be in either econtext->ecxt_outertuple or
 914  * econtext->ecxt_innertuple.  Vars in the hashkeys expressions should have
 915  * varno either OUTER_VAR or INNER_VAR.
 916  *
 917  * A TRUE result means the tuple's hash value has been successfully computed
 918  * and stored at *hashvalue.  A FALSE result means the tuple cannot match
 919  * because it contains a null attribute, and hence it should be discarded
 920  * immediately.  (If keep_nulls is true then FALSE is never returned.)
 921  */
 922 bool
 923 ExecHashGetHashValue(HashJoinTable hashtable,
 924                                          ExprContext *econtext,
 925                                          List *hashkeys,
 926                                          bool outer_tuple,
 927                                          bool keep_nulls,
 928                                          uint32 *hashvalue)
 929 {
 930         uint32          hashkey = 0;
 931         FmgrInfo   *hashfunctions;
 932         ListCell   *hk;
 933         int                     i = 0;
 934         MemoryContext oldContext;
 935
 936         /*
 937          * We reset the eval context each time to reclaim any memory leaked in the
 938          * hashkey expressions.
 939          */
 940         ResetExprContext(econtext);
 941
 942         oldContext = MemoryContextSwitchTo(econtext->ecxt_per_tuple_memory);
 943
 944         if (outer_tuple)
 945                 hashfunctions = hashtable->outer_hashfunctions;
 946         else
 947                 hashfunctions = hashtable->inner_hashfunctions;
 948
 949         foreach(hk, hashkeys)
 950         {
 951                 ExprState  *keyexpr = (ExprState *) lfirst(hk);
 952                 Datum           keyval;
 953                 bool            isNull;
 954
 955                 /* rotate hashkey left 1 bit at each step */
 956                 hashkey = (hashkey << 1) | ((hashkey & 0x80000000) ? 1 : 0);
 957
 958                 /*
 959                  * Get the join attribute value of the tuple
 960                  */
 961                 keyval = ExecEvalExpr(keyexpr, econtext, &isNull);
 962
 963                 /*
 964                  * If the attribute is NULL, and the join operator is strict, then
 965                  * this tuple cannot pass the join qual so we can reject it
 966                  * immediately (unless we're scanning the outside of an outer join, in
 967                  * which case we must not reject it).  Otherwise we act like the
 968                  * hashcode of NULL is zero (this will support operators that act like
 969                  * IS NOT DISTINCT, though not any more-random behavior).  We treat
 970                  * the hash support function as strict even if the operator is not.
 971                  *
 972                  * Note: currently, all hashjoinable operators must be strict since
 973                  * the hash index AM assumes that.  However, it takes so little extra
 974                  * code here to allow non-strict that we may as well do it.
 975                  */
 976                 if (isNull)
 977                 {
 978                         if (hashtable->hashStrict[i] && !keep_nulls)
 979                         {
 980                                 MemoryContextSwitchTo(oldContext);
 981                                 return false;   /* cannot match */
 982                         }
 983                         /* else, leave hashkey unmodified, equivalent to hashcode 0 */
 984                 }
 985                 else
 986                 {
 987                         /* Compute the hash function */
 988                         uint32          hkey;
 989
 990                         hkey = DatumGetUInt32(FunctionCall1(&hashfunctions[i], keyval));
 991                         hashkey ^= hkey;
 992                 }
 993
 994                 i++;
 995         }
 996
 997         MemoryContextSwitchTo(oldContext);
 998
 999         *hashvalue = hashkey;
1000         return true;
1001 }
1002
1003 /*
1004  * ExecHashGetBucketAndBatch
1005  *              Determine the bucket number and batch number for a hash value
1006  *
1007  * Note: on-the-fly increases of nbatch must not change the bucket number
1008  * for a given hash code (since we don't move tuples to different hash
1009  * chains), and must only cause the batch number to remain the same or
1010  * increase.  Our algorithm is
1011  *              bucketno = hashvalue MOD nbuckets
1012  *              batchno = (hashvalue DIV nbuckets) MOD nbatch
1013  * where nbuckets and nbatch are both expected to be powers of 2, so we can
1014  * do the computations by shifting and masking.  (This assumes that all hash
1015  * functions are good about randomizing all their output bits, else we are
1016  * likely to have very skewed bucket or batch occupancy.)
1017  *
1018  * nbuckets and log2_nbuckets may change while nbatch == 1 because of dynamic
1019  * bucket count growth.  Once we start batching, the value is fixed and does
1020  * not change over the course of the join (making it possible to compute batch
1021  * number the way we do here).
1022  *
1023  * nbatch is always a power of 2; we increase it only by doubling it.  This
1024  * effectively adds one more bit to the top of the batchno.
1025  */
1026 void
1027 ExecHashGetBucketAndBatch(HashJoinTable hashtable,
1028                                                   uint32 hashvalue,
1029                                                   int *bucketno,
1030                                                   int *batchno)
1031 {
1032         uint32          nbuckets = (uint32) hashtable->nbuckets;
1033         uint32          nbatch = (uint32) hashtable->nbatch;
1034
1035         if (nbatch > 1)
1036         {
1037                 /* we can do MOD by masking, DIV by shifting */
1038                 *bucketno = hashvalue & (nbuckets - 1);
1039                 *batchno = (hashvalue >> hashtable->log2_nbuckets) & (nbatch - 1);
1040         }
1041         else
1042         {
1043                 *bucketno = hashvalue & (nbuckets - 1);
1044                 *batchno = 0;
1045         }
1046 }
1047
1048 /*
1049  * ExecScanHashBucket
1050  *              scan a hash bucket for matches to the current outer tuple
1051  *
1052  * The current outer tuple must be stored in econtext->ecxt_outertuple.
1053  *
1054  * On success, the inner tuple is stored into hjstate->hj_CurTuple and
1055  * econtext->ecxt_innertuple, using hjstate->hj_HashTupleSlot as the slot
1056  * for the latter.
1057  */
1058 bool
1059 ExecScanHashBucket(HashJoinState *hjstate,
1060                                    ExprContext *econtext)
1061 {
1062         ExprState  *hjclauses = hjstate->hashclauses;
1063         HashJoinTable hashtable = hjstate->hj_HashTable;
1064         HashJoinTuple hashTuple = hjstate->hj_CurTuple;
1065         uint32          hashvalue = hjstate->hj_CurHashValue;
1066
1067         /*
1068          * hj_CurTuple is the address of the tuple last returned from the current
1069          * bucket, or NULL if it's time to start scanning a new bucket.
1070          *
1071          * If the tuple hashed to a skew bucket then scan the skew bucket
1072          * otherwise scan the standard hashtable bucket.
1073          */
1074         if (hashTuple != NULL)
1075                 hashTuple = hashTuple->next;
1076         else if (hjstate->hj_CurSkewBucketNo != INVALID_SKEW_BUCKET_NO)
1077                 hashTuple = hashtable->skewBucket[hjstate->hj_CurSkewBucketNo]->tuples;
1078         else
1079                 hashTuple = hashtable->buckets[hjstate->hj_CurBucketNo];
1080
1081         while (hashTuple != NULL)
1082         {
1083                 if (hashTuple->hashvalue == hashvalue)
1084                 {
1085                         TupleTableSlot *inntuple;
1086
1087                         /* insert hashtable's tuple into exec slot so ExecQual sees it */
1088                         inntuple = ExecStoreMinimalTuple(HJTUPLE_MINTUPLE(hashTuple),
1089                                                                                          hjstate->hj_HashTupleSlot,
1090                                                                                          false);        /* do not pfree */
1091                         econtext->ecxt_innertuple = inntuple;
1092
1093                         /* reset temp memory each time to avoid leaks from qual expr */
1094                         ResetExprContext(econtext);
1095
1096                         if (ExecQual(hjclauses, econtext))
1097                         {
1098                                 hjstate->hj_CurTuple = hashTuple;
1099                                 return true;
1100                         }
1101                 }
1102
1103                 hashTuple = hashTuple->next;
1104         }
1105
1106         /*
1107          * no match
1108          */
1109         return false;
1110 }
1111
1112 /*
1113  * ExecPrepHashTableForUnmatched
1114  *              set up for a series of ExecScanHashTableForUnmatched calls
1115  */
1116 void
1117 ExecPrepHashTableForUnmatched(HashJoinState *hjstate)
1118 {
1119         /*----------
1120          * During this scan we use the HashJoinState fields as follows:
1121          *
1122          * hj_CurBucketNo: next regular bucket to scan
1123          * hj_CurSkewBucketNo: next skew bucket (an index into skewBucketNums)
1124          * hj_CurTuple: last tuple returned, or NULL to start next bucket
1125          *----------
1126          */
1127         hjstate->hj_CurBucketNo = 0;
1128         hjstate->hj_CurSkewBucketNo = 0;
1129         hjstate->hj_CurTuple = NULL;
1130 }
1131
1132 /*
1133  * ExecScanHashTableForUnmatched
1134  *              scan the hash table for unmatched inner tuples
1135  *
1136  * On success, the inner tuple is stored into hjstate->hj_CurTuple and
1137  * econtext->ecxt_innertuple, using hjstate->hj_HashTupleSlot as the slot
1138  * for the latter.
1139  */
1140 bool
1141 ExecScanHashTableForUnmatched(HashJoinState *hjstate, ExprContext *econtext)
1142 {
1143         HashJoinTable hashtable = hjstate->hj_HashTable;
1144         HashJoinTuple hashTuple = hjstate->hj_CurTuple;
1145
1146         for (;;)
1147         {
1148                 /*
1149                  * hj_CurTuple is the address of the tuple last returned from the
1150                  * current bucket, or NULL if it's time to start scanning a new
1151                  * bucket.
1152                  */
1153                 if (hashTuple != NULL)
1154                         hashTuple = hashTuple->next;
1155                 else if (hjstate->hj_CurBucketNo < hashtable->nbuckets)
1156                 {
1157                         hashTuple = hashtable->buckets[hjstate->hj_CurBucketNo];
1158                         hjstate->hj_CurBucketNo++;
1159                 }
1160                 else if (hjstate->hj_CurSkewBucketNo < hashtable->nSkewBuckets)
1161                 {
1162                         int                     j = hashtable->skewBucketNums[hjstate->hj_CurSkewBucketNo];
1163
1164                         hashTuple = hashtable->skewBucket[j]->tuples;
1165                         hjstate->hj_CurSkewBucketNo++;
1166                 }
1167                 else
1168                         break;                          /* finished all buckets */
1169
1170                 while (hashTuple != NULL)
1171                 {
1172                         if (!HeapTupleHeaderHasMatch(HJTUPLE_MINTUPLE(hashTuple)))
1173                         {
1174                                 TupleTableSlot *inntuple;
1175
1176                                 /* insert hashtable's tuple into exec slot */
1177                                 inntuple = ExecStoreMinimalTuple(HJTUPLE_MINTUPLE(hashTuple),
1178                                                                                                  hjstate->hj_HashTupleSlot,
1179                                                                                                  false);                /* do not pfree */
1180                                 econtext->ecxt_innertuple = inntuple;
1181
1182                                 /*
1183                                  * Reset temp memory each time; although this function doesn't
1184                                  * do any qual eval, the caller will, so let's keep it
1185                                  * parallel to ExecScanHashBucket.
1186                                  */
1187                                 ResetExprContext(econtext);
1188
1189                                 hjstate->hj_CurTuple = hashTuple;
1190                                 return true;
1191                         }
1192
1193                         hashTuple = hashTuple->next;
1194                 }
1195         }
1196
1197         /*
1198          * no more unmatched tuples
1199          */
1200         return false;
1201 }
1202
1203 /*
1204  * ExecHashTableReset
1205  *
1206  *              reset hash table header for new batch
1207  */
1208 void
1209 ExecHashTableReset(HashJoinTable hashtable)
1210 {
1211         MemoryContext oldcxt;
1212         int                     nbuckets = hashtable->nbuckets;
1213
1214         /*
1215          * Release all the hash buckets and tuples acquired in the prior pass, and
1216          * reinitialize the context for a new pass.
1217          */
1218         MemoryContextReset(hashtable->batchCxt);
1219         oldcxt = MemoryContextSwitchTo(hashtable->batchCxt);
1220
1221         /* Reallocate and reinitialize the hash bucket headers. */
1222         hashtable->buckets = (HashJoinTuple *)
1223                 palloc0(nbuckets * sizeof(HashJoinTuple));
1224
1225         hashtable->spaceUsed = 0;
1226
1227         MemoryContextSwitchTo(oldcxt);
1228
1229         /* Forget the chunks (the memory was freed by the context reset above). */
1230         hashtable->chunks = NULL;
1231 }
1232
1233 /*
1234  * ExecHashTableResetMatchFlags
1235  *              Clear all the HeapTupleHeaderHasMatch flags in the table
1236  */
1237 void
1238 ExecHashTableResetMatchFlags(HashJoinTable hashtable)
1239 {
1240         HashJoinTuple tuple;
1241         int                     i;
1242
1243         /* Reset all flags in the main table ... */
1244         for (i = 0; i < hashtable->nbuckets; i++)
1245         {
1246                 for (tuple = hashtable->buckets[i]; tuple != NULL; tuple = tuple->next)
1247                         HeapTupleHeaderClearMatch(HJTUPLE_MINTUPLE(tuple));
1248         }
1249
1250         /* ... and the same for the skew buckets, if any */
1251         for (i = 0; i < hashtable->nSkewBuckets; i++)
1252         {
1253                 int                     j = hashtable->skewBucketNums[i];
1254                 HashSkewBucket *skewBucket = hashtable->skewBucket[j];
1255
1256                 for (tuple = skewBucket->tuples; tuple != NULL; tuple = tuple->next)
1257                         HeapTupleHeaderClearMatch(HJTUPLE_MINTUPLE(tuple));
1258         }
1259 }
1260
1261
1262 void
1263 ExecReScanHash(HashState *node)
1264 {
1265         /*
1266          * if chgParam of subnode is not null then plan will be re-scanned by
1267          * first ExecProcNode.
1268          */
1269         if (node->ps.lefttree->chgParam == NULL)
1270                 ExecReScan(node->ps.lefttree);
1271 }
1272
1273
1274 /*
1275  * ExecHashBuildSkewHash
1276  *
1277  *              Set up for skew optimization if we can identify the most common values
1278  *              (MCVs) of the outer relation's join key.  We make a skew hash bucket
1279  *              for the hash value of each MCV, up to the number of slots allowed
1280  *              based on available memory.
1281  */
1282 static void
1283 ExecHashBuildSkewHash(HashJoinTable hashtable, Hash *node, int mcvsToUse)
1284 {
1285         HeapTupleData *statsTuple;
1286         Datum      *values;
1287         int                     nvalues;
1288         float4     *numbers;
1289         int                     nnumbers;
1290
1291         /* Do nothing if planner didn't identify the outer relation's join key */
1292         if (!OidIsValid(node->skewTable))
1293                 return;
1294         /* Also, do nothing if we don't have room for at least one skew bucket */
1295         if (mcvsToUse <= 0)
1296                 return;
1297
1298         /*
1299          * Try to find the MCV statistics for the outer relation's join key.
1300          */
1301         statsTuple = SearchSysCache3(STATRELATTINH,
1302                                                                  ObjectIdGetDatum(node->skewTable),
1303                                                                  Int16GetDatum(node->skewColumn),
1304                                                                  BoolGetDatum(node->skewInherit));
1305         if (!HeapTupleIsValid(statsTuple))
1306                 return;
1307
1308         if (get_attstatsslot(statsTuple, node->skewColType, node->skewColTypmod,
1309                                                  STATISTIC_KIND_MCV, InvalidOid,
1310                                                  NULL,
1311                                                  &values, &nvalues,
1312                                                  &numbers, &nnumbers))
1313         {
1314                 double          frac;
1315                 int                     nbuckets;
1316                 FmgrInfo   *hashfunctions;
1317                 int                     i;
1318
1319                 if (mcvsToUse > nvalues)
1320                         mcvsToUse = nvalues;
1321
1322                 /*
1323                  * Calculate the expected fraction of outer relation that will
1324                  * participate in the skew optimization.  If this isn't at least
1325                  * SKEW_MIN_OUTER_FRACTION, don't use skew optimization.
1326                  */
1327                 frac = 0;
1328                 for (i = 0; i < mcvsToUse; i++)
1329                         frac += numbers[i];
1330                 if (frac < SKEW_MIN_OUTER_FRACTION)
1331                 {
1332                         free_attstatsslot(node->skewColType,
1333                                                           values, nvalues, numbers, nnumbers);
1334                         ReleaseSysCache(statsTuple);
1335                         return;
1336                 }
1337
1338                 /*
1339                  * Okay, set up the skew hashtable.
1340                  *
1341                  * skewBucket[] is an open addressing hashtable with a power of 2 size
1342                  * that is greater than the number of MCV values.  (This ensures there
1343                  * will be at least one null entry, so searches will always
1344                  * terminate.)
1345                  *
1346                  * Note: this code could fail if mcvsToUse exceeds INT_MAX/8 or
1347                  * MaxAllocSize/sizeof(void *)/8, but that is not currently possible
1348                  * since we limit pg_statistic entries to much less than that.
1349                  */
1350                 nbuckets = 2;
1351                 while (nbuckets <= mcvsToUse)
1352                         nbuckets <<= 1;
1353                 /* use two more bits just to help avoid collisions */
1354                 nbuckets <<= 2;
1355
1356                 hashtable->skewEnabled = true;
1357                 hashtable->skewBucketLen = nbuckets;
1358
1359                 /*
1360                  * We allocate the bucket memory in the hashtable's batch context. It
1361                  * is only needed during the first batch, and this ensures it will be
1362                  * automatically removed once the first batch is done.
1363                  */
1364                 hashtable->skewBucket = (HashSkewBucket **)
1365                         MemoryContextAllocZero(hashtable->batchCxt,
1366                                                                    nbuckets * sizeof(HashSkewBucket *));
1367                 hashtable->skewBucketNums = (int *)
1368                         MemoryContextAllocZero(hashtable->batchCxt,
1369                                                                    mcvsToUse * sizeof(int));
1370
1371                 hashtable->spaceUsed += nbuckets * sizeof(HashSkewBucket *)
1372                         + mcvsToUse * sizeof(int);
1373                 hashtable->spaceUsedSkew += nbuckets * sizeof(HashSkewBucket *)
1374                         + mcvsToUse * sizeof(int);
1375                 if (hashtable->spaceUsed > hashtable->spacePeak)
1376                         hashtable->spacePeak = hashtable->spaceUsed;
1377
1378                 /*
1379                  * Create a skew bucket for each MCV hash value.
1380                  *
1381                  * Note: it is very important that we create the buckets in order of
1382                  * decreasing MCV frequency.  If we have to remove some buckets, they
1383                  * must be removed in reverse order of creation (see notes in
1384                  * ExecHashRemoveNextSkewBucket) and we want the least common MCVs to
1385                  * be removed first.
1386                  */
1387                 hashfunctions = hashtable->outer_hashfunctions;
1388
1389                 for (i = 0; i < mcvsToUse; i++)
1390                 {
1391                         uint32          hashvalue;
1392                         int                     bucket;
1393
1394                         hashvalue = DatumGetUInt32(FunctionCall1(&hashfunctions[0],
1395                                                                                                          values[i]));
1396
1397                         /*
1398                          * While we have not hit a hole in the hashtable and have not hit
1399                          * the desired bucket, we have collided with some previous hash
1400                          * value, so try the next bucket location.  NB: this code must
1401                          * match ExecHashGetSkewBucket.
1402                          */
1403                         bucket = hashvalue & (nbuckets - 1);
1404                         while (hashtable->skewBucket[bucket] != NULL &&
1405                                    hashtable->skewBucket[bucket]->hashvalue != hashvalue)
1406                                 bucket = (bucket + 1) & (nbuckets - 1);
1407
1408                         /*
1409                          * If we found an existing bucket with the same hashvalue, leave
1410                          * it alone.  It's okay for two MCVs to share a hashvalue.
1411                          */
1412                         if (hashtable->skewBucket[bucket] != NULL)
1413                                 continue;
1414
1415                         /* Okay, create a new skew bucket for this hashvalue. */
1416                         hashtable->skewBucket[bucket] = (HashSkewBucket *)
1417                                 MemoryContextAlloc(hashtable->batchCxt,
1418                                                                    sizeof(HashSkewBucket));
1419                         hashtable->skewBucket[bucket]->hashvalue = hashvalue;
1420                         hashtable->skewBucket[bucket]->tuples = NULL;
1421                         hashtable->skewBucketNums[hashtable->nSkewBuckets] = bucket;
1422                         hashtable->nSkewBuckets++;
1423                         hashtable->spaceUsed += SKEW_BUCKET_OVERHEAD;
1424                         hashtable->spaceUsedSkew += SKEW_BUCKET_OVERHEAD;
1425                         if (hashtable->spaceUsed > hashtable->spacePeak)
1426                                 hashtable->spacePeak = hashtable->spaceUsed;
1427                 }
1428
1429                 free_attstatsslot(node->skewColType,
1430                                                   values, nvalues, numbers, nnumbers);
1431         }
1432
1433         ReleaseSysCache(statsTuple);
1434 }
1435
1436 /*
1437  * ExecHashGetSkewBucket
1438  *
1439  *              Returns the index of the skew bucket for this hashvalue,
1440  *              or INVALID_SKEW_BUCKET_NO if the hashvalue is not
1441  *              associated with any active skew bucket.
1442  */
1443 int
1444 ExecHashGetSkewBucket(HashJoinTable hashtable, uint32 hashvalue)
1445 {
1446         int                     bucket;
1447
1448         /*
1449          * Always return INVALID_SKEW_BUCKET_NO if not doing skew optimization (in
1450          * particular, this happens after the initial batch is done).
1451          */
1452         if (!hashtable->skewEnabled)
1453                 return INVALID_SKEW_BUCKET_NO;
1454
1455         /*
1456          * Since skewBucketLen is a power of 2, we can do a modulo by ANDing.
1457          */
1458         bucket = hashvalue & (hashtable->skewBucketLen - 1);
1459
1460         /*
1461          * While we have not hit a hole in the hashtable and have not hit the
1462          * desired bucket, we have collided with some other hash value, so try the
1463          * next bucket location.
1464          */
1465         while (hashtable->skewBucket[bucket] != NULL &&
1466                    hashtable->skewBucket[bucket]->hashvalue != hashvalue)
1467                 bucket = (bucket + 1) & (hashtable->skewBucketLen - 1);
1468
1469         /*
1470          * Found the desired bucket?
1471          */
1472         if (hashtable->skewBucket[bucket] != NULL)
1473                 return bucket;
1474
1475         /*
1476          * There must not be any hashtable entry for this hash value.
1477          */
1478         return INVALID_SKEW_BUCKET_NO;
1479 }
1480
1481 /*
1482  * ExecHashSkewTableInsert
1483  *
1484  *              Insert a tuple into the skew hashtable.
1485  *
1486  * This should generally match up with the current-batch case in
1487  * ExecHashTableInsert.
1488  */
1489 static void
1490 ExecHashSkewTableInsert(HashJoinTable hashtable,
1491                                                 TupleTableSlot *slot,
1492                                                 uint32 hashvalue,
1493                                                 int bucketNumber)
1494 {
1495         MinimalTuple tuple = ExecFetchSlotMinimalTuple(slot);
1496         HashJoinTuple hashTuple;
1497         int                     hashTupleSize;
1498
1499         /* Create the HashJoinTuple */
1500         hashTupleSize = HJTUPLE_OVERHEAD + tuple->t_len;
1501         hashTuple = (HashJoinTuple) MemoryContextAlloc(hashtable->batchCxt,
1502                                                                                                    hashTupleSize);
1503         hashTuple->hashvalue = hashvalue;
1504         memcpy(HJTUPLE_MINTUPLE(hashTuple), tuple, tuple->t_len);
1505         HeapTupleHeaderClearMatch(HJTUPLE_MINTUPLE(hashTuple));
1506
1507         /* Push it onto the front of the skew bucket's list */
1508         hashTuple->next = hashtable->skewBucket[bucketNumber]->tuples;
1509         hashtable->skewBucket[bucketNumber]->tuples = hashTuple;
1510
1511         /* Account for space used, and back off if we've used too much */
1512         hashtable->spaceUsed += hashTupleSize;
1513         hashtable->spaceUsedSkew += hashTupleSize;
1514         if (hashtable->spaceUsed > hashtable->spacePeak)
1515                 hashtable->spacePeak = hashtable->spaceUsed;
1516         while (hashtable->spaceUsedSkew > hashtable->spaceAllowedSkew)
1517                 ExecHashRemoveNextSkewBucket(hashtable);
1518
1519         /* Check we are not over the total spaceAllowed, either */
1520         if (hashtable->spaceUsed > hashtable->spaceAllowed)
1521                 ExecHashIncreaseNumBatches(hashtable);
1522 }
1523
1524 /*
1525  *              ExecHashRemoveNextSkewBucket
1526  *
1527  *              Remove the least valuable skew bucket by pushing its tuples into
1528  *              the main hash table.
1529  */
1530 static void
1531 ExecHashRemoveNextSkewBucket(HashJoinTable hashtable)
1532 {
1533         int                     bucketToRemove;
1534         HashSkewBucket *bucket;
1535         uint32          hashvalue;
1536         int                     bucketno;
1537         int                     batchno;
1538         HashJoinTuple hashTuple;
1539
1540         /* Locate the bucket to remove */
1541         bucketToRemove = hashtable->skewBucketNums[hashtable->nSkewBuckets - 1];
1542         bucket = hashtable->skewBucket[bucketToRemove];
1543
1544         /*
1545          * Calculate which bucket and batch the tuples belong to in the main
1546          * hashtable.  They all have the same hash value, so it's the same for all
1547          * of them.  Also note that it's not possible for nbatch to increase while
1548          * we are processing the tuples.
1549          */
1550         hashvalue = bucket->hashvalue;
1551         ExecHashGetBucketAndBatch(hashtable, hashvalue, &bucketno, &batchno);
1552
1553         /* Process all tuples in the bucket */
1554         hashTuple = bucket->tuples;
1555         while (hashTuple != NULL)
1556         {
1557                 HashJoinTuple nextHashTuple = hashTuple->next;
1558                 MinimalTuple tuple;
1559                 Size            tupleSize;
1560
1561                 /*
1562                  * This code must agree with ExecHashTableInsert.  We do not use
1563                  * ExecHashTableInsert directly as ExecHashTableInsert expects a
1564                  * TupleTableSlot while we already have HashJoinTuples.
1565                  */
1566                 tuple = HJTUPLE_MINTUPLE(hashTuple);
1567                 tupleSize = HJTUPLE_OVERHEAD + tuple->t_len;
1568
1569                 /* Decide whether to put the tuple in the hash table or a temp file */
1570                 if (batchno == hashtable->curbatch)
1571                 {
1572                         /* Move the tuple to the main hash table */
1573                         HashJoinTuple copyTuple;
1574
1575                         /*
1576                          * We must copy the tuple into the dense storage, else it will not
1577                          * be found by, eg, ExecHashIncreaseNumBatches.
1578                          */
1579                         copyTuple = (HashJoinTuple) dense_alloc(hashtable, tupleSize);
1580                         memcpy(copyTuple, hashTuple, tupleSize);
1581                         pfree(hashTuple);
1582
1583                         copyTuple->next = hashtable->buckets[bucketno];
1584                         hashtable->buckets[bucketno] = copyTuple;
1585
1586                         /* We have reduced skew space, but overall space doesn't change */
1587                         hashtable->spaceUsedSkew -= tupleSize;
1588                 }
1589                 else
1590                 {
1591                         /* Put the tuple into a temp file for later batches */
1592                         Assert(batchno > hashtable->curbatch);
1593                         ExecHashJoinSaveTuple(tuple, hashvalue,
1594                                                                   &hashtable->innerBatchFile[batchno]);
1595                         pfree(hashTuple);
1596                         hashtable->spaceUsed -= tupleSize;
1597                         hashtable->spaceUsedSkew -= tupleSize;
1598                 }
1599
1600                 hashTuple = nextHashTuple;
1601
1602                 /* allow this loop to be cancellable */
1603                 CHECK_FOR_INTERRUPTS();
1604         }
1605
1606         /*
1607          * Free the bucket struct itself and reset the hashtable entry to NULL.
1608          *
1609          * NOTE: this is not nearly as simple as it looks on the surface, because
1610          * of the possibility of collisions in the hashtable.  Suppose that hash
1611          * values A and B collide at a particular hashtable entry, and that A was
1612          * entered first so B gets shifted to a different table entry.  If we were
1613          * to remove A first then ExecHashGetSkewBucket would mistakenly start
1614          * reporting that B is not in the hashtable, because it would hit the NULL
1615          * before finding B.  However, we always remove entries in the reverse
1616          * order of creation, so this failure cannot happen.
1617          */
1618         hashtable->skewBucket[bucketToRemove] = NULL;
1619         hashtable->nSkewBuckets--;
1620         pfree(bucket);
1621         hashtable->spaceUsed -= SKEW_BUCKET_OVERHEAD;
1622         hashtable->spaceUsedSkew -= SKEW_BUCKET_OVERHEAD;
1623
1624         /*
1625          * If we have removed all skew buckets then give up on skew optimization.
1626          * Release the arrays since they aren't useful any more.
1627          */
1628         if (hashtable->nSkewBuckets == 0)
1629         {
1630                 hashtable->skewEnabled = false;
1631                 pfree(hashtable->skewBucket);
1632                 pfree(hashtable->skewBucketNums);
1633                 hashtable->skewBucket = NULL;
1634                 hashtable->skewBucketNums = NULL;
1635                 hashtable->spaceUsed -= hashtable->spaceUsedSkew;
1636                 hashtable->spaceUsedSkew = 0;
1637         }
1638 }
1639
1640 /*
1641  * Allocate 'size' bytes from the currently active HashMemoryChunk
1642  */
1643 static void *
1644 dense_alloc(HashJoinTable hashtable, Size size)
1645 {
1646         HashMemoryChunk newChunk;
1647         char       *ptr;
1648
1649         /* just in case the size is not already aligned properly */
1650         size = MAXALIGN(size);
1651
1652         /*
1653          * If tuple size is larger than of 1/4 of chunk size, allocate a separate
1654          * chunk.
1655          */
1656         if (size > HASH_CHUNK_THRESHOLD)
1657         {
1658                 /* allocate new chunk and put it at the beginning of the list */
1659                 newChunk = (HashMemoryChunk) MemoryContextAlloc(hashtable->batchCxt,
1660                                                                  offsetof(HashMemoryChunkData, data) + size);
1661                 newChunk->maxlen = size;
1662                 newChunk->used = 0;
1663                 newChunk->ntuples = 0;
1664
1665                 /*
1666                  * Add this chunk to the list after the first existing chunk, so that
1667                  * we don't lose the remaining space in the "current" chunk.
1668                  */
1669                 if (hashtable->chunks != NULL)
1670                 {
1671                         newChunk->next = hashtable->chunks->next;
1672                         hashtable->chunks->next = newChunk;
1673                 }
1674                 else
1675                 {
1676                         newChunk->next = hashtable->chunks;
1677                         hashtable->chunks = newChunk;
1678                 }
1679
1680                 newChunk->used += size;
1681                 newChunk->ntuples += 1;
1682
1683                 return newChunk->data;
1684         }
1685
1686         /*
1687          * See if we have enough space for it in the current chunk (if any). If
1688          * not, allocate a fresh chunk.
1689          */
1690         if ((hashtable->chunks == NULL) ||
1691                 (hashtable->chunks->maxlen - hashtable->chunks->used) < size)
1692         {
1693                 /* allocate new chunk and put it at the beginning of the list */
1694                 newChunk = (HashMemoryChunk) MemoryContextAlloc(hashtable->batchCxt,
1695                                           offsetof(HashMemoryChunkData, data) + HASH_CHUNK_SIZE);
1696
1697                 newChunk->maxlen = HASH_CHUNK_SIZE;
1698                 newChunk->used = size;
1699                 newChunk->ntuples = 1;
1700
1701                 newChunk->next = hashtable->chunks;
1702                 hashtable->chunks = newChunk;
1703
1704                 return newChunk->data;
1705         }
1706
1707         /* There is enough space in the current chunk, let's add the tuple */
1708         ptr = hashtable->chunks->data + hashtable->chunks->used;
1709         hashtable->chunks->used += size;
1710         hashtable->chunks->ntuples += 1;
1711
1712         /* return pointer to the start of the tuple memory */
1713         return ptr;
1714 }