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 given in kilobytes by the external variable SortMem.      Initially,
  35  * we absorb tuples and simply store them in an unsorted array as long as
  36  * we haven't exceeded SortMem.  If we reach the end of the input without
  37  * exceeding SortMem, we sort the array using qsort() and subsequently return
  38  * tuples just by scanning the tuple array sequentially.  If we do exceed
  39  * SortMem, 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 SortMem 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 at most six in the present code.
  52  * However, we can still make good use of our full SortMem 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 SortMem/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_gettuple; this
  74  * saves one cycle of writing all the data out to disk and reading it in.
  75  *
  76  *
  77  * Portions Copyright (c) 1996-2003, PostgreSQL Global Development Group
  78  * Portions Copyright (c) 1994, Regents of the University of California
  79  *
  80  * IDENTIFICATION
  81  *        $Header: /cvsroot/pgsql/src/backend/utils/sort/tuplesort.c,v 1.37 2003/08/17 19:58:06 tgl Exp $
  82  *
  83  *-------------------------------------------------------------------------
  84  */
  85
  86 #include "postgres.h"
  87
  88 #include "access/heapam.h"
  89 #include "access/nbtree.h"
  90 #include "catalog/pg_amop.h"
  91 #include "catalog/pg_operator.h"
  92 #include "miscadmin.h"
  93 #include "utils/catcache.h"
  94 #include "utils/datum.h"
  95 #include "utils/logtape.h"
  96 #include "utils/lsyscache.h"
  97 #include "utils/syscache.h"
  98 #include "utils/tuplesort.h"
  99
 100
 101 /*
 102  * Possible states of a Tuplesort object.  These denote the states that
 103  * persist between calls of Tuplesort routines.
 104  */
 105 typedef enum
 106 {
 107         TSS_INITIAL,                            /* Loading tuples; still within memory
 108                                                                  * limit */
 109         TSS_BUILDRUNS,                          /* Loading tuples; writing to tape */
 110         TSS_SORTEDINMEM,                        /* Sort completed entirely in memory */
 111         TSS_SORTEDONTAPE,                       /* Sort completed, final run is on tape */
 112         TSS_FINALMERGE                          /* Performing final merge on-the-fly */
 113 } TupSortStatus;
 114
 115 /*
 116  * We use a seven-tape polyphase merge, which is the "sweet spot" on the
 117  * tapes-to-passes curve according to Knuth's figure 70 (section 5.4.2).
 118  */
 119 #define MAXTAPES                7               /* Knuth's T */
 120 #define TAPERANGE               (MAXTAPES-1)    /* Knuth's P */
 121
 122 /*
 123  * Private state of a Tuplesort operation.
 124  */
 125 struct Tuplesortstate
 126 {
 127         TupSortStatus status;           /* enumerated value as shown above */
 128         bool            randomAccess;   /* did caller request random access? */
 129         long            availMem;               /* remaining memory available, in bytes */
 130         LogicalTapeSet *tapeset;        /* logtape.c object for tapes in a temp
 131                                                                  * file */
 132
 133         /*
 134          * These function pointers decouple the routines that must know what
 135          * kind of tuple we are sorting from the routines that don't need to
 136          * know it. They are set up by the tuplesort_begin_xxx routines.
 137          *
 138          * Function to compare two tuples; result is per qsort() convention, ie:
 139          *
 140          * <0, 0, >0 according as a<b, a=b, a>b.
 141          */
 142         int                     (*comparetup) (Tuplesortstate *state, const void *a, const void *b);
 143
 144         /*
 145          * Function to copy a supplied input tuple into palloc'd space. (NB:
 146          * we assume that a single pfree() is enough to release the tuple
 147          * later, so the representation must be "flat" in one palloc chunk.)
 148          * state->availMem must be decreased by the amount of space used.
 149          */
 150         void       *(*copytup) (Tuplesortstate *state, void *tup);
 151
 152         /*
 153          * Function to write a stored tuple onto tape.  The representation of
 154          * the tuple on tape need not be the same as it is in memory;
 155          * requirements on the tape representation are given below.  After
 156          * writing the tuple, pfree() it, and increase state->availMem by the
 157          * amount of memory space thereby released.
 158          */
 159         void            (*writetup) (Tuplesortstate *state, int tapenum, void *tup);
 160
 161         /*
 162          * Function to read a stored tuple from tape back into memory. 'len'
 163          * is the already-read length of the stored tuple.      Create and return
 164          * a palloc'd copy, and decrease state->availMem by the amount of
 165          * memory space consumed.
 166          */
 167         void       *(*readtup) (Tuplesortstate *state, int tapenum, unsigned int len);
 168
 169         /*
 170          * This array holds pointers to tuples in sort memory.  If we are in
 171          * state INITIAL, the tuples are in no particular order; if we are in
 172          * state SORTEDINMEM, the tuples are in final sorted order; in states
 173          * BUILDRUNS and FINALMERGE, the tuples are organized in "heap" order
 174          * per Algorithm H.  (Note that memtupcount only counts the tuples
 175          * that are part of the heap --- during merge passes, memtuples[]
 176          * entries beyond TAPERANGE are never in the heap and are used to hold
 177          * pre-read tuples.)  In state SORTEDONTAPE, the array is not used.
 178          */
 179         void      **memtuples;          /* array of pointers to palloc'd tuples */
 180         int                     memtupcount;    /* number of tuples currently present */
 181         int                     memtupsize;             /* allocated length of memtuples array */
 182
 183         /*
 184          * While building initial runs, this array holds the run number for
 185          * each tuple in memtuples[].  During merge passes, we re-use it to
 186          * hold the input tape number that each tuple in the heap was read
 187          * from, or to hold the index of the next tuple pre-read from the same
 188          * tape in the case of pre-read entries.  This array is never
 189          * allocated unless we need to use tapes.  Whenever it is allocated,
 190          * it has the same length as memtuples[].
 191          */
 192         int                *memtupindex;        /* index value associated with
 193                                                                  * memtuples[i] */
 194
 195         /*
 196          * While building initial runs, this is the current output run number
 197          * (starting at 0).  Afterwards, it is the number of initial runs we
 198          * made.
 199          */
 200         int                     currentRun;
 201
 202         /*
 203          * These variables are only used during merge passes.  mergeactive[i]
 204          * is true if we are reading an input run from (actual) tape number i
 205          * and have not yet exhausted that run.  mergenext[i] is the memtuples
 206          * index of the next pre-read tuple (next to be loaded into the heap)
 207          * for tape i, or 0 if we are out of pre-read tuples.  mergelast[i]
 208          * similarly points to the last pre-read tuple from each tape.
 209          * mergeavailmem[i] is the amount of unused space allocated for tape
 210          * i. mergefreelist and mergefirstfree keep track of unused locations
 211          * in the memtuples[] array.  memtupindex[] links together pre-read
 212          * tuples for each tape as well as recycled locations in
 213          * mergefreelist. It is OK to use 0 as a null link in these lists,
 214          * because memtuples[0] is part of the merge heap and is never a
 215          * pre-read tuple.
 216          */
 217         bool            mergeactive[MAXTAPES];  /* Active input run source? */
 218         int                     mergenext[MAXTAPES];    /* first preread tuple for each
 219                                                                                  * source */
 220         int                     mergelast[MAXTAPES];    /* last preread tuple for each
 221                                                                                  * source */
 222         long            mergeavailmem[MAXTAPES];                /* availMem for prereading
 223                                                                                                  * tapes */
 224         long            spacePerTape;   /* actual per-tape target usage */
 225         int                     mergefreelist;  /* head of freelist of recycled slots */
 226         int                     mergefirstfree; /* first slot never used in this merge */
 227
 228         /*
 229          * Variables for Algorithm D.  Note that destTape is a "logical" tape
 230          * number, ie, an index into the tp_xxx[] arrays.  Be careful to keep
 231          * "logical" and "actual" tape numbers straight!
 232          */
 233         int                     Level;                  /* Knuth's l */
 234         int                     destTape;               /* current output tape (Knuth's j, less 1) */
 235         int                     tp_fib[MAXTAPES];               /* Target Fibonacci run counts
 236                                                                                  * (A[]) */
 237         int                     tp_runs[MAXTAPES];              /* # of real runs on each tape */
 238         int                     tp_dummy[MAXTAPES];             /* # of dummy runs for each tape
 239                                                                                  * (D[]) */
 240         int                     tp_tapenum[MAXTAPES];   /* Actual tape numbers (TAPE[]) */
 241
 242         /*
 243          * These variables are used after completion of sorting to keep track
 244          * of the next tuple to return.  (In the tape case, the tape's current
 245          * read position is also critical state.)
 246          */
 247         int                     result_tape;    /* actual tape number of finished output */
 248         int                     current;                /* array index (only used if SORTEDINMEM) */
 249         bool            eof_reached;    /* reached EOF (needed for cursors) */
 250
 251         /* markpos_xxx holds marked position for mark and restore */
 252         long            markpos_block;  /* tape block# (only used if SORTEDONTAPE) */
 253         int                     markpos_offset; /* saved "current", or offset in tape
 254                                                                  * block */
 255         bool            markpos_eof;    /* saved "eof_reached" */
 256
 257         /*
 258          * These variables are specific to the HeapTuple case; they are set by
 259          * tuplesort_begin_heap and used only by the HeapTuple routines.
 260          */
 261         TupleDesc       tupDesc;
 262         int                     nKeys;
 263         ScanKey         scanKeys;
 264         SortFunctionKind *sortFnKinds;
 265
 266         /*
 267          * These variables are specific to the IndexTuple case; they are set
 268          * by tuplesort_begin_index and used only by the IndexTuple routines.
 269          */
 270         Relation        indexRel;
 271         ScanKey         indexScanKey;
 272         bool            enforceUnique;  /* complain if we find duplicate tuples */
 273
 274         /*
 275          * These variables are specific to the Datum case; they are set by
 276          * tuplesort_begin_datum and used only by the DatumTuple routines.
 277          */
 278         Oid                     datumType;
 279         Oid                     sortOperator;
 280         FmgrInfo        sortOpFn;               /* cached lookup data for sortOperator */
 281         SortFunctionKind sortFnKind;
 282         /* we need typelen and byval in order to know how to copy the Datums. */
 283         int                     datumTypeLen;
 284         bool            datumTypeByVal;
 285 };
 286
 287 #define COMPARETUP(state,a,b)   ((*(state)->comparetup) (state, a, b))
 288 #define COPYTUP(state,tup)      ((*(state)->copytup) (state, tup))
 289 #define WRITETUP(state,tape,tup)        ((*(state)->writetup) (state, tape, tup))
 290 #define READTUP(state,tape,len) ((*(state)->readtup) (state, tape, len))
 291 #define LACKMEM(state)          ((state)->availMem < 0)
 292 #define USEMEM(state,amt)       ((state)->availMem -= (amt))
 293 #define FREEMEM(state,amt)      ((state)->availMem += (amt))
 294
 295 /*--------------------
 296  *
 297  * NOTES about on-tape representation of tuples:
 298  *
 299  * We require the first "unsigned int" of a stored tuple to be the total size
 300  * on-tape of the tuple, including itself (so it is never zero; an all-zero
 301  * unsigned int is used to delimit runs).  The remainder of the stored tuple
 302  * may or may not match the in-memory representation of the tuple ---
 303  * any conversion needed is the job of the writetup and readtup routines.
 304  *
 305  * If state->randomAccess is true, then the stored representation of the
 306  * tuple must be followed by another "unsigned int" that is a copy of the
 307  * length --- so the total tape space used is actually sizeof(unsigned int)
 308  * more than the stored length value.  This allows read-backwards.      When
 309  * randomAccess is not true, the write/read routines may omit the extra
 310  * length word.
 311  *
 312  * writetup is expected to write both length words as well as the tuple
 313  * data.  When readtup is called, the tape is positioned just after the
 314  * front length word; readtup must read the tuple data and advance past
 315  * the back length word (if present).
 316  *
 317  * The write/read routines can make use of the tuple description data
 318  * stored in the Tuplesortstate record, if needed.      They are also expected
 319  * to adjust state->availMem by the amount of memory space (not tape space!)
 320  * released or consumed.  There is no error return from either writetup
 321  * or readtup; they should ereport() on failure.
 322  *
 323  *
 324  * NOTES about memory consumption calculations:
 325  *
 326  * We count space allocated for tuples against the SortMem limit, plus
 327  * the space used by the variable-size arrays memtuples and memtupindex.
 328  * Fixed-size space (primarily the LogicalTapeSet I/O buffers) is not
 329  * counted.
 330  *
 331  * Note that we count actual space used (as shown by GetMemoryChunkSpace)
 332  * rather than the originally-requested size.  This is important since
 333  * palloc can add substantial overhead.  It's not a complete answer since
 334  * we won't count any wasted space in palloc allocation blocks, but it's
 335  * a lot better than what we were doing before 7.3.
 336  *
 337  *--------------------
 338  */
 339
 340 /*
 341  * For sorting single Datums, we build "pseudo tuples" that just carry
 342  * the datum's value and null flag.  For pass-by-reference data types,
 343  * the actual data value appears after the DatumTupleHeader (MAXALIGNed,
 344  * of course), and the value field in the header is just a pointer to it.
 345  */
 346
 347 typedef struct
 348 {
 349         Datum           val;
 350         bool            isNull;
 351 } DatumTuple;
 352
 353
 354 static Tuplesortstate *tuplesort_begin_common(bool randomAccess);
 355 static void puttuple_common(Tuplesortstate *state, void *tuple);
 356 static void inittapes(Tuplesortstate *state);
 357 static void selectnewtape(Tuplesortstate *state);
 358 static void mergeruns(Tuplesortstate *state);
 359 static void mergeonerun(Tuplesortstate *state);
 360 static void beginmerge(Tuplesortstate *state);
 361 static void mergepreread(Tuplesortstate *state);
 362 static void dumptuples(Tuplesortstate *state, bool alltuples);
 363 static void tuplesort_heap_insert(Tuplesortstate *state, void *tuple,
 364                                           int tupleindex, bool checkIndex);
 365 static void tuplesort_heap_siftup(Tuplesortstate *state, bool checkIndex);
 366 static unsigned int getlen(Tuplesortstate *state, int tapenum, bool eofOK);
 367 static void markrunend(Tuplesortstate *state, int tapenum);
 368 static int      qsort_comparetup(const void *a, const void *b);
 369 static int comparetup_heap(Tuplesortstate *state,
 370                                 const void *a, const void *b);
 371 static void *copytup_heap(Tuplesortstate *state, void *tup);
 372 static void writetup_heap(Tuplesortstate *state, int tapenum, void *tup);
 373 static void *readtup_heap(Tuplesortstate *state, int tapenum,
 374                          unsigned int len);
 375 static int comparetup_index(Tuplesortstate *state,
 376                                  const void *a, const void *b);
 377 static void *copytup_index(Tuplesortstate *state, void *tup);
 378 static void writetup_index(Tuplesortstate *state, int tapenum, void *tup);
 379 static void *readtup_index(Tuplesortstate *state, int tapenum,
 380                           unsigned int len);
 381 static int comparetup_datum(Tuplesortstate *state,
 382                                  const void *a, const void *b);
 383 static void *copytup_datum(Tuplesortstate *state, void *tup);
 384 static void writetup_datum(Tuplesortstate *state, int tapenum, void *tup);
 385 static void *readtup_datum(Tuplesortstate *state, int tapenum,
 386                           unsigned int len);
 387
 388 /*
 389  * Since qsort(3) will not pass any context info to qsort_comparetup(),
 390  * we have to use this ugly static variable.  It is set to point to the
 391  * active Tuplesortstate object just before calling qsort.      It should
 392  * not be used directly by anything except qsort_comparetup().
 393  */
 394 static Tuplesortstate *qsort_tuplesortstate;
 395
 396
 397 /*
 398  *              tuplesort_begin_xxx
 399  *
 400  * Initialize for a tuple sort operation.
 401  *
 402  * After calling tuplesort_begin, the caller should call tuplesort_puttuple
 403  * zero or more times, then call tuplesort_performsort when all the tuples
 404  * have been supplied.  After performsort, retrieve the tuples in sorted
 405  * order by calling tuplesort_gettuple until it returns NULL.  (If random
 406  * access was requested, rescan, markpos, and restorepos can also be called.)
 407  * For Datum sorts, putdatum/getdatum are used instead of puttuple/gettuple.
 408  * Call tuplesort_end to terminate the operation and release memory/disk space.
 409  */
 410
 411 static Tuplesortstate *
 412 tuplesort_begin_common(bool randomAccess)
 413 {
 414         Tuplesortstate *state;
 415
 416         state = (Tuplesortstate *) palloc0(sizeof(Tuplesortstate));
 417
 418         state->status = TSS_INITIAL;
 419         state->randomAccess = randomAccess;
 420         state->availMem = SortMem * 1024L;
 421         state->tapeset = NULL;
 422
 423         state->memtupcount = 0;
 424         state->memtupsize = 1024;       /* initial guess */
 425         state->memtuples = (void **) palloc(state->memtupsize * sizeof(void *));
 426
 427         state->memtupindex = NULL;      /* until and unless needed */
 428
 429         USEMEM(state, GetMemoryChunkSpace(state->memtuples));
 430
 431         state->currentRun = 0;
 432
 433         /* Algorithm D variables will be initialized by inittapes, if needed */
 434
 435         state->result_tape = -1;        /* flag that result tape has not been
 436                                                                  * formed */
 437
 438         return state;
 439 }
 440
 441 Tuplesortstate *
 442 tuplesort_begin_heap(TupleDesc tupDesc,
 443                                          int nkeys,
 444                                          Oid *sortOperators, AttrNumber *attNums,
 445                                          bool randomAccess)
 446 {
 447         Tuplesortstate *state = tuplesort_begin_common(randomAccess);
 448         int                     i;
 449
 450         AssertArg(nkeys > 0);
 451
 452         state->comparetup = comparetup_heap;
 453         state->copytup = copytup_heap;
 454         state->writetup = writetup_heap;
 455         state->readtup = readtup_heap;
 456
 457         state->tupDesc = tupDesc;
 458         state->nKeys = nkeys;
 459         state->scanKeys = (ScanKey) palloc0(nkeys * sizeof(ScanKeyData));
 460         state->sortFnKinds = (SortFunctionKind *)
 461                 palloc0(nkeys * sizeof(SortFunctionKind));
 462
 463         for (i = 0; i < nkeys; i++)
 464         {
 465                 RegProcedure sortFunction;
 466
 467                 AssertArg(sortOperators[i] != 0);
 468                 AssertArg(attNums[i] != 0);
 469
 470                 /* select a function that implements the sort operator */
 471                 SelectSortFunction(sortOperators[i], &sortFunction,
 472                                                    &state->sortFnKinds[i]);
 473
 474                 ScanKeyEntryInitialize(&state->scanKeys[i],
 475                                                            0x0,
 476                                                            attNums[i],
 477                                                            sortFunction,
 478                                                            (Datum) 0);
 479         }
 480
 481         return state;
 482 }
 483
 484 Tuplesortstate *
 485 tuplesort_begin_index(Relation indexRel,
 486                                           bool enforceUnique,
 487                                           bool randomAccess)
 488 {
 489         Tuplesortstate *state = tuplesort_begin_common(randomAccess);
 490
 491         state->comparetup = comparetup_index;
 492         state->copytup = copytup_index;
 493         state->writetup = writetup_index;
 494         state->readtup = readtup_index;
 495
 496         state->indexRel = indexRel;
 497         /* see comments below about btree dependence of this code... */
 498         state->indexScanKey = _bt_mkscankey_nodata(indexRel);
 499         state->enforceUnique = enforceUnique;
 500
 501         return state;
 502 }
 503
 504 Tuplesortstate *
 505 tuplesort_begin_datum(Oid datumType,
 506                                           Oid sortOperator,
 507                                           bool randomAccess)
 508 {
 509         Tuplesortstate *state = tuplesort_begin_common(randomAccess);
 510         RegProcedure sortFunction;
 511         int16           typlen;
 512         bool            typbyval;
 513
 514         state->comparetup = comparetup_datum;
 515         state->copytup = copytup_datum;
 516         state->writetup = writetup_datum;
 517         state->readtup = readtup_datum;
 518
 519         state->datumType = datumType;
 520         state->sortOperator = sortOperator;
 521
 522         /* select a function that implements the sort operator */
 523         SelectSortFunction(sortOperator, &sortFunction, &state->sortFnKind);
 524         /* and look up the function */
 525         fmgr_info(sortFunction, &state->sortOpFn);
 526
 527         /* lookup necessary attributes of the datum type */
 528         get_typlenbyval(datumType, &typlen, &typbyval);
 529         state->datumTypeLen = typlen;
 530         state->datumTypeByVal = typbyval;
 531
 532         return state;
 533 }
 534
 535 /*
 536  * tuplesort_end
 537  *
 538  *      Release resources and clean up.
 539  */
 540 void
 541 tuplesort_end(Tuplesortstate *state)
 542 {
 543         int                     i;
 544
 545         if (state->tapeset)
 546                 LogicalTapeSetClose(state->tapeset);
 547         if (state->memtuples)
 548         {
 549                 for (i = 0; i < state->memtupcount; i++)
 550                         pfree(state->memtuples[i]);
 551                 pfree(state->memtuples);
 552         }
 553         if (state->memtupindex)
 554                 pfree(state->memtupindex);
 555
 556         /*
 557          * this stuff might better belong in a variant-specific shutdown
 558          * routine
 559          */
 560         if (state->scanKeys)
 561                 pfree(state->scanKeys);
 562         if (state->sortFnKinds)
 563                 pfree(state->sortFnKinds);
 564
 565         pfree(state);
 566 }
 567
 568 /*
 569  * Accept one tuple while collecting input data for sort.
 570  *
 571  * Note that the input tuple is always copied; the caller need not save it.
 572  */
 573 void
 574 tuplesort_puttuple(Tuplesortstate *state, void *tuple)
 575 {
 576         /*
 577          * Copy the given tuple into memory we control, and decrease availMem.
 578          * Then call the code shared with the Datum case.
 579          */
 580         tuple = COPYTUP(state, tuple);
 581
 582         puttuple_common(state, tuple);
 583 }
 584
 585 /*
 586  * Accept one Datum while collecting input data for sort.
 587  *
 588  * If the Datum is pass-by-ref type, the value will be copied.
 589  */
 590 void
 591 tuplesort_putdatum(Tuplesortstate *state, Datum val, bool isNull)
 592 {
 593         DatumTuple *tuple;
 594
 595         /*
 596          * Build pseudo-tuple carrying the datum, and decrease availMem.
 597          */
 598         if (isNull || state->datumTypeByVal)
 599         {
 600                 tuple = (DatumTuple *) palloc(sizeof(DatumTuple));
 601                 tuple->val = val;
 602                 tuple->isNull = isNull;
 603         }
 604         else
 605         {
 606                 Size            datalen;
 607                 Size            tuplelen;
 608                 char       *newVal;
 609
 610                 datalen = datumGetSize(val, false, state->datumTypeLen);
 611                 tuplelen = datalen + MAXALIGN(sizeof(DatumTuple));
 612                 tuple = (DatumTuple *) palloc(tuplelen);
 613                 newVal = ((char *) tuple) + MAXALIGN(sizeof(DatumTuple));
 614                 memcpy(newVal, DatumGetPointer(val), datalen);
 615                 tuple->val = PointerGetDatum(newVal);
 616                 tuple->isNull = false;
 617         }
 618
 619         USEMEM(state, GetMemoryChunkSpace(tuple));
 620
 621         puttuple_common(state, (void *) tuple);
 622 }
 623
 624 /*
 625  * Shared code for tuple and datum cases.
 626  */
 627 static void
 628 puttuple_common(Tuplesortstate *state, void *tuple)
 629 {
 630         switch (state->status)
 631         {
 632                 case TSS_INITIAL:
 633
 634                         /*
 635                          * Save the copied tuple into the unsorted array.
 636                          */
 637                         if (state->memtupcount >= state->memtupsize)
 638                         {
 639                                 /* Grow the unsorted array as needed. */
 640                                 FREEMEM(state, GetMemoryChunkSpace(state->memtuples));
 641                                 state->memtupsize *= 2;
 642                                 state->memtuples = (void **)
 643                                         repalloc(state->memtuples,
 644                                                          state->memtupsize * sizeof(void *));
 645                                 USEMEM(state, GetMemoryChunkSpace(state->memtuples));
 646                         }
 647                         state->memtuples[state->memtupcount++] = tuple;
 648
 649                         /*
 650                          * Done if we still fit in available memory.
 651                          */
 652                         if (!LACKMEM(state))
 653                                 return;
 654
 655                         /*
 656                          * Nope; time to switch to tape-based operation.
 657                          */
 658                         inittapes(state);
 659
 660                         /*
 661                          * Dump tuples until we are back under the limit.
 662                          */
 663                         dumptuples(state, false);
 664                         break;
 665                 case TSS_BUILDRUNS:
 666
 667                         /*
 668                          * Insert the copied tuple into the heap, with run number
 669                          * currentRun if it can go into the current run, else run
 670                          * number currentRun+1.  The tuple can go into the current run
 671                          * if it is >= the first not-yet-output tuple.  (Actually, it
 672                          * could go into the current run if it is >= the most recently
 673                          * output tuple ... but that would require keeping around the
 674                          * tuple we last output, and it's simplest to let writetup
 675                          * free each tuple as soon as it's written.)
 676                          *
 677                          * Note there will always be at least one tuple in the heap at
 678                          * this point; see dumptuples.
 679                          */
 680                         Assert(state->memtupcount > 0);
 681                         if (COMPARETUP(state, tuple, state->memtuples[0]) >= 0)
 682                                 tuplesort_heap_insert(state, tuple, state->currentRun, true);
 683                         else
 684                                 tuplesort_heap_insert(state, tuple, state->currentRun + 1, true);
 685
 686                         /*
 687                          * If we are over the memory limit, dump tuples till we're
 688                          * under.
 689                          */
 690                         dumptuples(state, false);
 691                         break;
 692                 default:
 693                         elog(ERROR, "invalid tuplesort state");
 694                         break;
 695         }
 696 }
 697
 698 /*
 699  * All tuples have been provided; finish the sort.
 700  */
 701 void
 702 tuplesort_performsort(Tuplesortstate *state)
 703 {
 704         switch (state->status)
 705         {
 706                 case TSS_INITIAL:
 707
 708                         /*
 709                          * We were able to accumulate all the tuples within the
 710                          * allowed amount of memory.  Just qsort 'em and we're done.
 711                          */
 712                         if (state->memtupcount > 1)
 713                         {
 714                                 qsort_tuplesortstate = state;
 715                                 qsort((void *) state->memtuples, state->memtupcount,
 716                                           sizeof(void *), qsort_comparetup);
 717                         }
 718                         state->current = 0;
 719                         state->eof_reached = false;
 720                         state->markpos_offset = 0;
 721                         state->markpos_eof = false;
 722                         state->status = TSS_SORTEDINMEM;
 723                         break;
 724                 case TSS_BUILDRUNS:
 725
 726                         /*
 727                          * Finish tape-based sort.      First, flush all tuples remaining
 728                          * in memory out to tape; then merge until we have a single
 729                          * remaining run (or, if !randomAccess, one run per tape).
 730                          * Note that mergeruns sets the correct state->status.
 731                          */
 732                         dumptuples(state, true);
 733                         mergeruns(state);
 734                         state->eof_reached = false;
 735                         state->markpos_block = 0L;
 736                         state->markpos_offset = 0;
 737                         state->markpos_eof = false;
 738                         break;
 739                 default:
 740                         elog(ERROR, "invalid tuplesort state");
 741                         break;
 742         }
 743 }
 744
 745 /*
 746  * Fetch the next tuple in either forward or back direction.
 747  * Returns NULL if no more tuples.      If should_free is set, the
 748  * caller must pfree the returned tuple when done with it.
 749  */
 750 void *
 751 tuplesort_gettuple(Tuplesortstate *state, bool forward,
 752                                    bool *should_free)
 753 {
 754         unsigned int tuplen;
 755         void       *tup;
 756
 757         switch (state->status)
 758         {
 759                 case TSS_SORTEDINMEM:
 760                         Assert(forward || state->randomAccess);
 761                         *should_free = false;
 762                         if (forward)
 763                         {
 764                                 if (state->current < state->memtupcount)
 765                                         return state->memtuples[state->current++];
 766                                 state->eof_reached = true;
 767                                 return NULL;
 768                         }
 769                         else
 770                         {
 771                                 if (state->current <= 0)
 772                                         return NULL;
 773
 774                                 /*
 775                                  * if all tuples are fetched already then we return last
 776                                  * tuple, else - tuple before last returned.
 777                                  */
 778                                 if (state->eof_reached)
 779                                         state->eof_reached = false;
 780                                 else
 781                                 {
 782                                         state->current--;       /* last returned tuple */
 783                                         if (state->current <= 0)
 784                                                 return NULL;
 785                                 }
 786                                 return state->memtuples[state->current - 1];
 787                         }
 788                         break;
 789
 790                 case TSS_SORTEDONTAPE:
 791                         Assert(forward || state->randomAccess);
 792                         *should_free = true;
 793                         if (forward)
 794                         {
 795                                 if (state->eof_reached)
 796                                         return NULL;
 797                                 if ((tuplen = getlen(state, state->result_tape, true)) != 0)
 798                                 {
 799                                         tup = READTUP(state, state->result_tape, tuplen);
 800                                         return tup;
 801                                 }
 802                                 else
 803                                 {
 804                                         state->eof_reached = true;
 805                                         return NULL;
 806                                 }
 807                         }
 808
 809                         /*
 810                          * Backward.
 811                          *
 812                          * if all tuples are fetched already then we return last tuple,
 813                          * else - tuple before last returned.
 814                          */
 815                         if (state->eof_reached)
 816                         {
 817                                 /*
 818                                  * Seek position is pointing just past the zero tuplen at
 819                                  * the end of file; back up to fetch last tuple's ending
 820                                  * length word.  If seek fails we must have a completely
 821                                  * empty file.
 822                                  */
 823                                 if (!LogicalTapeBackspace(state->tapeset,
 824                                                                                   state->result_tape,
 825                                                                                   2 * sizeof(unsigned int)))
 826                                         return NULL;
 827                                 state->eof_reached = false;
 828                         }
 829                         else
 830                         {
 831                                 /*
 832                                  * Back up and fetch previously-returned tuple's ending
 833                                  * length word.  If seek fails, assume we are at start of
 834                                  * file.
 835                                  */
 836                                 if (!LogicalTapeBackspace(state->tapeset,
 837                                                                                   state->result_tape,
 838                                                                                   sizeof(unsigned int)))
 839                                         return NULL;
 840                                 tuplen = getlen(state, state->result_tape, false);
 841
 842                                 /*
 843                                  * Back up to get ending length word of tuple before it.
 844                                  */
 845                                 if (!LogicalTapeBackspace(state->tapeset,
 846                                                                                   state->result_tape,
 847                                                                           tuplen + 2 * sizeof(unsigned int)))
 848                                 {
 849                                         /*
 850                                          * If that fails, presumably the prev tuple is the
 851                                          * first in the file.  Back up so that it becomes next
 852                                          * to read in forward direction (not obviously right,
 853                                          * but that is what in-memory case does).
 854                                          */
 855                                         if (!LogicalTapeBackspace(state->tapeset,
 856                                                                                           state->result_tape,
 857                                                                                   tuplen + sizeof(unsigned int)))
 858                                                 elog(ERROR, "bogus tuple length in backward scan");
 859                                         return NULL;
 860                                 }
 861                         }
 862
 863                         tuplen = getlen(state, state->result_tape, false);
 864
 865                         /*
 866                          * Now we have the length of the prior tuple, back up and read
 867                          * it. Note: READTUP expects we are positioned after the
 868                          * initial length word of the tuple, so back up to that point.
 869                          */
 870                         if (!LogicalTapeBackspace(state->tapeset,
 871                                                                           state->result_tape,
 872                                                                           tuplen))
 873                                 elog(ERROR, "bogus tuple length in backward scan");
 874                         tup = READTUP(state, state->result_tape, tuplen);
 875                         return tup;
 876
 877                 case TSS_FINALMERGE:
 878                         Assert(forward);
 879                         *should_free = true;
 880
 881                         /*
 882                          * This code should match the inner loop of mergeonerun().
 883                          */
 884                         if (state->memtupcount > 0)
 885                         {
 886                                 int                     srcTape = state->memtupindex[0];
 887                                 Size            tuplen;
 888                                 int                     tupIndex;
 889                                 void       *newtup;
 890
 891                                 tup = state->memtuples[0];
 892                                 /* returned tuple is no longer counted in our memory space */
 893                                 tuplen = GetMemoryChunkSpace(tup);
 894                                 state->availMem += tuplen;
 895                                 state->mergeavailmem[srcTape] += tuplen;
 896                                 tuplesort_heap_siftup(state, false);
 897                                 if ((tupIndex = state->mergenext[srcTape]) == 0)
 898                                 {
 899                                         /*
 900                                          * out of preloaded data on this tape, try to read
 901                                          * more
 902                                          */
 903                                         mergepreread(state);
 904
 905                                         /*
 906                                          * if still no data, we've reached end of run on this
 907                                          * tape
 908                                          */
 909                                         if ((tupIndex = state->mergenext[srcTape]) == 0)
 910                                                 return tup;
 911                                 }
 912                                 /* pull next preread tuple from list, insert in heap */
 913                                 newtup = state->memtuples[tupIndex];
 914                                 state->mergenext[srcTape] = state->memtupindex[tupIndex];
 915                                 if (state->mergenext[srcTape] == 0)
 916                                         state->mergelast[srcTape] = 0;
 917                                 state->memtupindex[tupIndex] = state->mergefreelist;
 918                                 state->mergefreelist = tupIndex;
 919                                 tuplesort_heap_insert(state, newtup, srcTape, false);
 920                                 return tup;
 921                         }
 922                         return NULL;
 923
 924                 default:
 925                         elog(ERROR, "invalid tuplesort state");
 926                         return NULL;            /* keep compiler quiet */
 927         }
 928 }
 929
 930 /*
 931  * Fetch the next Datum in either forward or back direction.
 932  * Returns FALSE if no more datums.
 933  *
 934  * If the Datum is pass-by-ref type, the returned value is freshly palloc'd
 935  * and is now owned by the caller.
 936  */
 937 bool
 938 tuplesort_getdatum(Tuplesortstate *state, bool forward,
 939                                    Datum *val, bool *isNull)
 940 {
 941         DatumTuple *tuple;
 942         bool            should_free;
 943
 944         tuple = (DatumTuple *) tuplesort_gettuple(state, forward, &should_free);
 945
 946         if (tuple == NULL)
 947                 return false;
 948
 949         if (tuple->isNull || state->datumTypeByVal)
 950         {
 951                 *val = tuple->val;
 952                 *isNull = tuple->isNull;
 953         }
 954         else
 955         {
 956                 *val = datumCopy(tuple->val, false, state->datumTypeLen);
 957                 *isNull = false;
 958         }
 959
 960         if (should_free)
 961                 pfree(tuple);
 962
 963         return true;
 964 }
 965
 966
 967 /*
 968  * inittapes - initialize for tape sorting.
 969  *
 970  * This is called only if we have found we don't have room to sort in memory.
 971  */
 972 static void
 973 inittapes(Tuplesortstate *state)
 974 {
 975         int                     ntuples,
 976                                 j;
 977
 978         state->tapeset = LogicalTapeSetCreate(MAXTAPES);
 979
 980         /*
 981          * Allocate the memtupindex array, same size as memtuples.
 982          */
 983         state->memtupindex = (int *) palloc(state->memtupsize * sizeof(int));
 984
 985         USEMEM(state, GetMemoryChunkSpace(state->memtupindex));
 986
 987         /*
 988          * Convert the unsorted contents of memtuples[] into a heap. Each
 989          * tuple is marked as belonging to run number zero.
 990          *
 991          * NOTE: we pass false for checkIndex since there's no point in comparing
 992          * indexes in this step, even though we do intend the indexes to be
 993          * part of the sort key...
 994          */
 995         ntuples = state->memtupcount;
 996         state->memtupcount = 0;         /* make the heap empty */
 997         for (j = 0; j < ntuples; j++)
 998                 tuplesort_heap_insert(state, state->memtuples[j], 0, false);
 999         Assert(state->memtupcount == ntuples);
1000
1001         state->currentRun = 0;
1002
1003         /*
1004          * Initialize variables of Algorithm D (step D1).
1005          */
1006         for (j = 0; j < MAXTAPES; j++)
1007         {
1008                 state->tp_fib[j] = 1;
1009                 state->tp_runs[j] = 0;
1010                 state->tp_dummy[j] = 1;
1011                 state->tp_tapenum[j] = j;
1012         }
1013         state->tp_fib[TAPERANGE] = 0;
1014         state->tp_dummy[TAPERANGE] = 0;
1015
1016         state->Level = 1;
1017         state->destTape = 0;
1018
1019         state->status = TSS_BUILDRUNS;
1020 }
1021
1022 /*
1023  * selectnewtape -- select new tape for new initial run.
1024  *
1025  * This is called after finishing a run when we know another run
1026  * must be started.  This implements steps D3, D4 of Algorithm D.
1027  */
1028 static void
1029 selectnewtape(Tuplesortstate *state)
1030 {
1031         int                     j;
1032         int                     a;
1033
1034         /* Step D3: advance j (destTape) */
1035         if (state->tp_dummy[state->destTape] < state->tp_dummy[state->destTape + 1])
1036         {
1037                 state->destTape++;
1038                 return;
1039         }
1040         if (state->tp_dummy[state->destTape] != 0)
1041         {
1042                 state->destTape = 0;
1043                 return;
1044         }
1045
1046         /* Step D4: increase level */
1047         state->Level++;
1048         a = state->tp_fib[0];
1049         for (j = 0; j < TAPERANGE; j++)
1050         {
1051                 state->tp_dummy[j] = a + state->tp_fib[j + 1] - state->tp_fib[j];
1052                 state->tp_fib[j] = a + state->tp_fib[j + 1];
1053         }
1054         state->destTape = 0;
1055 }
1056
1057 /*
1058  * mergeruns -- merge all the completed initial runs.
1059  *
1060  * This implements steps D5, D6 of Algorithm D.  All input data has
1061  * already been written to initial runs on tape (see dumptuples).
1062  */
1063 static void
1064 mergeruns(Tuplesortstate *state)
1065 {
1066         int                     tapenum,
1067                                 svTape,
1068                                 svRuns,
1069                                 svDummy;
1070
1071         Assert(state->status == TSS_BUILDRUNS);
1072         Assert(state->memtupcount == 0);
1073
1074         /*
1075          * If we produced only one initial run (quite likely if the total data
1076          * volume is between 1X and 2X SortMem), we can just use that tape as
1077          * the finished output, rather than doing a useless merge.
1078          */
1079         if (state->currentRun == 1)
1080         {
1081                 state->result_tape = state->tp_tapenum[state->destTape];
1082                 /* must freeze and rewind the finished output tape */
1083                 LogicalTapeFreeze(state->tapeset, state->result_tape);
1084                 state->status = TSS_SORTEDONTAPE;
1085                 return;
1086         }
1087
1088         /* End of step D2: rewind all output tapes to prepare for merging */
1089         for (tapenum = 0; tapenum < TAPERANGE; tapenum++)
1090                 LogicalTapeRewind(state->tapeset, tapenum, false);
1091
1092         for (;;)
1093         {
1094                 /* Step D5: merge runs onto tape[T] until tape[P] is empty */
1095                 while (state->tp_runs[TAPERANGE - 1] || state->tp_dummy[TAPERANGE - 1])
1096                 {
1097                         bool            allDummy = true;
1098                         bool            allOneRun = true;
1099
1100                         for (tapenum = 0; tapenum < TAPERANGE; tapenum++)
1101                         {
1102                                 if (state->tp_dummy[tapenum] == 0)
1103                                         allDummy = false;
1104                                 if (state->tp_runs[tapenum] + state->tp_dummy[tapenum] != 1)
1105                                         allOneRun = false;
1106                         }
1107
1108                         /*
1109                          * If we don't have to produce a materialized sorted tape,
1110                          * quit as soon as we're down to one real/dummy run per tape.
1111                          */
1112                         if (!state->randomAccess && allOneRun)
1113                         {
1114                                 Assert(!allDummy);
1115                                 /* Initialize for the final merge pass */
1116                                 beginmerge(state);
1117                                 state->status = TSS_FINALMERGE;
1118                                 return;
1119                         }
1120                         if (allDummy)
1121                         {
1122                                 state->tp_dummy[TAPERANGE]++;
1123                                 for (tapenum = 0; tapenum < TAPERANGE; tapenum++)
1124                                         state->tp_dummy[tapenum]--;
1125                         }
1126                         else
1127                                 mergeonerun(state);
1128                 }
1129                 /* Step D6: decrease level */
1130                 if (--state->Level == 0)
1131                         break;
1132                 /* rewind output tape T to use as new input */
1133                 LogicalTapeRewind(state->tapeset, state->tp_tapenum[TAPERANGE],
1134                                                   false);
1135                 /* rewind used-up input tape P, and prepare it for write pass */
1136                 LogicalTapeRewind(state->tapeset, state->tp_tapenum[TAPERANGE - 1],
1137                                                   true);
1138                 state->tp_runs[TAPERANGE - 1] = 0;
1139
1140                 /*
1141                  * reassign tape units per step D6; note we no longer care about
1142                  * A[]
1143                  */
1144                 svTape = state->tp_tapenum[TAPERANGE];
1145                 svDummy = state->tp_dummy[TAPERANGE];
1146                 svRuns = state->tp_runs[TAPERANGE];
1147                 for (tapenum = TAPERANGE; tapenum > 0; tapenum--)
1148                 {
1149                         state->tp_tapenum[tapenum] = state->tp_tapenum[tapenum - 1];
1150                         state->tp_dummy[tapenum] = state->tp_dummy[tapenum - 1];
1151                         state->tp_runs[tapenum] = state->tp_runs[tapenum - 1];
1152                 }
1153                 state->tp_tapenum[0] = svTape;
1154                 state->tp_dummy[0] = svDummy;
1155                 state->tp_runs[0] = svRuns;
1156         }
1157
1158         /*
1159          * Done.  Knuth says that the result is on TAPE[1], but since we
1160          * exited the loop without performing the last iteration of step D6,
1161          * we have not rearranged the tape unit assignment, and therefore the
1162          * result is on TAPE[T].  We need to do it this way so that we can
1163          * freeze the final output tape while rewinding it.  The last
1164          * iteration of step D6 would be a waste of cycles anyway...
1165          */
1166         state->result_tape = state->tp_tapenum[TAPERANGE];
1167         LogicalTapeFreeze(state->tapeset, state->result_tape);
1168         state->status = TSS_SORTEDONTAPE;
1169 }
1170
1171 /*
1172  * Merge one run from each input tape, except ones with dummy runs.
1173  *
1174  * This is the inner loop of Algorithm D step D5.  We know that the
1175  * output tape is TAPE[T].
1176  */
1177 static void
1178 mergeonerun(Tuplesortstate *state)
1179 {
1180         int                     destTape = state->tp_tapenum[TAPERANGE];
1181         int                     srcTape;
1182         int                     tupIndex;
1183         void       *tup;
1184         long            priorAvail,
1185                                 spaceFreed;
1186
1187         /*
1188          * Start the merge by loading one tuple from each active source tape
1189          * into the heap.  We can also decrease the input run/dummy run
1190          * counts.
1191          */
1192         beginmerge(state);
1193
1194         /*
1195          * Execute merge by repeatedly extracting lowest tuple in heap,
1196          * writing it out, and replacing it with next tuple from same tape (if
1197          * there is another one).
1198          */
1199         while (state->memtupcount > 0)
1200         {
1201                 CHECK_FOR_INTERRUPTS();
1202                 /* write the tuple to destTape */
1203                 priorAvail = state->availMem;
1204                 srcTape = state->memtupindex[0];
1205                 WRITETUP(state, destTape, state->memtuples[0]);
1206                 /* writetup adjusted total free space, now fix per-tape space */
1207                 spaceFreed = state->availMem - priorAvail;
1208                 state->mergeavailmem[srcTape] += spaceFreed;
1209                 /* compact the heap */
1210                 tuplesort_heap_siftup(state, false);
1211                 if ((tupIndex = state->mergenext[srcTape]) == 0)
1212                 {
1213                         /* out of preloaded data on this tape, try to read more */
1214                         mergepreread(state);
1215                         /* if still no data, we've reached end of run on this tape */
1216                         if ((tupIndex = state->mergenext[srcTape]) == 0)
1217                                 continue;
1218                 }
1219                 /* pull next preread tuple from list, insert in heap */
1220                 tup = state->memtuples[tupIndex];
1221                 state->mergenext[srcTape] = state->memtupindex[tupIndex];
1222                 if (state->mergenext[srcTape] == 0)
1223                         state->mergelast[srcTape] = 0;
1224                 state->memtupindex[tupIndex] = state->mergefreelist;
1225                 state->mergefreelist = tupIndex;
1226                 tuplesort_heap_insert(state, tup, srcTape, false);
1227         }
1228
1229         /*
1230          * When the heap empties, we're done.  Write an end-of-run marker on
1231          * the output tape, and increment its count of real runs.
1232          */
1233         markrunend(state, destTape);
1234         state->tp_runs[TAPERANGE]++;
1235 }
1236
1237 /*
1238  * beginmerge - initialize for a merge pass
1239  *
1240  * We decrease the counts of real and dummy runs for each tape, and mark
1241  * which tapes contain active input runs in mergeactive[].      Then, load
1242  * as many tuples as we can from each active input tape, and finally
1243  * fill the merge heap with the first tuple from each active tape.
1244  */
1245 static void
1246 beginmerge(Tuplesortstate *state)
1247 {
1248         int                     activeTapes;
1249         int                     tapenum;
1250         int                     srcTape;
1251
1252         /* Heap should be empty here */
1253         Assert(state->memtupcount == 0);
1254
1255         /* Clear merge-pass state variables */
1256         memset(state->mergeactive, 0, sizeof(state->mergeactive));
1257         memset(state->mergenext, 0, sizeof(state->mergenext));
1258         memset(state->mergelast, 0, sizeof(state->mergelast));
1259         memset(state->mergeavailmem, 0, sizeof(state->mergeavailmem));
1260         state->mergefreelist = 0;       /* nothing in the freelist */
1261         state->mergefirstfree = MAXTAPES;       /* first slot available for
1262                                                                                  * preread */
1263
1264         /* Adjust run counts and mark the active tapes */
1265         activeTapes = 0;
1266         for (tapenum = 0; tapenum < TAPERANGE; tapenum++)
1267         {
1268                 if (state->tp_dummy[tapenum] > 0)
1269                         state->tp_dummy[tapenum]--;
1270                 else
1271                 {
1272                         Assert(state->tp_runs[tapenum] > 0);
1273                         state->tp_runs[tapenum]--;
1274                         srcTape = state->tp_tapenum[tapenum];
1275                         state->mergeactive[srcTape] = true;
1276                         activeTapes++;
1277                 }
1278         }
1279
1280         /*
1281          * Initialize space allocation to let each active input tape have an
1282          * equal share of preread space.
1283          */
1284         Assert(activeTapes > 0);
1285         state->spacePerTape = state->availMem / activeTapes;
1286         for (srcTape = 0; srcTape < MAXTAPES; srcTape++)
1287         {
1288                 if (state->mergeactive[srcTape])
1289                         state->mergeavailmem[srcTape] = state->spacePerTape;
1290         }
1291
1292         /*
1293          * Preread as many tuples as possible (and at least one) from each
1294          * active tape
1295          */
1296         mergepreread(state);
1297
1298         /* Load the merge heap with the first tuple from each input tape */
1299         for (srcTape = 0; srcTape < MAXTAPES; srcTape++)
1300         {
1301                 int                     tupIndex = state->mergenext[srcTape];
1302                 void       *tup;
1303
1304                 if (tupIndex)
1305                 {
1306                         tup = state->memtuples[tupIndex];
1307                         state->mergenext[srcTape] = state->memtupindex[tupIndex];
1308                         if (state->mergenext[srcTape] == 0)
1309                                 state->mergelast[srcTape] = 0;
1310                         state->memtupindex[tupIndex] = state->mergefreelist;
1311                         state->mergefreelist = tupIndex;
1312                         tuplesort_heap_insert(state, tup, srcTape, false);
1313                 }
1314         }
1315 }
1316
1317 /*
1318  * mergepreread - load tuples from merge input tapes
1319  *
1320  * This routine exists to improve sequentiality of reads during a merge pass,
1321  * as explained in the header comments of this file.  Load tuples from each
1322  * active source tape until the tape's run is exhausted or it has used up
1323  * its fair share of available memory.  In any case, we guarantee that there
1324  * is at one preread tuple available from each unexhausted input tape.
1325  */
1326 static void
1327 mergepreread(Tuplesortstate *state)
1328 {
1329         int                     srcTape;
1330         unsigned int tuplen;
1331         void       *tup;
1332         int                     tupIndex;
1333         long            priorAvail,
1334                                 spaceUsed;
1335
1336         for (srcTape = 0; srcTape < MAXTAPES; srcTape++)
1337         {
1338                 if (!state->mergeactive[srcTape])
1339                         continue;
1340
1341                 /*
1342                  * Skip reading from any tape that still has at least half of its
1343                  * target memory filled with tuples (threshold fraction may need
1344                  * adjustment?).  This avoids reading just a few tuples when the
1345                  * incoming runs are not being consumed evenly.
1346                  */
1347                 if (state->mergenext[srcTape] != 0 &&
1348                         state->mergeavailmem[srcTape] <= state->spacePerTape / 2)
1349                         continue;
1350
1351                 /*
1352                  * Read tuples from this tape until it has used up its free
1353                  * memory, but ensure that we have at least one.
1354                  */
1355                 priorAvail = state->availMem;
1356                 state->availMem = state->mergeavailmem[srcTape];
1357                 while (!LACKMEM(state) || state->mergenext[srcTape] == 0)
1358                 {
1359                         /* read next tuple, if any */
1360                         if ((tuplen = getlen(state, srcTape, true)) == 0)
1361                         {
1362                                 state->mergeactive[srcTape] = false;
1363                                 break;
1364                         }
1365                         tup = READTUP(state, srcTape, tuplen);
1366                         /* find or make a free slot in memtuples[] for it */
1367                         tupIndex = state->mergefreelist;
1368                         if (tupIndex)
1369                                 state->mergefreelist = state->memtupindex[tupIndex];
1370                         else
1371                         {
1372                                 tupIndex = state->mergefirstfree++;
1373                                 /* Might need to enlarge arrays! */
1374                                 if (tupIndex >= state->memtupsize)
1375                                 {
1376                                         FREEMEM(state, GetMemoryChunkSpace(state->memtuples));
1377                                         FREEMEM(state, GetMemoryChunkSpace(state->memtupindex));
1378                                         state->memtupsize *= 2;
1379                                         state->memtuples = (void **)
1380                                                 repalloc(state->memtuples,
1381                                                                  state->memtupsize * sizeof(void *));
1382                                         state->memtupindex = (int *)
1383                                                 repalloc(state->memtupindex,
1384                                                                  state->memtupsize * sizeof(int));
1385                                         USEMEM(state, GetMemoryChunkSpace(state->memtuples));
1386                                         USEMEM(state, GetMemoryChunkSpace(state->memtupindex));
1387                                 }
1388                         }
1389                         /* store tuple, append to list for its tape */
1390                         state->memtuples[tupIndex] = tup;
1391                         state->memtupindex[tupIndex] = 0;
1392                         if (state->mergelast[srcTape])
1393                                 state->memtupindex[state->mergelast[srcTape]] = tupIndex;
1394                         else
1395                                 state->mergenext[srcTape] = tupIndex;
1396                         state->mergelast[srcTape] = tupIndex;
1397                 }
1398                 /* update per-tape and global availmem counts */
1399                 spaceUsed = state->mergeavailmem[srcTape] - state->availMem;
1400                 state->mergeavailmem[srcTape] = state->availMem;
1401                 state->availMem = priorAvail - spaceUsed;
1402         }
1403 }
1404
1405 /*
1406  * dumptuples - remove tuples from heap and write to tape
1407  *
1408  * This is used during initial-run building, but not during merging.
1409  *
1410  * When alltuples = false, dump only enough tuples to get under the
1411  * availMem limit (and leave at least one tuple in the heap in any case,
1412  * since puttuple assumes it always has a tuple to compare to).
1413  *
1414  * When alltuples = true, dump everything currently in memory.
1415  * (This case is only used at end of input data.)
1416  *
1417  * If we empty the heap, close out the current run and return (this should
1418  * only happen at end of input data).  If we see that the tuple run number
1419  * at the top of the heap has changed, start a new run.
1420  */
1421 static void
1422 dumptuples(Tuplesortstate *state, bool alltuples)
1423 {
1424         while (alltuples ||
1425                    (LACKMEM(state) && state->memtupcount > 1))
1426         {
1427                 /*
1428                  * Dump the heap's frontmost entry, and sift up to remove it from
1429                  * the heap.
1430                  */
1431                 Assert(state->memtupcount > 0);
1432                 WRITETUP(state, state->tp_tapenum[state->destTape],
1433                                  state->memtuples[0]);
1434                 tuplesort_heap_siftup(state, true);
1435
1436                 /*
1437                  * If the heap is empty *or* top run number has changed, we've
1438                  * finished the current run.
1439                  */
1440                 if (state->memtupcount == 0 ||
1441                         state->currentRun != state->memtupindex[0])
1442                 {
1443                         markrunend(state, state->tp_tapenum[state->destTape]);
1444                         state->currentRun++;
1445                         state->tp_runs[state->destTape]++;
1446                         state->tp_dummy[state->destTape]--; /* per Alg D step D2 */
1447
1448                         /*
1449                          * Done if heap is empty, else prepare for new run.
1450                          */
1451                         if (state->memtupcount == 0)
1452                                 break;
1453                         Assert(state->currentRun == state->memtupindex[0]);
1454                         selectnewtape(state);
1455                 }
1456         }
1457 }
1458
1459 /*
1460  * tuplesort_rescan             - rewind and replay the scan
1461  */
1462 void
1463 tuplesort_rescan(Tuplesortstate *state)
1464 {
1465         Assert(state->randomAccess);
1466
1467         switch (state->status)
1468         {
1469                 case TSS_SORTEDINMEM:
1470                         state->current = 0;
1471                         state->eof_reached = false;
1472                         state->markpos_offset = 0;
1473                         state->markpos_eof = false;
1474                         break;
1475                 case TSS_SORTEDONTAPE:
1476                         LogicalTapeRewind(state->tapeset,
1477                                                           state->result_tape,
1478                                                           false);
1479                         state->eof_reached = false;
1480                         state->markpos_block = 0L;
1481                         state->markpos_offset = 0;
1482                         state->markpos_eof = false;
1483                         break;
1484                 default:
1485                         elog(ERROR, "invalid tuplesort state");
1486                         break;
1487         }
1488 }
1489
1490 /*
1491  * tuplesort_markpos    - saves current position in the merged sort file
1492  */
1493 void
1494 tuplesort_markpos(Tuplesortstate *state)
1495 {
1496         Assert(state->randomAccess);
1497
1498         switch (state->status)
1499         {
1500                 case TSS_SORTEDINMEM:
1501                         state->markpos_offset = state->current;
1502                         state->markpos_eof = state->eof_reached;
1503                         break;
1504                 case TSS_SORTEDONTAPE:
1505                         LogicalTapeTell(state->tapeset,
1506                                                         state->result_tape,
1507                                                         &state->markpos_block,
1508                                                         &state->markpos_offset);
1509                         state->markpos_eof = state->eof_reached;
1510                         break;
1511                 default:
1512                         elog(ERROR, "invalid tuplesort state");
1513                         break;
1514         }
1515 }
1516
1517 /*
1518  * tuplesort_restorepos - restores current position in merged sort file to
1519  *                                                last saved position
1520  */
1521 void
1522 tuplesort_restorepos(Tuplesortstate *state)
1523 {
1524         Assert(state->randomAccess);
1525
1526         switch (state->status)
1527         {
1528                 case TSS_SORTEDINMEM:
1529                         state->current = state->markpos_offset;
1530                         state->eof_reached = state->markpos_eof;
1531                         break;
1532                 case TSS_SORTEDONTAPE:
1533                         if (!LogicalTapeSeek(state->tapeset,
1534                                                                  state->result_tape,
1535                                                                  state->markpos_block,
1536                                                                  state->markpos_offset))
1537                                 elog(ERROR, "tuplesort_restorepos failed");
1538                         state->eof_reached = state->markpos_eof;
1539                         break;
1540                 default:
1541                         elog(ERROR, "invalid tuplesort state");
1542                         break;
1543         }
1544 }
1545
1546
1547 /*
1548  * Heap manipulation routines, per Knuth's Algorithm 5.2.3H.
1549  *
1550  * The heap lives in state->memtuples[], with parallel data storage
1551  * for indexes in state->memtupindex[].  If checkIndex is true, use
1552  * the tuple index as the front of the sort key; otherwise, no.
1553  */
1554
1555 #define HEAPCOMPARE(tup1,index1,tup2,index2) \
1556         (checkIndex && (index1 != index2) ? index1 - index2 : \
1557          COMPARETUP(state, tup1, tup2))
1558
1559 /*
1560  * Insert a new tuple into an empty or existing heap, maintaining the
1561  * heap invariant.
1562  */
1563 static void
1564 tuplesort_heap_insert(Tuplesortstate *state, void *tuple,
1565                                           int tupleindex, bool checkIndex)
1566 {
1567         void      **memtuples;
1568         int                *memtupindex;
1569         int                     j;
1570
1571         /*
1572          * Make sure memtuples[] can handle another entry.
1573          */
1574         if (state->memtupcount >= state->memtupsize)
1575         {
1576                 FREEMEM(state, GetMemoryChunkSpace(state->memtuples));
1577                 FREEMEM(state, GetMemoryChunkSpace(state->memtupindex));
1578                 state->memtupsize *= 2;
1579                 state->memtuples = (void **)
1580                         repalloc(state->memtuples,
1581                                          state->memtupsize * sizeof(void *));
1582                 state->memtupindex = (int *)
1583                         repalloc(state->memtupindex,
1584                                          state->memtupsize * sizeof(int));
1585                 USEMEM(state, GetMemoryChunkSpace(state->memtuples));
1586                 USEMEM(state, GetMemoryChunkSpace(state->memtupindex));
1587         }
1588         memtuples = state->memtuples;
1589         memtupindex = state->memtupindex;
1590
1591         /*
1592          * Sift-up the new entry, per Knuth 5.2.3 exercise 16. Note that Knuth
1593          * is using 1-based array indexes, not 0-based.
1594          */
1595         j = state->memtupcount++;
1596         while (j > 0)
1597         {
1598                 int                     i = (j - 1) >> 1;
1599
1600                 if (HEAPCOMPARE(tuple, tupleindex,
1601                                                 memtuples[i], memtupindex[i]) >= 0)
1602                         break;
1603                 memtuples[j] = memtuples[i];
1604                 memtupindex[j] = memtupindex[i];
1605                 j = i;
1606         }
1607         memtuples[j] = tuple;
1608         memtupindex[j] = tupleindex;
1609 }
1610
1611 /*
1612  * The tuple at state->memtuples[0] has been removed from the heap.
1613  * Decrement memtupcount, and sift up to maintain the heap invariant.
1614  */
1615 static void
1616 tuplesort_heap_siftup(Tuplesortstate *state, bool checkIndex)
1617 {
1618         void      **memtuples = state->memtuples;
1619         int                *memtupindex = state->memtupindex;
1620         void       *tuple;
1621         int                     tupindex,
1622                                 i,
1623                                 n;
1624
1625         if (--state->memtupcount <= 0)
1626                 return;
1627         n = state->memtupcount;
1628         tuple = memtuples[n];           /* tuple that must be reinserted */
1629         tupindex = memtupindex[n];
1630         i = 0;                                          /* i is where the "hole" is */
1631         for (;;)
1632         {
1633                 int                     j = 2 * i + 1;
1634
1635                 if (j >= n)
1636                         break;
1637                 if (j + 1 < n &&
1638                         HEAPCOMPARE(memtuples[j], memtupindex[j],
1639                                                 memtuples[j + 1], memtupindex[j + 1]) > 0)
1640                         j++;
1641                 if (HEAPCOMPARE(tuple, tupindex,
1642                                                 memtuples[j], memtupindex[j]) <= 0)
1643                         break;
1644                 memtuples[i] = memtuples[j];
1645                 memtupindex[i] = memtupindex[j];
1646                 i = j;
1647         }
1648         memtuples[i] = tuple;
1649         memtupindex[i] = tupindex;
1650 }
1651
1652
1653 /*
1654  * Tape interface routines
1655  */
1656
1657 static unsigned int
1658 getlen(Tuplesortstate *state, int tapenum, bool eofOK)
1659 {
1660         unsigned int len;
1661
1662         if (LogicalTapeRead(state->tapeset, tapenum, (void *) &len,
1663                                                 sizeof(len)) != sizeof(len))
1664                 elog(ERROR, "unexpected end of tape");
1665         if (len == 0 && !eofOK)
1666                 elog(ERROR, "unexpected end of data");
1667         return len;
1668 }
1669
1670 static void
1671 markrunend(Tuplesortstate *state, int tapenum)
1672 {
1673         unsigned int len = 0;
1674
1675         LogicalTapeWrite(state->tapeset, tapenum, (void *) &len, sizeof(len));
1676 }
1677
1678
1679 /*
1680  * qsort interface
1681  */
1682
1683 static int
1684 qsort_comparetup(const void *a, const void *b)
1685 {
1686         /* The passed pointers are pointers to void * ... */
1687
1688         return COMPARETUP(qsort_tuplesortstate, *(void **) a, *(void **) b);
1689 }
1690
1691
1692 /*
1693  * This routine selects an appropriate sorting function to implement
1694  * a sort operator as efficiently as possible.  The straightforward
1695  * method is to use the operator's implementation proc --- ie, "<"
1696  * comparison.  However, that way often requires two calls of the function
1697  * per comparison.      If we can find a btree three-way comparator function
1698  * associated with the operator, we can use it to do the comparisons
1699  * more efficiently.  We also support the possibility that the operator
1700  * is ">" (descending sort), in which case we have to reverse the output
1701  * of the btree comparator.
1702  *
1703  * Possibly this should live somewhere else (backend/catalog/, maybe?).
1704  */
1705 void
1706 SelectSortFunction(Oid sortOperator,
1707                                    RegProcedure *sortFunction,
1708                                    SortFunctionKind *kind)
1709 {
1710         CatCList   *catlist;
1711         int                     i;
1712         HeapTuple       tuple;
1713         Form_pg_operator optup;
1714         Oid                     opclass = InvalidOid;
1715
1716         /*
1717          * Search pg_amop to see if the target operator is registered as the
1718          * "<" or ">" operator of any btree opclass.  It's possible that it
1719          * might be registered both ways (eg, if someone were to build a
1720          * "reverse sort" opclass for some reason); prefer the "<" case if so.
1721          * If the operator is registered the same way in multiple opclasses,
1722          * assume we can use the associated comparator function from any one.
1723          */
1724         catlist = SearchSysCacheList(AMOPOPID, 1,
1725                                                                  ObjectIdGetDatum(sortOperator),
1726                                                                  0, 0, 0);
1727
1728         for (i = 0; i < catlist->n_members; i++)
1729         {
1730                 Form_pg_amop aform;
1731
1732                 tuple = &catlist->members[i]->tuple;
1733                 aform = (Form_pg_amop) GETSTRUCT(tuple);
1734
1735                 if (!opclass_is_btree(aform->amopclaid))
1736                         continue;
1737                 if (aform->amopstrategy == BTLessStrategyNumber)
1738                 {
1739                         opclass = aform->amopclaid;
1740                         *kind = SORTFUNC_CMP;
1741                         break;                          /* done looking */
1742                 }
1743                 else if (aform->amopstrategy == BTGreaterStrategyNumber)
1744                 {
1745                         opclass = aform->amopclaid;
1746                         *kind = SORTFUNC_REVCMP;
1747                         /* keep scanning in hopes of finding a BTLess entry */
1748                 }
1749         }
1750
1751         ReleaseSysCacheList(catlist);
1752
1753         if (OidIsValid(opclass))
1754         {
1755                 /* Found a suitable opclass, get its comparator support function */
1756                 *sortFunction = get_opclass_proc(opclass, BTORDER_PROC);
1757                 Assert(RegProcedureIsValid(*sortFunction));
1758                 return;
1759         }
1760
1761         /*
1762          * Can't find a comparator, so use the operator as-is.  Decide whether
1763          * it is forward or reverse sort by looking at its name (grotty, but
1764          * this only matters for deciding which end NULLs should get sorted
1765          * to).  XXX possibly better idea: see whether its selectivity function
1766          * is scalargtcmp?
1767          */
1768         tuple = SearchSysCache(OPEROID,
1769                                                    ObjectIdGetDatum(sortOperator),
1770                                                    0, 0, 0);
1771         if (!HeapTupleIsValid(tuple))
1772                 elog(ERROR, "cache lookup failed for operator %u", sortOperator);
1773         optup = (Form_pg_operator) GETSTRUCT(tuple);
1774         if (strcmp(NameStr(optup->oprname), ">") == 0)
1775                 *kind = SORTFUNC_REVLT;
1776         else
1777                 *kind = SORTFUNC_LT;
1778         *sortFunction = optup->oprcode;
1779         ReleaseSysCache(tuple);
1780
1781         Assert(RegProcedureIsValid(*sortFunction));
1782 }
1783
1784 /*
1785  * Inline-able copy of FunctionCall2() to save some cycles in sorting.
1786  */
1787 static inline Datum
1788 myFunctionCall2(FmgrInfo *flinfo, Datum arg1, Datum arg2)
1789 {
1790         FunctionCallInfoData fcinfo;
1791         Datum           result;
1792
1793         /* MemSet(&fcinfo, 0, sizeof(fcinfo)); */
1794         fcinfo.context = NULL;
1795         fcinfo.resultinfo = NULL;
1796         fcinfo.isnull = false;
1797
1798         fcinfo.flinfo = flinfo;
1799         fcinfo.nargs = 2;
1800         fcinfo.arg[0] = arg1;
1801         fcinfo.arg[1] = arg2;
1802         fcinfo.argnull[0] = false;
1803         fcinfo.argnull[1] = false;
1804
1805         result = FunctionCallInvoke(&fcinfo);
1806
1807         /* Check for null result, since caller is clearly not expecting one */
1808         if (fcinfo.isnull)
1809                 elog(ERROR, "function %u returned NULL", fcinfo.flinfo->fn_oid);
1810
1811         return result;
1812 }
1813
1814 /*
1815  * Apply a sort function (by now converted to fmgr lookup form)
1816  * and return a 3-way comparison result.  This takes care of handling
1817  * NULLs and sort ordering direction properly.
1818  */
1819 static inline int32
1820 inlineApplySortFunction(FmgrInfo *sortFunction, SortFunctionKind kind,
1821                                                 Datum datum1, bool isNull1,
1822                                                 Datum datum2, bool isNull2)
1823 {
1824         switch (kind)
1825         {
1826                 case SORTFUNC_LT:
1827                         if (isNull1)
1828                         {
1829                                 if (isNull2)
1830                                         return 0;
1831                                 return 1;               /* NULL sorts after non-NULL */
1832                         }
1833                         if (isNull2)
1834                                 return -1;
1835                         if (DatumGetBool(myFunctionCall2(sortFunction, datum1, datum2)))
1836                                 return -1;              /* a < b */
1837                         if (DatumGetBool(myFunctionCall2(sortFunction, datum2, datum1)))
1838                                 return 1;               /* a > b */
1839                         return 0;
1840
1841                 case SORTFUNC_REVLT:
1842                         /* We reverse the ordering of NULLs, but not the operator */
1843                         if (isNull1)
1844                         {
1845                                 if (isNull2)
1846                                         return 0;
1847                                 return -1;              /* NULL sorts before non-NULL */
1848                         }
1849                         if (isNull2)
1850                                 return 1;
1851                         if (DatumGetBool(myFunctionCall2(sortFunction, datum1, datum2)))
1852                                 return -1;              /* a < b */
1853                         if (DatumGetBool(myFunctionCall2(sortFunction, datum2, datum1)))
1854                                 return 1;               /* a > b */
1855                         return 0;
1856
1857                 case SORTFUNC_CMP:
1858                         if (isNull1)
1859                         {
1860                                 if (isNull2)
1861                                         return 0;
1862                                 return 1;               /* NULL sorts after non-NULL */
1863                         }
1864                         if (isNull2)
1865                                 return -1;
1866                         return DatumGetInt32(myFunctionCall2(sortFunction,
1867                                                                                                  datum1, datum2));
1868
1869                 case SORTFUNC_REVCMP:
1870                         if (isNull1)
1871                         {
1872                                 if (isNull2)
1873                                         return 0;
1874                                 return -1;              /* NULL sorts before non-NULL */
1875                         }
1876                         if (isNull2)
1877                                 return 1;
1878                         return -DatumGetInt32(myFunctionCall2(sortFunction,
1879                                                                                                   datum1, datum2));
1880
1881                 default:
1882                         elog(ERROR, "unrecognized SortFunctionKind: %d", (int) kind);
1883                         return 0;                       /* can't get here, but keep compiler quiet */
1884         }
1885 }
1886
1887 /*
1888  * Non-inline ApplySortFunction() --- this is needed only to conform to
1889  * C99's brain-dead notions about how to implement inline functions...
1890  */
1891 int32
1892 ApplySortFunction(FmgrInfo *sortFunction, SortFunctionKind kind,
1893                                   Datum datum1, bool isNull1,
1894                                   Datum datum2, bool isNull2)
1895 {
1896         return inlineApplySortFunction(sortFunction, kind,
1897                                                                    datum1, isNull1,
1898                                                                    datum2, isNull2);
1899 }
1900
1901
1902 /*
1903  * Routines specialized for HeapTuple case
1904  */
1905
1906 static int
1907 comparetup_heap(Tuplesortstate *state, const void *a, const void *b)
1908 {
1909         HeapTuple       ltup = (HeapTuple) a;
1910         HeapTuple       rtup = (HeapTuple) b;
1911         TupleDesc       tupDesc = state->tupDesc;
1912         int                     nkey;
1913
1914         for (nkey = 0; nkey < state->nKeys; nkey++)
1915         {
1916                 ScanKey         scanKey = state->scanKeys + nkey;
1917                 AttrNumber      attno = scanKey->sk_attno;
1918                 Datum           datum1,
1919                                         datum2;
1920                 bool            isnull1,
1921                                         isnull2;
1922                 int32           compare;
1923
1924                 datum1 = heap_getattr(ltup, attno, tupDesc, &isnull1);
1925                 datum2 = heap_getattr(rtup, attno, tupDesc, &isnull2);
1926
1927                 compare = inlineApplySortFunction(&scanKey->sk_func,
1928                                                                                   state->sortFnKinds[nkey],
1929                                                                                   datum1, isnull1,
1930                                                                                   datum2, isnull2);
1931                 if (compare != 0)
1932                 {
1933                         /* dead code? SK_COMMUTE can't actually be set here, can it? */
1934                         if (scanKey->sk_flags & SK_COMMUTE)
1935                                 compare = -compare;
1936                         return compare;
1937                 }
1938         }
1939
1940         return 0;
1941 }
1942
1943 static void *
1944 copytup_heap(Tuplesortstate *state, void *tup)
1945 {
1946         HeapTuple       tuple = (HeapTuple) tup;
1947
1948         tuple = heap_copytuple(tuple);
1949         USEMEM(state, GetMemoryChunkSpace(tuple));
1950         return (void *) tuple;
1951 }
1952
1953 /*
1954  * We don't bother to write the HeapTupleData part of the tuple.
1955  */
1956
1957 static void
1958 writetup_heap(Tuplesortstate *state, int tapenum, void *tup)
1959 {
1960         HeapTuple       tuple = (HeapTuple) tup;
1961         unsigned int tuplen;
1962
1963         tuplen = tuple->t_len + sizeof(tuplen);
1964         LogicalTapeWrite(state->tapeset, tapenum,
1965                                          (void *) &tuplen, sizeof(tuplen));
1966         LogicalTapeWrite(state->tapeset, tapenum,
1967                                          (void *) tuple->t_data, tuple->t_len);
1968         if (state->randomAccess)        /* need trailing length word? */
1969                 LogicalTapeWrite(state->tapeset, tapenum,
1970                                                  (void *) &tuplen, sizeof(tuplen));
1971
1972         FREEMEM(state, GetMemoryChunkSpace(tuple));
1973         heap_freetuple(tuple);
1974 }
1975
1976 static void *
1977 readtup_heap(Tuplesortstate *state, int tapenum, unsigned int len)
1978 {
1979         unsigned int tuplen = len - sizeof(unsigned int) + HEAPTUPLESIZE;
1980         HeapTuple       tuple = (HeapTuple) palloc(tuplen);
1981
1982         USEMEM(state, GetMemoryChunkSpace(tuple));
1983         /* reconstruct the HeapTupleData portion */
1984         tuple->t_len = len - sizeof(unsigned int);
1985         ItemPointerSetInvalid(&(tuple->t_self));
1986         tuple->t_datamcxt = CurrentMemoryContext;
1987         tuple->t_data = (HeapTupleHeader) (((char *) tuple) + HEAPTUPLESIZE);
1988         /* read in the tuple proper */
1989         if (LogicalTapeRead(state->tapeset, tapenum, (void *) tuple->t_data,
1990                                                 tuple->t_len) != tuple->t_len)
1991                 elog(ERROR, "unexpected end of data");
1992         if (state->randomAccess)        /* need trailing length word? */
1993                 if (LogicalTapeRead(state->tapeset, tapenum, (void *) &tuplen,
1994                                                         sizeof(tuplen)) != sizeof(tuplen))
1995                         elog(ERROR, "unexpected end of data");
1996         return (void *) tuple;
1997 }
1998
1999
2000 /*
2001  * Routines specialized for IndexTuple case
2002  *
2003  * NOTE: actually, these are specialized for the btree case; it's not
2004  * clear whether you could use them for a non-btree index.      Possibly
2005  * you'd need to make another set of routines if you needed to sort
2006  * according to another kind of index.
2007  */
2008
2009 static int
2010 comparetup_index(Tuplesortstate *state, const void *a, const void *b)
2011 {
2012         /*
2013          * This is almost the same as _bt_tuplecompare(), but we need to keep
2014          * track of whether any null fields are present.
2015          */
2016         IndexTuple      tuple1 = (IndexTuple) a;
2017         IndexTuple      tuple2 = (IndexTuple) b;
2018         Relation        rel = state->indexRel;
2019         int                     keysz = RelationGetNumberOfAttributes(rel);
2020         ScanKey         scankey = state->indexScanKey;
2021         TupleDesc       tupDes;
2022         int                     i;
2023         bool            equal_hasnull = false;
2024
2025         tupDes = RelationGetDescr(rel);
2026
2027         for (i = 1; i <= keysz; i++)
2028         {
2029                 ScanKey         entry = &scankey[i - 1];
2030                 Datum           datum1,
2031                                         datum2;
2032                 bool            isnull1,
2033                                         isnull2;
2034                 int32           compare;
2035
2036                 datum1 = index_getattr(tuple1, i, tupDes, &isnull1);
2037                 datum2 = index_getattr(tuple2, i, tupDes, &isnull2);
2038
2039                 /* see comments about NULLs handling in btbuild */
2040
2041                 /* the comparison function is always of CMP type */
2042                 compare = inlineApplySortFunction(&entry->sk_func, SORTFUNC_CMP,
2043                                                                                   datum1, isnull1,
2044                                                                                   datum2, isnull2);
2045
2046                 if (compare != 0)
2047                         return (int) compare;           /* done when we find unequal
2048                                                                                  * attributes */
2049
2050                 /* they are equal, so we only need to examine one null flag */
2051                 if (isnull1)
2052                         equal_hasnull = true;
2053         }
2054
2055         /*
2056          * If btree has asked us to enforce uniqueness, complain if two equal
2057          * tuples are detected (unless there was at least one NULL field).
2058          *
2059          * It is sufficient to make the test here, because if two tuples are
2060          * equal they *must* get compared at some stage of the sort ---
2061          * otherwise the sort algorithm wouldn't have checked whether one must
2062          * appear before the other.
2063          *
2064          * Some rather brain-dead implementations of qsort will sometimes call
2065          * the comparison routine to compare a value to itself.  (At this
2066          * writing only QNX 4 is known to do such silly things.)  Don't raise
2067          * a bogus error in that case.
2068          */
2069         if (state->enforceUnique && !equal_hasnull && tuple1 != tuple2)
2070                 ereport(ERROR,
2071                                 (errcode(ERRCODE_UNIQUE_VIOLATION),
2072                                  errmsg("could not create unique index"),
2073                                  errdetail("Table contains duplicated values.")));
2074
2075         return 0;
2076 }
2077
2078 static void *
2079 copytup_index(Tuplesortstate *state, void *tup)
2080 {
2081         IndexTuple      tuple = (IndexTuple) tup;
2082         unsigned int tuplen = IndexTupleSize(tuple);
2083         IndexTuple      newtuple;
2084
2085         newtuple = (IndexTuple) palloc(tuplen);
2086         USEMEM(state, GetMemoryChunkSpace(newtuple));
2087
2088         memcpy(newtuple, tuple, tuplen);
2089
2090         return (void *) newtuple;
2091 }
2092
2093 static void
2094 writetup_index(Tuplesortstate *state, int tapenum, void *tup)
2095 {
2096         IndexTuple      tuple = (IndexTuple) tup;
2097         unsigned int tuplen;
2098
2099         tuplen = IndexTupleSize(tuple) + sizeof(tuplen);
2100         LogicalTapeWrite(state->tapeset, tapenum,
2101                                          (void *) &tuplen, sizeof(tuplen));
2102         LogicalTapeWrite(state->tapeset, tapenum,
2103                                          (void *) tuple, IndexTupleSize(tuple));
2104         if (state->randomAccess)        /* need trailing length word? */
2105                 LogicalTapeWrite(state->tapeset, tapenum,
2106                                                  (void *) &tuplen, sizeof(tuplen));
2107
2108         FREEMEM(state, GetMemoryChunkSpace(tuple));
2109         pfree(tuple);
2110 }
2111
2112 static void *
2113 readtup_index(Tuplesortstate *state, int tapenum, unsigned int len)
2114 {
2115         unsigned int tuplen = len - sizeof(unsigned int);
2116         IndexTuple      tuple = (IndexTuple) palloc(tuplen);
2117
2118         USEMEM(state, GetMemoryChunkSpace(tuple));
2119         if (LogicalTapeRead(state->tapeset, tapenum, (void *) tuple,
2120                                                 tuplen) != tuplen)
2121                 elog(ERROR, "unexpected end of data");
2122         if (state->randomAccess)        /* need trailing length word? */
2123                 if (LogicalTapeRead(state->tapeset, tapenum, (void *) &tuplen,
2124                                                         sizeof(tuplen)) != sizeof(tuplen))
2125                         elog(ERROR, "unexpected end of data");
2126         return (void *) tuple;
2127 }
2128
2129
2130 /*
2131  * Routines specialized for DatumTuple case
2132  */
2133
2134 static int
2135 comparetup_datum(Tuplesortstate *state, const void *a, const void *b)
2136 {
2137         DatumTuple *ltup = (DatumTuple *) a;
2138         DatumTuple *rtup = (DatumTuple *) b;
2139
2140         return inlineApplySortFunction(&state->sortOpFn, state->sortFnKind,
2141                                                                    ltup->val, ltup->isNull,
2142                                                                    rtup->val, rtup->isNull);
2143 }
2144
2145 static void *
2146 copytup_datum(Tuplesortstate *state, void *tup)
2147 {
2148         /* Not currently needed */
2149         elog(ERROR, "copytup_datum() should not be called");
2150         return NULL;
2151 }
2152
2153 static void
2154 writetup_datum(Tuplesortstate *state, int tapenum, void *tup)
2155 {
2156         DatumTuple *tuple = (DatumTuple *) tup;
2157         unsigned int tuplen;
2158         unsigned int writtenlen;
2159
2160         if (tuple->isNull || state->datumTypeByVal)
2161                 tuplen = sizeof(DatumTuple);
2162         else
2163         {
2164                 Size            datalen;
2165
2166                 datalen = datumGetSize(tuple->val, false, state->datumTypeLen);
2167                 tuplen = datalen + MAXALIGN(sizeof(DatumTuple));
2168         }
2169
2170         writtenlen = tuplen + sizeof(unsigned int);
2171
2172         LogicalTapeWrite(state->tapeset, tapenum,
2173                                          (void *) &writtenlen, sizeof(writtenlen));
2174         LogicalTapeWrite(state->tapeset, tapenum,
2175                                          (void *) tuple, tuplen);
2176         if (state->randomAccess)        /* need trailing length word? */
2177                 LogicalTapeWrite(state->tapeset, tapenum,
2178                                                  (void *) &writtenlen, sizeof(writtenlen));
2179
2180         FREEMEM(state, GetMemoryChunkSpace(tuple));
2181         pfree(tuple);
2182 }
2183
2184 static void *
2185 readtup_datum(Tuplesortstate *state, int tapenum, unsigned int len)
2186 {
2187         unsigned int tuplen = len - sizeof(unsigned int);
2188         DatumTuple *tuple = (DatumTuple *) palloc(tuplen);
2189
2190         USEMEM(state, GetMemoryChunkSpace(tuple));
2191         if (LogicalTapeRead(state->tapeset, tapenum, (void *) tuple,
2192                                                 tuplen) != tuplen)
2193                 elog(ERROR, "unexpected end of data");
2194         if (state->randomAccess)        /* need trailing length word? */
2195                 if (LogicalTapeRead(state->tapeset, tapenum, (void *) &tuplen,
2196                                                         sizeof(tuplen)) != sizeof(tuplen))
2197                         elog(ERROR, "unexpected end of data");
2198
2199         if (!tuple->isNull && !state->datumTypeByVal)
2200                 tuple->val = PointerGetDatum(((char *) tuple) +
2201                                                                          MAXALIGN(sizeof(DatumTuple)));
2202         return (void *) tuple;
2203 }