granicus.if.org Git - postgresql/blob - src/backend/utils/sort/tuplesort.c

   1 /*-------------------------------------------------------------------------
   2  *
   3  * tuplesort.c
   4  *        Generalized tuple sorting routines.
   5  *
   6  * This module handles sorting of heap tuples, index tuples, or single
   7  * Datums (and could easily support other kinds of sortable objects,
   8  * if necessary).  It works efficiently for both small and large amounts
   9  * of data.  Small amounts are sorted in-memory using qsort().  Large
  10  * amounts are sorted using temporary files and a standard external sort
  11  * algorithm.
  12  *
  13  * See Knuth, volume 3, for more than you want to know about the external
  14  * sorting algorithm.  We divide the input into sorted runs using replacement
  15  * selection, in the form of a priority tree implemented as a heap
  16  * (essentially his Algorithm 5.2.3H), then merge the runs using polyphase
  17  * merge, Knuth's Algorithm 5.4.2D.  The logical "tapes" used by Algorithm D
  18  * are implemented by logtape.c, which avoids space wastage by recycling
  19  * disk space as soon as each block is read from its "tape".
  20  *
  21  * We do not form the initial runs using Knuth's recommended replacement
  22  * selection data structure (Algorithm 5.4.1R), because it uses a fixed
  23  * number of records in memory at all times.  Since we are dealing with
  24  * tuples that may vary considerably in size, we want to be able to vary
  25  * the number of records kept in memory to ensure full utilization of the
  26  * allowed sort memory space.  So, we keep the tuples in a variable-size
  27  * heap, with the next record to go out at the top of the heap.  Like
  28  * Algorithm 5.4.1R, each record is stored with the run number that it
  29  * must go into, and we use (run number, key) as the ordering key for the
  30  * heap.  When the run number at the top of the heap changes, we know that
  31  * no more records of the prior run are left in the heap.
  32  *
  33  * The approximate amount of memory allowed for any one sort operation
  34  * is specified in kilobytes by the caller (most pass work_mem).  Initially,
  35  * we absorb tuples and simply store them in an unsorted array as long as
  36  * we haven't exceeded workMem.  If we reach the end of the input without
  37  * exceeding workMem, we sort the array using qsort() and subsequently return
  38  * tuples just by scanning the tuple array sequentially.  If we do exceed
  39  * workMem, we construct a heap using Algorithm H and begin to emit tuples
  40  * into sorted runs in temporary tapes, emitting just enough tuples at each
  41  * step to get back within the workMem limit.  Whenever the run number at
  42  * the top of the heap changes, we begin a new run with a new output tape
  43  * (selected per Algorithm D).  After the end of the input is reached,
  44  * we dump out remaining tuples in memory into a final run (or two),
  45  * then merge the runs using Algorithm D.
  46  *
  47  * When merging runs, we use a heap containing just the frontmost tuple from
  48  * each source run; we repeatedly output the smallest tuple and insert the
  49  * next tuple from its source tape (if any).  When the heap empties, the merge
  50  * is complete.  The basic merge algorithm thus needs very little memory ---
  51  * only M tuples for an M-way merge, and M is constrained to a small number.
  52  * However, we can still make good use of our full workMem allocation by
  53  * pre-reading additional tuples from each source tape.  Without prereading,
  54  * our access pattern to the temporary file would be very erratic; on average
  55  * we'd read one block from each of M source tapes during the same time that
  56  * we're writing M blocks to the output tape, so there is no sequentiality of
  57  * access at all, defeating the read-ahead methods used by most Unix kernels.
  58  * Worse, the output tape gets written into a very random sequence of blocks
  59  * of the temp file, ensuring that things will be even worse when it comes
  60  * time to read that tape.  A straightforward merge pass thus ends up doing a
  61  * lot of waiting for disk seeks.  We can improve matters by prereading from
  62  * each source tape sequentially, loading about workMem/M bytes from each tape
  63  * in turn.  Then we run the merge algorithm, writing but not reading until
  64  * one of the preloaded tuple series runs out.  Then we switch back to preread
  65  * mode, fill memory again, and repeat.  This approach helps to localize both
  66  * read and write accesses.
  67  *
  68  * When the caller requests random access to the sort result, we form
  69  * the final sorted run on a logical tape which is then "frozen", so
  70  * that we can access it randomly.  When the caller does not need random
  71  * access, we return from tuplesort_performsort() as soon as we are down
  72  * to one run per logical tape.  The final merge is then performed
  73  * on-the-fly as the caller repeatedly calls tuplesort_getXXX; this
  74  * saves one cycle of writing all the data out to disk and reading it in.
  75  *
  76  * Before Postgres 8.2, we always used a seven-tape polyphase merge, on the
  77  * grounds that 7 is the "sweet spot" on the tapes-to-passes curve according
  78  * to Knuth's figure 70 (section 5.4.2).  However, Knuth is assuming that
  79  * tape drives are expensive beasts, and in particular that there will always
  80  * be many more runs than tape drives.  In our implementation a "tape drive"
  81  * doesn't cost much more than a few Kb of memory buffers, so we can afford
  82  * to have lots of them.  In particular, if we can have as many tape drives
  83  * as sorted runs, we can eliminate any repeated I/O at all.  In the current
  84  * code we determine the number of tapes M on the basis of workMem: we want
  85  * workMem/M to be large enough that we read a fair amount of data each time
  86  * we preread from a tape, so as to maintain the locality of access described
  87  * above.  Nonetheless, with large workMem we can have many tapes.
  88  *
  89  *
  90  * Portions Copyright (c) 1996-2014, PostgreSQL Global Development Group
  91  * Portions Copyright (c) 1994, Regents of the University of California
  92  *
  93  * IDENTIFICATION
  94  *        src/backend/utils/sort/tuplesort.c
  95  *
  96  *-------------------------------------------------------------------------
  97  */
  98
  99 #include "postgres.h"
 100
 101 #include <limits.h>
 102
 103 #include "access/htup_details.h"
 104 #include "access/nbtree.h"
 105 #include "catalog/index.h"
 106 #include "commands/tablespace.h"
 107 #include "executor/executor.h"
 108 #include "miscadmin.h"
 109 #include "pg_trace.h"
 110 #include "utils/datum.h"
 111 #include "utils/logtape.h"
 112 #include "utils/lsyscache.h"
 113 #include "utils/memutils.h"
 114 #include "utils/pg_rusage.h"
 115 #include "utils/rel.h"
 116 #include "utils/sortsupport.h"
 117 #include "utils/tuplesort.h"
 118
 119
 120 /* sort-type codes for sort__start probes */
 121 #define HEAP_SORT               0
 122 #define INDEX_SORT              1
 123 #define DATUM_SORT              2
 124 #define CLUSTER_SORT    3
 125
 126 /* GUC variables */
 127 #ifdef TRACE_SORT
 128 bool            trace_sort = false;
 129 #endif
 130
 131 #ifdef DEBUG_BOUNDED_SORT
 132 bool            optimize_bounded_sort = true;
 133 #endif
 134
 135
 136 /*
 137  * The objects we actually sort are SortTuple structs.  These contain
 138  * a pointer to the tuple proper (might be a MinimalTuple or IndexTuple),
 139  * which is a separate palloc chunk --- we assume it is just one chunk and
 140  * can be freed by a simple pfree().  SortTuples also contain the tuple's
 141  * first key column in Datum/nullflag format, and an index integer.
 142  *
 143  * Storing the first key column lets us save heap_getattr or index_getattr
 144  * calls during tuple comparisons.  We could extract and save all the key
 145  * columns not just the first, but this would increase code complexity and
 146  * overhead, and wouldn't actually save any comparison cycles in the common
 147  * case where the first key determines the comparison result.  Note that
 148  * for a pass-by-reference datatype, datum1 points into the "tuple" storage.
 149  *
 150  * When sorting single Datums, the data value is represented directly by
 151  * datum1/isnull1.  If the datatype is pass-by-reference and isnull1 is false,
 152  * then datum1 points to a separately palloc'd data value that is also pointed
 153  * to by the "tuple" pointer; otherwise "tuple" is NULL.
 154  *
 155  * While building initial runs, tupindex holds the tuple's run number.  During
 156  * merge passes, we re-use it to hold the input tape number that each tuple in
 157  * the heap was read from, or to hold the index of the next tuple pre-read
 158  * from the same tape in the case of pre-read entries.  tupindex goes unused
 159  * if the sort occurs entirely in memory.
 160  */
 161 typedef struct
 162 {
 163         void       *tuple;                      /* the tuple proper */
 164         Datum           datum1;                 /* value of first key column */
 165         bool            isnull1;                /* is first key column NULL? */
 166         int                     tupindex;               /* see notes above */
 167 } SortTuple;
 168
 169
 170 /*
 171  * Possible states of a Tuplesort object.  These denote the states that
 172  * persist between calls of Tuplesort routines.
 173  */
 174 typedef enum
 175 {
 176         TSS_INITIAL,                            /* Loading tuples; still within memory limit */
 177         TSS_BOUNDED,                            /* Loading tuples into bounded-size heap */
 178         TSS_BUILDRUNS,                          /* Loading tuples; writing to tape */
 179         TSS_SORTEDINMEM,                        /* Sort completed entirely in memory */
 180         TSS_SORTEDONTAPE,                       /* Sort completed, final run is on tape */
 181         TSS_FINALMERGE                          /* Performing final merge on-the-fly */
 182 } TupSortStatus;
 183
 184 /*
 185  * Parameters for calculation of number of tapes to use --- see inittapes()
 186  * and tuplesort_merge_order().
 187  *
 188  * In this calculation we assume that each tape will cost us about 3 blocks
 189  * worth of buffer space (which is an underestimate for very large data
 190  * volumes, but it's probably close enough --- see logtape.c).
 191  *
 192  * MERGE_BUFFER_SIZE is how much data we'd like to read from each input
 193  * tape during a preread cycle (see discussion at top of file).
 194  */
 195 #define MINORDER                6               /* minimum merge order */
 196 #define TAPE_BUFFER_OVERHEAD            (BLCKSZ * 3)
 197 #define MERGE_BUFFER_SIZE                       (BLCKSZ * 32)
 198
 199 typedef int (*SortTupleComparator) (const SortTuple *a, const SortTuple *b,
 200                                                                                                 Tuplesortstate *state);
 201
 202 /*
 203  * Private state of a Tuplesort operation.
 204  */
 205 struct Tuplesortstate
 206 {
 207         TupSortStatus status;           /* enumerated value as shown above */
 208         int                     nKeys;                  /* number of columns in sort key */
 209         bool            randomAccess;   /* did caller request random access? */
 210         bool            bounded;                /* did caller specify a maximum number of
 211                                                                  * tuples to return? */
 212         bool            boundUsed;              /* true if we made use of a bounded heap */
 213         int                     bound;                  /* if bounded, the maximum number of tuples */
 214         int64           availMem;               /* remaining memory available, in bytes */
 215         int64           allowedMem;             /* total memory allowed, in bytes */
 216         int                     maxTapes;               /* number of tapes (Knuth's T) */
 217         int                     tapeRange;              /* maxTapes-1 (Knuth's P) */
 218         MemoryContext sortcontext;      /* memory context holding all sort data */
 219         LogicalTapeSet *tapeset;        /* logtape.c object for tapes in a temp file */
 220
 221         /*
 222          * These function pointers decouple the routines that must know what kind
 223          * of tuple we are sorting from the routines that don't need to know it.
 224          * They are set up by the tuplesort_begin_xxx routines.
 225          *
 226          * Function to compare two tuples; result is per qsort() convention, ie:
 227          * <0, 0, >0 according as a<b, a=b, a>b.  The API must match
 228          * qsort_arg_comparator.
 229          */
 230         SortTupleComparator comparetup;
 231
 232         /*
 233          * Function to copy a supplied input tuple into palloc'd space and set up
 234          * its SortTuple representation (ie, set tuple/datum1/isnull1).  Also,
 235          * state->availMem must be decreased by the amount of space used for the
 236          * tuple copy (note the SortTuple struct itself is not counted).
 237          */
 238         void            (*copytup) (Tuplesortstate *state, SortTuple *stup, void *tup);
 239
 240         /*
 241          * Function to write a stored tuple onto tape.  The representation of the
 242          * tuple on tape need not be the same as it is in memory; requirements on
 243          * the tape representation are given below.  After writing the tuple,
 244          * pfree() the out-of-line data (not the SortTuple struct!), and increase
 245          * state->availMem by the amount of memory space thereby released.
 246          */
 247         void            (*writetup) (Tuplesortstate *state, int tapenum,
 248                                                                                  SortTuple *stup);
 249
 250         /*
 251          * Function to read a stored tuple from tape back into memory. 'len' is
 252          * the already-read length of the stored tuple.  Create a palloc'd copy,
 253          * initialize tuple/datum1/isnull1 in the target SortTuple struct, and
 254          * decrease state->availMem by the amount of memory space consumed.
 255          */
 256         void            (*readtup) (Tuplesortstate *state, SortTuple *stup,
 257                                                                                 int tapenum, unsigned int len);
 258
 259         /*
 260          * Function to reverse the sort direction from its current state. (We
 261          * could dispense with this if we wanted to enforce that all variants
 262          * represent the sort key information alike.)
 263          */
 264         void            (*reversedirection) (Tuplesortstate *state);
 265
 266         /*
 267          * This array holds the tuples now in sort memory.  If we are in state
 268          * INITIAL, the tuples are in no particular order; if we are in state
 269          * SORTEDINMEM, the tuples are in final sorted order; in states BUILDRUNS
 270          * and FINALMERGE, the tuples are organized in "heap" order per Algorithm
 271          * H.  (Note that memtupcount only counts the tuples that are part of the
 272          * heap --- during merge passes, memtuples[] entries beyond tapeRange are
 273          * never in the heap and are used to hold pre-read tuples.)  In state
 274          * SORTEDONTAPE, the array is not used.
 275          */
 276         SortTuple  *memtuples;          /* array of SortTuple structs */
 277         int                     memtupcount;    /* number of tuples currently present */
 278         int                     memtupsize;             /* allocated length of memtuples array */
 279         bool            growmemtuples;  /* memtuples' growth still underway? */
 280
 281         /*
 282          * While building initial runs, this is the current output run number
 283          * (starting at 0).  Afterwards, it is the number of initial runs we made.
 284          */
 285         int                     currentRun;
 286
 287         /*
 288          * Unless otherwise noted, all pointer variables below are pointers to
 289          * arrays of length maxTapes, holding per-tape data.
 290          */
 291
 292         /*
 293          * These variables are only used during merge passes.  mergeactive[i] is
 294          * true if we are reading an input run from (actual) tape number i and
 295          * have not yet exhausted that run.  mergenext[i] is the memtuples index
 296          * of the next pre-read tuple (next to be loaded into the heap) for tape
 297          * i, or 0 if we are out of pre-read tuples.  mergelast[i] similarly
 298          * points to the last pre-read tuple from each tape.  mergeavailslots[i]
 299          * is the number of unused memtuples[] slots reserved for tape i, and
 300          * mergeavailmem[i] is the amount of unused space allocated for tape i.
 301          * mergefreelist and mergefirstfree keep track of unused locations in the
 302          * memtuples[] array.  The memtuples[].tupindex fields link together
 303          * pre-read tuples for each tape as well as recycled locations in
 304          * mergefreelist. It is OK to use 0 as a null link in these lists, because
 305          * memtuples[0] is part of the merge heap and is never a pre-read tuple.
 306          */
 307         bool       *mergeactive;        /* active input run source? */
 308         int                *mergenext;          /* first preread tuple for each source */
 309         int                *mergelast;          /* last preread tuple for each source */
 310         int                *mergeavailslots;    /* slots left for prereading each tape */
 311         int64      *mergeavailmem;      /* availMem for prereading each tape */
 312         int                     mergefreelist;  /* head of freelist of recycled slots */
 313         int                     mergefirstfree; /* first slot never used in this merge */
 314
 315         /*
 316          * Variables for Algorithm D.  Note that destTape is a "logical" tape
 317          * number, ie, an index into the tp_xxx[] arrays.  Be careful to keep
 318          * "logical" and "actual" tape numbers straight!
 319          */
 320         int                     Level;                  /* Knuth's l */
 321         int                     destTape;               /* current output tape (Knuth's j, less 1) */
 322         int                *tp_fib;                     /* Target Fibonacci run counts (A[]) */
 323         int                *tp_runs;            /* # of real runs on each tape */
 324         int                *tp_dummy;           /* # of dummy runs for each tape (D[]) */
 325         int                *tp_tapenum;         /* Actual tape numbers (TAPE[]) */
 326         int                     activeTapes;    /* # of active input tapes in merge pass */
 327
 328         /*
 329          * These variables are used after completion of sorting to keep track of
 330          * the next tuple to return.  (In the tape case, the tape's current read
 331          * position is also critical state.)
 332          */
 333         int                     result_tape;    /* actual tape number of finished output */
 334         int                     current;                /* array index (only used if SORTEDINMEM) */
 335         bool            eof_reached;    /* reached EOF (needed for cursors) */
 336
 337         /* markpos_xxx holds marked position for mark and restore */
 338         long            markpos_block;  /* tape block# (only used if SORTEDONTAPE) */
 339         int                     markpos_offset; /* saved "current", or offset in tape block */
 340         bool            markpos_eof;    /* saved "eof_reached" */
 341
 342         /*
 343          * These variables are specific to the MinimalTuple case; they are set by
 344          * tuplesort_begin_heap and used only by the MinimalTuple routines.
 345          */
 346         TupleDesc       tupDesc;
 347         SortSupport sortKeys;           /* array of length nKeys */
 348
 349         /*
 350          * This variable is shared by the single-key MinimalTuple case and the
 351          * Datum case (which both use qsort_ssup()).  Otherwise it's NULL.
 352          */
 353         SortSupport onlyKey;
 354
 355         /*
 356          * These variables are specific to the CLUSTER case; they are set by
 357          * tuplesort_begin_cluster.  Note CLUSTER also uses tupDesc and
 358          * indexScanKey.
 359          */
 360         IndexInfo  *indexInfo;          /* info about index being used for reference */
 361         EState     *estate;                     /* for evaluating index expressions */
 362
 363         /*
 364          * These variables are specific to the IndexTuple case; they are set by
 365          * tuplesort_begin_index_xxx and used only by the IndexTuple routines.
 366          */
 367         Relation        heapRel;                /* table the index is being built on */
 368         Relation        indexRel;               /* index being built */
 369
 370         /* These are specific to the index_btree subcase: */
 371         ScanKey         indexScanKey;
 372         bool            enforceUnique;  /* complain if we find duplicate tuples */
 373
 374         /* These are specific to the index_hash subcase: */
 375         uint32          hash_mask;              /* mask for sortable part of hash code */
 376
 377         /*
 378          * These variables are specific to the Datum case; they are set by
 379          * tuplesort_begin_datum and used only by the DatumTuple routines.
 380          */
 381         Oid                     datumType;
 382         /* we need typelen and byval in order to know how to copy the Datums. */
 383         int                     datumTypeLen;
 384         bool            datumTypeByVal;
 385
 386         /*
 387          * Resource snapshot for time of sort start.
 388          */
 389 #ifdef TRACE_SORT
 390         PGRUsage        ru_start;
 391 #endif
 392 };
 393
 394 #define COMPARETUP(state,a,b)   ((*(state)->comparetup) (a, b, state))
 395 #define COPYTUP(state,stup,tup) ((*(state)->copytup) (state, stup, tup))
 396 #define WRITETUP(state,tape,stup)       ((*(state)->writetup) (state, tape, stup))
 397 #define READTUP(state,stup,tape,len) ((*(state)->readtup) (state, stup, tape, len))
 398 #define REVERSEDIRECTION(state) ((*(state)->reversedirection) (state))
 399 #define LACKMEM(state)          ((state)->availMem < 0)
 400 #define USEMEM(state,amt)       ((state)->availMem -= (amt))
 401 #define FREEMEM(state,amt)      ((state)->availMem += (amt))
 402
 403 /*
 404  * NOTES about on-tape representation of tuples:
 405  *
 406  * We require the first "unsigned int" of a stored tuple to be the total size
 407  * on-tape of the tuple, including itself (so it is never zero; an all-zero
 408  * unsigned int is used to delimit runs).  The remainder of the stored tuple
 409  * may or may not match the in-memory representation of the tuple ---
 410  * any conversion needed is the job of the writetup and readtup routines.
 411  *
 412  * If state->randomAccess is true, then the stored representation of the
 413  * tuple must be followed by another "unsigned int" that is a copy of the
 414  * length --- so the total tape space used is actually sizeof(unsigned int)
 415  * more than the stored length value.  This allows read-backwards.  When
 416  * randomAccess is not true, the write/read routines may omit the extra
 417  * length word.
 418  *
 419  * writetup is expected to write both length words as well as the tuple
 420  * data.  When readtup is called, the tape is positioned just after the
 421  * front length word; readtup must read the tuple data and advance past
 422  * the back length word (if present).
 423  *
 424  * The write/read routines can make use of the tuple description data
 425  * stored in the Tuplesortstate record, if needed.  They are also expected
 426  * to adjust state->availMem by the amount of memory space (not tape space!)
 427  * released or consumed.  There is no error return from either writetup
 428  * or readtup; they should ereport() on failure.
 429  *
 430  *
 431  * NOTES about memory consumption calculations:
 432  *
 433  * We count space allocated for tuples against the workMem limit, plus
 434  * the space used by the variable-size memtuples array.  Fixed-size space
 435  * is not counted; it's small enough to not be interesting.
 436  *
 437  * Note that we count actual space used (as shown by GetMemoryChunkSpace)
 438  * rather than the originally-requested size.  This is important since
 439  * palloc can add substantial overhead.  It's not a complete answer since
 440  * we won't count any wasted space in palloc allocation blocks, but it's
 441  * a lot better than what we were doing before 7.3.
 442  */
 443
 444 /* When using this macro, beware of double evaluation of len */
 445 #define LogicalTapeReadExact(tapeset, tapenum, ptr, len) \
 446         do { \
 447                 if (LogicalTapeRead(tapeset, tapenum, ptr, len) != (size_t) (len)) \
 448                         elog(ERROR, "unexpected end of data"); \
 449         } while(0)
 450
 451
 452 static Tuplesortstate *tuplesort_begin_common(int workMem, bool randomAccess);
 453 static void puttuple_common(Tuplesortstate *state, SortTuple *tuple);
 454 static void inittapes(Tuplesortstate *state);
 455 static void selectnewtape(Tuplesortstate *state);
 456 static void mergeruns(Tuplesortstate *state);
 457 static void mergeonerun(Tuplesortstate *state);
 458 static void beginmerge(Tuplesortstate *state);
 459 static void mergepreread(Tuplesortstate *state);
 460 static void mergeprereadone(Tuplesortstate *state, int srcTape);
 461 static void dumptuples(Tuplesortstate *state, bool alltuples);
 462 static void make_bounded_heap(Tuplesortstate *state);
 463 static void sort_bounded_heap(Tuplesortstate *state);
 464 static void tuplesort_heap_insert(Tuplesortstate *state, SortTuple *tuple,
 465                                           int tupleindex, bool checkIndex);
 466 static void tuplesort_heap_siftup(Tuplesortstate *state, bool checkIndex);
 467 static unsigned int getlen(Tuplesortstate *state, int tapenum, bool eofOK);
 468 static void markrunend(Tuplesortstate *state, int tapenum);
 469 static int comparetup_heap(const SortTuple *a, const SortTuple *b,
 470                                 Tuplesortstate *state);
 471 static void copytup_heap(Tuplesortstate *state, SortTuple *stup, void *tup);
 472 static void writetup_heap(Tuplesortstate *state, int tapenum,
 473                           SortTuple *stup);
 474 static void readtup_heap(Tuplesortstate *state, SortTuple *stup,
 475                          int tapenum, unsigned int len);
 476 static void reversedirection_heap(Tuplesortstate *state);
 477 static int comparetup_cluster(const SortTuple *a, const SortTuple *b,
 478                                    Tuplesortstate *state);
 479 static void copytup_cluster(Tuplesortstate *state, SortTuple *stup, void *tup);
 480 static void writetup_cluster(Tuplesortstate *state, int tapenum,
 481                                  SortTuple *stup);
 482 static void readtup_cluster(Tuplesortstate *state, SortTuple *stup,
 483                                 int tapenum, unsigned int len);
 484 static int comparetup_index_btree(const SortTuple *a, const SortTuple *b,
 485                                            Tuplesortstate *state);
 486 static int comparetup_index_hash(const SortTuple *a, const SortTuple *b,
 487                                           Tuplesortstate *state);
 488 static void copytup_index(Tuplesortstate *state, SortTuple *stup, void *tup);
 489 static void writetup_index(Tuplesortstate *state, int tapenum,
 490                            SortTuple *stup);
 491 static void readtup_index(Tuplesortstate *state, SortTuple *stup,
 492                           int tapenum, unsigned int len);
 493 static void reversedirection_index_btree(Tuplesortstate *state);
 494 static void reversedirection_index_hash(Tuplesortstate *state);
 495 static int comparetup_datum(const SortTuple *a, const SortTuple *b,
 496                                  Tuplesortstate *state);
 497 static void copytup_datum(Tuplesortstate *state, SortTuple *stup, void *tup);
 498 static void writetup_datum(Tuplesortstate *state, int tapenum,
 499                            SortTuple *stup);
 500 static void readtup_datum(Tuplesortstate *state, SortTuple *stup,
 501                           int tapenum, unsigned int len);
 502 static void reversedirection_datum(Tuplesortstate *state);
 503 static void free_sort_tuple(Tuplesortstate *state, SortTuple *stup);
 504
 505 /*
 506  * Special versions of qsort just for SortTuple objects.  qsort_tuple() sorts
 507  * any variant of SortTuples, using the appropriate comparetup function.
 508  * qsort_ssup() is specialized for the case where the comparetup function
 509  * reduces to ApplySortComparator(), that is single-key MinimalTuple sorts
 510  * and Datum sorts.
 511  */
 512 #include "qsort_tuple.c"
 513
 514
 515 /*
 516  *              tuplesort_begin_xxx
 517  *
 518  * Initialize for a tuple sort operation.
 519  *
 520  * After calling tuplesort_begin, the caller should call tuplesort_putXXX
 521  * zero or more times, then call tuplesort_performsort when all the tuples
 522  * have been supplied.  After performsort, retrieve the tuples in sorted
 523  * order by calling tuplesort_getXXX until it returns false/NULL.  (If random
 524  * access was requested, rescan, markpos, and restorepos can also be called.)
 525  * Call tuplesort_end to terminate the operation and release memory/disk space.
 526  *
 527  * Each variant of tuplesort_begin has a workMem parameter specifying the
 528  * maximum number of kilobytes of RAM to use before spilling data to disk.
 529  * (The normal value of this parameter is work_mem, but some callers use
 530  * other values.)  Each variant also has a randomAccess parameter specifying
 531  * whether the caller needs non-sequential access to the sort result.
 532  */
 533
 534 static Tuplesortstate *
 535 tuplesort_begin_common(int workMem, bool randomAccess)
 536 {
 537         Tuplesortstate *state;
 538         MemoryContext sortcontext;
 539         MemoryContext oldcontext;
 540
 541         /*
 542          * Create a working memory context for this sort operation. All data
 543          * needed by the sort will live inside this context.
 544          */
 545         sortcontext = AllocSetContextCreate(CurrentMemoryContext,
 546                                                                                 "TupleSort",
 547                                                                                 ALLOCSET_DEFAULT_MINSIZE,
 548                                                                                 ALLOCSET_DEFAULT_INITSIZE,
 549                                                                                 ALLOCSET_DEFAULT_MAXSIZE);
 550
 551         /*
 552          * Make the Tuplesortstate within the per-sort context.  This way, we
 553          * don't need a separate pfree() operation for it at shutdown.
 554          */
 555         oldcontext = MemoryContextSwitchTo(sortcontext);
 556
 557         state = (Tuplesortstate *) palloc0(sizeof(Tuplesortstate));
 558
 559 #ifdef TRACE_SORT
 560         if (trace_sort)
 561                 pg_rusage_init(&state->ru_start);
 562 #endif
 563
 564         state->status = TSS_INITIAL;
 565         state->randomAccess = randomAccess;
 566         state->bounded = false;
 567         state->boundUsed = false;
 568         state->allowedMem = workMem * (int64) 1024;
 569         state->availMem = state->allowedMem;
 570         state->sortcontext = sortcontext;
 571         state->tapeset = NULL;
 572
 573         state->memtupcount = 0;
 574         state->memtupsize = 1024;       /* initial guess */
 575         state->growmemtuples = true;
 576         state->memtuples = (SortTuple *) palloc(state->memtupsize * sizeof(SortTuple));
 577
 578         USEMEM(state, GetMemoryChunkSpace(state->memtuples));
 579
 580         /* workMem must be large enough for the minimal memtuples array */
 581         if (LACKMEM(state))
 582                 elog(ERROR, "insufficient memory allowed for sort");
 583
 584         state->currentRun = 0;
 585
 586         /*
 587          * maxTapes, tapeRange, and Algorithm D variables will be initialized by
 588          * inittapes(), if needed
 589          */
 590
 591         state->result_tape = -1;        /* flag that result tape has not been formed */
 592
 593         MemoryContextSwitchTo(oldcontext);
 594
 595         return state;
 596 }
 597
 598 Tuplesortstate *
 599 tuplesort_begin_heap(TupleDesc tupDesc,
 600                                          int nkeys, AttrNumber *attNums,
 601                                          Oid *sortOperators, Oid *sortCollations,
 602                                          bool *nullsFirstFlags,
 603                                          int workMem, bool randomAccess)
 604 {
 605         Tuplesortstate *state = tuplesort_begin_common(workMem, randomAccess);
 606         MemoryContext oldcontext;
 607         int                     i;
 608
 609         oldcontext = MemoryContextSwitchTo(state->sortcontext);
 610
 611         AssertArg(nkeys > 0);
 612
 613 #ifdef TRACE_SORT
 614         if (trace_sort)
 615                 elog(LOG,
 616                          "begin tuple sort: nkeys = %d, workMem = %d, randomAccess = %c",
 617                          nkeys, workMem, randomAccess ? 't' : 'f');
 618 #endif
 619
 620         state->nKeys = nkeys;
 621
 622         TRACE_POSTGRESQL_SORT_START(HEAP_SORT,
 623                                                                 false,  /* no unique check */
 624                                                                 nkeys,
 625                                                                 workMem,
 626                                                                 randomAccess);
 627
 628         state->comparetup = comparetup_heap;
 629         state->copytup = copytup_heap;
 630         state->writetup = writetup_heap;
 631         state->readtup = readtup_heap;
 632         state->reversedirection = reversedirection_heap;
 633
 634         state->tupDesc = tupDesc;       /* assume we need not copy tupDesc */
 635
 636         /* Prepare SortSupport data for each column */
 637         state->sortKeys = (SortSupport) palloc0(nkeys * sizeof(SortSupportData));
 638
 639         for (i = 0; i < nkeys; i++)
 640         {
 641                 SortSupport sortKey = state->sortKeys + i;
 642
 643                 AssertArg(attNums[i] != 0);
 644                 AssertArg(sortOperators[i] != 0);
 645
 646                 sortKey->ssup_cxt = CurrentMemoryContext;
 647                 sortKey->ssup_collation = sortCollations[i];
 648                 sortKey->ssup_nulls_first = nullsFirstFlags[i];
 649                 sortKey->ssup_attno = attNums[i];
 650
 651                 PrepareSortSupportFromOrderingOp(sortOperators[i], sortKey);
 652         }
 653
 654         if (nkeys == 1)
 655                 state->onlyKey = state->sortKeys;
 656
 657         MemoryContextSwitchTo(oldcontext);
 658
 659         return state;
 660 }
 661
 662 Tuplesortstate *
 663 tuplesort_begin_cluster(TupleDesc tupDesc,
 664                                                 Relation indexRel,
 665                                                 int workMem, bool randomAccess)
 666 {
 667         Tuplesortstate *state = tuplesort_begin_common(workMem, randomAccess);
 668         MemoryContext oldcontext;
 669
 670         Assert(indexRel->rd_rel->relam == BTREE_AM_OID);
 671
 672         oldcontext = MemoryContextSwitchTo(state->sortcontext);
 673
 674 #ifdef TRACE_SORT
 675         if (trace_sort)
 676                 elog(LOG,
 677                          "begin tuple sort: nkeys = %d, workMem = %d, randomAccess = %c",
 678                          RelationGetNumberOfAttributes(indexRel),
 679                          workMem, randomAccess ? 't' : 'f');
 680 #endif
 681
 682         state->nKeys = RelationGetNumberOfAttributes(indexRel);
 683
 684         TRACE_POSTGRESQL_SORT_START(CLUSTER_SORT,
 685                                                                 false,  /* no unique check */
 686                                                                 state->nKeys,
 687                                                                 workMem,
 688                                                                 randomAccess);
 689
 690         state->comparetup = comparetup_cluster;
 691         state->copytup = copytup_cluster;
 692         state->writetup = writetup_cluster;
 693         state->readtup = readtup_cluster;
 694         state->reversedirection = reversedirection_index_btree;
 695
 696         state->indexInfo = BuildIndexInfo(indexRel);
 697         state->indexScanKey = _bt_mkscankey_nodata(indexRel);
 698
 699         state->tupDesc = tupDesc;       /* assume we need not copy tupDesc */
 700
 701         if (state->indexInfo->ii_Expressions != NULL)
 702         {
 703                 TupleTableSlot *slot;
 704                 ExprContext *econtext;
 705
 706                 /*
 707                  * We will need to use FormIndexDatum to evaluate the index
 708                  * expressions.  To do that, we need an EState, as well as a
 709                  * TupleTableSlot to put the table tuples into.  The econtext's
 710                  * scantuple has to point to that slot, too.
 711                  */
 712                 state->estate = CreateExecutorState();
 713                 slot = MakeSingleTupleTableSlot(tupDesc);
 714                 econtext = GetPerTupleExprContext(state->estate);
 715                 econtext->ecxt_scantuple = slot;
 716         }
 717
 718         MemoryContextSwitchTo(oldcontext);
 719
 720         return state;
 721 }
 722
 723 Tuplesortstate *
 724 tuplesort_begin_index_btree(Relation heapRel,
 725                                                         Relation indexRel,
 726                                                         bool enforceUnique,
 727                                                         int workMem, bool randomAccess)
 728 {
 729         Tuplesortstate *state = tuplesort_begin_common(workMem, randomAccess);
 730         MemoryContext oldcontext;
 731
 732         oldcontext = MemoryContextSwitchTo(state->sortcontext);
 733
 734 #ifdef TRACE_SORT
 735         if (trace_sort)
 736                 elog(LOG,
 737                          "begin index sort: unique = %c, workMem = %d, randomAccess = %c",
 738                          enforceUnique ? 't' : 'f',
 739                          workMem, randomAccess ? 't' : 'f');
 740 #endif
 741
 742         state->nKeys = RelationGetNumberOfAttributes(indexRel);
 743
 744         TRACE_POSTGRESQL_SORT_START(INDEX_SORT,
 745                                                                 enforceUnique,
 746                                                                 state->nKeys,
 747                                                                 workMem,
 748                                                                 randomAccess);
 749
 750         state->comparetup = comparetup_index_btree;
 751         state->copytup = copytup_index;
 752         state->writetup = writetup_index;
 753         state->readtup = readtup_index;
 754         state->reversedirection = reversedirection_index_btree;
 755
 756         state->heapRel = heapRel;
 757         state->indexRel = indexRel;
 758         state->indexScanKey = _bt_mkscankey_nodata(indexRel);
 759         state->enforceUnique = enforceUnique;
 760
 761         MemoryContextSwitchTo(oldcontext);
 762
 763         return state;
 764 }
 765
 766 Tuplesortstate *
 767 tuplesort_begin_index_hash(Relation heapRel,
 768                                                    Relation indexRel,
 769                                                    uint32 hash_mask,
 770                                                    int workMem, bool randomAccess)
 771 {
 772         Tuplesortstate *state = tuplesort_begin_common(workMem, randomAccess);
 773         MemoryContext oldcontext;
 774
 775         oldcontext = MemoryContextSwitchTo(state->sortcontext);
 776
 777 #ifdef TRACE_SORT
 778         if (trace_sort)
 779                 elog(LOG,
 780                 "begin index sort: hash_mask = 0x%x, workMem = %d, randomAccess = %c",
 781                          hash_mask,
 782                          workMem, randomAccess ? 't' : 'f');
 783 #endif
 784
 785         state->nKeys = 1;                       /* Only one sort column, the hash code */
 786
 787         state->comparetup = comparetup_index_hash;
 788         state->copytup = copytup_index;
 789         state->writetup = writetup_index;
 790         state->readtup = readtup_index;
 791         state->reversedirection = reversedirection_index_hash;
 792
 793         state->heapRel = heapRel;
 794         state->indexRel = indexRel;
 795
 796         state->hash_mask = hash_mask;
 797
 798         MemoryContextSwitchTo(oldcontext);
 799
 800         return state;
 801 }
 802
 803 Tuplesortstate *
 804 tuplesort_begin_datum(Oid datumType, Oid sortOperator, Oid sortCollation,
 805                                           bool nullsFirstFlag,
 806                                           int workMem, bool randomAccess)
 807 {
 808         Tuplesortstate *state = tuplesort_begin_common(workMem, randomAccess);
 809         MemoryContext oldcontext;
 810         int16           typlen;
 811         bool            typbyval;
 812
 813         oldcontext = MemoryContextSwitchTo(state->sortcontext);
 814
 815 #ifdef TRACE_SORT
 816         if (trace_sort)
 817                 elog(LOG,
 818                          "begin datum sort: workMem = %d, randomAccess = %c",
 819                          workMem, randomAccess ? 't' : 'f');
 820 #endif
 821
 822         state->nKeys = 1;                       /* always a one-column sort */
 823
 824         TRACE_POSTGRESQL_SORT_START(DATUM_SORT,
 825                                                                 false,  /* no unique check */
 826                                                                 1,
 827                                                                 workMem,
 828                                                                 randomAccess);
 829
 830         state->comparetup = comparetup_datum;
 831         state->copytup = copytup_datum;
 832         state->writetup = writetup_datum;
 833         state->readtup = readtup_datum;
 834         state->reversedirection = reversedirection_datum;
 835
 836         state->datumType = datumType;
 837
 838         /* Prepare SortSupport data */
 839         state->onlyKey = (SortSupport) palloc0(sizeof(SortSupportData));
 840
 841         state->onlyKey->ssup_cxt = CurrentMemoryContext;
 842         state->onlyKey->ssup_collation = sortCollation;
 843         state->onlyKey->ssup_nulls_first = nullsFirstFlag;
 844
 845         PrepareSortSupportFromOrderingOp(sortOperator, state->onlyKey);
 846
 847         /* lookup necessary attributes of the datum type */
 848         get_typlenbyval(datumType, &typlen, &typbyval);
 849         state->datumTypeLen = typlen;
 850         state->datumTypeByVal = typbyval;
 851
 852         MemoryContextSwitchTo(oldcontext);
 853
 854         return state;
 855 }
 856
 857 /*
 858  * tuplesort_set_bound
 859  *
 860  *      Advise tuplesort that at most the first N result tuples are required.
 861  *
 862  * Must be called before inserting any tuples.  (Actually, we could allow it
 863  * as long as the sort hasn't spilled to disk, but there seems no need for
 864  * delayed calls at the moment.)
 865  *
 866  * This is a hint only. The tuplesort may still return more tuples than
 867  * requested.
 868  */
 869 void
 870 tuplesort_set_bound(Tuplesortstate *state, int64 bound)
 871 {
 872         /* Assert we're called before loading any tuples */
 873         Assert(state->status == TSS_INITIAL);
 874         Assert(state->memtupcount == 0);
 875         Assert(!state->bounded);
 876
 877 #ifdef DEBUG_BOUNDED_SORT
 878         /* Honor GUC setting that disables the feature (for easy testing) */
 879         if (!optimize_bounded_sort)
 880                 return;
 881 #endif
 882
 883         /* We want to be able to compute bound * 2, so limit the setting */
 884         if (bound > (int64) (INT_MAX / 2))
 885                 return;
 886
 887         state->bounded = true;
 888         state->bound = (int) bound;
 889 }
 890
 891 /*
 892  * tuplesort_end
 893  *
 894  *      Release resources and clean up.
 895  *
 896  * NOTE: after calling this, any pointers returned by tuplesort_getXXX are
 897  * pointing to garbage.  Be careful not to attempt to use or free such
 898  * pointers afterwards!
 899  */
 900 void
 901 tuplesort_end(Tuplesortstate *state)
 902 {
 903         /* context swap probably not needed, but let's be safe */
 904         MemoryContext oldcontext = MemoryContextSwitchTo(state->sortcontext);
 905
 906 #ifdef TRACE_SORT
 907         long            spaceUsed;
 908
 909         if (state->tapeset)
 910                 spaceUsed = LogicalTapeSetBlocks(state->tapeset);
 911         else
 912                 spaceUsed = (state->allowedMem - state->availMem + 1023) / 1024;
 913 #endif
 914
 915         /*
 916          * Delete temporary "tape" files, if any.
 917          *
 918          * Note: want to include this in reported total cost of sort, hence need
 919          * for two #ifdef TRACE_SORT sections.
 920          */
 921         if (state->tapeset)
 922                 LogicalTapeSetClose(state->tapeset);
 923
 924 #ifdef TRACE_SORT
 925         if (trace_sort)
 926         {
 927                 if (state->tapeset)
 928                         elog(LOG, "external sort ended, %ld disk blocks used: %s",
 929                                  spaceUsed, pg_rusage_show(&state->ru_start));
 930                 else
 931                         elog(LOG, "internal sort ended, %ld KB used: %s",
 932                                  spaceUsed, pg_rusage_show(&state->ru_start));
 933         }
 934
 935         TRACE_POSTGRESQL_SORT_DONE(state->tapeset != NULL, spaceUsed);
 936 #else
 937
 938         /*
 939          * If you disabled TRACE_SORT, you can still probe sort__done, but you
 940          * ain't getting space-used stats.
 941          */
 942         TRACE_POSTGRESQL_SORT_DONE(state->tapeset != NULL, 0L);
 943 #endif
 944
 945         /* Free any execution state created for CLUSTER case */
 946         if (state->estate != NULL)
 947         {
 948                 ExprContext *econtext = GetPerTupleExprContext(state->estate);
 949
 950                 ExecDropSingleTupleTableSlot(econtext->ecxt_scantuple);
 951                 FreeExecutorState(state->estate);
 952         }
 953
 954         MemoryContextSwitchTo(oldcontext);
 955
 956         /*
 957          * Free the per-sort memory context, thereby releasing all working memory,
 958          * including the Tuplesortstate struct itself.
 959          */
 960         MemoryContextDelete(state->sortcontext);
 961 }
 962
 963 /*
 964  * Grow the memtuples[] array, if possible within our memory constraint.  We
 965  * must not exceed INT_MAX tuples in memory or the caller-provided memory
 966  * limit.  Return TRUE if we were able to enlarge the array, FALSE if not.
 967  *
 968  * Normally, at each increment we double the size of the array.  When doing
 969  * that would exceed a limit, we attempt one last, smaller increase (and then
 970  * clear the growmemtuples flag so we don't try any more).  That allows us to
 971  * use memory as fully as permitted; sticking to the pure doubling rule could
 972  * result in almost half going unused.  Because availMem moves around with
 973  * tuple addition/removal, we need some rule to prevent making repeated small
 974  * increases in memtupsize, which would just be useless thrashing.  The
 975  * growmemtuples flag accomplishes that and also prevents useless
 976  * recalculations in this function.
 977  */
 978 static bool
 979 grow_memtuples(Tuplesortstate *state)
 980 {
 981         int                     newmemtupsize;
 982         int                     memtupsize = state->memtupsize;
 983         int64           memNowUsed = state->allowedMem - state->availMem;
 984
 985         /* Forget it if we've already maxed out memtuples, per comment above */
 986         if (!state->growmemtuples)
 987                 return false;
 988
 989         /* Select new value of memtupsize */
 990         if (memNowUsed <= state->availMem)
 991         {
 992                 /*
 993                  * We've used no more than half of allowedMem; double our usage,
 994                  * clamping at INT_MAX tuples.
 995                  */
 996                 if (memtupsize < INT_MAX / 2)
 997                         newmemtupsize = memtupsize * 2;
 998                 else
 999                 {
1000                         newmemtupsize = INT_MAX;
1001                         state->growmemtuples = false;
1002                 }
1003         }
1004         else
1005         {
1006                 /*
1007                  * This will be the last increment of memtupsize.  Abandon doubling
1008                  * strategy and instead increase as much as we safely can.
1009                  *
1010                  * To stay within allowedMem, we can't increase memtupsize by more
1011                  * than availMem / sizeof(SortTuple) elements.  In practice, we want
1012                  * to increase it by considerably less, because we need to leave some
1013                  * space for the tuples to which the new array slots will refer.  We
1014                  * assume the new tuples will be about the same size as the tuples
1015                  * we've already seen, and thus we can extrapolate from the space
1016                  * consumption so far to estimate an appropriate new size for the
1017                  * memtuples array.  The optimal value might be higher or lower than
1018                  * this estimate, but it's hard to know that in advance.  We again
1019                  * clamp at INT_MAX tuples.
1020                  *
1021                  * This calculation is safe against enlarging the array so much that
1022                  * LACKMEM becomes true, because the memory currently used includes
1023                  * the present array; thus, there would be enough allowedMem for the
1024                  * new array elements even if no other memory were currently used.
1025                  *
1026                  * We do the arithmetic in float8, because otherwise the product of
1027                  * memtupsize and allowedMem could overflow.  Any inaccuracy in the
1028                  * result should be insignificant; but even if we computed a
1029                  * completely insane result, the checks below will prevent anything
1030                  * really bad from happening.
1031                  */
1032                 double          grow_ratio;
1033
1034                 grow_ratio = (double) state->allowedMem / (double) memNowUsed;
1035                 if (memtupsize * grow_ratio < INT_MAX)
1036                         newmemtupsize = (int) (memtupsize * grow_ratio);
1037                 else
1038                         newmemtupsize = INT_MAX;
1039
1040                 /* We won't make any further enlargement attempts */
1041                 state->growmemtuples = false;
1042         }
1043
1044         /* Must enlarge array by at least one element, else report failure */
1045         if (newmemtupsize <= memtupsize)
1046                 goto noalloc;
1047
1048         /*
1049          * On a 32-bit machine, allowedMem could exceed MaxAllocHugeSize.  Clamp
1050          * to ensure our request won't be rejected.  Note that we can easily
1051          * exhaust address space before facing this outcome.  (This is presently
1052          * impossible due to guc.c's MAX_KILOBYTES limitation on work_mem, but
1053          * don't rely on that at this distance.)
1054          */
1055         if ((Size) newmemtupsize >= MaxAllocHugeSize / sizeof(SortTuple))
1056         {
1057                 newmemtupsize = (int) (MaxAllocHugeSize / sizeof(SortTuple));
1058                 state->growmemtuples = false;   /* can't grow any more */
1059         }
1060
1061         /*
1062          * We need to be sure that we do not cause LACKMEM to become true, else
1063          * the space management algorithm will go nuts.  The code above should
1064          * never generate a dangerous request, but to be safe, check explicitly
1065          * that the array growth fits within availMem.  (We could still cause
1066          * LACKMEM if the memory chunk overhead associated with the memtuples
1067          * array were to increase.  That shouldn't happen with any sane value of
1068          * allowedMem, because at any array size large enough to risk LACKMEM,
1069          * palloc would be treating both old and new arrays as separate chunks.
1070          * But we'll check LACKMEM explicitly below just in case.)
1071          */
1072         if (state->availMem < (int64) ((newmemtupsize - memtupsize) * sizeof(SortTuple)))
1073                 goto noalloc;
1074
1075         /* OK, do it */
1076         FREEMEM(state, GetMemoryChunkSpace(state->memtuples));
1077         state->memtupsize = newmemtupsize;
1078         state->memtuples = (SortTuple *)
1079                 repalloc_huge(state->memtuples,
1080                                           state->memtupsize * sizeof(SortTuple));
1081         USEMEM(state, GetMemoryChunkSpace(state->memtuples));
1082         if (LACKMEM(state))
1083                 elog(ERROR, "unexpected out-of-memory situation during sort");
1084         return true;
1085
1086 noalloc:
1087         /* If for any reason we didn't realloc, shut off future attempts */
1088         state->growmemtuples = false;
1089         return false;
1090 }
1091
1092 /*
1093  * Accept one tuple while collecting input data for sort.
1094  *
1095  * Note that the input data is always copied; the caller need not save it.
1096  */
1097 void
1098 tuplesort_puttupleslot(Tuplesortstate *state, TupleTableSlot *slot)
1099 {
1100         MemoryContext oldcontext = MemoryContextSwitchTo(state->sortcontext);
1101         SortTuple       stup;
1102
1103         /*
1104          * Copy the given tuple into memory we control, and decrease availMem.
1105          * Then call the common code.
1106          */
1107         COPYTUP(state, &stup, (void *) slot);
1108
1109         puttuple_common(state, &stup);
1110
1111         MemoryContextSwitchTo(oldcontext);
1112 }
1113
1114 /*
1115  * Accept one tuple while collecting input data for sort.
1116  *
1117  * Note that the input data is always copied; the caller need not save it.
1118  */
1119 void
1120 tuplesort_putheaptuple(Tuplesortstate *state, HeapTuple tup)
1121 {
1122         MemoryContext oldcontext = MemoryContextSwitchTo(state->sortcontext);
1123         SortTuple       stup;
1124
1125         /*
1126          * Copy the given tuple into memory we control, and decrease availMem.
1127          * Then call the common code.
1128          */
1129         COPYTUP(state, &stup, (void *) tup);
1130
1131         puttuple_common(state, &stup);
1132
1133         MemoryContextSwitchTo(oldcontext);
1134 }
1135
1136 /*
1137  * Accept one index tuple while collecting input data for sort.
1138  *
1139  * Note that the input tuple is always copied; the caller need not save it.
1140  */
1141 void
1142 tuplesort_putindextuple(Tuplesortstate *state, IndexTuple tuple)
1143 {
1144         MemoryContext oldcontext = MemoryContextSwitchTo(state->sortcontext);
1145         SortTuple       stup;
1146
1147         /*
1148          * Copy the given tuple into memory we control, and decrease availMem.
1149          * Then call the common code.
1150          */
1151         COPYTUP(state, &stup, (void *) tuple);
1152
1153         puttuple_common(state, &stup);
1154
1155         MemoryContextSwitchTo(oldcontext);
1156 }
1157
1158 /*
1159  * Accept one Datum while collecting input data for sort.
1160  *
1161  * If the Datum is pass-by-ref type, the value will be copied.
1162  */
1163 void
1164 tuplesort_putdatum(Tuplesortstate *state, Datum val, bool isNull)
1165 {
1166         MemoryContext oldcontext = MemoryContextSwitchTo(state->sortcontext);
1167         SortTuple       stup;
1168
1169         /*
1170          * If it's a pass-by-reference value, copy it into memory we control, and
1171          * decrease availMem.  Then call the common code.
1172          */
1173         if (isNull || state->datumTypeByVal)
1174         {
1175                 stup.datum1 = val;
1176                 stup.isnull1 = isNull;
1177                 stup.tuple = NULL;              /* no separate storage */
1178         }
1179         else
1180         {
1181                 stup.datum1 = datumCopy(val, false, state->datumTypeLen);
1182                 stup.isnull1 = false;
1183                 stup.tuple = DatumGetPointer(stup.datum1);
1184                 USEMEM(state, GetMemoryChunkSpace(stup.tuple));
1185         }
1186
1187         puttuple_common(state, &stup);
1188
1189         MemoryContextSwitchTo(oldcontext);
1190 }
1191
1192 /*
1193  * Shared code for tuple and datum cases.
1194  */
1195 static void
1196 puttuple_common(Tuplesortstate *state, SortTuple *tuple)
1197 {
1198         switch (state->status)
1199         {
1200                 case TSS_INITIAL:
1201
1202                         /*
1203                          * Save the tuple into the unsorted array.  First, grow the array
1204                          * as needed.  Note that we try to grow the array when there is
1205                          * still one free slot remaining --- if we fail, there'll still be
1206                          * room to store the incoming tuple, and then we'll switch to
1207                          * tape-based operation.
1208                          */
1209                         if (state->memtupcount >= state->memtupsize - 1)
1210                         {
1211                                 (void) grow_memtuples(state);
1212                                 Assert(state->memtupcount < state->memtupsize);
1213                         }
1214                         state->memtuples[state->memtupcount++] = *tuple;
1215
1216                         /*
1217                          * Check if it's time to switch over to a bounded heapsort. We do
1218                          * so if the input tuple count exceeds twice the desired tuple
1219                          * count (this is a heuristic for where heapsort becomes cheaper
1220                          * than a quicksort), or if we've just filled workMem and have
1221                          * enough tuples to meet the bound.
1222                          *
1223                          * Note that once we enter TSS_BOUNDED state we will always try to
1224                          * complete the sort that way.  In the worst case, if later input
1225                          * tuples are larger than earlier ones, this might cause us to
1226                          * exceed workMem significantly.
1227                          */
1228                         if (state->bounded &&
1229                                 (state->memtupcount > state->bound * 2 ||
1230                                  (state->memtupcount > state->bound && LACKMEM(state))))
1231                         {
1232 #ifdef TRACE_SORT
1233                                 if (trace_sort)
1234                                         elog(LOG, "switching to bounded heapsort at %d tuples: %s",
1235                                                  state->memtupcount,
1236                                                  pg_rusage_show(&state->ru_start));
1237 #endif
1238                                 make_bounded_heap(state);
1239                                 return;
1240                         }
1241
1242                         /*
1243                          * Done if we still fit in available memory and have array slots.
1244                          */
1245                         if (state->memtupcount < state->memtupsize && !LACKMEM(state))
1246                                 return;
1247
1248                         /*
1249                          * Nope; time to switch to tape-based operation.
1250                          */
1251                         inittapes(state);
1252
1253                         /*
1254                          * Dump tuples until we are back under the limit.
1255                          */
1256                         dumptuples(state, false);
1257                         break;
1258
1259                 case TSS_BOUNDED:
1260
1261                         /*
1262                          * We don't want to grow the array here, so check whether the new
1263                          * tuple can be discarded before putting it in.  This should be a
1264                          * good speed optimization, too, since when there are many more
1265                          * input tuples than the bound, most input tuples can be discarded
1266                          * with just this one comparison.  Note that because we currently
1267                          * have the sort direction reversed, we must check for <= not >=.
1268                          */
1269                         if (COMPARETUP(state, tuple, &state->memtuples[0]) <= 0)
1270                         {
1271                                 /* new tuple <= top of the heap, so we can discard it */
1272                                 free_sort_tuple(state, tuple);
1273                                 CHECK_FOR_INTERRUPTS();
1274                         }
1275                         else
1276                         {
1277                                 /* discard top of heap, sift up, insert new tuple */
1278                                 free_sort_tuple(state, &state->memtuples[0]);
1279                                 tuplesort_heap_siftup(state, false);
1280                                 tuplesort_heap_insert(state, tuple, 0, false);
1281                         }
1282                         break;
1283
1284                 case TSS_BUILDRUNS:
1285
1286                         /*
1287                          * Insert the tuple into the heap, with run number currentRun if
1288                          * it can go into the current run, else run number currentRun+1.
1289                          * The tuple can go into the current run if it is >= the first
1290                          * not-yet-output tuple.  (Actually, it could go into the current
1291                          * run if it is >= the most recently output tuple ... but that
1292                          * would require keeping around the tuple we last output, and it's
1293                          * simplest to let writetup free each tuple as soon as it's
1294                          * written.)
1295                          *
1296                          * Note there will always be at least one tuple in the heap at
1297                          * this point; see dumptuples.
1298                          */
1299                         Assert(state->memtupcount > 0);
1300                         if (COMPARETUP(state, tuple, &state->memtuples[0]) >= 0)
1301                                 tuplesort_heap_insert(state, tuple, state->currentRun, true);
1302                         else
1303                                 tuplesort_heap_insert(state, tuple, state->currentRun + 1, true);
1304
1305                         /*
1306                          * If we are over the memory limit, dump tuples till we're under.
1307                          */
1308                         dumptuples(state, false);
1309                         break;
1310
1311                 default:
1312                         elog(ERROR, "invalid tuplesort state");
1313                         break;
1314         }
1315 }
1316
1317 /*
1318  * All tuples have been provided; finish the sort.
1319  */
1320 void
1321 tuplesort_performsort(Tuplesortstate *state)
1322 {
1323         MemoryContext oldcontext = MemoryContextSwitchTo(state->sortcontext);
1324
1325 #ifdef TRACE_SORT
1326         if (trace_sort)
1327                 elog(LOG, "performsort starting: %s",
1328                          pg_rusage_show(&state->ru_start));
1329 #endif
1330
1331         switch (state->status)
1332         {
1333                 case TSS_INITIAL:
1334
1335                         /*
1336                          * We were able to accumulate all the tuples within the allowed
1337                          * amount of memory.  Just qsort 'em and we're done.
1338                          */
1339                         if (state->memtupcount > 1)
1340                         {
1341                                 /* Can we use the single-key sort function? */
1342                                 if (state->onlyKey != NULL)
1343                                         qsort_ssup(state->memtuples, state->memtupcount,
1344                                                            state->onlyKey);
1345                                 else
1346                                         qsort_tuple(state->memtuples,
1347                                                                 state->memtupcount,
1348                                                                 state->comparetup,
1349                                                                 state);
1350                         }
1351                         state->current = 0;
1352                         state->eof_reached = false;
1353                         state->markpos_offset = 0;
1354                         state->markpos_eof = false;
1355                         state->status = TSS_SORTEDINMEM;
1356                         break;
1357
1358                 case TSS_BOUNDED:
1359
1360                         /*
1361                          * We were able to accumulate all the tuples required for output
1362                          * in memory, using a heap to eliminate excess tuples.  Now we
1363                          * have to transform the heap to a properly-sorted array.
1364                          */
1365                         sort_bounded_heap(state);
1366                         state->current = 0;
1367                         state->eof_reached = false;
1368                         state->markpos_offset = 0;
1369                         state->markpos_eof = false;
1370                         state->status = TSS_SORTEDINMEM;
1371                         break;
1372
1373                 case TSS_BUILDRUNS:
1374
1375                         /*
1376                          * Finish tape-based sort.  First, flush all tuples remaining in
1377                          * memory out to tape; then merge until we have a single remaining
1378                          * run (or, if !randomAccess, one run per tape). Note that
1379                          * mergeruns sets the correct state->status.
1380                          */
1381                         dumptuples(state, true);
1382                         mergeruns(state);
1383                         state->eof_reached = false;
1384                         state->markpos_block = 0L;
1385                         state->markpos_offset = 0;
1386                         state->markpos_eof = false;
1387                         break;
1388
1389                 default:
1390                         elog(ERROR, "invalid tuplesort state");
1391                         break;
1392         }
1393
1394 #ifdef TRACE_SORT
1395         if (trace_sort)
1396         {
1397                 if (state->status == TSS_FINALMERGE)
1398                         elog(LOG, "performsort done (except %d-way final merge): %s",
1399                                  state->activeTapes,
1400                                  pg_rusage_show(&state->ru_start));
1401                 else
1402                         elog(LOG, "performsort done: %s",
1403                                  pg_rusage_show(&state->ru_start));
1404         }
1405 #endif
1406
1407         MemoryContextSwitchTo(oldcontext);
1408 }
1409
1410 /*
1411  * Internal routine to fetch the next tuple in either forward or back
1412  * direction into *stup.  Returns FALSE if no more tuples.
1413  * If *should_free is set, the caller must pfree stup.tuple when done with it.
1414  */
1415 static bool
1416 tuplesort_gettuple_common(Tuplesortstate *state, bool forward,
1417                                                   SortTuple *stup, bool *should_free)
1418 {
1419         unsigned int tuplen;
1420
1421         switch (state->status)
1422         {
1423                 case TSS_SORTEDINMEM:
1424                         Assert(forward || state->randomAccess);
1425                         *should_free = false;
1426                         if (forward)
1427                         {
1428                                 if (state->current < state->memtupcount)
1429                                 {
1430                                         *stup = state->memtuples[state->current++];
1431                                         return true;
1432                                 }
1433                                 state->eof_reached = true;
1434
1435                                 /*
1436                                  * Complain if caller tries to retrieve more tuples than
1437                                  * originally asked for in a bounded sort.  This is because
1438                                  * returning EOF here might be the wrong thing.
1439                                  */
1440                                 if (state->bounded && state->current >= state->bound)
1441                                         elog(ERROR, "retrieved too many tuples in a bounded sort");
1442
1443                                 return false;
1444                         }
1445                         else
1446                         {
1447                                 if (state->current <= 0)
1448                                         return false;
1449
1450                                 /*
1451                                  * if all tuples are fetched already then we return last
1452                                  * tuple, else - tuple before last returned.
1453                                  */
1454                                 if (state->eof_reached)
1455                                         state->eof_reached = false;
1456                                 else
1457                                 {
1458                                         state->current--;       /* last returned tuple */
1459                                         if (state->current <= 0)
1460                                                 return false;
1461                                 }
1462                                 *stup = state->memtuples[state->current - 1];
1463                                 return true;
1464                         }
1465                         break;
1466
1467                 case TSS_SORTEDONTAPE:
1468                         Assert(forward || state->randomAccess);
1469                         *should_free = true;
1470                         if (forward)
1471                         {
1472                                 if (state->eof_reached)
1473                                         return false;
1474                                 if ((tuplen = getlen(state, state->result_tape, true)) != 0)
1475                                 {
1476                                         READTUP(state, stup, state->result_tape, tuplen);
1477                                         return true;
1478                                 }
1479                                 else
1480                                 {
1481                                         state->eof_reached = true;
1482                                         return false;
1483                                 }
1484                         }
1485
1486                         /*
1487                          * Backward.
1488                          *
1489                          * if all tuples are fetched already then we return last tuple,
1490                          * else - tuple before last returned.
1491                          */
1492                         if (state->eof_reached)
1493                         {
1494                                 /*
1495                                  * Seek position is pointing just past the zero tuplen at the
1496                                  * end of file; back up to fetch last tuple's ending length
1497                                  * word.  If seek fails we must have a completely empty file.
1498                                  */
1499                                 if (!LogicalTapeBackspace(state->tapeset,
1500                                                                                   state->result_tape,
1501                                                                                   2 * sizeof(unsigned int)))
1502                                         return false;
1503                                 state->eof_reached = false;
1504                         }
1505                         else
1506                         {
1507                                 /*
1508                                  * Back up and fetch previously-returned tuple's ending length
1509                                  * word.  If seek fails, assume we are at start of file.
1510                                  */
1511                                 if (!LogicalTapeBackspace(state->tapeset,
1512                                                                                   state->result_tape,
1513                                                                                   sizeof(unsigned int)))
1514                                         return false;
1515                                 tuplen = getlen(state, state->result_tape, false);
1516
1517                                 /*
1518                                  * Back up to get ending length word of tuple before it.
1519                                  */
1520                                 if (!LogicalTapeBackspace(state->tapeset,
1521                                                                                   state->result_tape,
1522                                                                                   tuplen + 2 * sizeof(unsigned int)))
1523                                 {
1524                                         /*
1525                                          * If that fails, presumably the prev tuple is the first
1526                                          * in the file.  Back up so that it becomes next to read
1527                                          * in forward direction (not obviously right, but that is
1528                                          * what in-memory case does).
1529                                          */
1530                                         if (!LogicalTapeBackspace(state->tapeset,
1531                                                                                           state->result_tape,
1532                                                                                           tuplen + sizeof(unsigned int)))
1533                                                 elog(ERROR, "bogus tuple length in backward scan");
1534                                         return false;
1535                                 }
1536                         }
1537
1538                         tuplen = getlen(state, state->result_tape, false);
1539
1540                         /*
1541                          * Now we have the length of the prior tuple, back up and read it.
1542                          * Note: READTUP expects we are positioned after the initial
1543                          * length word of the tuple, so back up to that point.
1544                          */
1545                         if (!LogicalTapeBackspace(state->tapeset,
1546                                                                           state->result_tape,
1547                                                                           tuplen))
1548                                 elog(ERROR, "bogus tuple length in backward scan");
1549                         READTUP(state, stup, state->result_tape, tuplen);
1550                         return true;
1551
1552                 case TSS_FINALMERGE:
1553                         Assert(forward);
1554                         *should_free = true;
1555
1556                         /*
1557                          * This code should match the inner loop of mergeonerun().
1558                          */
1559                         if (state->memtupcount > 0)
1560                         {
1561                                 int                     srcTape = state->memtuples[0].tupindex;
1562                                 Size            tuplen;
1563                                 int                     tupIndex;
1564                                 SortTuple  *newtup;
1565
1566                                 *stup = state->memtuples[0];
1567                                 /* returned tuple is no longer counted in our memory space */
1568                                 if (stup->tuple)
1569                                 {
1570                                         tuplen = GetMemoryChunkSpace(stup->tuple);
1571                                         state->availMem += tuplen;
1572                                         state->mergeavailmem[srcTape] += tuplen;
1573                                 }
1574                                 tuplesort_heap_siftup(state, false);
1575                                 if ((tupIndex = state->mergenext[srcTape]) == 0)
1576                                 {
1577                                         /*
1578                                          * out of preloaded data on this tape, try to read more
1579                                          *
1580                                          * Unlike mergeonerun(), we only preload from the single
1581                                          * tape that's run dry.  See mergepreread() comments.
1582                                          */
1583                                         mergeprereadone(state, srcTape);
1584
1585                                         /*
1586                                          * if still no data, we've reached end of run on this tape
1587                                          */
1588                                         if ((tupIndex = state->mergenext[srcTape]) == 0)
1589                                                 return true;
1590                                 }
1591                                 /* pull next preread tuple from list, insert in heap */
1592                                 newtup = &state->memtuples[tupIndex];
1593                                 state->mergenext[srcTape] = newtup->tupindex;
1594                                 if (state->mergenext[srcTape] == 0)
1595                                         state->mergelast[srcTape] = 0;
1596                                 tuplesort_heap_insert(state, newtup, srcTape, false);
1597                                 /* put the now-unused memtuples entry on the freelist */
1598                                 newtup->tupindex = state->mergefreelist;
1599                                 state->mergefreelist = tupIndex;
1600                                 state->mergeavailslots[srcTape]++;
1601                                 return true;
1602                         }
1603                         return false;
1604
1605                 default:
1606                         elog(ERROR, "invalid tuplesort state");
1607                         return false;           /* keep compiler quiet */
1608         }
1609 }
1610
1611 /*
1612  * Fetch the next tuple in either forward or back direction.
1613  * If successful, put tuple in slot and return TRUE; else, clear the slot
1614  * and return FALSE.
1615  */
1616 bool
1617 tuplesort_gettupleslot(Tuplesortstate *state, bool forward,
1618                                            TupleTableSlot *slot)
1619 {
1620         MemoryContext oldcontext = MemoryContextSwitchTo(state->sortcontext);
1621         SortTuple       stup;
1622         bool            should_free;
1623
1624         if (!tuplesort_gettuple_common(state, forward, &stup, &should_free))
1625                 stup.tuple = NULL;
1626
1627         MemoryContextSwitchTo(oldcontext);
1628
1629         if (stup.tuple)
1630         {
1631                 ExecStoreMinimalTuple((MinimalTuple) stup.tuple, slot, should_free);
1632                 return true;
1633         }
1634         else
1635         {
1636                 ExecClearTuple(slot);
1637                 return false;
1638         }
1639 }
1640
1641 /*
1642  * Fetch the next tuple in either forward or back direction.
1643  * Returns NULL if no more tuples.  If *should_free is set, the
1644  * caller must pfree the returned tuple when done with it.
1645  */
1646 HeapTuple
1647 tuplesort_getheaptuple(Tuplesortstate *state, bool forward, bool *should_free)
1648 {
1649         MemoryContext oldcontext = MemoryContextSwitchTo(state->sortcontext);
1650         SortTuple       stup;
1651
1652         if (!tuplesort_gettuple_common(state, forward, &stup, should_free))
1653                 stup.tuple = NULL;
1654
1655         MemoryContextSwitchTo(oldcontext);
1656
1657         return stup.tuple;
1658 }
1659
1660 /*
1661  * Fetch the next index tuple in either forward or back direction.
1662  * Returns NULL if no more tuples.  If *should_free is set, the
1663  * caller must pfree the returned tuple when done with it.
1664  */
1665 IndexTuple
1666 tuplesort_getindextuple(Tuplesortstate *state, bool forward,
1667                                                 bool *should_free)
1668 {
1669         MemoryContext oldcontext = MemoryContextSwitchTo(state->sortcontext);
1670         SortTuple       stup;
1671
1672         if (!tuplesort_gettuple_common(state, forward, &stup, should_free))
1673                 stup.tuple = NULL;
1674
1675         MemoryContextSwitchTo(oldcontext);
1676
1677         return (IndexTuple) stup.tuple;
1678 }
1679
1680 /*
1681  * Fetch the next Datum in either forward or back direction.
1682  * Returns FALSE if no more datums.
1683  *
1684  * If the Datum is pass-by-ref type, the returned value is freshly palloc'd
1685  * and is now owned by the caller.
1686  */
1687 bool
1688 tuplesort_getdatum(Tuplesortstate *state, bool forward,
1689                                    Datum *val, bool *isNull)
1690 {
1691         MemoryContext oldcontext = MemoryContextSwitchTo(state->sortcontext);
1692         SortTuple       stup;
1693         bool            should_free;
1694
1695         if (!tuplesort_gettuple_common(state, forward, &stup, &should_free))
1696         {
1697                 MemoryContextSwitchTo(oldcontext);
1698                 return false;
1699         }
1700
1701         if (stup.isnull1 || state->datumTypeByVal)
1702         {
1703                 *val = stup.datum1;
1704                 *isNull = stup.isnull1;
1705         }
1706         else
1707         {
1708                 if (should_free)
1709                         *val = stup.datum1;
1710                 else
1711                         *val = datumCopy(stup.datum1, false, state->datumTypeLen);
1712                 *isNull = false;
1713         }
1714
1715         MemoryContextSwitchTo(oldcontext);
1716
1717         return true;
1718 }
1719
1720 /*
1721  * Advance over N tuples in either forward or back direction,
1722  * without returning any data.  N==0 is a no-op.
1723  * Returns TRUE if successful, FALSE if ran out of tuples.
1724  */
1725 bool
1726 tuplesort_skiptuples(Tuplesortstate *state, int64 ntuples, bool forward)
1727 {
1728         MemoryContext oldcontext;
1729
1730         /*
1731          * We don't actually support backwards skip yet, because no callers need
1732          * it.  The API is designed to allow for that later, though.
1733          */
1734         Assert(forward);
1735         Assert(ntuples >= 0);
1736
1737         switch (state->status)
1738         {
1739                 case TSS_SORTEDINMEM:
1740                         if (state->memtupcount - state->current >= ntuples)
1741                         {
1742                                 state->current += ntuples;
1743                                 return true;
1744                         }
1745                         state->current = state->memtupcount;
1746                         state->eof_reached = true;
1747
1748                         /*
1749                          * Complain if caller tries to retrieve more tuples than
1750                          * originally asked for in a bounded sort.  This is because
1751                          * returning EOF here might be the wrong thing.
1752                          */
1753                         if (state->bounded && state->current >= state->bound)
1754                                 elog(ERROR, "retrieved too many tuples in a bounded sort");
1755
1756                         return false;
1757
1758                 case TSS_SORTEDONTAPE:
1759                 case TSS_FINALMERGE:
1760
1761                         /*
1762                          * We could probably optimize these cases better, but for now it's
1763                          * not worth the trouble.
1764                          */
1765                         oldcontext = MemoryContextSwitchTo(state->sortcontext);
1766                         while (ntuples-- > 0)
1767                         {
1768                                 SortTuple       stup;
1769                                 bool            should_free;
1770
1771                                 if (!tuplesort_gettuple_common(state, forward,
1772                                                                                            &stup, &should_free))
1773                                 {
1774                                         MemoryContextSwitchTo(oldcontext);
1775                                         return false;
1776                                 }
1777                                 if (should_free && stup.tuple)
1778                                         pfree(stup.tuple);
1779                                 CHECK_FOR_INTERRUPTS();
1780                         }
1781                         MemoryContextSwitchTo(oldcontext);
1782                         return true;
1783
1784                 default:
1785                         elog(ERROR, "invalid tuplesort state");
1786                         return false;           /* keep compiler quiet */
1787         }
1788 }
1789
1790 /*
1791  * tuplesort_merge_order - report merge order we'll use for given memory
1792  * (note: "merge order" just means the number of input tapes in the merge).
1793  *
1794  * This is exported for use by the planner.  allowedMem is in bytes.
1795  */
1796 int
1797 tuplesort_merge_order(int64 allowedMem)
1798 {
1799         int                     mOrder;
1800
1801         /*
1802          * We need one tape for each merge input, plus another one for the output,
1803          * and each of these tapes needs buffer space.  In addition we want
1804          * MERGE_BUFFER_SIZE workspace per input tape (but the output tape doesn't
1805          * count).
1806          *
1807          * Note: you might be thinking we need to account for the memtuples[]
1808          * array in this calculation, but we effectively treat that as part of the
1809          * MERGE_BUFFER_SIZE workspace.
1810          */
1811         mOrder = (allowedMem - TAPE_BUFFER_OVERHEAD) /
1812                 (MERGE_BUFFER_SIZE + TAPE_BUFFER_OVERHEAD);
1813
1814         /* Even in minimum memory, use at least a MINORDER merge */
1815         mOrder = Max(mOrder, MINORDER);
1816
1817         return mOrder;
1818 }
1819
1820 /*
1821  * inittapes - initialize for tape sorting.
1822  *
1823  * This is called only if we have found we don't have room to sort in memory.
1824  */
1825 static void
1826 inittapes(Tuplesortstate *state)
1827 {
1828         int                     maxTapes,
1829                                 ntuples,
1830                                 j;
1831         int64           tapeSpace;
1832
1833         /* Compute number of tapes to use: merge order plus 1 */
1834         maxTapes = tuplesort_merge_order(state->allowedMem) + 1;
1835
1836         /*
1837          * We must have at least 2*maxTapes slots in the memtuples[] array, else
1838          * we'd not have room for merge heap plus preread.  It seems unlikely that
1839          * this case would ever occur, but be safe.
1840          */
1841         maxTapes = Min(maxTapes, state->memtupsize / 2);
1842
1843         state->maxTapes = maxTapes;
1844         state->tapeRange = maxTapes - 1;
1845
1846 #ifdef TRACE_SORT
1847         if (trace_sort)
1848                 elog(LOG, "switching to external sort with %d tapes: %s",
1849                          maxTapes, pg_rusage_show(&state->ru_start));
1850 #endif
1851
1852         /*
1853          * Decrease availMem to reflect the space needed for tape buffers; but
1854          * don't decrease it to the point that we have no room for tuples. (That
1855          * case is only likely to occur if sorting pass-by-value Datums; in all
1856          * other scenarios the memtuples[] array is unlikely to occupy more than
1857          * half of allowedMem.  In the pass-by-value case it's not important to
1858          * account for tuple space, so we don't care if LACKMEM becomes
1859          * inaccurate.)
1860          */
1861         tapeSpace = (int64) maxTapes *TAPE_BUFFER_OVERHEAD;
1862
1863         if (tapeSpace + GetMemoryChunkSpace(state->memtuples) < state->allowedMem)
1864                 USEMEM(state, tapeSpace);
1865
1866         /*
1867          * Make sure that the temp file(s) underlying the tape set are created in
1868          * suitable temp tablespaces.
1869          */
1870         PrepareTempTablespaces();
1871
1872         /*
1873          * Create the tape set and allocate the per-tape data arrays.
1874          */
1875         state->tapeset = LogicalTapeSetCreate(maxTapes);
1876
1877         state->mergeactive = (bool *) palloc0(maxTapes * sizeof(bool));
1878         state->mergenext = (int *) palloc0(maxTapes * sizeof(int));
1879         state->mergelast = (int *) palloc0(maxTapes * sizeof(int));
1880         state->mergeavailslots = (int *) palloc0(maxTapes * sizeof(int));
1881         state->mergeavailmem = (int64 *) palloc0(maxTapes * sizeof(int64));
1882         state->tp_fib = (int *) palloc0(maxTapes * sizeof(int));
1883         state->tp_runs = (int *) palloc0(maxTapes * sizeof(int));
1884         state->tp_dummy = (int *) palloc0(maxTapes * sizeof(int));
1885         state->tp_tapenum = (int *) palloc0(maxTapes * sizeof(int));
1886
1887         /*
1888          * Convert the unsorted contents of memtuples[] into a heap. Each tuple is
1889          * marked as belonging to run number zero.
1890          *
1891          * NOTE: we pass false for checkIndex since there's no point in comparing
1892          * indexes in this step, even though we do intend the indexes to be part
1893          * of the sort key...
1894          */
1895         ntuples = state->memtupcount;
1896         state->memtupcount = 0;         /* make the heap empty */
1897         for (j = 0; j < ntuples; j++)
1898         {
1899                 /* Must copy source tuple to avoid possible overwrite */
1900                 SortTuple       stup = state->memtuples[j];
1901
1902                 tuplesort_heap_insert(state, &stup, 0, false);
1903         }
1904         Assert(state->memtupcount == ntuples);
1905
1906         state->currentRun = 0;
1907
1908         /*
1909          * Initialize variables of Algorithm D (step D1).
1910          */
1911         for (j = 0; j < maxTapes; j++)
1912         {
1913                 state->tp_fib[j] = 1;
1914                 state->tp_runs[j] = 0;
1915                 state->tp_dummy[j] = 1;
1916                 state->tp_tapenum[j] = j;
1917         }
1918         state->tp_fib[state->tapeRange] = 0;
1919         state->tp_dummy[state->tapeRange] = 0;
1920
1921         state->Level = 1;
1922         state->destTape = 0;
1923
1924         state->status = TSS_BUILDRUNS;
1925 }
1926
1927 /*
1928  * selectnewtape -- select new tape for new initial run.
1929  *
1930  * This is called after finishing a run when we know another run
1931  * must be started.  This implements steps D3, D4 of Algorithm D.
1932  */
1933 static void
1934 selectnewtape(Tuplesortstate *state)
1935 {
1936         int                     j;
1937         int                     a;
1938
1939         /* Step D3: advance j (destTape) */
1940         if (state->tp_dummy[state->destTape] < state->tp_dummy[state->destTape + 1])
1941         {
1942                 state->destTape++;
1943                 return;
1944         }
1945         if (state->tp_dummy[state->destTape] != 0)
1946         {
1947                 state->destTape = 0;
1948                 return;
1949         }
1950
1951         /* Step D4: increase level */
1952         state->Level++;
1953         a = state->tp_fib[0];
1954         for (j = 0; j < state->tapeRange; j++)
1955         {
1956                 state->tp_dummy[j] = a + state->tp_fib[j + 1] - state->tp_fib[j];
1957                 state->tp_fib[j] = a + state->tp_fib[j + 1];
1958         }
1959         state->destTape = 0;
1960 }
1961
1962 /*
1963  * mergeruns -- merge all the completed initial runs.
1964  *
1965  * This implements steps D5, D6 of Algorithm D.  All input data has
1966  * already been written to initial runs on tape (see dumptuples).
1967  */
1968 static void
1969 mergeruns(Tuplesortstate *state)
1970 {
1971         int                     tapenum,
1972                                 svTape,
1973                                 svRuns,
1974                                 svDummy;
1975
1976         Assert(state->status == TSS_BUILDRUNS);
1977         Assert(state->memtupcount == 0);
1978
1979         /*
1980          * If we produced only one initial run (quite likely if the total data
1981          * volume is between 1X and 2X workMem), we can just use that tape as the
1982          * finished output, rather than doing a useless merge.  (This obvious
1983          * optimization is not in Knuth's algorithm.)
1984          */
1985         if (state->currentRun == 1)
1986         {
1987                 state->result_tape = state->tp_tapenum[state->destTape];
1988                 /* must freeze and rewind the finished output tape */
1989                 LogicalTapeFreeze(state->tapeset, state->result_tape);
1990                 state->status = TSS_SORTEDONTAPE;
1991                 return;
1992         }
1993
1994         /* End of step D2: rewind all output tapes to prepare for merging */
1995         for (tapenum = 0; tapenum < state->tapeRange; tapenum++)
1996                 LogicalTapeRewind(state->tapeset, tapenum, false);
1997
1998         for (;;)
1999         {
2000                 /*
2001                  * At this point we know that tape[T] is empty.  If there's just one
2002                  * (real or dummy) run left on each input tape, then only one merge
2003                  * pass remains.  If we don't have to produce a materialized sorted
2004                  * tape, we can stop at this point and do the final merge on-the-fly.
2005                  */
2006                 if (!state->randomAccess)
2007                 {
2008                         bool            allOneRun = true;
2009
2010                         Assert(state->tp_runs[state->tapeRange] == 0);
2011                         for (tapenum = 0; tapenum < state->tapeRange; tapenum++)
2012                         {
2013                                 if (state->tp_runs[tapenum] + state->tp_dummy[tapenum] != 1)
2014                                 {
2015                                         allOneRun = false;
2016                                         break;
2017                                 }
2018                         }
2019                         if (allOneRun)
2020                         {
2021                                 /* Tell logtape.c we won't be writing anymore */
2022                                 LogicalTapeSetForgetFreeSpace(state->tapeset);
2023                                 /* Initialize for the final merge pass */
2024                                 beginmerge(state);
2025                                 state->status = TSS_FINALMERGE;
2026                                 return;
2027                         }
2028                 }
2029
2030                 /* Step D5: merge runs onto tape[T] until tape[P] is empty */
2031                 while (state->tp_runs[state->tapeRange - 1] ||
2032                            state->tp_dummy[state->tapeRange - 1])
2033                 {
2034                         bool            allDummy = true;
2035
2036                         for (tapenum = 0; tapenum < state->tapeRange; tapenum++)
2037                         {
2038                                 if (state->tp_dummy[tapenum] == 0)
2039                                 {
2040                                         allDummy = false;
2041                                         break;
2042                                 }
2043                         }
2044
2045                         if (allDummy)
2046                         {
2047                                 state->tp_dummy[state->tapeRange]++;
2048                                 for (tapenum = 0; tapenum < state->tapeRange; tapenum++)
2049                                         state->tp_dummy[tapenum]--;
2050                         }
2051                         else
2052                                 mergeonerun(state);
2053                 }
2054
2055                 /* Step D6: decrease level */
2056                 if (--state->Level == 0)
2057                         break;
2058                 /* rewind output tape T to use as new input */
2059                 LogicalTapeRewind(state->tapeset, state->tp_tapenum[state->tapeRange],
2060                                                   false);
2061                 /* rewind used-up input tape P, and prepare it for write pass */
2062                 LogicalTapeRewind(state->tapeset, state->tp_tapenum[state->tapeRange - 1],
2063                                                   true);
2064                 state->tp_runs[state->tapeRange - 1] = 0;
2065
2066                 /*
2067                  * reassign tape units per step D6; note we no longer care about A[]
2068                  */
2069                 svTape = state->tp_tapenum[state->tapeRange];
2070                 svDummy = state->tp_dummy[state->tapeRange];
2071                 svRuns = state->tp_runs[state->tapeRange];
2072                 for (tapenum = state->tapeRange; tapenum > 0; tapenum--)
2073                 {
2074                         state->tp_tapenum[tapenum] = state->tp_tapenum[tapenum - 1];
2075                         state->tp_dummy[tapenum] = state->tp_dummy[tapenum - 1];
2076                         state->tp_runs[tapenum] = state->tp_runs[tapenum - 1];
2077                 }
2078                 state->tp_tapenum[0] = svTape;
2079                 state->tp_dummy[0] = svDummy;
2080                 state->tp_runs[0] = svRuns;
2081         }
2082
2083         /*
2084          * Done.  Knuth says that the result is on TAPE[1], but since we exited
2085          * the loop without performing the last iteration of step D6, we have not
2086          * rearranged the tape unit assignment, and therefore the result is on
2087          * TAPE[T].  We need to do it this way so that we can freeze the final
2088          * output tape while rewinding it.  The last iteration of step D6 would be
2089          * a waste of cycles anyway...
2090          */
2091         state->result_tape = state->tp_tapenum[state->tapeRange];
2092         LogicalTapeFreeze(state->tapeset, state->result_tape);
2093         state->status = TSS_SORTEDONTAPE;
2094 }
2095
2096 /*
2097  * Merge one run from each input tape, except ones with dummy runs.
2098  *
2099  * This is the inner loop of Algorithm D step D5.  We know that the
2100  * output tape is TAPE[T].
2101  */
2102 static void
2103 mergeonerun(Tuplesortstate *state)
2104 {
2105         int                     destTape = state->tp_tapenum[state->tapeRange];
2106         int                     srcTape;
2107         int                     tupIndex;
2108         SortTuple  *tup;
2109         int64           priorAvail,
2110                                 spaceFreed;
2111
2112         /*
2113          * Start the merge by loading one tuple from each active source tape into
2114          * the heap.  We can also decrease the input run/dummy run counts.
2115          */
2116         beginmerge(state);
2117
2118         /*
2119          * Execute merge by repeatedly extracting lowest tuple in heap, writing it
2120          * out, and replacing it with next tuple from same tape (if there is
2121          * another one).
2122          */
2123         while (state->memtupcount > 0)
2124         {
2125                 /* write the tuple to destTape */
2126                 priorAvail = state->availMem;
2127                 srcTape = state->memtuples[0].tupindex;
2128                 WRITETUP(state, destTape, &state->memtuples[0]);
2129                 /* writetup adjusted total free space, now fix per-tape space */
2130                 spaceFreed = state->availMem - priorAvail;
2131                 state->mergeavailmem[srcTape] += spaceFreed;
2132                 /* compact the heap */
2133                 tuplesort_heap_siftup(state, false);
2134                 if ((tupIndex = state->mergenext[srcTape]) == 0)
2135                 {
2136                         /* out of preloaded data on this tape, try to read more */
2137                         mergepreread(state);
2138                         /* if still no data, we've reached end of run on this tape */
2139                         if ((tupIndex = state->mergenext[srcTape]) == 0)
2140                                 continue;
2141                 }
2142                 /* pull next preread tuple from list, insert in heap */
2143                 tup = &state->memtuples[tupIndex];
2144                 state->mergenext[srcTape] = tup->tupindex;
2145                 if (state->mergenext[srcTape] == 0)
2146                         state->mergelast[srcTape] = 0;
2147                 tuplesort_heap_insert(state, tup, srcTape, false);
2148                 /* put the now-unused memtuples entry on the freelist */
2149                 tup->tupindex = state->mergefreelist;
2150                 state->mergefreelist = tupIndex;
2151                 state->mergeavailslots[srcTape]++;
2152         }
2153
2154         /*
2155          * When the heap empties, we're done.  Write an end-of-run marker on the
2156          * output tape, and increment its count of real runs.
2157          */
2158         markrunend(state, destTape);
2159         state->tp_runs[state->tapeRange]++;
2160
2161 #ifdef TRACE_SORT
2162         if (trace_sort)
2163                 elog(LOG, "finished %d-way merge step: %s", state->activeTapes,
2164                          pg_rusage_show(&state->ru_start));
2165 #endif
2166 }
2167
2168 /*
2169  * beginmerge - initialize for a merge pass
2170  *
2171  * We decrease the counts of real and dummy runs for each tape, and mark
2172  * which tapes contain active input runs in mergeactive[].  Then, load
2173  * as many tuples as we can from each active input tape, and finally
2174  * fill the merge heap with the first tuple from each active tape.
2175  */
2176 static void
2177 beginmerge(Tuplesortstate *state)
2178 {
2179         int                     activeTapes;
2180         int                     tapenum;
2181         int                     srcTape;
2182         int                     slotsPerTape;
2183         int64           spacePerTape;
2184
2185         /* Heap should be empty here */
2186         Assert(state->memtupcount == 0);
2187
2188         /* Adjust run counts and mark the active tapes */
2189         memset(state->mergeactive, 0,
2190                    state->maxTapes * sizeof(*state->mergeactive));
2191         activeTapes = 0;
2192         for (tapenum = 0; tapenum < state->tapeRange; tapenum++)
2193         {
2194                 if (state->tp_dummy[tapenum] > 0)
2195                         state->tp_dummy[tapenum]--;
2196                 else
2197                 {
2198                         Assert(state->tp_runs[tapenum] > 0);
2199                         state->tp_runs[tapenum]--;
2200                         srcTape = state->tp_tapenum[tapenum];
2201                         state->mergeactive[srcTape] = true;
2202                         activeTapes++;
2203                 }
2204         }
2205         state->activeTapes = activeTapes;
2206
2207         /* Clear merge-pass state variables */
2208         memset(state->mergenext, 0,
2209                    state->maxTapes * sizeof(*state->mergenext));
2210         memset(state->mergelast, 0,
2211                    state->maxTapes * sizeof(*state->mergelast));
2212         state->mergefreelist = 0;       /* nothing in the freelist */
2213         state->mergefirstfree = activeTapes;            /* 1st slot avail for preread */
2214
2215         /*
2216          * Initialize space allocation to let each active input tape have an equal
2217          * share of preread space.
2218          */
2219         Assert(activeTapes > 0);
2220         slotsPerTape = (state->memtupsize - state->mergefirstfree) / activeTapes;
2221         Assert(slotsPerTape > 0);
2222         spacePerTape = state->availMem / activeTapes;
2223         for (srcTape = 0; srcTape < state->maxTapes; srcTape++)
2224         {
2225                 if (state->mergeactive[srcTape])
2226                 {
2227                         state->mergeavailslots[srcTape] = slotsPerTape;
2228                         state->mergeavailmem[srcTape] = spacePerTape;
2229                 }
2230         }
2231
2232         /*
2233          * Preread as many tuples as possible (and at least one) from each active
2234          * tape
2235          */
2236         mergepreread(state);
2237
2238         /* Load the merge heap with the first tuple from each input tape */
2239         for (srcTape = 0; srcTape < state->maxTapes; srcTape++)
2240         {
2241                 int                     tupIndex = state->mergenext[srcTape];
2242                 SortTuple  *tup;
2243
2244                 if (tupIndex)
2245                 {
2246                         tup = &state->memtuples[tupIndex];
2247                         state->mergenext[srcTape] = tup->tupindex;
2248                         if (state->mergenext[srcTape] == 0)
2249                                 state->mergelast[srcTape] = 0;
2250                         tuplesort_heap_insert(state, tup, srcTape, false);
2251                         /* put the now-unused memtuples entry on the freelist */
2252                         tup->tupindex = state->mergefreelist;
2253                         state->mergefreelist = tupIndex;
2254                         state->mergeavailslots[srcTape]++;
2255                 }
2256         }
2257 }
2258
2259 /*
2260  * mergepreread - load tuples from merge input tapes
2261  *
2262  * This routine exists to improve sequentiality of reads during a merge pass,
2263  * as explained in the header comments of this file.  Load tuples from each
2264  * active source tape until the tape's run is exhausted or it has used up
2265  * its fair share of available memory.  In any case, we guarantee that there
2266  * is at least one preread tuple available from each unexhausted input tape.
2267  *
2268  * We invoke this routine at the start of a merge pass for initial load,
2269  * and then whenever any tape's preread data runs out.  Note that we load
2270  * as much data as possible from all tapes, not just the one that ran out.
2271  * This is because logtape.c works best with a usage pattern that alternates
2272  * between reading a lot of data and writing a lot of data, so whenever we
2273  * are forced to read, we should fill working memory completely.
2274  *
2275  * In FINALMERGE state, we *don't* use this routine, but instead just preread
2276  * from the single tape that ran dry.  There's no read/write alternation in
2277  * that state and so no point in scanning through all the tapes to fix one.
2278  * (Moreover, there may be quite a lot of inactive tapes in that state, since
2279  * we might have had many fewer runs than tapes.  In a regular tape-to-tape
2280  * merge we can expect most of the tapes to be active.)
2281  */
2282 static void
2283 mergepreread(Tuplesortstate *state)
2284 {
2285         int                     srcTape;
2286
2287         for (srcTape = 0; srcTape < state->maxTapes; srcTape++)
2288                 mergeprereadone(state, srcTape);
2289 }
2290
2291 /*
2292  * mergeprereadone - load tuples from one merge input tape
2293  *
2294  * Read tuples from the specified tape until it has used up its free memory
2295  * or array slots; but ensure that we have at least one tuple, if any are
2296  * to be had.
2297  */
2298 static void
2299 mergeprereadone(Tuplesortstate *state, int srcTape)
2300 {
2301         unsigned int tuplen;
2302         SortTuple       stup;
2303         int                     tupIndex;
2304         int64           priorAvail,
2305                                 spaceUsed;
2306
2307         if (!state->mergeactive[srcTape])
2308                 return;                                 /* tape's run is already exhausted */
2309         priorAvail = state->availMem;
2310         state->availMem = state->mergeavailmem[srcTape];
2311         while ((state->mergeavailslots[srcTape] > 0 && !LACKMEM(state)) ||
2312                    state->mergenext[srcTape] == 0)
2313         {
2314                 /* read next tuple, if any */
2315                 if ((tuplen = getlen(state, srcTape, true)) == 0)
2316                 {
2317                         state->mergeactive[srcTape] = false;
2318                         break;
2319                 }
2320                 READTUP(state, &stup, srcTape, tuplen);
2321                 /* find a free slot in memtuples[] for it */
2322                 tupIndex = state->mergefreelist;
2323                 if (tupIndex)
2324                         state->mergefreelist = state->memtuples[tupIndex].tupindex;
2325                 else
2326                 {
2327                         tupIndex = state->mergefirstfree++;
2328                         Assert(tupIndex < state->memtupsize);
2329                 }
2330                 state->mergeavailslots[srcTape]--;
2331                 /* store tuple, append to list for its tape */
2332                 stup.tupindex = 0;
2333                 state->memtuples[tupIndex] = stup;
2334                 if (state->mergelast[srcTape])
2335                         state->memtuples[state->mergelast[srcTape]].tupindex = tupIndex;
2336                 else
2337                         state->mergenext[srcTape] = tupIndex;
2338                 state->mergelast[srcTape] = tupIndex;
2339         }
2340         /* update per-tape and global availmem counts */
2341         spaceUsed = state->mergeavailmem[srcTape] - state->availMem;
2342         state->mergeavailmem[srcTape] = state->availMem;
2343         state->availMem = priorAvail - spaceUsed;
2344 }
2345
2346 /*
2347  * dumptuples - remove tuples from heap and write to tape
2348  *
2349  * This is used during initial-run building, but not during merging.
2350  *
2351  * When alltuples = false, dump only enough tuples to get under the
2352  * availMem limit (and leave at least one tuple in the heap in any case,
2353  * since puttuple assumes it always has a tuple to compare to).  We also
2354  * insist there be at least one free slot in the memtuples[] array.
2355  *
2356  * When alltuples = true, dump everything currently in memory.
2357  * (This case is only used at end of input data.)
2358  *
2359  * If we empty the heap, close out the current run and return (this should
2360  * only happen at end of input data).  If we see that the tuple run number
2361  * at the top of the heap has changed, start a new run.
2362  */
2363 static void
2364 dumptuples(Tuplesortstate *state, bool alltuples)
2365 {
2366         while (alltuples ||
2367                    (LACKMEM(state) && state->memtupcount > 1) ||
2368                    state->memtupcount >= state->memtupsize)
2369         {
2370                 /*
2371                  * Dump the heap's frontmost entry, and sift up to remove it from the
2372                  * heap.
2373                  */
2374                 Assert(state->memtupcount > 0);
2375                 WRITETUP(state, state->tp_tapenum[state->destTape],
2376                                  &state->memtuples[0]);
2377                 tuplesort_heap_siftup(state, true);
2378
2379                 /*
2380                  * If the heap is empty *or* top run number has changed, we've
2381                  * finished the current run.
2382                  */
2383                 if (state->memtupcount == 0 ||
2384                         state->currentRun != state->memtuples[0].tupindex)
2385                 {
2386                         markrunend(state, state->tp_tapenum[state->destTape]);
2387                         state->currentRun++;
2388                         state->tp_runs[state->destTape]++;
2389                         state->tp_dummy[state->destTape]--; /* per Alg D step D2 */
2390
2391 #ifdef TRACE_SORT
2392                         if (trace_sort)
2393                                 elog(LOG, "finished writing%s run %d to tape %d: %s",
2394                                          (state->memtupcount == 0) ? " final" : "",
2395                                          state->currentRun, state->destTape,
2396                                          pg_rusage_show(&state->ru_start));
2397 #endif
2398
2399                         /*
2400                          * Done if heap is empty, else prepare for new run.
2401                          */
2402                         if (state->memtupcount == 0)
2403                                 break;
2404                         Assert(state->currentRun == state->memtuples[0].tupindex);
2405                         selectnewtape(state);
2406                 }
2407         }
2408 }
2409
2410 /*
2411  * tuplesort_rescan             - rewind and replay the scan
2412  */
2413 void
2414 tuplesort_rescan(Tuplesortstate *state)
2415 {
2416         MemoryContext oldcontext = MemoryContextSwitchTo(state->sortcontext);
2417
2418         Assert(state->randomAccess);
2419
2420         switch (state->status)
2421         {
2422                 case TSS_SORTEDINMEM:
2423                         state->current = 0;
2424                         state->eof_reached = false;
2425                         state->markpos_offset = 0;
2426                         state->markpos_eof = false;
2427                         break;
2428                 case TSS_SORTEDONTAPE:
2429                         LogicalTapeRewind(state->tapeset,
2430                                                           state->result_tape,
2431                                                           false);
2432                         state->eof_reached = false;
2433                         state->markpos_block = 0L;
2434                         state->markpos_offset = 0;
2435                         state->markpos_eof = false;
2436                         break;
2437                 default:
2438                         elog(ERROR, "invalid tuplesort state");
2439                         break;
2440         }
2441
2442         MemoryContextSwitchTo(oldcontext);
2443 }
2444
2445 /*
2446  * tuplesort_markpos    - saves current position in the merged sort file
2447  */
2448 void
2449 tuplesort_markpos(Tuplesortstate *state)
2450 {
2451         MemoryContext oldcontext = MemoryContextSwitchTo(state->sortcontext);
2452
2453         Assert(state->randomAccess);
2454
2455         switch (state->status)
2456         {
2457                 case TSS_SORTEDINMEM:
2458                         state->markpos_offset = state->current;
2459                         state->markpos_eof = state->eof_reached;
2460                         break;
2461                 case TSS_SORTEDONTAPE:
2462                         LogicalTapeTell(state->tapeset,
2463                                                         state->result_tape,
2464                                                         &state->markpos_block,
2465                                                         &state->markpos_offset);
2466                         state->markpos_eof = state->eof_reached;
2467                         break;
2468                 default:
2469                         elog(ERROR, "invalid tuplesort state");
2470                         break;
2471         }
2472
2473         MemoryContextSwitchTo(oldcontext);
2474 }
2475
2476 /*
2477  * tuplesort_restorepos - restores current position in merged sort file to
2478  *                                                last saved position
2479  */
2480 void
2481 tuplesort_restorepos(Tuplesortstate *state)
2482 {
2483         MemoryContext oldcontext = MemoryContextSwitchTo(state->sortcontext);
2484
2485         Assert(state->randomAccess);
2486
2487         switch (state->status)
2488         {
2489                 case TSS_SORTEDINMEM:
2490                         state->current = state->markpos_offset;
2491                         state->eof_reached = state->markpos_eof;
2492                         break;
2493                 case TSS_SORTEDONTAPE:
2494                         if (!LogicalTapeSeek(state->tapeset,
2495                                                                  state->result_tape,
2496                                                                  state->markpos_block,
2497                                                                  state->markpos_offset))
2498                                 elog(ERROR, "tuplesort_restorepos failed");
2499                         state->eof_reached = state->markpos_eof;
2500                         break;
2501                 default:
2502                         elog(ERROR, "invalid tuplesort state");
2503                         break;
2504         }
2505
2506         MemoryContextSwitchTo(oldcontext);
2507 }
2508
2509 /*
2510  * tuplesort_get_stats - extract summary statistics
2511  *
2512  * This can be called after tuplesort_performsort() finishes to obtain
2513  * printable summary information about how the sort was performed.
2514  * spaceUsed is measured in kilobytes.
2515  */
2516 void
2517 tuplesort_get_stats(Tuplesortstate *state,
2518                                         const char **sortMethod,
2519                                         const char **spaceType,
2520                                         long *spaceUsed)
2521 {
2522         /*
2523          * Note: it might seem we should provide both memory and disk usage for a
2524          * disk-based sort.  However, the current code doesn't track memory space
2525          * accurately once we have begun to return tuples to the caller (since we
2526          * don't account for pfree's the caller is expected to do), so we cannot
2527          * rely on availMem in a disk sort.  This does not seem worth the overhead
2528          * to fix.  Is it worth creating an API for the memory context code to
2529          * tell us how much is actually used in sortcontext?
2530          */
2531         if (state->tapeset)
2532         {
2533                 *spaceType = "Disk";
2534                 *spaceUsed = LogicalTapeSetBlocks(state->tapeset) * (BLCKSZ / 1024);
2535         }
2536         else
2537         {
2538                 *spaceType = "Memory";
2539                 *spaceUsed = (state->allowedMem - state->availMem + 1023) / 1024;
2540         }
2541
2542         switch (state->status)
2543         {
2544                 case TSS_SORTEDINMEM:
2545                         if (state->boundUsed)
2546                                 *sortMethod = "top-N heapsort";
2547                         else
2548                                 *sortMethod = "quicksort";
2549                         break;
2550                 case TSS_SORTEDONTAPE:
2551                         *sortMethod = "external sort";
2552                         break;
2553                 case TSS_FINALMERGE:
2554                         *sortMethod = "external merge";
2555                         break;
2556                 default:
2557                         *sortMethod = "still in progress";
2558                         break;
2559         }
2560 }
2561
2562
2563 /*
2564  * Heap manipulation routines, per Knuth's Algorithm 5.2.3H.
2565  *
2566  * Compare two SortTuples.  If checkIndex is true, use the tuple index
2567  * as the front of the sort key; otherwise, no.
2568  */
2569
2570 #define HEAPCOMPARE(tup1,tup2) \
2571         (checkIndex && ((tup1)->tupindex != (tup2)->tupindex) ? \
2572          ((tup1)->tupindex) - ((tup2)->tupindex) : \
2573          COMPARETUP(state, tup1, tup2))
2574
2575 /*
2576  * Convert the existing unordered array of SortTuples to a bounded heap,
2577  * discarding all but the smallest "state->bound" tuples.
2578  *
2579  * When working with a bounded heap, we want to keep the largest entry
2580  * at the root (array entry zero), instead of the smallest as in the normal
2581  * sort case.  This allows us to discard the largest entry cheaply.
2582  * Therefore, we temporarily reverse the sort direction.
2583  *
2584  * We assume that all entries in a bounded heap will always have tupindex
2585  * zero; it therefore doesn't matter that HEAPCOMPARE() doesn't reverse
2586  * the direction of comparison for tupindexes.
2587  */
2588 static void
2589 make_bounded_heap(Tuplesortstate *state)
2590 {
2591         int                     tupcount = state->memtupcount;
2592         int                     i;
2593
2594         Assert(state->status == TSS_INITIAL);
2595         Assert(state->bounded);
2596         Assert(tupcount >= state->bound);
2597
2598         /* Reverse sort direction so largest entry will be at root */
2599         REVERSEDIRECTION(state);
2600
2601         state->memtupcount = 0;         /* make the heap empty */
2602         for (i = 0; i < tupcount; i++)
2603         {
2604                 if (state->memtupcount >= state->bound &&
2605                   COMPARETUP(state, &state->memtuples[i], &state->memtuples[0]) <= 0)
2606                 {
2607                         /* New tuple would just get thrown out, so skip it */
2608                         free_sort_tuple(state, &state->memtuples[i]);
2609                         CHECK_FOR_INTERRUPTS();
2610                 }
2611                 else
2612                 {
2613                         /* Insert next tuple into heap */
2614                         /* Must copy source tuple to avoid possible overwrite */
2615                         SortTuple       stup = state->memtuples[i];
2616
2617                         tuplesort_heap_insert(state, &stup, 0, false);
2618
2619                         /* If heap too full, discard largest entry */
2620                         if (state->memtupcount > state->bound)
2621                         {
2622                                 free_sort_tuple(state, &state->memtuples[0]);
2623                                 tuplesort_heap_siftup(state, false);
2624                         }
2625                 }
2626         }
2627
2628         Assert(state->memtupcount == state->bound);
2629         state->status = TSS_BOUNDED;
2630 }
2631
2632 /*
2633  * Convert the bounded heap to a properly-sorted array
2634  */
2635 static void
2636 sort_bounded_heap(Tuplesortstate *state)
2637 {
2638         int                     tupcount = state->memtupcount;
2639
2640         Assert(state->status == TSS_BOUNDED);
2641         Assert(state->bounded);
2642         Assert(tupcount == state->bound);
2643
2644         /*
2645          * We can unheapify in place because each sift-up will remove the largest
2646          * entry, which we can promptly store in the newly freed slot at the end.
2647          * Once we're down to a single-entry heap, we're done.
2648          */
2649         while (state->memtupcount > 1)
2650         {
2651                 SortTuple       stup = state->memtuples[0];
2652
2653                 /* this sifts-up the next-largest entry and decreases memtupcount */
2654                 tuplesort_heap_siftup(state, false);
2655                 state->memtuples[state->memtupcount] = stup;
2656         }
2657         state->memtupcount = tupcount;
2658
2659         /*
2660          * Reverse sort direction back to the original state.  This is not
2661          * actually necessary but seems like a good idea for tidiness.
2662          */
2663         REVERSEDIRECTION(state);
2664
2665         state->status = TSS_SORTEDINMEM;
2666         state->boundUsed = true;
2667 }
2668
2669 /*
2670  * Insert a new tuple into an empty or existing heap, maintaining the
2671  * heap invariant.  Caller is responsible for ensuring there's room.
2672  *
2673  * Note: we assume *tuple is a temporary variable that can be scribbled on.
2674  * For some callers, tuple actually points to a memtuples[] entry above the
2675  * end of the heap.  This is safe as long as it's not immediately adjacent
2676  * to the end of the heap (ie, in the [memtupcount] array entry) --- if it
2677  * is, it might get overwritten before being moved into the heap!
2678  */
2679 static void
2680 tuplesort_heap_insert(Tuplesortstate *state, SortTuple *tuple,
2681                                           int tupleindex, bool checkIndex)
2682 {
2683         SortTuple  *memtuples;
2684         int                     j;
2685
2686         /*
2687          * Save the tupleindex --- see notes above about writing on *tuple. It's a
2688          * historical artifact that tupleindex is passed as a separate argument
2689          * and not in *tuple, but it's notationally convenient so let's leave it
2690          * that way.
2691          */
2692         tuple->tupindex = tupleindex;
2693
2694         memtuples = state->memtuples;
2695         Assert(state->memtupcount < state->memtupsize);
2696
2697         CHECK_FOR_INTERRUPTS();
2698
2699         /*
2700          * Sift-up the new entry, per Knuth 5.2.3 exercise 16. Note that Knuth is
2701          * using 1-based array indexes, not 0-based.
2702          */
2703         j = state->memtupcount++;
2704         while (j > 0)
2705         {
2706                 int                     i = (j - 1) >> 1;
2707
2708                 if (HEAPCOMPARE(tuple, &memtuples[i]) >= 0)
2709                         break;
2710                 memtuples[j] = memtuples[i];
2711                 j = i;
2712         }
2713         memtuples[j] = *tuple;
2714 }
2715
2716 /*
2717  * The tuple at state->memtuples[0] has been removed from the heap.
2718  * Decrement memtupcount, and sift up to maintain the heap invariant.
2719  */
2720 static void
2721 tuplesort_heap_siftup(Tuplesortstate *state, bool checkIndex)
2722 {
2723         SortTuple  *memtuples = state->memtuples;
2724         SortTuple  *tuple;
2725         int                     i,
2726                                 n;
2727
2728         if (--state->memtupcount <= 0)
2729                 return;
2730
2731         CHECK_FOR_INTERRUPTS();
2732
2733         n = state->memtupcount;
2734         tuple = &memtuples[n];          /* tuple that must be reinserted */
2735         i = 0;                                          /* i is where the "hole" is */
2736         for (;;)
2737         {
2738                 int                     j = 2 * i + 1;
2739
2740                 if (j >= n)
2741                         break;
2742                 if (j + 1 < n &&
2743                         HEAPCOMPARE(&memtuples[j], &memtuples[j + 1]) > 0)
2744                         j++;
2745                 if (HEAPCOMPARE(tuple, &memtuples[j]) <= 0)
2746                         break;
2747                 memtuples[i] = memtuples[j];
2748                 i = j;
2749         }
2750         memtuples[i] = *tuple;
2751 }
2752
2753
2754 /*
2755  * Tape interface routines
2756  */
2757
2758 static unsigned int
2759 getlen(Tuplesortstate *state, int tapenum, bool eofOK)
2760 {
2761         unsigned int len;
2762
2763         if (LogicalTapeRead(state->tapeset, tapenum,
2764                                                 &len, sizeof(len)) != sizeof(len))
2765                 elog(ERROR, "unexpected end of tape");
2766         if (len == 0 && !eofOK)
2767                 elog(ERROR, "unexpected end of data");
2768         return len;
2769 }
2770
2771 static void
2772 markrunend(Tuplesortstate *state, int tapenum)
2773 {
2774         unsigned int len = 0;
2775
2776         LogicalTapeWrite(state->tapeset, tapenum, (void *) &len, sizeof(len));
2777 }
2778
2779
2780 /*
2781  * Inline-able copy of FunctionCall2Coll() to save some cycles in sorting.
2782  */
2783 static inline Datum
2784 myFunctionCall2Coll(FmgrInfo *flinfo, Oid collation, Datum arg1, Datum arg2)
2785 {
2786         FunctionCallInfoData fcinfo;
2787         Datum           result;
2788
2789         InitFunctionCallInfoData(fcinfo, flinfo, 2, collation, NULL, NULL);
2790
2791         fcinfo.arg[0] = arg1;
2792         fcinfo.arg[1] = arg2;
2793         fcinfo.argnull[0] = false;
2794         fcinfo.argnull[1] = false;
2795
2796         result = FunctionCallInvoke(&fcinfo);
2797
2798         /* Check for null result, since caller is clearly not expecting one */
2799         if (fcinfo.isnull)
2800                 elog(ERROR, "function %u returned NULL", fcinfo.flinfo->fn_oid);
2801
2802         return result;
2803 }
2804
2805 /*
2806  * Apply a sort function (by now converted to fmgr lookup form)
2807  * and return a 3-way comparison result.  This takes care of handling
2808  * reverse-sort and NULLs-ordering properly.  We assume that DESC and
2809  * NULLS_FIRST options are encoded in sk_flags the same way btree does it.
2810  */
2811 static inline int32
2812 inlineApplySortFunction(FmgrInfo *sortFunction, int sk_flags, Oid collation,
2813                                                 Datum datum1, bool isNull1,
2814                                                 Datum datum2, bool isNull2)
2815 {
2816         int32           compare;
2817
2818         if (isNull1)
2819         {
2820                 if (isNull2)
2821                         compare = 0;            /* NULL "=" NULL */
2822                 else if (sk_flags & SK_BT_NULLS_FIRST)
2823                         compare = -1;           /* NULL "<" NOT_NULL */
2824                 else
2825                         compare = 1;            /* NULL ">" NOT_NULL */
2826         }
2827         else if (isNull2)
2828         {
2829                 if (sk_flags & SK_BT_NULLS_FIRST)
2830                         compare = 1;            /* NOT_NULL ">" NULL */
2831                 else
2832                         compare = -1;           /* NOT_NULL "<" NULL */
2833         }
2834         else
2835         {
2836                 compare = DatumGetInt32(myFunctionCall2Coll(sortFunction, collation,
2837                                                                                                         datum1, datum2));
2838
2839                 if (sk_flags & SK_BT_DESC)
2840                         compare = -compare;
2841         }
2842
2843         return compare;
2844 }
2845
2846
2847 /*
2848  * Routines specialized for HeapTuple (actually MinimalTuple) case
2849  */
2850
2851 static int
2852 comparetup_heap(const SortTuple *a, const SortTuple *b, Tuplesortstate *state)
2853 {
2854         SortSupport sortKey = state->sortKeys;
2855         HeapTupleData ltup;
2856         HeapTupleData rtup;
2857         TupleDesc       tupDesc;
2858         int                     nkey;
2859         int32           compare;
2860
2861         /* Compare the leading sort key */
2862         compare = ApplySortComparator(a->datum1, a->isnull1,
2863                                                                   b->datum1, b->isnull1,
2864                                                                   sortKey);
2865         if (compare != 0)
2866                 return compare;
2867
2868         /* Compare additional sort keys */
2869         ltup.t_len = ((MinimalTuple) a->tuple)->t_len + MINIMAL_TUPLE_OFFSET;
2870         ltup.t_data = (HeapTupleHeader) ((char *) a->tuple - MINIMAL_TUPLE_OFFSET);
2871         rtup.t_len = ((MinimalTuple) b->tuple)->t_len + MINIMAL_TUPLE_OFFSET;
2872         rtup.t_data = (HeapTupleHeader) ((char *) b->tuple - MINIMAL_TUPLE_OFFSET);
2873         tupDesc = state->tupDesc;
2874         sortKey++;
2875         for (nkey = 1; nkey < state->nKeys; nkey++, sortKey++)
2876         {
2877                 AttrNumber      attno = sortKey->ssup_attno;
2878                 Datum           datum1,
2879                                         datum2;
2880                 bool            isnull1,
2881                                         isnull2;
2882
2883                 datum1 = heap_getattr(&ltup, attno, tupDesc, &isnull1);
2884                 datum2 = heap_getattr(&rtup, attno, tupDesc, &isnull2);
2885
2886                 compare = ApplySortComparator(datum1, isnull1,
2887                                                                           datum2, isnull2,
2888                                                                           sortKey);
2889                 if (compare != 0)
2890                         return compare;
2891         }
2892
2893         return 0;
2894 }
2895
2896 static void
2897 copytup_heap(Tuplesortstate *state, SortTuple *stup, void *tup)
2898 {
2899         /*
2900          * We expect the passed "tup" to be a TupleTableSlot, and form a
2901          * MinimalTuple using the exported interface for that.
2902          */
2903         TupleTableSlot *slot = (TupleTableSlot *) tup;
2904         MinimalTuple tuple;
2905         HeapTupleData htup;
2906
2907         /* copy the tuple into sort storage */
2908         tuple = ExecCopySlotMinimalTuple(slot);
2909         stup->tuple = (void *) tuple;
2910         USEMEM(state, GetMemoryChunkSpace(tuple));
2911         /* set up first-column key value */
2912         htup.t_len = tuple->t_len + MINIMAL_TUPLE_OFFSET;
2913         htup.t_data = (HeapTupleHeader) ((char *) tuple - MINIMAL_TUPLE_OFFSET);
2914         stup->datum1 = heap_getattr(&htup,
2915                                                                 state->sortKeys[0].ssup_attno,
2916                                                                 state->tupDesc,
2917                                                                 &stup->isnull1);
2918 }
2919
2920 static void
2921 writetup_heap(Tuplesortstate *state, int tapenum, SortTuple *stup)
2922 {
2923         MinimalTuple tuple = (MinimalTuple) stup->tuple;
2924
2925         /* the part of the MinimalTuple we'll write: */
2926         char       *tupbody = (char *) tuple + MINIMAL_TUPLE_DATA_OFFSET;
2927         unsigned int tupbodylen = tuple->t_len - MINIMAL_TUPLE_DATA_OFFSET;
2928
2929         /* total on-disk footprint: */
2930         unsigned int tuplen = tupbodylen + sizeof(int);
2931
2932         LogicalTapeWrite(state->tapeset, tapenum,
2933                                          (void *) &tuplen, sizeof(tuplen));
2934         LogicalTapeWrite(state->tapeset, tapenum,
2935                                          (void *) tupbody, tupbodylen);
2936         if (state->randomAccess)        /* need trailing length word? */
2937                 LogicalTapeWrite(state->tapeset, tapenum,
2938                                                  (void *) &tuplen, sizeof(tuplen));
2939
2940         FREEMEM(state, GetMemoryChunkSpace(tuple));
2941         heap_free_minimal_tuple(tuple);
2942 }
2943
2944 static void
2945 readtup_heap(Tuplesortstate *state, SortTuple *stup,
2946                          int tapenum, unsigned int len)
2947 {
2948         unsigned int tupbodylen = len - sizeof(int);
2949         unsigned int tuplen = tupbodylen + MINIMAL_TUPLE_DATA_OFFSET;
2950         MinimalTuple tuple = (MinimalTuple) palloc(tuplen);
2951         char       *tupbody = (char *) tuple + MINIMAL_TUPLE_DATA_OFFSET;
2952         HeapTupleData htup;
2953
2954         USEMEM(state, GetMemoryChunkSpace(tuple));
2955         /* read in the tuple proper */
2956         tuple->t_len = tuplen;
2957         LogicalTapeReadExact(state->tapeset, tapenum,
2958                                                  tupbody, tupbodylen);
2959         if (state->randomAccess)        /* need trailing length word? */
2960                 LogicalTapeReadExact(state->tapeset, tapenum,
2961                                                          &tuplen, sizeof(tuplen));
2962         stup->tuple = (void *) tuple;
2963         /* set up first-column key value */
2964         htup.t_len = tuple->t_len + MINIMAL_TUPLE_OFFSET;
2965         htup.t_data = (HeapTupleHeader) ((char *) tuple - MINIMAL_TUPLE_OFFSET);
2966         stup->datum1 = heap_getattr(&htup,
2967                                                                 state->sortKeys[0].ssup_attno,
2968                                                                 state->tupDesc,
2969                                                                 &stup->isnull1);
2970 }
2971
2972 static void
2973 reversedirection_heap(Tuplesortstate *state)
2974 {
2975         SortSupport sortKey = state->sortKeys;
2976         int                     nkey;
2977
2978         for (nkey = 0; nkey < state->nKeys; nkey++, sortKey++)
2979         {
2980                 sortKey->ssup_reverse = !sortKey->ssup_reverse;
2981                 sortKey->ssup_nulls_first = !sortKey->ssup_nulls_first;
2982         }
2983 }
2984
2985
2986 /*
2987  * Routines specialized for the CLUSTER case (HeapTuple data, with
2988  * comparisons per a btree index definition)
2989  */
2990
2991 static int
2992 comparetup_cluster(const SortTuple *a, const SortTuple *b,
2993                                    Tuplesortstate *state)
2994 {
2995         ScanKey         scanKey = state->indexScanKey;
2996         HeapTuple       ltup;
2997         HeapTuple       rtup;
2998         TupleDesc       tupDesc;
2999         int                     nkey;
3000         int32           compare;
3001
3002         /* Compare the leading sort key, if it's simple */
3003         if (state->indexInfo->ii_KeyAttrNumbers[0] != 0)
3004         {
3005                 compare = inlineApplySortFunction(&scanKey->sk_func, scanKey->sk_flags,
3006                                                                                   scanKey->sk_collation,
3007                                                                                   a->datum1, a->isnull1,
3008                                                                                   b->datum1, b->isnull1);
3009                 if (compare != 0 || state->nKeys == 1)
3010                         return compare;
3011                 /* Compare additional columns the hard way */
3012                 scanKey++;
3013                 nkey = 1;
3014         }
3015         else
3016         {
3017                 /* Must compare all keys the hard way */
3018                 nkey = 0;
3019         }
3020
3021         /* Compare additional sort keys */
3022         ltup = (HeapTuple) a->tuple;
3023         rtup = (HeapTuple) b->tuple;
3024
3025         if (state->indexInfo->ii_Expressions == NULL)
3026         {
3027                 /* If not expression index, just compare the proper heap attrs */
3028                 tupDesc = state->tupDesc;
3029
3030                 for (; nkey < state->nKeys; nkey++, scanKey++)
3031                 {
3032                         AttrNumber      attno = state->indexInfo->ii_KeyAttrNumbers[nkey];
3033                         Datum           datum1,
3034                                                 datum2;
3035                         bool            isnull1,
3036                                                 isnull2;
3037
3038                         datum1 = heap_getattr(ltup, attno, tupDesc, &isnull1);
3039                         datum2 = heap_getattr(rtup, attno, tupDesc, &isnull2);
3040
3041                         compare = inlineApplySortFunction(&scanKey->sk_func,
3042                                                                                           scanKey->sk_flags,
3043                                                                                           scanKey->sk_collation,
3044                                                                                           datum1, isnull1,
3045                                                                                           datum2, isnull2);
3046                         if (compare != 0)
3047                                 return compare;
3048                 }
3049         }
3050         else
3051         {
3052                 /*
3053                  * In the expression index case, compute the whole index tuple and
3054                  * then compare values.  It would perhaps be faster to compute only as
3055                  * many columns as we need to compare, but that would require
3056                  * duplicating all the logic in FormIndexDatum.
3057                  */
3058                 Datum           l_index_values[INDEX_MAX_KEYS];
3059                 bool            l_index_isnull[INDEX_MAX_KEYS];
3060                 Datum           r_index_values[INDEX_MAX_KEYS];
3061                 bool            r_index_isnull[INDEX_MAX_KEYS];
3062                 TupleTableSlot *ecxt_scantuple;
3063
3064                 /* Reset context each time to prevent memory leakage */
3065                 ResetPerTupleExprContext(state->estate);
3066
3067                 ecxt_scantuple = GetPerTupleExprContext(state->estate)->ecxt_scantuple;
3068
3069                 ExecStoreTuple(ltup, ecxt_scantuple, InvalidBuffer, false);
3070                 FormIndexDatum(state->indexInfo, ecxt_scantuple, state->estate,
3071                                            l_index_values, l_index_isnull);
3072
3073                 ExecStoreTuple(rtup, ecxt_scantuple, InvalidBuffer, false);
3074                 FormIndexDatum(state->indexInfo, ecxt_scantuple, state->estate,
3075                                            r_index_values, r_index_isnull);
3076
3077                 for (; nkey < state->nKeys; nkey++, scanKey++)
3078                 {
3079                         compare = inlineApplySortFunction(&scanKey->sk_func,
3080                                                                                           scanKey->sk_flags,
3081                                                                                           scanKey->sk_collation,
3082                                                                                           l_index_values[nkey],
3083                                                                                           l_index_isnull[nkey],
3084                                                                                           r_index_values[nkey],
3085                                                                                           r_index_isnull[nkey]);
3086                         if (compare != 0)
3087                                 return compare;
3088                 }
3089         }
3090
3091         return 0;
3092 }
3093
3094 static void
3095 copytup_cluster(Tuplesortstate *state, SortTuple *stup, void *tup)
3096 {
3097         HeapTuple       tuple = (HeapTuple) tup;
3098
3099         /* copy the tuple into sort storage */
3100         tuple = heap_copytuple(tuple);
3101         stup->tuple = (void *) tuple;
3102         USEMEM(state, GetMemoryChunkSpace(tuple));
3103         /* set up first-column key value, if it's a simple column */
3104         if (state->indexInfo->ii_KeyAttrNumbers[0] != 0)
3105                 stup->datum1 = heap_getattr(tuple,
3106                                                                         state->indexInfo->ii_KeyAttrNumbers[0],
3107                                                                         state->tupDesc,
3108                                                                         &stup->isnull1);
3109 }
3110
3111 static void
3112 writetup_cluster(Tuplesortstate *state, int tapenum, SortTuple *stup)
3113 {
3114         HeapTuple       tuple = (HeapTuple) stup->tuple;
3115         unsigned int tuplen = tuple->t_len + sizeof(ItemPointerData) + sizeof(int);
3116
3117         /* We need to store t_self, but not other fields of HeapTupleData */
3118         LogicalTapeWrite(state->tapeset, tapenum,
3119                                          &tuplen, sizeof(tuplen));
3120         LogicalTapeWrite(state->tapeset, tapenum,
3121                                          &tuple->t_self, sizeof(ItemPointerData));
3122         LogicalTapeWrite(state->tapeset, tapenum,
3123                                          tuple->t_data, tuple->t_len);
3124         if (state->randomAccess)        /* need trailing length word? */
3125                 LogicalTapeWrite(state->tapeset, tapenum,
3126                                                  &tuplen, sizeof(tuplen));
3127
3128         FREEMEM(state, GetMemoryChunkSpace(tuple));
3129         heap_freetuple(tuple);
3130 }
3131
3132 static void
3133 readtup_cluster(Tuplesortstate *state, SortTuple *stup,
3134                                 int tapenum, unsigned int tuplen)
3135 {
3136         unsigned int t_len = tuplen - sizeof(ItemPointerData) - sizeof(int);
3137         HeapTuple       tuple = (HeapTuple) palloc(t_len + HEAPTUPLESIZE);
3138
3139         USEMEM(state, GetMemoryChunkSpace(tuple));
3140         /* Reconstruct the HeapTupleData header */
3141         tuple->t_data = (HeapTupleHeader) ((char *) tuple + HEAPTUPLESIZE);
3142         tuple->t_len = t_len;
3143         LogicalTapeReadExact(state->tapeset, tapenum,
3144                                                  &tuple->t_self, sizeof(ItemPointerData));
3145         /* We don't currently bother to reconstruct t_tableOid */
3146         tuple->t_tableOid = InvalidOid;
3147         /* Read in the tuple body */
3148         LogicalTapeReadExact(state->tapeset, tapenum,
3149                                                  tuple->t_data, tuple->t_len);
3150         if (state->randomAccess)        /* need trailing length word? */
3151                 LogicalTapeReadExact(state->tapeset, tapenum,
3152                                                          &tuplen, sizeof(tuplen));
3153         stup->tuple = (void *) tuple;
3154         /* set up first-column key value, if it's a simple column */
3155         if (state->indexInfo->ii_KeyAttrNumbers[0] != 0)
3156                 stup->datum1 = heap_getattr(tuple,
3157                                                                         state->indexInfo->ii_KeyAttrNumbers[0],
3158                                                                         state->tupDesc,
3159                                                                         &stup->isnull1);
3160 }
3161
3162
3163 /*
3164  * Routines specialized for IndexTuple case
3165  *
3166  * The btree and hash cases require separate comparison functions, but the
3167  * IndexTuple representation is the same so the copy/write/read support
3168  * functions can be shared.
3169  */
3170
3171 static int
3172 comparetup_index_btree(const SortTuple *a, const SortTuple *b,
3173                                            Tuplesortstate *state)
3174 {
3175         /*
3176          * This is similar to _bt_tuplecompare(), but we have already done the
3177          * index_getattr calls for the first column, and we need to keep track of
3178          * whether any null fields are present.  Also see the special treatment
3179          * for equal keys at the end.
3180          */
3181         ScanKey         scanKey = state->indexScanKey;
3182         IndexTuple      tuple1;
3183         IndexTuple      tuple2;
3184         int                     keysz;
3185         TupleDesc       tupDes;
3186         bool            equal_hasnull = false;
3187         int                     nkey;
3188         int32           compare;
3189
3190         /* Compare the leading sort key */
3191         compare = inlineApplySortFunction(&scanKey->sk_func, scanKey->sk_flags,
3192                                                                           scanKey->sk_collation,
3193                                                                           a->datum1, a->isnull1,
3194                                                                           b->datum1, b->isnull1);
3195         if (compare != 0)
3196                 return compare;
3197
3198         /* they are equal, so we only need to examine one null flag */
3199         if (a->isnull1)
3200                 equal_hasnull = true;
3201
3202         /* Compare additional sort keys */
3203         tuple1 = (IndexTuple) a->tuple;
3204         tuple2 = (IndexTuple) b->tuple;
3205         keysz = state->nKeys;
3206         tupDes = RelationGetDescr(state->indexRel);
3207         scanKey++;
3208         for (nkey = 2; nkey <= keysz; nkey++, scanKey++)
3209         {
3210                 Datum           datum1,
3211                                         datum2;
3212                 bool            isnull1,
3213                                         isnull2;
3214
3215                 datum1 = index_getattr(tuple1, nkey, tupDes, &isnull1);
3216                 datum2 = index_getattr(tuple2, nkey, tupDes, &isnull2);
3217
3218                 compare = inlineApplySortFunction(&scanKey->sk_func, scanKey->sk_flags,
3219                                                                                   scanKey->sk_collation,
3220                                                                                   datum1, isnull1,
3221                                                                                   datum2, isnull2);
3222                 if (compare != 0)
3223                         return compare;         /* done when we find unequal attributes */
3224
3225                 /* they are equal, so we only need to examine one null flag */
3226                 if (isnull1)
3227                         equal_hasnull = true;
3228         }
3229
3230         /*
3231          * If btree has asked us to enforce uniqueness, complain if two equal
3232          * tuples are detected (unless there was at least one NULL field).
3233          *
3234          * It is sufficient to make the test here, because if two tuples are equal
3235          * they *must* get compared at some stage of the sort --- otherwise the
3236          * sort algorithm wouldn't have checked whether one must appear before the
3237          * other.
3238          */
3239         if (state->enforceUnique && !equal_hasnull)
3240         {
3241                 Datum           values[INDEX_MAX_KEYS];
3242                 bool            isnull[INDEX_MAX_KEYS];
3243
3244                 /*
3245                  * Some rather brain-dead implementations of qsort (such as the one in
3246                  * QNX 4) will sometimes call the comparison routine to compare a
3247                  * value to itself, but we always use our own implementation, which
3248                  * does not.
3249                  */
3250                 Assert(tuple1 != tuple2);
3251
3252                 index_deform_tuple(tuple1, tupDes, values, isnull);
3253                 ereport(ERROR,
3254                                 (errcode(ERRCODE_UNIQUE_VIOLATION),
3255                                  errmsg("could not create unique index \"%s\"",
3256                                                 RelationGetRelationName(state->indexRel)),
3257                                  errdetail("Key %s is duplicated.",
3258                                                    BuildIndexValueDescription(state->indexRel,
3259                                                                                                           values, isnull)),
3260                                  errtableconstraint(state->heapRel,
3261                                                                  RelationGetRelationName(state->indexRel))));
3262         }
3263
3264         /*
3265          * If key values are equal, we sort on ItemPointer.  This does not affect
3266          * validity of the finished index, but it may be useful to have index
3267          * scans in physical order.
3268          */
3269         {
3270                 BlockNumber blk1 = ItemPointerGetBlockNumber(&tuple1->t_tid);
3271                 BlockNumber blk2 = ItemPointerGetBlockNumber(&tuple2->t_tid);
3272
3273                 if (blk1 != blk2)
3274                         return (blk1 < blk2) ? -1 : 1;
3275         }
3276         {
3277                 OffsetNumber pos1 = ItemPointerGetOffsetNumber(&tuple1->t_tid);
3278                 OffsetNumber pos2 = ItemPointerGetOffsetNumber(&tuple2->t_tid);
3279
3280                 if (pos1 != pos2)
3281                         return (pos1 < pos2) ? -1 : 1;
3282         }
3283
3284         return 0;
3285 }
3286
3287 static int
3288 comparetup_index_hash(const SortTuple *a, const SortTuple *b,
3289                                           Tuplesortstate *state)
3290 {
3291         uint32          hash1;
3292         uint32          hash2;
3293         IndexTuple      tuple1;
3294         IndexTuple      tuple2;
3295
3296         /*
3297          * Fetch hash keys and mask off bits we don't want to sort by. We know
3298          * that the first column of the index tuple is the hash key.
3299          */
3300         Assert(!a->isnull1);
3301         hash1 = DatumGetUInt32(a->datum1) & state->hash_mask;
3302         Assert(!b->isnull1);
3303         hash2 = DatumGetUInt32(b->datum1) & state->hash_mask;
3304
3305         if (hash1 > hash2)
3306                 return 1;
3307         else if (hash1 < hash2)
3308                 return -1;
3309
3310         /*
3311          * If hash values are equal, we sort on ItemPointer.  This does not affect
3312          * validity of the finished index, but it may be useful to have index
3313          * scans in physical order.
3314          */
3315         tuple1 = (IndexTuple) a->tuple;
3316         tuple2 = (IndexTuple) b->tuple;
3317
3318         {
3319                 BlockNumber blk1 = ItemPointerGetBlockNumber(&tuple1->t_tid);
3320                 BlockNumber blk2 = ItemPointerGetBlockNumber(&tuple2->t_tid);
3321
3322                 if (blk1 != blk2)
3323                         return (blk1 < blk2) ? -1 : 1;
3324         }
3325         {
3326                 OffsetNumber pos1 = ItemPointerGetOffsetNumber(&tuple1->t_tid);
3327                 OffsetNumber pos2 = ItemPointerGetOffsetNumber(&tuple2->t_tid);
3328
3329                 if (pos1 != pos2)
3330                         return (pos1 < pos2) ? -1 : 1;
3331         }
3332
3333         return 0;
3334 }
3335
3336 static void
3337 copytup_index(Tuplesortstate *state, SortTuple *stup, void *tup)
3338 {
3339         IndexTuple      tuple = (IndexTuple) tup;
3340         unsigned int tuplen = IndexTupleSize(tuple);
3341         IndexTuple      newtuple;
3342
3343         /* copy the tuple into sort storage */
3344         newtuple = (IndexTuple) palloc(tuplen);
3345         memcpy(newtuple, tuple, tuplen);
3346         USEMEM(state, GetMemoryChunkSpace(newtuple));
3347         stup->tuple = (void *) newtuple;
3348         /* set up first-column key value */
3349         stup->datum1 = index_getattr(newtuple,
3350                                                                  1,
3351                                                                  RelationGetDescr(state->indexRel),
3352                                                                  &stup->isnull1);
3353 }
3354
3355 static void
3356 writetup_index(Tuplesortstate *state, int tapenum, SortTuple *stup)
3357 {
3358         IndexTuple      tuple = (IndexTuple) stup->tuple;
3359         unsigned int tuplen;
3360
3361         tuplen = IndexTupleSize(tuple) + sizeof(tuplen);
3362         LogicalTapeWrite(state->tapeset, tapenum,
3363                                          (void *) &tuplen, sizeof(tuplen));
3364         LogicalTapeWrite(state->tapeset, tapenum,
3365                                          (void *) tuple, IndexTupleSize(tuple));
3366         if (state->randomAccess)        /* need trailing length word? */
3367                 LogicalTapeWrite(state->tapeset, tapenum,
3368                                                  (void *) &tuplen, sizeof(tuplen));
3369
3370         FREEMEM(state, GetMemoryChunkSpace(tuple));
3371         pfree(tuple);
3372 }
3373
3374 static void
3375 readtup_index(Tuplesortstate *state, SortTuple *stup,
3376                           int tapenum, unsigned int len)
3377 {
3378         unsigned int tuplen = len - sizeof(unsigned int);
3379         IndexTuple      tuple = (IndexTuple) palloc(tuplen);
3380
3381         USEMEM(state, GetMemoryChunkSpace(tuple));
3382         LogicalTapeReadExact(state->tapeset, tapenum,
3383                                                  tuple, tuplen);
3384         if (state->randomAccess)        /* need trailing length word? */
3385                 LogicalTapeReadExact(state->tapeset, tapenum,
3386                                                          &tuplen, sizeof(tuplen));
3387         stup->tuple = (void *) tuple;
3388         /* set up first-column key value */
3389         stup->datum1 = index_getattr(tuple,
3390                                                                  1,
3391                                                                  RelationGetDescr(state->indexRel),
3392                                                                  &stup->isnull1);
3393 }
3394
3395 static void
3396 reversedirection_index_btree(Tuplesortstate *state)
3397 {
3398         ScanKey         scanKey = state->indexScanKey;
3399         int                     nkey;
3400
3401         for (nkey = 0; nkey < state->nKeys; nkey++, scanKey++)
3402         {
3403                 scanKey->sk_flags ^= (SK_BT_DESC | SK_BT_NULLS_FIRST);
3404         }
3405 }
3406
3407 static void
3408 reversedirection_index_hash(Tuplesortstate *state)
3409 {
3410         /* We don't support reversing direction in a hash index sort */
3411         elog(ERROR, "reversedirection_index_hash is not implemented");
3412 }
3413
3414
3415 /*
3416  * Routines specialized for DatumTuple case
3417  */
3418
3419 static int
3420 comparetup_datum(const SortTuple *a, const SortTuple *b, Tuplesortstate *state)
3421 {
3422         return ApplySortComparator(a->datum1, a->isnull1,
3423                                                            b->datum1, b->isnull1,
3424                                                            state->onlyKey);
3425 }
3426
3427 static void
3428 copytup_datum(Tuplesortstate *state, SortTuple *stup, void *tup)
3429 {
3430         /* Not currently needed */
3431         elog(ERROR, "copytup_datum() should not be called");
3432 }
3433
3434 static void
3435 writetup_datum(Tuplesortstate *state, int tapenum, SortTuple *stup)
3436 {
3437         void       *waddr;
3438         unsigned int tuplen;
3439         unsigned int writtenlen;
3440
3441         if (stup->isnull1)
3442         {
3443                 waddr = NULL;
3444                 tuplen = 0;
3445         }
3446         else if (state->datumTypeByVal)
3447         {
3448                 waddr = &stup->datum1;
3449                 tuplen = sizeof(Datum);
3450         }
3451         else
3452         {
3453                 waddr = DatumGetPointer(stup->datum1);
3454                 tuplen = datumGetSize(stup->datum1, false, state->datumTypeLen);
3455                 Assert(tuplen != 0);
3456         }
3457
3458         writtenlen = tuplen + sizeof(unsigned int);
3459
3460         LogicalTapeWrite(state->tapeset, tapenum,
3461                                          (void *) &writtenlen, sizeof(writtenlen));
3462         LogicalTapeWrite(state->tapeset, tapenum,
3463                                          waddr, tuplen);
3464         if (state->randomAccess)        /* need trailing length word? */
3465                 LogicalTapeWrite(state->tapeset, tapenum,
3466                                                  (void *) &writtenlen, sizeof(writtenlen));
3467
3468         if (stup->tuple)
3469         {
3470                 FREEMEM(state, GetMemoryChunkSpace(stup->tuple));
3471                 pfree(stup->tuple);
3472         }
3473 }
3474
3475 static void
3476 readtup_datum(Tuplesortstate *state, SortTuple *stup,
3477                           int tapenum, unsigned int len)
3478 {
3479         unsigned int tuplen = len - sizeof(unsigned int);
3480
3481         if (tuplen == 0)
3482         {
3483                 /* it's NULL */
3484                 stup->datum1 = (Datum) 0;
3485                 stup->isnull1 = true;
3486                 stup->tuple = NULL;
3487         }
3488         else if (state->datumTypeByVal)
3489         {
3490                 Assert(tuplen == sizeof(Datum));
3491                 LogicalTapeReadExact(state->tapeset, tapenum,
3492                                                          &stup->datum1, tuplen);
3493                 stup->isnull1 = false;
3494                 stup->tuple = NULL;
3495         }
3496         else
3497         {
3498                 void       *raddr = palloc(tuplen);
3499
3500                 LogicalTapeReadExact(state->tapeset, tapenum,
3501                                                          raddr, tuplen);
3502                 stup->datum1 = PointerGetDatum(raddr);
3503                 stup->isnull1 = false;
3504                 stup->tuple = raddr;
3505                 USEMEM(state, GetMemoryChunkSpace(raddr));
3506         }
3507
3508         if (state->randomAccess)        /* need trailing length word? */
3509                 LogicalTapeReadExact(state->tapeset, tapenum,
3510                                                          &tuplen, sizeof(tuplen));
3511 }
3512
3513 static void
3514 reversedirection_datum(Tuplesortstate *state)
3515 {
3516         state->onlyKey->ssup_reverse = !state->onlyKey->ssup_reverse;
3517         state->onlyKey->ssup_nulls_first = !state->onlyKey->ssup_nulls_first;
3518 }
3519
3520 /*
3521  * Convenience routine to free a tuple previously loaded into sort memory
3522  */
3523 static void
3524 free_sort_tuple(Tuplesortstate *state, SortTuple *stup)
3525 {
3526         FREEMEM(state, GetMemoryChunkSpace(stup->tuple));
3527         pfree(stup->tuple);
3528 }