Google Git
Sign in
apache / hawq / 91c45cab0e5844abfe0f398f2378637f4028fe45 / . / src / backend / utils / sort / tuplesort_mk.c
blob: c95308f513bfb4b9475780aa12bb9d74de657fe3 [file] [log] [blame]
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
/*-------------------------------------------------------------------------
*
* tuplesort.c
* Generalized tuple sorting routines.
*
* This module handles sorting of heap tuples, index tuples, or single
* Datums (and could easily support other kinds of sortable objects,
* if necessary). It works efficiently for both small and large amounts
* of data. Small amounts are sorted in-memory using qsort(). Large
* amounts are sorted using temporary files and a standard external sort
* algorithm.
*
* See Knuth, volume 3, for more than you want to know about the external
* sorting algorithm. We divide the input into sorted runs using replacement
* selection, in the form of a priority tree implemented as a heap
* (essentially his Algorithm 5.2.3H), then merge the runs using polyphase
* merge, Knuth's Algorithm 5.4.2D. The logical "tapes" used by Algorithm D
* are implemented by logtape.c, which avoids space wastage by recycling
* disk space as soon as each block is read from its "tape".
*
* We do not form the initial runs using Knuth's recommended replacement
* selection data structure (Algorithm 5.4.1R), because it uses a fixed
* number of records in memory at all times. Since we are dealing with
* tuples that may vary considerably in size, we want to be able to vary
* the number of records kept in memory to ensure full utilization of the
* allowed sort memory space. So, we keep the tuples in a variable-size
* heap, with the next record to go out at the top of the heap. Like
* Algorithm 5.4.1R, each record is stored with the run number that it
* must go into, and we use (run number, key) as the ordering key for the
* heap. When the run number at the top of the heap changes, we know that
* no more records of the prior run are left in the heap.
*
* The approximate amount of memory allowed for any one sort operation
* is specified in kilobytes by the caller (most pass work_mem). Initially,
* we absorb tuples and simply store them in an unsorted array as long as
* we haven't exceeded workMem. If we reach the end of the input without
* exceeding workMem, we sort the array using qsort() and subsequently return
* tuples just by scanning the tuple array sequentially. If we do exceed
* workMem, we construct a heap using Algorithm H and begin to emit tuples
* into sorted runs in temporary tapes, emitting just enough tuples at each
* step to get back within the workMem limit. Whenever the run number at
* the top of the heap changes, we begin a new run with a new output tape
* (selected per Algorithm D). After the end of the input is reached,
* we dump out remaining tuples in memory into a final run (or two),
* then merge the runs using Algorithm D.
*
* When merging runs, we use a heap containing just the frontmost tuple from
* each source run; we repeatedly output the smallest tuple and insert the
* next tuple from its source tape (if any). When the heap empties, the merge
* is complete. The basic merge algorithm thus needs very little memory ---
* only M tuples for an M-way merge, and M is constrained to a small number.
* However, we can still make good use of our full workMem allocation by
* pre-reading additional tuples from each source tape. Without prereading,
* our access pattern to the temporary file would be very erratic; on average
* we'd read one block from each of M source tapes during the same time that
* we're writing M blocks to the output tape, so there is no sequentiality of
* access at all, defeating the read-ahead methods used by most Unix kernels.
* Worse, the output tape gets written into a very random sequence of blocks
* of the temp file, ensuring that things will be even worse when it comes
* time to read that tape. A straightforward merge pass thus ends up doing a
* lot of waiting for disk seeks. We can improve matters by prereading from
* each source tape sequentially, loading about workMem/M bytes from each tape
* in turn. Then we run the merge algorithm, writing but not reading until
* one of the preloaded tuple series runs out. Then we switch back to preread
* mode, fill memory again, and repeat. This approach helps to localize both
* read and write accesses.
*
* When the caller requests random access to the sort result, we form
* the final sorted run on a logical tape which is then "frozen", so
* that we can access it randomly. When the caller does not need random
* access, we return from tuplesort_performsort() as soon as we are down
* to one run per logical tape. The final merge is then performed
* on-the-fly as the caller repeatedly calls tuplesort_getXXX; this
* saves one cycle of writing all the data out to disk and reading it in.
*
* Before Postgres 8.2, we always used a seven-tape polyphase merge, on the
* grounds that 7 is the "sweet spot" on the tapes-to-passes curve according
* to Knuth's figure 70 (section 5.4.2). However, Knuth is assuming that
* tape drives are expensive beasts, and in particular that there will always
* be many more runs than tape drives. In our implementation a "tape drive"
* doesn't cost much more than a few Kb of memory buffers, so we can afford
* to have lots of them. In particular, if we can have as many tape drives
* as sorted runs, we can eliminate any repeated I/O at all. In the current
* code we determine the number of tapes M on the basis of workMem: we want
* workMem/M to be large enough that we read a fair amount of data each time
* we preread from a tape, so as to maintain the locality of access described
* above. Nonetheless, with large workMem we can have many tapes.
*
*
* Portions Copyright (c) 2007-2008, Greenplum inc
* Portions Copyright (c) 1996-2008, PostgreSQL Global Development Group
* Portions Copyright (c) 1994, Regents of the University of California
*
* IDENTIFICATION
* $PostgreSQL: pgsql/src/backend/utils/sort/tuplesort.c,v 1.70 2006/10/04 00:30:04 momjian Exp $
*
*-------------------------------------------------------------------------
*/
#include "postgres.h"
#include "access/heapam.h"
#include "access/nbtree.h"
#include "access/tuptoaster.h"
#include "catalog/pg_type.h"
#include "catalog/pg_amop.h"
#include "catalog/pg_operator.h"
#include "executor/instrument.h" /* Instrumentation */
#include "lib/stringinfo.h" /* StringInfo */
#include "executor/nodeSort.h" /* gpmon */
#include "miscadmin.h"
#include "utils/datum.h"
#include "executor/execWorkfile.h"
#include "utils/logtape.h"
#include "utils/lsyscache.h"
#include "utils/memutils.h"
#include "utils/pg_rusage.h"
#include "utils/syscache.h"
#include "utils/tuplesort.h"
#include "utils/pg_locale.h"
#include "utils/builtins.h"
#include "utils/tuplesort_mk.h"
#include "utils/string_wrapper.h"
#include "utils/faultinjector.h"
#include "cdb/cdbvars.h"
/*
* Possible states of a Tuplesort object. These denote the states that
* persist between calls of Tuplesort routines.
*/
typedef enum
{
TSS_INITIAL, /* Loading tuples; still within memory limit */
TSS_BUILDRUNS, /* Loading tuples; writing to tape */
TSS_SORTEDINMEM, /* Sort completed entirely in memory */
TSS_SORTEDONTAPE, /* Sort completed, final run is on tape */
TSS_FINALMERGE /* Performing final merge on-the-fly */
} TupSortStatus;
/*
* Parameters for calculation of number of tapes to use --- see inittapes()
* and tuplesort_merge_order().
*
* In this calculation we assume that each tape will cost us about 3 blocks
* worth of buffer space (which is an underestimate for very large data
* volumes, but it's probably close enough --- see logtape.c).
*
* MERGE_BUFFER_SIZE is how much data we'd like to read from each input
* tape during a preread cycle (see discussion at top of file).
*/
#define MINORDER 6 /* minimum merge order */
#define MAXORDER 250 /* maximum merge order */
#define TAPE_BUFFER_OVERHEAD (BLCKSZ * 3)
#define MERGE_BUFFER_SIZE (BLCKSZ * 32)
// #define PRINT_SPILL_AND_MEMORY_MESSAGES
/*
* Current position of Tuplesort operation.
*/
struct TuplesortPos_mk
{
/*
* These variables are used after completion of sorting to keep track of
* the next tuple to return. (In the tape case, the tape's current read
* position is also critical state.)
*/
int current; /* array index (only used if SORTEDINMEM) */
bool eof_reached; /* reached EOF (needed for cursors) */
/* markpos_xxx holds marked position for mark and restore */
union {
LogicalTapePos tapepos;
long mempos;
} markpos;
bool markpos_eof;
LogicalTape* cur_work_tape; /* current tape that I am working on */
};
/* Merge reader (read back from runs) context for mk_heap */
typedef struct TupsortMergeReadCtxt
{
Tuplesortstate_mk *tsstate;
TuplesortPos_mk pos;
/* Buffer for preread */
MKEntry *p;
int allocsz;
int cnt;
int cur;
int mem_allowed;
int mem_used;
} TupsortMergeReadCtxt;
/*
* Private state of a Tuplesort operation.
*/
struct Tuplesortstate_mk
{
TupSortStatus status; /* enumerated value as shown above */
int nKeys; /* number of columns in sort key */
bool randomAccess; /* did caller request random access? */
long memAllowed;
int maxTapes; /* number of tapes (Knuth's T) */
int tapeRange; /* maxTapes-1 (Knuth's P) */
MemoryContext sortcontext; /* memory context holding all sort data */
LogicalTapeSet *tapeset; /* logtape.c object for tapes in a temp file */
ScanState *ss;
/* Representation of all spill file names, for spill file reuse */
workfile_set *work_set;
/*
* MUST be set
*
* Function to copy a supplied input tuple into palloc'd space and set up
* its SortTuple representation (ie, set tuple/datum1/isnull1). Also,
* state->availMem must be decreased by the amount of space used for the
* tuple copy (note the SortTuple struct itself is not counted).
*/
void (*copytup) (Tuplesortstate_mk *state, MKEntry *e, void *tup);
/*
* MUST be set
*
* Function to write a stored tuple onto tape. The representation of the
* tuple on tape need not be the same as it is in memory; requirements on
* the tape representation are given below. After writing the tuple,
* pfree() the out-of-line data (not the SortTuple struct!), and increase
* state->availMem by the amount of memory space thereby released.
*/
long (*writetup) (Tuplesortstate_mk *state, LogicalTape *lt, MKEntry *e);
/*
* MUST be set
*
* Function to read a stored tuple from tape back into memory. 'len' is
* the already-read length of the stored tuple. Create a palloc'd copy,
* initialize tuple/datum1/isnull1 in the target SortTuple struct, and
* decrease state->availMem by the amount of memory space consumed.
*/
void (*readtup) (Tuplesortstate_mk *state, TuplesortPos_mk *pos, MKEntry *e,
LogicalTape *lt, uint32 len);
/*
* This array holds the tuples now in sort memory. If we are in state
* INITIAL, the tuples are in no particular order; if we are in state
* SORTEDINMEM, the tuples are in final sorted order;
*/
MKEntry *entries;
long entry_allocsize;
long entry_count;
/*
* BUILDRUNS and FINALMERGE: Heap for producing and merging runs
*/
MKHeap *mkheap;
MKHeapReader *mkhreader;
TupsortMergeReadCtxt *mkhreader_ctxt;
int mkhreader_allocsize;
/*
* MK context, holds info needed by compare/prepare functions
*/
MKContext mkctxt;
/*
* A flag to indicate whether the stats for this tuplesort
* has been finalized.
*/
bool statsFinalized;
int currentRun;
/*
* Unless otherwise noted, all pointer variables below are pointers to
* arrays of length maxTapes, holding per-tape data.
*/
/*
* These variables are only used during merge passes. mergeactive[i] is
* true if we are reading an input run from (actual) tape number i and
* have not yet exhausted that run. mergenext[i] is the memtuples index
* of the next pre-read tuple (next to be loaded into the heap) for tape
* i, or 0 if we are out of pre-read tuples. mergelast[i] similarly
* points to the last pre-read tuple from each tape. mergeavailslots[i]
* is the number of unused memtuples[] slots reserved for tape i, and
* mergeavailmem[i] is the amount of unused space allocated for tape i.
* mergefreelist and mergefirstfree keep track of unused locations in the
* memtuples[] array. The memtuples[].tupindex fields link together
* pre-read tuples for each tape as well as recycled locations in
* mergefreelist. It is OK to use 0 as a null link in these lists, because
* memtuples[0] is part of the merge heap and is never a pre-read tuple.
*/
bool *mergeactive; /* active input run source? */
int *mergenext; /* first preread tuple for each source */
int *mergelast; /* last preread tuple for each source */
int *mergeavailslots; /* slots left for prereading each tape */
long *mergeavailmem; /* availMem for prereading each tape */
int mergefreelist; /* head of freelist of recycled slots */
int mergefirstfree; /* first slot never used in this merge */
/*
* Variables for Algorithm D. Note that destTape is a "logical" tape
* number, ie, an index into the tp_xxx[] arrays. Be careful to keep
* "logical" and "actual" tape numbers straight!
*/
int Level; /* Knuth's l */
int destTape; /* current output tape (Knuth's j, less 1) */
int *tp_fib; /* Target Fibonacci run counts (A[]) */
int *tp_runs; /* # of real runs on each tape */
int *tp_dummy; /* # of dummy runs for each tape (D[]) */
int *tp_tapenum; /* Actual tape numbers (TAPE[]) */
int activeTapes; /* # of active input tapes in merge pass */
LogicalTape *result_tape; /* actual tape of finished output */
TuplesortPos_mk pos; /* current postion */
/*
* Tuple desc and binding, for MemTuple.
*/
TupleDesc tupDesc;
MemTupleBinding *mt_bind;
ScanKey cmpScanKey;
/*
* These variables are specific to the IndexTuple case; they are set by
* tuplesort_begin_index and used only by the IndexTuple routines.
*/
Relation indexRel;
/*
* These variables are specific to the Datum case; they are set by
* tuplesort_begin_datum and used only by the DatumTuple routines.
*/
Oid datumType;
Oid sortOperator;
bool nullfirst;
/* we need typelen and byval in order to know how to copy the Datums. */
int datumTypeLen;
bool datumTypeByVal;
/*
* CDB: EXPLAIN ANALYZE reporting interface and statistics.
*/
struct Instrumentation *instrument;
struct StringInfoData *explainbuf;
uint64 totalTupleBytes;
uint64 totalNumTuples;
uint64 numTuplesInMem;
uint64 memUsedBeforeSpill; /* memory that is used by Sort at the time of spilling */
long arraySizeBeforeSpill; /* the value for entry_allocsize at the time of spilling */
/*
* File for dump/load logical tape set. Used by sharing sort across slice
*/
char *tapeset_file_prefix;
/*
* State file used to load a logical tape set. Used by sharing sort across slice
*/
ExecWorkFile *tapeset_state_file;
/* Gpmon */
gpmon_packet_t *gpmon_pkt;
int *gpmon_sort_tick;
};
static bool is_sortstate_rwfile(Tuplesortstate_mk *state)
{
return state->tapeset_state_file != NULL;
}
#ifdef USE_ASSERT_CHECKING
static bool is_under_sort_ctxt(Tuplesortstate_mk *state)
{
return CurrentMemoryContext == state->sortcontext;
}
#endif
/**
* Any strings that are STRXFRM_INPUT_LENGTH_LIMIT or larger will store only the
* first STRXFRM_INPUT_LENGTH_LIMIT bytes of the transformed string.
*
* Note that this is actually less transformed data in some cases than if the string were
* just a little smaller than STRXFRM_INPUT_LENGTH_LIMIT. We can probably make the
* transition more gradual but will still want this fading off -- for long strings
* that differ within the first STRXFRM_INPUT_LENGTH_LIMIT bytes then the prefix
* will be sufficient for all but equal cases -- in which case a longer prefix does
* not help (we must resort to datum for comparison)
*
* If less than the whole transformed size is stored then the transformed string itself is also
* not copied -- so for large strings we must resort to datum-based comparison.
*/
#define STRXFRM_INPUT_LENGTH_LIMIT (512)
#define COPYTUP(state,stup,tup) ((*(state)->copytup) (state, stup, tup))
#define WRITETUP(state,tape,stup) ((*(state)->writetup) (state, tape, stup))
#define READTUP(state,pos,stup,tape,len) ((*(state)->readtup) (state, pos, stup, tape, len))
static inline bool LACKMEM_WITH_ESTIMATE(Tuplesortstate_mk *state)
{
/* Assert: lackmem is only effective during initial run build because
* we don't maintain the estimate after that. */
Assert(state->status == TSS_INITIAL);
return MemoryContextGetCurrentSpace(state->sortcontext) + state->mkctxt.estimatedExtraForPrep > state->memAllowed;
}
/*
* NOTES about on-tape representation of tuples:
*
* We require the first "unsigned int" of a stored tuple to be the total size
* on-tape of the tuple, including itself (so it is never zero; an all-zero
* unsigned int is used to delimit runs). The remainder of the stored tuple
* may or may not match the in-memory representation of the tuple ---
* any conversion needed is the job of the writetup and readtup routines.
*
* If state->randomAccess is true, then the stored representation of the
* tuple must be followed by another "unsigned int" that is a copy of the
* length --- so the total tape space used is actually sizeof(unsigned int)
* more than the stored length value. This allows read-backwards. When
* randomAccess is not true, the write/read routines may omit the extra
* length word.
*
* writetup is expected to write both length words as well as the tuple
* data. When readtup is called, the tape is positioned just after the
* front length word; readtup must read the tuple data and advance past
* the back length word (if present).
*
* The write/read routines can make use of the tuple description data
* stored in the Tuplesortstate_mk record, if needed. They are also expected
* to adjust state->availMem by the amount of memory space (not tape space!)
* released or consumed. There is no error return from either writetup
* or readtup; they should ereport() on failure.
*
*
* NOTES about memory consumption calculations:
*
* We count space allocated for tuples against the workMem limit, plus
* the space used by the variable-size memtuples array. Fixed-size space
* is not counted; it's small enough to not be interesting.
*
* Note that we count actual space used (as shown by GetMemoryChunkSpace)
* rather than the originally-requested size. This is important since
* palloc can add substantial overhead. It's not a complete answer since
* we won't count any wasted space in palloc allocation blocks, but it's
* a lot better than what we were doing before 7.3.
*/
static Tuplesortstate_mk *tuplesort_begin_common(ScanState * ss, int workMem, bool randomAccess, bool allocmemtuple);
static void puttuple_common(Tuplesortstate_mk *state, MKEntry *e);
static void selectnewtape_mk(Tuplesortstate_mk *state);
static void mergeruns(Tuplesortstate_mk *state);
static void beginmerge(Tuplesortstate_mk *state);
static uint32 getlen(Tuplesortstate_mk *state, TuplesortPos_mk *pos, LogicalTape *lt, bool eofOK);
static void markrunend(Tuplesortstate_mk *state, int tapenum);
static void copytup_heap(Tuplesortstate_mk *state, MKEntry *e, void *tup);
static long writetup_heap(Tuplesortstate_mk *state, LogicalTape *lt, MKEntry *e);
static void freetup_heap(MKEntry *e);
static void readtup_heap(Tuplesortstate_mk *state, TuplesortPos_mk *pos, MKEntry *e,
LogicalTape *lt, uint32 len);
static void copytup_index(Tuplesortstate_mk *state, MKEntry *e, void *tup);
static long writetup_index(Tuplesortstate_mk *state, LogicalTape *lt, MKEntry *e);
static void freetup_index(MKEntry *e);
static void readtup_index(Tuplesortstate_mk *state, TuplesortPos_mk *pos, MKEntry *e,
LogicalTape *lt, uint32 len);
static void freetup_noop(MKEntry *e);
static void copytup_datum(Tuplesortstate_mk *state, MKEntry *e, void *tup);
static long writetup_datum(Tuplesortstate_mk *state, LogicalTape *lt, MKEntry *e);
static void freetup_datum(MKEntry *e);
static void readtup_datum(Tuplesortstate_mk *state, TuplesortPos_mk *pos, MKEntry *e,
LogicalTape *lt, uint32 len);
static void tupsort_prepare_char(MKEntry *a, bool isChar);
static int tupsort_compare_char(MKEntry *v1, MKEntry *v2, MKLvContext *lvctxt, MKContext *mkContext);
static Datum tupsort_fetch_datum_mtup(MKEntry *a, MKContext *mkctxt, MKLvContext *lvctxt, bool *isNullOut);
static Datum tupsort_fetch_datum_itup(MKEntry *a, MKContext *mkctxt, MKLvContext *lvctxt, bool *isNullOut);
static int32 estimateMaxPrepareSizeForEntry(MKEntry *a, struct MKContext *mkctxt);
static int32 estimatePrepareSpaceForChar(struct MKContext *mkContext, MKEntry *e, Datum d, bool isCHAR);
static void tuplesort_inmem_limit_insert(Tuplesortstate_mk *state, MKEntry *e);
static void tuplesort_inmem_nolimit_insert(Tuplesortstate_mk * state, MKEntry * e);
static void tuplesort_heap_insert(Tuplesortstate_mk *state, MKEntry *e);
static void tuplesort_limit_sort(Tuplesortstate_mk *state);
static void tupsort_refcnt(void *vp, int ref);
/* Declare the following as extern so that older dtrace will not complain */
extern void inittapes_mk(Tuplesortstate_mk *state, const char* rwfile_prefix);
extern void dumptuples_mk(Tuplesortstate_mk *state, bool alltuples);
extern void mergeonerun_mk(Tuplesortstate_mk *state);
extern void mkheap_verify_heap(MKHeap *heap, int top);
/*
* tuplesort_begin_xxx
*
* Initialize for a tuple sort operation.
*
* After calling tuplesort_begin, the caller should call tuplesort_putXXX
* zero or more times, then call tuplesort_performsort when all the tuples
* have been supplied. After performsort, retrieve the tuples in sorted
* order by calling tuplesort_getXXX until it returns false/NULL. (If random
* access was requested, rescan, markpos, and restorepos can also be called.)
* Call tuplesort_end to terminate the operation and release memory/disk space.
*
* Each variant of tuplesort_begin has a workMem parameter specifying the
* maximum number of kilobytes of RAM to use before spilling data to disk.
* (The normal value of this parameter is work_mem, but some callers use
* other values.) Each variant also has a randomAccess parameter specifying
* whether the caller needs non-sequential access to the sort result.
*
* CDB: During EXPLAIN ANALYZE, after tuplesort_begin_xxx() the caller should
* use tuplesort_set_instrument() (q.v.) to enable statistical reporting.
*/
static Tuplesortstate_mk *
tuplesort_begin_common(ScanState * ss, int workMem, bool randomAccess, bool allocmemtuple)
{
Tuplesortstate_mk *state;
MemoryContext sortcontext;
MemoryContext oldcontext;
/*
* Create a working memory context for this sort operation. All data
* needed by the sort will live inside this context.
*/
sortcontext = AllocSetContextCreate(CurrentMemoryContext,
"TupleSort",
ALLOCSET_DEFAULT_MINSIZE,
ALLOCSET_DEFAULT_INITSIZE,
ALLOCSET_DEFAULT_MAXSIZE);
/*
* Make the Tuplesortstate_mk within the per-sort context. This way, we
* don't need a separate pfree() operation for it at shutdown.
*/
oldcontext = MemoryContextSwitchTo(sortcontext);
/*
* palloc0, need to zero out memeories, so we do not need to set
* every var of the tuplestate to 0 below.
*/
state = (Tuplesortstate_mk *) palloc0(sizeof(Tuplesortstate_mk));
state->status = TSS_INITIAL;
state->randomAccess = randomAccess;
state->memAllowed = workMem * 1024L;
state->work_set = NULL;
state->ss = ss;
state->sortcontext = sortcontext;
if(allocmemtuple)
{
int i;
state->entry_allocsize = 1024;
state->entries = (MKEntry *) palloc0(state->entry_allocsize * sizeof(MKEntry));
for(i=0; i<state->entry_allocsize; ++i)
mke_blank(state->entries+i);
/* workMem must be large enough for the minimal memtuples array */
if (LACKMEM_WITH_ESTIMATE(state))
elog(ERROR, "insufficient memory allowed for sort");
}
/*
* maxTapes, tapeRange, and Algorithm D variables will be initialized by
* inittapes(), if needed
*/
MemoryContextSwitchTo(oldcontext);
Assert(!state->statsFinalized);
return state;
}
/*
* Initialize some extra CDB attributes for the sort, including limit
* and uniqueness. Should do this after begin_heap.
*
*/
void
cdb_tuplesort_init_mk(Tuplesortstate_mk *state,
int64 offset, int64 limit, int unique, int sort_flags,
int64 maxdistinct)
{
UnusedArg(sort_flags);
UnusedArg(maxdistinct);
if(limit)
{
int64 uselimit = offset + limit;
/* Only use limit for less than 10 million */
if(uselimit < 10000000)
{
state->mkctxt.limit = (int32) uselimit;
state->mkctxt.limitmask = -1;
}
}
if(unique)
state->mkctxt.unique = true;
}
/* make a copy of current state pos */
void tuplesort_begin_pos_mk(Tuplesortstate_mk *st, TuplesortPos_mk **pos)
{
TuplesortPos_mk *st_pos;
Assert(st);
st_pos = (TuplesortPos_mk *) palloc(sizeof(TuplesortPos_mk));
memcpy(st_pos, &(st->pos), sizeof(TuplesortPos_mk));
if(st->tapeset)
st_pos->cur_work_tape = LogicalTapeSetDuplicateTape(st->tapeset, st->result_tape);
*pos = st_pos;
}
void
create_mksort_context(
MKContext *mkctxt,
int nkeys,
MKFetchDatumForPrepare fetchForPrep, MKFreeTuple freeTupleFn, TupleDesc tupdesc, bool tbyv, int tlen,
Oid *sortOperators,
AttrNumber *attNums,
ScanKey sk)
{
int i;
Assert(mkctxt);
mkctxt->total_lv = nkeys;
mkctxt->lvctxt = (MKLvContext *) palloc0(sizeof(MKLvContext) * nkeys);
AssertEquivalent(sortOperators == NULL, sk != NULL);
AssertEquivalent(fetchForPrep==NULL, tupdesc==NULL);
mkctxt->fetchForPrep = fetchForPrep;
mkctxt->tupdesc = tupdesc;
if (tupdesc)
mkctxt->mt_bind = create_memtuple_binding(tupdesc);
mkctxt->cpfr = tupsort_cpfr;
mkctxt->freeTup = freeTupleFn;
mkctxt->estimatedExtraForPrep = 0;
lc_guess_strxfrm_scaling_factor(&mkctxt->strxfrmScaleFactor, &mkctxt->strxfrmConstantFactor);
for(i=0; i<nkeys; ++i)
{
RegProcedure sortFunction;
MKLvContext *sinfo = mkctxt->lvctxt+i;
if (sortOperators)
{
Assert(sortOperators[i] != 0);
/* Select a sort function */
SelectSortFunction(sortOperators[i], &sortFunction, &sinfo->sortfnkind);
fmgr_info(sortFunction, &sinfo->fmgrinfo);
/* Hacking in null first/null last. Untill parser implements null first/last,
* we will keep the old postgres behaviour.
*/
if(sinfo->sortfnkind == SORTFUNC_REVLT || sinfo->sortfnkind == SORTFUNC_REVCMP)
sinfo->nullfirst = true;
else
sinfo->nullfirst = false;
}
else
{
sinfo->sortfnkind = SORTFUNC_CMP;
sinfo->nullfirst = false;
sinfo->fmgrinfo = sk[i].sk_func; /* structural copy */
}
AssertImply(attNums, attNums[i] != 0);
/* Per lv context info. Need later for prep */
sinfo->attno = attNums ? attNums[i] : i+1;
sinfo->lvtype = MKLV_TYPE_NONE;
if (tupdesc)
{
sinfo->typByVal = tupdesc->attrs[sinfo->attno-1]->attbyval;
sinfo->typLen = tupdesc->attrs[sinfo->attno-1]->attlen;
if (sinfo->fmgrinfo.fn_addr == btint4cmp)
sinfo->lvtype = MKLV_TYPE_INT32;
if (!lc_collate_is_c())
{
if(sinfo->fmgrinfo.fn_addr == bpcharcmp)
sinfo->lvtype = MKLV_TYPE_CHAR;
else if (sinfo->fmgrinfo.fn_addr == bttextcmp)
sinfo->lvtype = MKLV_TYPE_TEXT;
}
}
else
{
sinfo->typByVal = tbyv;
sinfo->typLen = tlen;
}
sinfo->mkctxt = mkctxt;
}
}
Tuplesortstate_mk *
tuplesort_begin_heap_mk(ScanState *ss,
TupleDesc tupDesc,
int nkeys,
Oid *sortOperators, AttrNumber *attNums,
int workMem, bool randomAccess)
{
Tuplesortstate_mk *state = tuplesort_begin_common(ss, workMem, randomAccess, true);
MemoryContext oldcontext;
oldcontext = MemoryContextSwitchTo(state->sortcontext);
AssertArg(nkeys > 0);
if (trace_sort)
PG_TRACE3(tuplesort__begin, nkeys, workMem, randomAccess);
state->nKeys = nkeys;
state->copytup = copytup_heap;
state->writetup = writetup_heap;
state->readtup = readtup_heap;
state->tupDesc = tupDesc; /* assume we need not copy tupDesc */
state->mt_bind = create_memtuple_binding(tupDesc);
state->cmpScanKey = NULL;
create_mksort_context(
&state->mkctxt,
nkeys,
tupsort_fetch_datum_mtup,
freetup_heap,
tupDesc, 0, 0,
sortOperators, attNums,
NULL);
MemoryContextSwitchTo(oldcontext);
return state;
}
Tuplesortstate_mk *
tuplesort_begin_heap_file_readerwriter_mk(ScanState *ss,
const char *rwfile_prefix, bool isWriter,
TupleDesc tupDesc,
int nkeys,
Oid *sortOperators, AttrNumber *attNums,
int workMem, bool randomAccess)
{
Tuplesortstate_mk *state;
char statedump[MAXPGPATH];
char full_prefix[MAXPGPATH];
Assert(randomAccess);
int len = snprintf(statedump, sizeof(statedump), "%s/%s_sortstate",
PG_TEMP_FILES_DIR,
rwfile_prefix);
insist_log(len <= MAXPGPATH - 1, "could not generate temporary file name");
len = snprintf(full_prefix, sizeof(full_prefix), "%s/%s",
PG_TEMP_FILES_DIR,
rwfile_prefix);
insist_log(len <= MAXPGPATH - 1, "could not generate temporary file name");
if(isWriter)
{
/*
* Writer is a ordinary tuplesort, except the underlying buf file are named by
* rwfile_prefix.
*/
state = tuplesort_begin_heap_mk(ss, tupDesc, nkeys, sortOperators, attNums, workMem, randomAccess);
state->tapeset_file_prefix = MemoryContextStrdup(state->sortcontext, full_prefix);
state->tapeset_state_file = ExecWorkFile_Create(statedump,
BUFFILE,
true /* delOnClose */ ,
0 /* compressType */ );
Assert(state->tapeset_state_file != NULL);
return state;
}
else
{
/*
* For reader, we really don't know anything about sort op, attNums, etc.
* All the readers cares are the data on the logical tape set. The state
* of the logical tape set has been dumped, so we load it back and that is
* it.
*/
MemoryContext oldctxt;
state = tuplesort_begin_common(ss, workMem, randomAccess, false);
state->status = TSS_SORTEDONTAPE;
state->randomAccess = true;
state->readtup = readtup_heap;
oldctxt = MemoryContextSwitchTo(state->sortcontext);
state->tapeset_file_prefix = MemoryContextStrdup(state->sortcontext, full_prefix);
state->tapeset_state_file = ExecWorkFile_Open(statedump,
BUFFILE,
false /* delOnClose */,
0 /* compressType */);
ExecWorkFile *tapefile = ExecWorkFile_Open(full_prefix,
BUFFILE,
false /* delOnClose */,
0 /* compressType */);
state->tapeset = LoadLogicalTapeSetState(state->tapeset_state_file, tapefile);
state->currentRun = 0;
state->result_tape = LogicalTapeSetGetTape(state->tapeset, 0);
state->pos.eof_reached =false;
state->pos.markpos.tapepos.blkNum = 0;
state->pos.markpos.tapepos.offset = 0;
state->pos.markpos.mempos = 0;
state->pos.markpos_eof = false;
state->pos.cur_work_tape = NULL;
MemoryContextSwitchTo(oldctxt);
return state;
}
}
Tuplesortstate_mk *
tuplesort_begin_index_mk(Relation indexRel,
bool enforceUnique,
int workMem, bool randomAccess)
{
Tuplesortstate_mk *state = tuplesort_begin_common(NULL, workMem, randomAccess, true);
MemoryContext oldcontext;
TupleDesc tupdesc;
oldcontext = MemoryContextSwitchTo(state->sortcontext);
if (trace_sort)
PG_TRACE3(tuplesort__begin, enforceUnique, workMem, randomAccess);
state->nKeys = RelationGetNumberOfAttributes(indexRel);
tupdesc = RelationGetDescr(indexRel);
state->copytup = copytup_index;
state->writetup = writetup_index;
state->readtup = readtup_index;
state->indexRel = indexRel;
state->cmpScanKey = _bt_mkscankey_nodata(indexRel);
create_mksort_context(
&state->mkctxt,
state->nKeys,
tupsort_fetch_datum_itup,
freetup_index,
tupdesc, 0, 0,
NULL, NULL, state->cmpScanKey);
state->mkctxt.enforceUnique = enforceUnique;
MemoryContextSwitchTo(oldcontext);
return state;
}
Tuplesortstate_mk *
tuplesort_begin_datum_mk(ScanState * ss,
Oid datumType,
Oid sortOperator,
int workMem, bool randomAccess)
{
Tuplesortstate_mk *state = tuplesort_begin_common(ss, workMem, randomAccess, true);
MemoryContext oldcontext;
int16 typlen;
bool typbyval;
oldcontext = MemoryContextSwitchTo(state->sortcontext);
if (trace_sort)
PG_TRACE3(tuplesort__begin, datumType, workMem, randomAccess);
state->nKeys = 1; /* always a one-column sort */
state->copytup = copytup_datum;
state->writetup = writetup_datum;
state->readtup = readtup_datum;
state->datumType = datumType;
/* lookup necessary attributes of the datum type */
get_typlenbyval(datumType, &typlen, &typbyval);
state->datumTypeLen = typlen;
state->datumTypeByVal = typbyval;
state->sortOperator = sortOperator;
state->cmpScanKey = NULL;
create_mksort_context(
&state->mkctxt,
1,
NULL, /* tupsort_prepare_datum, */
typbyval ? freetup_noop : freetup_datum,
NULL, typbyval, typlen,
&sortOperator, NULL, NULL);
MemoryContextSwitchTo(oldcontext);
return state;
}
/*
* tuplesort_end
*
* Release resources and clean up.
*
* NOTE: after calling this, any pointers returned by tuplesort_getXXX are
* pointing to garbage. Be careful not to attempt to use or free such
* pointers afterwards!
*/
void
tuplesort_end_mk(Tuplesortstate_mk *state)
{
long spaceUsed;
if (state->tapeset)
spaceUsed = LogicalTapeSetBlocks(state->tapeset);
else
spaceUsed = (MemoryContextGetCurrentSpace(state->sortcontext) + 1024) / 1024;
/*
* Delete temporary "tape" files, if any.
*
* Note: want to include this in reported total cost of sort, hence need
* for two #ifdef TRACE_SORT sections.
*/
if (state->tapeset)
{
LogicalTapeSetClose(state->tapeset, state->work_set);
state->tapeset = NULL;
if (state->tapeset_state_file)
{
workfile_mgr_close_file(state->work_set, state->tapeset_state_file);
}
}
if (state->work_set)
{
workfile_mgr_close_set(state->work_set);
}
tuplesort_finalize_stats_mk(state);
if (trace_sort)
PG_TRACE2(tuplesort__end, state->tapeset ? 1 : 0, spaceUsed);
/*
* Free the per-sort memory context, thereby releasing all working memory,
* including the Tuplesortstate_mk struct itself.
*/
MemoryContextDelete(state->sortcontext);
}
/*
* tuplesort_finalize_stats_mk
*
* Finalize the EXPLAIN ANALYZE stats.
*/
void
tuplesort_finalize_stats_mk(Tuplesortstate_mk *state)
{
if (state->instrument && !state->statsFinalized)
{
Size maxSpaceUsedOnSort = MemoryContextGetPeakSpace(state->sortcontext);
/* Report executor memory used by our memory context. */
state->instrument->execmemused += (double)maxSpaceUsedOnSort;
if (state->instrument->workmemused < maxSpaceUsedOnSort)
{
state->instrument->workmemused = maxSpaceUsedOnSort;
}
if (state->numTuplesInMem < state->totalNumTuples)
{
uint64 mem_for_metadata = sizeof(Tuplesortstate_mk) +
state->arraySizeBeforeSpill * sizeof(MKEntry);
double tupleRatio = ((double)state->totalNumTuples) / ((double)state->numTuplesInMem);
/*
* The memwanted is summed up of the following:
* (1) metadata
* (2) the array size
* (3) the prorated number of bytes for all tuples, estimated from
* the memUsedBeforeSpill. Note that because of our memory allocation
* algorithm, the used memory for tuples may be much larger than
* the actual bytes needed for tuples.
* (4) the prorated number of bytes for extra space needed.
*/
uint64 memwanted =
sizeof(Tuplesortstate_mk) /* (1) */ +
state->totalNumTuples * sizeof(MKEntry) /* (2) */ +
(uint64)(tupleRatio * (double)(state->memUsedBeforeSpill - mem_for_metadata)) /* (3) */ +
(uint64)(tupleRatio * (double)state->mkctxt.estimatedExtraForPrep) /* (4) */ ;
state->instrument->workmemwanted =
Max(state->instrument->workmemwanted, memwanted);
}
state->statsFinalized = true;
}
}
/*
* tuplesort_set_instrument
*
* May be called after tuplesort_begin_xxx() to enable reporting of
* statistics and events for EXPLAIN ANALYZE.
*
* The 'instr' and 'explainbuf' ptrs are retained in the 'state' object for
* possible use anytime during the sort, up to and including tuplesort_end().
* The caller must ensure that the referenced objects remain allocated and
* valid for the life of the Tuplesortstate_mk object; or if they are to be
* freed early, disconnect them by calling again with NULL pointers.
*/
void
tuplesort_set_instrument_mk(Tuplesortstate_mk *state,
struct Instrumentation *instrument,
struct StringInfoData *explainbuf)
{
state->instrument = instrument;
state->explainbuf = explainbuf;
} /* tuplesort_set_instrument */
/*
* Accept one tuple while collecting input data for sort.
*
* Note that the input data is always copied; the caller need not save it.
*/
void
tuplesort_puttupleslot_mk(Tuplesortstate_mk *state, TupleTableSlot *slot)
{
MemoryContext oldcontext = MemoryContextSwitchTo(state->sortcontext);
MKEntry e;
mke_blank(&e);
COPYTUP(state, &e, (void *) slot);
puttuple_common(state, &e);
MemoryContextSwitchTo(oldcontext);
}
/*
* Accept one index tuple while collecting input data for sort.
*
* Note that the input tuple is always copied; the caller need not save it.
*/
void
tuplesort_putindextuple_mk(Tuplesortstate_mk *state, IndexTuple tuple)
{
MemoryContext oldcontext = MemoryContextSwitchTo(state->sortcontext);
MKEntry e;
mke_blank(&e);
COPYTUP(state, &e, (void *) tuple);
puttuple_common(state, &e);
MemoryContextSwitchTo(oldcontext);
}
/*
* Accept one Datum while collecting input data for sort.
*
* If the Datum is pass-by-ref type, the value will be copied.
*/
void
tuplesort_putdatum_mk(Tuplesortstate_mk *state, Datum val, bool isNull)
{
MemoryContext oldcontext = MemoryContextSwitchTo(state->sortcontext);
MKEntry e;
mke_blank(&e);
/*
* If it's a pass-by-reference value, copy it into memory we control, and
* decrease availMem. Then call the common code.
*/
if (isNull || state->datumTypeByVal)
{
e.d = val;
if(isNull)
mke_set_null(&e, state->nullfirst);
else
mke_set_not_null(&e);
}
else
{
mke_set_not_null(&e);
e.d = datumCopy(val, false, state->datumTypeLen);
state->totalTupleBytes += state->datumTypeLen;
}
puttuple_common(state, &e);
MemoryContextSwitchTo(oldcontext);
}
/*
* grow_unsorted_array
* Grow the unsorted array to allow more entries to be inserted later.
* If there are no enough memory available for the growth, this function
* returns false. Otherwsie, returns true.
*/
static bool
grow_unsorted_array(Tuplesortstate_mk *state)
{
/*
* We want to grow the array twice as large as the previous one. However,
* when we are close to the memory limit, we do not want to do so. Otherwise,
* too much memory got allocated for the metadata, not to the tuple itself.
* We estimate the maximum number of entries that is possible under the
* current memory limit by considering both metadata and tuple size.
*/
if (state->memAllowed < MemoryContextGetCurrentSpace(state->sortcontext))
return false;
uint64 availMem = state->memAllowed - MemoryContextGetCurrentSpace(state->sortcontext);
uint64 avgTupSize = (uint64)(((double)state->totalTupleBytes) / ((double)state->totalNumTuples));
Assert(avgTupSize >= 0);
uint64 avgExtraForPrep = (uint64) (((double)state->mkctxt.estimatedExtraForPrep) / ((double)state->totalNumTuples));
if ((availMem / (sizeof(MKEntry) + avgTupSize + avgExtraForPrep)) == 0)
return false;
int maxNumEntries = state->entry_allocsize + (availMem / (sizeof(MKEntry) + avgTupSize + avgExtraForPrep));
int newNumEntries = Min(maxNumEntries, state->entry_allocsize * 2);
state->entries = (MKEntry *)repalloc(state->entries, newNumEntries * sizeof(MKEntry));
for (int entryNo = state->entry_allocsize; entryNo < newNumEntries; entryNo++)
mke_blank(state->entries + entryNo);
state->entry_allocsize = newNumEntries;
return true;
}
/*
* Shared code for tuple and datum cases.
*/
static void
puttuple_common(Tuplesortstate_mk *state, MKEntry *e)
{
Assert(is_under_sort_ctxt(state));
state->totalNumTuples++;
if(state->gpmon_pkt)
Gpmon_M_Incr(state->gpmon_pkt, GPMON_QEXEC_M_ROWSIN);
bool growSucceed = true;
switch (state->status)
{
case TSS_INITIAL:
/*
* Save the tuple into the unsorted array. First, grow the array
* as needed. Note that we try to grow the array when there is
* still one free slot remaining --- if we fail, there'll still be
* room to store the incoming tuple, and then we'll switch to
* tape-based operation.
*/
if (!state->mkheap && state->entry_count >= state->entry_allocsize - 1)
{
growSucceed = grow_unsorted_array(state);
}
/* full sort? */
if (state->mkctxt.limit == 0)
{
tuplesort_inmem_nolimit_insert(state,e);
}
else
{
/* Limit sort insert */
tuplesort_inmem_limit_insert(state,e);
}
/* If out of work_mem, switch to diskmode */
if(!growSucceed)
{
if (state->mkheap != NULL)
{
/* Corner case. LIMIT == amount of entries in memory.
* In this case, we failed to grow array, but we just created
* the heap. No need to spill in this case, we'll just use the in-memory heap. */
Assert(state->mkctxt.limit != 0);
Assert(state->numTuplesInMem == state->mkheap->count);
}
else
{
Assert(state->mkheap == NULL);
Assert(state->entry_count > 0);
state->arraySizeBeforeSpill = state->entry_allocsize;
state->memUsedBeforeSpill = MemoryContextGetPeakSpace(state->sortcontext);
inittapes_mk(state, is_sortstate_rwfile(state) ? state->tapeset_file_prefix : NULL);
Assert(state->status == TSS_BUILDRUNS);
}
if (state->instrument)
{
state->instrument->workfileCreated = true;
}
}
break;
case TSS_BUILDRUNS:
/*
* Insert the tuple into the heap
*/
Assert(state->mkheap);
tuplesort_heap_insert(state, e);
break;
default:
elog(ERROR, "invalid tuplesort state");
break;
}
}
/*
* All tuples have been provided; finish the sort.
*/
void
tuplesort_performsort_mk(Tuplesortstate_mk *state)
{
MemoryContext oldcontext = MemoryContextSwitchTo(state->sortcontext);
if (trace_sort)
PG_TRACE(tuplesort__perform__sort);
switch (state->status)
{
case TSS_INITIAL:
/*
* We were able to accumulate all the tuples within the allowed
* amount of memory. Just qsort 'em and we're done.
*/
if(state->mkctxt.limit == 0)
mk_qsort(state->entries, state->entry_count, &state->mkctxt);
else
tuplesort_limit_sort(state);
state->pos.current = 0;
state->pos.eof_reached = false;
state->pos.markpos.mempos = 0;
state->pos.markpos_eof = false;
state->status = TSS_SORTEDINMEM;
/* Not shareinput sort, we are done. */
if(!is_sortstate_rwfile(state))
break;
/* Shareinput sort, need to put this stuff onto disk */
inittapes_mk(state, state->tapeset_file_prefix);
/* Fall through */
case TSS_BUILDRUNS:
#ifdef PRINT_SPILL_AND_MEMORY_MESSAGES
elog(INFO, "Done building runs. Mem peak is now %ld", (long)MemoryContextGetPeakSpace(state->sortcontext));
#endif // PRINT_SPILL_AND_MEMORY_MESSAGES
dumptuples_mk(state, true);
#ifdef PRINT_SPILL_AND_MEMORY_MESSAGES
elog(INFO, "Done tuple dump. Mem peak is now %ld", (long)MemoryContextGetPeakSpace(state->sortcontext));
#endif // PRINT_SPILL_AND_MEMORY_MESSAGES
HOLD_INTERRUPTS();
/* MPP-18288: Do not change this log message, it is used to test mksort query cancellation */
elog(DEBUG1,"ExecSort: mksort starting merge runs >>======== ");
RESUME_INTERRUPTS();
mergeruns(state);
HOLD_INTERRUPTS();
/* MPP-18288: Do not change this log message, it is used to test mksort query cancellation */
elog(DEBUG1, "ExecSort: mksort finished merge runs ++++++++>> ");
RESUME_INTERRUPTS();
state->pos.eof_reached = false;
state->pos.markpos.tapepos.blkNum = 0;
state->pos.markpos.tapepos.offset = 0;
state->pos.markpos_eof = false;
/*
* If we're planning to reuse the spill files from this sort,
* save metadata here and mark work_set complete.
*/
if (gp_workfile_caching && state->work_set)
{
tuplesort_write_spill_metadata_mk(state);
/* We don't know how to handle TSS_FINALMERGE yet */
Assert(state->status == TSS_SORTEDONTAPE);
Assert(state->work_set);
workfile_mgr_mark_complete(state->work_set);
}
break;
default:
elog(ERROR, "Invalid tuplesort state");
break;
}
MemoryContextSwitchTo(oldcontext);
}
void tuplesort_flush_mk(Tuplesortstate_mk *state)
{
Assert(state->status == TSS_SORTEDONTAPE);
Assert(state->tapeset && state->tapeset_state_file);
Assert(state->pos.cur_work_tape == NULL);
elog(gp_workfile_caching_loglevel, "tuplesort_mk: writing logical tape state to file");
LogicalTapeFlush(state->tapeset, state->result_tape, state->tapeset_state_file);
ExecWorkFile_Flush(state->tapeset_state_file);
}
/*
* Internal routine to fetch the next tuple in either forward or back
* direction into *stup. Returns FALSE if no more tuples.
* If *should_free is set, the caller must pfree stup.tuple when done with it.
*/
static bool
tuplesort_gettuple_common_pos(Tuplesortstate_mk *state, TuplesortPos_mk *pos,
bool forward, MKEntry *e, bool *should_free)
{
uint32 tuplen;
LogicalTape *work_tape;
bool fOK;
Assert(is_under_sort_ctxt(state));
switch (state->status)
{
case TSS_SORTEDINMEM:
Assert(forward || state->randomAccess);
*should_free = false;
if (forward)
{
if ( state->mkctxt.unique)
{
/**
* request to skip duplicates. qsort has already replaced all but one of each
* values with empty so skip the empties
* When limit heap sort was used, the limit heap sort should have done this
*/
while ( pos->current < state->entry_count &&
mke_is_empty( &state->entries[pos->current]))
{
pos->current++;
}
}
if (pos->current < state->entry_count)
{
*e = state->entries[pos->current];
pos->current++;
return true;
}
pos->eof_reached = true;
return false;
}
else
{
if (pos->current <= 0)
return false;
/*
* if all tuples are fetched already then we return last
* tuple, else - tuple before last returned.
*/
if (pos->eof_reached)
pos->eof_reached = false;
else
{
pos->current--; /* pos->current points to one above the last returned tuple */
if (pos->current <= 0)
return false;
}
if ( state->mkctxt.unique)
{
/**
* request to skip duplicates. qsort has already replaced all but one of each
* values with empty so skip the empties
* When limit heap sort was used, the limit heap sort should have done this
*/
while ( pos->current - 1 >= 0 &&
mke_is_empty( &state->entries[pos->current - 1]))
{
pos->current--;
}
if ( pos->current <= 0)
return false;
}
*e = state->entries[pos->current-1];
return true;
}
break;
case TSS_SORTEDONTAPE:
AssertEquivalent((pos == &state->pos), (pos->cur_work_tape == NULL));
Assert(forward || state->randomAccess);
*should_free = true;
work_tape = pos->cur_work_tape == NULL ? state->result_tape : pos->cur_work_tape;
if (forward)
{
if (pos->eof_reached)
return false;
if ((tuplen = getlen(state, pos, work_tape, true)) != 0)
{
READTUP(state, pos, e, work_tape, tuplen);
return true;
}
else
{
pos->eof_reached = true;
return false;
}
}
/*
* Backward.
*
* if all tuples are fetched already then we return last tuple,
* else - tuple before last returned.
*/
/*
* Seek position is pointing just past the zero tuplen at the
* end of file; back up to fetch last tuple's ending length
* word. If seek fails we must have a completely empty file.
*/
fOK = LogicalTapeBackspace(state->tapeset, work_tape, 2*sizeof(uint32));
if(!fOK)
return false;
if (pos->eof_reached)
{
pos->eof_reached = false;
}
else
{
tuplen = getlen(state, pos, work_tape, false);
/*
* Back up to get ending length word of tuple before it.
*/
fOK = LogicalTapeBackspace(state->tapeset, work_tape, tuplen + 2 *sizeof(uint32));
if (!fOK)
{
/*
* If that fails, presumably the prev tuple is the first
* in the file. Back up so that it becomes next to read
* in forward direction (not obviously right, but that is
* what in-memory case does).
*/
fOK = LogicalTapeBackspace(state->tapeset, work_tape, tuplen + 2 *sizeof(uint32));
if(!fOK)
elog(ERROR, "bogus tuple length in backward scan");
return false;
}
}
tuplen = getlen(state, pos, work_tape, false);
/*
* Now we have the length of the prior tuple, back up and read it.
* Note: READTUP expects we are positioned after the initial
* length word of the tuple, so back up to that point.
*/
fOK = LogicalTapeBackspace(state->tapeset, work_tape, tuplen);
if (!fOK)
elog(ERROR, "bogus tuple length in backward scan");
READTUP(state, pos, e, work_tape, tuplen);
return true;
case TSS_FINALMERGE:
Assert(forward);
Assert(pos == &state->pos && pos->cur_work_tape == NULL);
*should_free = true;
mkheap_putAndGet(state->mkheap, e);
return !mke_is_empty(e);
default:
elog(ERROR, "invalid tuplesort state");
return false; /* keep compiler quiet */
}
}
/*
* Fetch the next tuple in either forward or back direction.
* If successful, put tuple in slot and return TRUE; else, clear the slot
* and return FALSE.
*/
bool
tuplesort_gettupleslot_mk(Tuplesortstate_mk *state, bool forward,
TupleTableSlot *slot)
{
return tuplesort_gettupleslot_pos_mk(state, &state->pos, forward, slot);
}
bool
tuplesort_gettupleslot_pos_mk(Tuplesortstate_mk *state, TuplesortPos_mk *pos,
bool forward, TupleTableSlot *slot)
{
MemoryContext oldcontext = MemoryContextSwitchTo(state->sortcontext);
MKEntry e;
bool should_free = false;
bool fOK;
mke_set_empty(&e);
fOK = tuplesort_gettuple_common_pos(state, pos, forward, &e, &should_free);
MemoryContextSwitchTo(oldcontext);
if (fOK)
{
Assert(!mke_is_empty(&e));
ExecStoreMemTuple(e.ptr, slot, should_free);
#ifdef USE_ASSERT_CHECKING
if (should_free && state->mkheap != NULL && state->mkheap->count > 0)
{
Assert(e.ptr != NULL &&
(e.ptr != state->mkheap->lvtops->ptr));
}
#endif
if(state->gpmon_pkt)
Gpmon_M_Incr_Rows_Out(state->gpmon_pkt);
return true;
}
ExecClearTuple(slot);
return false;
}
/*
* Fetch the next index tuple in either forward or back direction.
* Returns NULL if no more tuples. If *should_free is set, the
* caller must pfree the returned tuple when done with it.
*/
IndexTuple
tuplesort_getindextuple_mk(Tuplesortstate_mk *state, bool forward,
bool *should_free)
{
MemoryContext oldcontext = MemoryContextSwitchTo(state->sortcontext);
MKEntry e;
bool fOK = tuplesort_gettuple_common_pos(state, &state->pos, forward, &e, should_free);
MemoryContextSwitchTo(oldcontext);
if(fOK)
return (IndexTuple) (e.ptr);
return NULL;
}
/*
* Fetch the next Datum in either forward or back direction.
* Returns FALSE if no more datums.
*
* If the Datum is pass-by-ref type, the returned value is freshly palloc'd
* and is now owned by the caller.
*/
bool
tuplesort_getdatum_mk(Tuplesortstate_mk *state, bool forward,
Datum *val, bool *isNull)
{
MemoryContext oldcontext = MemoryContextSwitchTo(state->sortcontext);
MKEntry e;
bool should_free = false;
bool fOK = tuplesort_gettuple_common_pos(state, &state->pos, forward, &e, &should_free);
if(fOK)
{
*isNull = mke_is_null(&e);
if(*isNull || state->datumTypeByVal || should_free)
*val = e.d;
else
*val = datumCopy(e.d, false, state->datumTypeLen);
}
MemoryContextSwitchTo(oldcontext);
return fOK;
}
/*
* tuplesort_merge_order - report merge order we'll use for given memory
* (note: "merge order" just means the number of input tapes in the merge).
*
* This is exported for use by the planner. allowedMem is in bytes.
*/
int
tuplesort_merge_order(long allowedMem)
{
int mOrder;
/*
* We need one tape for each merge input, plus another one for the output,
* and each of these tapes needs buffer space. In addition we want
* MERGE_BUFFER_SIZE workspace per input tape (but the output tape doesn't
* count).
*
* Note: you might be thinking we need to account for the memtuples[]
* array in this calculation, but we effectively treat that as part of the
* MERGE_BUFFER_SIZE workspace.
*/
mOrder = (allowedMem - TAPE_BUFFER_OVERHEAD) /
(MERGE_BUFFER_SIZE + TAPE_BUFFER_OVERHEAD);
mOrder = Max(mOrder, MINORDER);
mOrder = Min(mOrder, MAXORDER);
return mOrder;
}
/*
* inittapes - initialize for tape sorting.
*
* This is called only if we have found we don't have room to sort in memory.
*/
void
inittapes_mk(Tuplesortstate_mk *state, const char* rwfile_prefix)
{
int maxTapes;
int j;
long tapeSpace;
Assert(is_under_sort_ctxt(state));
/* Compute number of tapes to use: merge order plus 1 */
maxTapes = tuplesort_merge_order(state->memAllowed) + 1;
#ifdef PRINT_SPILL_AND_MEMORY_MESSAGES
elog(INFO, "Spilling after %d", (int) state->entry_count);
#endif // PRINT_SPILL_AND_MEMORY_MESSAGES
/*
* We must have at least 2*maxTapes slots in the memtuples[] array, else
* we'd not have room for merge heap plus preread. It seems unlikely that
* this case would ever occur, but be safe.
*/
maxTapes = Min(maxTapes, state->entry_count / 2);
maxTapes = Max(maxTapes, 1);
#ifdef PRINT_SPILL_AND_MEMORY_MESSAGES
elog(INFO, " maxtapes %d Mem peak is now %ld", maxTapes, (long)MemoryContextGetPeakSpace(state->sortcontext));
#endif // PRINT_SPILL_AND_MEMORY_MESSAGES
/* XXX XXX: with losers, only need 1x slots because we don't need a merge heap */
state->maxTapes = maxTapes;
state->tapeRange = maxTapes - 1;
if (trace_sort)
PG_TRACE1(tuplesort__switch__external, maxTapes);
/*
* Decrease availMem to reflect the space needed for tape buffers; but
* don't decrease it to the point that we have no room for tuples. (That
* case is only likely to occur if sorting pass-by-value Datums; in all
* other scenarios the memtuples[] array is unlikely to occupy more than
* half of allowedMem. In the pass-by-value case it's not important to
* account for tuple space, so we don't care if LACKMEM becomes
* inaccurate.)
*/
tapeSpace = maxTapes * TAPE_BUFFER_OVERHEAD;
Assert(state->work_set == NULL);
PlanState *ps = NULL;
bool can_be_reused = false;
if (state->ss != NULL)
{
ps = &state->ss->ps;
Sort *node = (Sort *) ps->plan;
if (node->share_type == SHARE_NOTSHARED)
{
/* Only attempt to cache when not shared under a ShareInputScan */
can_be_reused = true;
}
}
/*
* Create the tape set and allocate the per-tape data arrays.
*/
if(!rwfile_prefix)
{
state->work_set = workfile_mgr_create_set(BUFFILE, can_be_reused, ps, NULL_SNAPSHOT);
state->tapeset_state_file = workfile_mgr_create_fileno(state->work_set, WORKFILE_NUM_MKSORT_METADATA);
ExecWorkFile *tape_file = workfile_mgr_create_fileno(state->work_set, WORKFILE_NUM_MKSORT_TAPESET);
state->tapeset = LogicalTapeSetCreate_File(tape_file, maxTapes);
}
else
{
/* We are shared XSLICE, use given prefix to create files so that consumers can find them */
ExecWorkFile *tape_file = ExecWorkFile_Create(rwfile_prefix,
BUFFILE,
true /* delOnClose */,
0 /* compressType */);
state->tapeset = LogicalTapeSetCreate_File(tape_file, maxTapes);
}
state->mergeactive = (bool *) palloc0(maxTapes * sizeof(bool));
state->mergenext = (int *) palloc0(maxTapes * sizeof(int));
state->mergelast = (int *) palloc0(maxTapes * sizeof(int));
state->mergeavailslots = (int *) palloc0(maxTapes * sizeof(int));
state->mergeavailmem = (long *) palloc0(maxTapes * sizeof(long));
state->tp_fib = (int *) palloc0(maxTapes * sizeof(int));
state->tp_runs = (int *) palloc0(maxTapes * sizeof(int));
state->tp_dummy = (int *) palloc0(maxTapes * sizeof(int));
state->tp_tapenum = (int *) palloc0(maxTapes * sizeof(int));
/*
* Convert the unsorted contents of memtuples[] into a heap. Each tuple is
* marked as belonging to run number zero.
*
* NOTE: we pass false for checkIndex since there's no point in comparing
* indexes in this step, even though we do intend the indexes to be part
* of the sort key...
*/
Assert(state->mkheap == NULL);
Assert(state->status == TSS_INITIAL || state->status == TSS_SORTEDINMEM);
#ifdef PRINT_SPILL_AND_MEMORY_MESSAGES
elog(INFO, " about to make heap mem peak is now %ld", (long)MemoryContextGetPeakSpace(CurrentMemoryContext));
#endif // PRINT_SPILL_AND_MEMORY_MESSAGES
if(state->status == TSS_INITIAL)
{
/*
* We are now building heap from array for run-building using
* replacement-selection algorithm. Such run-building heap need
* to be a MIN-HEAP, but for limit sorting, we use a MAX-HEAP. The way we
* convert MIN-HEAP to MAX-HEAP is by setting the limitmask to -1, and then
* using the limitmask in mkheap_compare() in tuplesort_mkheap.c. So, to restore
* the MAX-HEAP to a MIN-HEAP, we can just revert the limitmask to 0.
* This is needed for MPP-19310 and MPP-19857
*/
state->mkctxt.limitmask = 0;
state->mkheap = mkheap_from_array(state->entries, state->entry_allocsize, state->entry_count, &state->mkctxt);
state->entries = NULL;
state->entry_allocsize = 0;
state->entry_count = 0;
}
state->currentRun = 0;
/*
* Initialize variables of Algorithm D (step D1).
*/
for (j = 0; j < maxTapes; j++)
{
state->tp_fib[j] = 1;
state->tp_runs[j] = 0;
state->tp_dummy[j] = 1;
state->tp_tapenum[j] = j;
}
state->tp_fib[state->tapeRange] = 0;
state->tp_dummy[state->tapeRange] = 0;
state->Level = 1;
state->destTape = 0;
state->status = TSS_BUILDRUNS;
#ifdef PRINT_SPILL_AND_MEMORY_MESSAGES
elog(INFO, " build-run ready mem peak is now %ld", (long)MemoryContextGetPeakSpace(state->sortcontext));
#endif // PRINT_SPILL_AND_MEMORY_MESSAGES
}
/*
* selectnewtape -- select new tape for new initial run.
*
* This is called after finishing a run when we know another run
* must be started. This implements steps D3, D4 of Algorithm D.
*/
static void
selectnewtape_mk(Tuplesortstate_mk *state)
{
int j;
int a;
/* Step D3: advance j (destTape) */
if (state->tp_dummy[state->destTape] < state->tp_dummy[state->destTape + 1])
{
state->destTape++;
return;
}
if (state->tp_dummy[state->destTape] != 0)
{
state->destTape = 0;
return;
}
/* Step D4: increase level */
state->Level++;
a = state->tp_fib[0];
for (j = 0; j < state->tapeRange; j++)
{
state->tp_dummy[j] = a + state->tp_fib[j + 1] - state->tp_fib[j];
state->tp_fib[j] = a + state->tp_fib[j + 1];
}
state->destTape = 0;
}
/*
* mergeruns -- merge all the completed initial runs.
*
* This implements steps D5, D6 of Algorithm D. All input data has
* already been written to initial runs on tape (see dumptuples).
*/
static void
mergeruns(Tuplesortstate_mk *state)
{
int tapenum,
svTape,
svRuns,
svDummy;
LogicalTape *lt = NULL;
Assert(state->status == TSS_BUILDRUNS);
#ifdef FAULT_INJECTOR
/*
* MPP-18288: We're injecting an interrupt here. We have to hold interrupts
* while we're injecting it to make sure the interrupt is not handled
* within the fault injector itself.
*/
HOLD_INTERRUPTS();
FaultInjector_InjectFaultIfSet(
ExecSortMKSortMergeRuns,
DDLNotSpecified,
"", // databaseName
""); // tableName
RESUME_INTERRUPTS();
#endif
/*
* If we produced only one initial run (quite likely if the total data
* volume is between 1X and 2X workMem), we can just use that tape as the
* finished output, rather than doing a useless merge. (This obvious
* optimization is not in Knuth's algorithm.)
*/
if (state->currentRun == 1)
{
state->result_tape = LogicalTapeSetGetTape(state->tapeset, state->tp_tapenum[state->destTape]);
/* must freeze and rewind the finished output tape */
LogicalTapeFreeze(state->tapeset, state->result_tape);
state->status = TSS_SORTEDONTAPE;
return;
}
/* End of step D2: rewind all output tapes to prepare for merging */
for (tapenum = 0; tapenum < state->tapeRange; tapenum++)
{
lt = LogicalTapeSetGetTape(state->tapeset, tapenum);
LogicalTapeRewind(state->tapeset, lt, false);
}
/* Clear gpmon for respilling data */
if(state->gpmon_pkt)
{
Gpmon_M_Incr(state->gpmon_pkt, GPMON_SORT_SPILLPASS);
Gpmon_M_Reset(state->gpmon_pkt, GPMON_SORT_CURRSPILLPASS_TUPLE);
Gpmon_M_Reset(state->gpmon_pkt, GPMON_SORT_CURRSPILLPASS_BYTE);
}
for (;;)
{
/*
* At this point we know that tape[T] is empty. If there's just one
* (real or dummy) run left on each input tape, then only one merge
* pass remains. If we don't have to produce a materialized sorted
* tape, we can stop at this point and do the final merge on-the-fly.
*/
/* If workfile caching is enabled, always do the final merging
* and store the sorted result on disk, instead of stopping before the
* last merge iteration.
* This can cause some slowdown compared to no workfile caching, but
* it enables us to re-use the mechanism to dump and restore logical
* tape set information as-is.
*/
if (!state->randomAccess && !gp_workfile_caching)
{
bool allOneRun = true;
Assert(state->tp_runs[state->tapeRange] == 0);
for (tapenum = 0; tapenum < state->tapeRange; tapenum++)
{
if (state->tp_runs[tapenum] + state->tp_dummy[tapenum] != 1)
{
allOneRun = false;
break;
}
}
if (allOneRun)
{
/* Tell logtape.c we won't be writing anymore */
LogicalTapeSetForgetFreeSpace(state->tapeset);
/* Initialize for the final merge pass */
beginmerge(state);
state->status = TSS_FINALMERGE;
return;
}
}
/* Step D5: merge runs onto tape[T] until tape[P] is empty */
while (state->tp_runs[state->tapeRange - 1] ||
state->tp_dummy[state->tapeRange - 1])
{
bool allDummy = true;
for (tapenum = 0; tapenum < state->tapeRange; tapenum++)
{
if (state->tp_dummy[tapenum] == 0)
{
allDummy = false;
break;
}
}
if (allDummy)
{
state->tp_dummy[state->tapeRange]++;
for (tapenum = 0; tapenum < state->tapeRange; tapenum++)
state->tp_dummy[tapenum]--;
}
else
mergeonerun_mk(state);
}
/* Step D6: decrease level */
if (--state->Level == 0)
break;
/* rewind output tape T to use as new input */
lt = LogicalTapeSetGetTape(state->tapeset, state->tp_tapenum[state->tapeRange]);
LogicalTapeRewind(state->tapeset, lt, false);
/* rewind used-up input tape P, and prepare it for write pass */
lt = LogicalTapeSetGetTape(state->tapeset, state->tp_tapenum[state->tapeRange - 1]);
LogicalTapeRewind(state->tapeset, lt, true);
state->tp_runs[state->tapeRange - 1] = 0;
/*
* reassign tape units per step D6; note we no longer care about A[]
*/
svTape = state->tp_tapenum[state->tapeRange];
svDummy = state->tp_dummy[state->tapeRange];
svRuns = state->tp_runs[state->tapeRange];
for (tapenum = state->tapeRange; tapenum > 0; tapenum--)
{
state->tp_tapenum[tapenum] = state->tp_tapenum[tapenum - 1];
state->tp_dummy[tapenum] = state->tp_dummy[tapenum - 1];
state->tp_runs[tapenum] = state->tp_runs[tapenum - 1];
}
state->tp_tapenum[0] = svTape;
state->tp_dummy[0] = svDummy;
state->tp_runs[0] = svRuns;
}
/*
* Done. Knuth says that the result is on TAPE[1], but since we exited
* the loop without performing the last iteration of step D6, we have not
* rearranged the tape unit assignment, and therefore the result is on
* TAPE[T]. We need to do it this way so that we can freeze the final
* output tape while rewinding it. The last iteration of step D6 would be
* a waste of cycles anyway...
*/
state->result_tape = LogicalTapeSetGetTape(state->tapeset, state->tp_tapenum[state->tapeRange]);
LogicalTapeFreeze(state->tapeset, state->result_tape);
state->status = TSS_SORTEDONTAPE;
}
/*
* Merge one run from each input tape, except ones with dummy runs.
*
* This is the inner loop of Algorithm D step D5. We know that the
* output tape is TAPE[T].
*/
void
mergeonerun_mk(Tuplesortstate_mk *state)
{
int destTape = state->tp_tapenum[state->tapeRange];
MKEntry e;
LogicalTape *lt = NULL;
/*
* Start the merge by loading one tuple from each active source tape into
* the heap. We can also decrease the input run/dummy run counts.
*/
beginmerge(state);
/*
* Execute merge by repeatedly extracting lowest tuple in heap, writing it
* out, and replacing it with next tuple from same tape (if there is
* another one).
*/
lt = LogicalTapeSetGetTape(state->tapeset, destTape);
Assert(state->mkheap);
while (mkheap_putAndGet(state->mkheap, &e) >= 0)
WRITETUP(state, lt, &e);
/*
* When the heap empties, we're done. Write an end-of-run marker on the
* output tape, and increment its count of real runs.
*/
markrunend(state, destTape);
state->tp_runs[state->tapeRange]++;
if (trace_sort)
PG_TRACE1(tuplesort__mergeonerun, state->activeTapes);
}
static bool tupsort_preread(TupsortMergeReadCtxt *ctxt)
{
uint32 tuplen;
Assert(ctxt->mem_allowed > 0);
if(!ctxt->pos.cur_work_tape || ctxt->pos.eof_reached)
return false;
ctxt->mem_used = 0;
Assert(ctxt->p && ctxt->allocsz > 0);
Assert(ctxt->mem_allowed > 0);
for(ctxt->cnt=0;
ctxt->cnt<ctxt->allocsz && ctxt->mem_used < ctxt->mem_allowed;
++ctxt->cnt)
{
tuplen = getlen(ctxt->tsstate, &ctxt->pos, ctxt->pos.cur_work_tape, true);
if(tuplen != 0)
{
MKEntry *e = ctxt->p + ctxt->cnt;
READTUP(ctxt->tsstate, &ctxt->pos, e, ctxt->pos.cur_work_tape, tuplen);
ctxt->mem_used += tuplen;
}
else
{
ctxt->pos.eof_reached = true;
break;
}
}
ctxt->cur = 0;
return ctxt->cnt > ctxt->cur;
}
static bool tupsort_mergeread(void *pvctxt, MKEntry *e)
{
TupsortMergeReadCtxt *ctxt = (TupsortMergeReadCtxt *) pvctxt;
Assert(ctxt->mem_allowed > 0);
if(ctxt->cur < ctxt->cnt)
{
*e = ctxt->p[ctxt->cur++];
return true;
}
if(!tupsort_preread(ctxt))
return false;
Assert(ctxt->cur == 0 && ctxt->cnt > 0);
*e = ctxt->p[ctxt->cur++];
return true;
}
/*
* beginmerge - initialize for a merge pass
*
* We decrease the counts of real and dummy runs for each tape, and mark
* which tapes contain active input runs in mergeactive[]. Then, load
* as many tuples as we can from each active input tape, and finally
* fill the merge heap with the first tuple from each active tape.
*/
static void
beginmerge(Tuplesortstate_mk *state)
{
int activeTapes;
int tapenum;
int srcTape;
int totalSlots;
int slotsPerTape;
long spacePerTape;
int i;
MemoryContext oldctxt;
/* Heap should be empty here */
Assert(mkheap_empty(state->mkheap));
/* Adjust run counts and mark the active tapes */
memset(state->mergeactive, 0,
state->maxTapes * sizeof(*state->mergeactive));
activeTapes = 0;
for (tapenum = 0; tapenum < state->tapeRange; tapenum++)
{
if (state->tp_dummy[tapenum] > 0)
state->tp_dummy[tapenum]--;
else
{
Assert(state->tp_runs[tapenum] > 0);
state->tp_runs[tapenum]--;
srcTape = state->tp_tapenum[tapenum];
state->mergeactive[srcTape] = true;
activeTapes++;
}
}
state->activeTapes = activeTapes;
/* Clear merge-pass state variables */
memset(state->mergenext, 0,
state->maxTapes * sizeof(*state->mergenext));
memset(state->mergelast, 0,
state->maxTapes * sizeof(*state->mergelast));
state->mergefreelist = 0; /* nothing in the freelist */
state->mergefirstfree = activeTapes; /* 1st slot avail for preread */
/*
* Initialize space allocation to let each active input tape have an equal
* share of preread space.
*/
Assert(activeTapes > 0);
totalSlots = (state->mkheap == NULL) ? state->entry_allocsize : state->mkheap->alloc_size;
slotsPerTape = (totalSlots - state->mergefirstfree) / activeTapes;
slotsPerTape = Max(slotsPerTape, 128);
spacePerTape = state->memAllowed / activeTapes;
oldctxt = MemoryContextSwitchTo(state->sortcontext);
if(state->mkheap)
{
mkheap_destroy(state->mkheap);
state->mkheap = NULL;
}
if(state->mkhreader)
{
Assert(state->mkhreader_ctxt);
for(i=0; i<state->mkhreader_allocsize; ++i)
{
TupsortMergeReadCtxt *mkhr_ctxt = state->mkhreader_ctxt+i;
AssertEquivalent(mkhr_ctxt->p!=NULL, mkhr_ctxt->allocsz>0);
if(mkhr_ctxt->p)
pfree(mkhr_ctxt->p);
}
pfree(state->mkhreader_ctxt);
pfree(state->mkhreader);
}
state->mkhreader = palloc0(sizeof(MKHeapReader) * activeTapes);
state->mkhreader_ctxt = palloc0(sizeof(TupsortMergeReadCtxt) * activeTapes);
for(i=0; i<activeTapes; ++i)
{
TupsortMergeReadCtxt *mkhr_ctxt = state->mkhreader_ctxt+i;
state->mkhreader[i].reader = tupsort_mergeread;
state->mkhreader[i].mkhr_ctxt = mkhr_ctxt;
mkhr_ctxt->tsstate = state;
}
state->mkhreader_allocsize = activeTapes;
for (i=0, srcTape = 0; srcTape < state->maxTapes; srcTape++)
{
if (state->mergeactive[srcTape])
{
TupsortMergeReadCtxt *mkhr_ctxt = state->mkhreader_ctxt+i;
mkhr_ctxt->pos.cur_work_tape = LogicalTapeSetGetTape(state->tapeset, srcTape);
mkhr_ctxt->pos.eof_reached = false;
mkhr_ctxt->mem_allowed = spacePerTape;
Assert(mkhr_ctxt->mem_allowed > 0);
mkhr_ctxt->mem_used = 0;
mkhr_ctxt->cur = 0;
mkhr_ctxt->cnt = 0;
Assert(mkhr_ctxt->p == NULL);
Assert(slotsPerTape > 0);
mkhr_ctxt->p = (MKEntry *) palloc(sizeof(MKEntry) * slotsPerTape);
mkhr_ctxt->allocsz = slotsPerTape;
++i;
}
}
Assert(i==activeTapes);
state->mkheap = mkheap_from_reader(state->mkhreader, i, &state->mkctxt);
Assert(state->mkheap);
MemoryContextSwitchTo(oldctxt);
}
/*
* dumptuples - remove tuples from heap and write to tape
*
* This is used during initial-run building, but not during merging.
*
* When alltuples = false, dump only enough tuples to get under the
* availMem limit (and leave at least one tuple in the heap in any case,
* since puttuple assumes it always has a tuple to compare to). We also
* insist there be at least one free slot in the memtuples[] array.
*
* When alltuples = true, dump everything currently in memory.
* (This case is only used at end of input data.)
*
* If we empty the heap, close out the current run and return (this should
* only happen at end of input data). If we see that the tuple run number
* at the top of the heap has changed, start a new run.
*/
void
dumptuples_mk(Tuplesortstate_mk *state, bool alltuples)
{
int i;
int bDumped = 0;
LogicalTape *lt = NULL;
MKEntry e;
Assert(is_under_sort_ctxt(state));
if(alltuples && !state->mkheap)
{
/* Dump a sorted array. Must be shared input that sends up here */
/* ShareInput or sort: The sort may sort nothing, we still
* need to handle it here
*/
if(state->entry_count != 0)
{
lt = LogicalTapeSetGetTape(state->tapeset, state->tp_tapenum[state->destTape]);
if ( state->mkctxt.unique)
{
for(i=0; i<state->entry_count; ++i)
if ( ! mke_is_empty(&state->entries[i])) /* can be empty because the qsort may have marked duplicates */
WRITETUP(state, lt, &state->entries[i]);
}
else
{
for(i=0; i<state->entry_count; ++i)
WRITETUP(state, lt, &state->entries[i]);
}
}
markrunend(state, state->tp_tapenum[state->destTape]);
state->currentRun++;
state->tp_runs[state->destTape]++;
state->tp_dummy[state->destTape]--; /* per Alg D step D2 */
return;
}
/* OK. Normal case */
Assert(state->mkheap);
while (1)
{
if (mkheap_empty(state->mkheap))
{
markrunend(state, state->tp_tapenum[state->destTape]);
state->currentRun++;
state->tp_runs[state->destTape]++;
state->tp_dummy[state->destTape]--; /* per Alg D step D2 */
break;
}
if (!bDumped)
bDumped = 1;
/*
* Dump the heap's frontmost entry, and sift up to remove it from the
* heap.
*/
Assert(!mkheap_empty(state->mkheap));
lt = LogicalTapeSetGetTape(state->tapeset, state->tp_tapenum[state->destTape]);
mke_set_empty(&e);
mke_set_run(&e, state->currentRun+1);
mkheap_putAndGet(state->mkheap, &e);
Assert(!mke_is_empty(&e));
WRITETUP(state, lt, &e);
/*
* If the heap is empty *or* top run number has changed, we've
* finished the current run.
*/
if(mkheap_empty(state->mkheap) || !mkheap_run_match(state->mkheap, state->currentRun))
{
#ifdef USE_ASSERT_CHECKING
mkheap_verify_heap(state->mkheap, 0);
#endif
markrunend(state, state->tp_tapenum[state->destTape]);
state->currentRun++;
state->tp_runs[state->destTape]++;
state->tp_dummy[state->destTape]--; /* per Alg D step D2 */
if (trace_sort)
PG_TRACE3(tuplesort__dumptuples, state->entry_count, state->currentRun, state->destTape);
/*
* Done if heap is empty, else prepare for new run.
*/
if (mkheap_empty(state->mkheap))
break;
Assert(mkheap_run_match(state->mkheap, state->currentRun));
selectnewtape_mk(state);
}
}
if(state->gpmon_pkt)
tuplesort_checksend_gpmonpkt(state->gpmon_pkt, state->gpmon_sort_tick);
}
/*
* Put pos at the begining of the tuplesort. Create pos->work_tape if necessary
*/
void
tuplesort_rescan_pos_mk(Tuplesortstate_mk *state, TuplesortPos_mk *pos)
{
MemoryContext oldcontext = MemoryContextSwitchTo(state->sortcontext);
Assert(state->randomAccess);
switch (state->status)
{
case TSS_SORTEDINMEM:
pos->current = 0;
pos->eof_reached = false;
pos->markpos.mempos = 0;
pos->markpos_eof = false;
pos->cur_work_tape = NULL;
break;
case TSS_SORTEDONTAPE:
if(pos == &state->pos)
{
Assert(pos->cur_work_tape == NULL);
LogicalTapeRewind(state->tapeset, state->result_tape, false);
}
else
{
if(pos->cur_work_tape == NULL)
pos->cur_work_tape = state->result_tape;
LogicalTapeRewind(state->tapeset, pos->cur_work_tape, false);
}
pos->eof_reached = false;
pos->markpos.tapepos.blkNum = 0L;
pos->markpos.tapepos.offset = 0;
pos->markpos_eof = false;
break;
default:
elog(ERROR, "invalid tuplesort state");
break;
}
MemoryContextSwitchTo(oldcontext);
}
/*
* tuplesort_rescan - rewind and replay the scan
*/
void tuplesort_rescan_mk(Tuplesortstate_mk *state)
{
tuplesort_rescan_pos_mk(state, &state->pos);
}
/*
* tuplesort_markpos - saves current position in the merged sort file
*/
void
tuplesort_markpos_pos_mk(Tuplesortstate_mk *state, TuplesortPos_mk *pos)
{
LogicalTape *work_tape = NULL;
MemoryContext oldcontext = MemoryContextSwitchTo(state->sortcontext);
Assert(state->randomAccess);
switch (state->status)
{
case TSS_SORTEDINMEM:
pos->markpos.mempos = pos->current;
pos->markpos_eof = pos->eof_reached;
break;
case TSS_SORTEDONTAPE:
AssertEquivalent(pos == &state->pos, pos->cur_work_tape == NULL);
work_tape = pos->cur_work_tape == NULL ? state->result_tape : pos->cur_work_tape;
LogicalTapeTell(state->tapeset, work_tape, &pos->markpos.tapepos);
pos->markpos_eof = pos->eof_reached;
break;
default:
elog(ERROR, "invalid tuplesort state");
break;
}
MemoryContextSwitchTo(oldcontext);
}
void
tuplesort_markpos_mk(Tuplesortstate_mk *state)
{
tuplesort_markpos_pos_mk(state, &state->pos);
}
/*
* tuplesort_restorepos - restores current position in merged sort file to
* last saved position
*/
void
tuplesort_restorepos_pos_mk(Tuplesortstate_mk *state, TuplesortPos_mk *pos)
{
MemoryContext oldcontext = MemoryContextSwitchTo(state->sortcontext);
Assert(state->randomAccess);
switch (state->status)
{
case TSS_SORTEDINMEM:
pos->current = pos->markpos.mempos;
pos->eof_reached = pos->markpos_eof;
break;
case TSS_SORTEDONTAPE:
AssertEquivalent(pos == &state->pos, pos->cur_work_tape == NULL);
{
LogicalTape *work_tape = pos->cur_work_tape == NULL ? state->result_tape : pos->cur_work_tape;
bool fSeekOK = LogicalTapeSeek(state->tapeset, work_tape, &pos->markpos.tapepos);
if(!fSeekOK)
elog(ERROR, "tuplesort_restorepos failed");
pos->eof_reached = pos->markpos_eof;
}
break;
default:
elog(ERROR, "invalid tuplesort state");
break;
}
MemoryContextSwitchTo(oldcontext);
}
void
tuplesort_restorepos_mk(Tuplesortstate_mk *state)
{
tuplesort_restorepos_pos_mk(state, &state->pos);
}
/*
* Tape interface routines
*/
static uint32
getlen(Tuplesortstate_mk *state, TuplesortPos_mk *pos, LogicalTape *lt, bool eofOK)
{
uint32 len;
size_t readSize;
Assert(lt);
readSize = LogicalTapeRead(state->tapeset, lt, (void *)&len, sizeof(len));
if(readSize != sizeof(len))
{
Assert(!"Catch me");
elog(ERROR, "unexpected end of tape");
}
if (len == 0 && !eofOK)
elog(ERROR, "unexpected end of data");
return len;
}
static void
markrunend(Tuplesortstate_mk *state, int tapenum)
{
uint32 len = 0;
LogicalTape *lt = LogicalTapeSetGetTape(state->tapeset, tapenum);
LogicalTapeWrite(state->tapeset, lt, (void *) &len, sizeof(len));
}
/*
* Inline-able copy of FunctionCall2() to save some cycles in sorting.
*/
static inline Datum
myFunctionCall2(FmgrInfo *flinfo, Datum arg1, Datum arg2)
{
FunctionCallInfoData fcinfo;
Datum result;
InitFunctionCallInfoData(fcinfo, flinfo, 2, NULL, NULL);
fcinfo.arg[0] = arg1;
fcinfo.arg[1] = arg2;
fcinfo.argnull[0] = false;
fcinfo.argnull[1] = false;
result = FunctionCallInvoke(&fcinfo);
/* Check for null result, since caller is clearly not expecting one */
if (fcinfo.isnull)
elog(ERROR, "function %u returned NULL", fcinfo.flinfo->fn_oid);
return result;
}
/*
* Apply a sort function (by now converted to fmgr lookup form)
* and return a 3-way comparison result. This takes care of handling
* NULLs and sort ordering direction properly.
*/
static inline int32
inlineApplySortFunction(FmgrInfo *sortFunction, SortFunctionKind kind,
Datum datum1, bool isNull1,
Datum datum2, bool isNull2)
{
switch (kind)
{
case SORTFUNC_LT:
if (isNull1)
{
if (isNull2)
return 0;
return 1; /* NULL sorts after non-NULL */
}
if (isNull2)
return -1;
if (DatumGetBool(myFunctionCall2(sortFunction, datum1, datum2)))
return -1; /* a < b */
if (DatumGetBool(myFunctionCall2(sortFunction, datum2, datum1)))
return 1; /* a > b */
return 0;
case SORTFUNC_REVLT:
/* We reverse the ordering of NULLs, but not the operator */
if (isNull1)
{
if (isNull2)
return 0;
return -1; /* NULL sorts before non-NULL */
}
if (isNull2)
return 1;
if (DatumGetBool(myFunctionCall2(sortFunction, datum1, datum2)))
return -1; /* a < b */
if (DatumGetBool(myFunctionCall2(sortFunction, datum2, datum1)))
return 1; /* a > b */
return 0;
case SORTFUNC_CMP:
if (isNull1)
{
if (isNull2)
return 0;
return 1; /* NULL sorts after non-NULL */
}
if (isNull2)
return -1;
return DatumGetInt32(myFunctionCall2(sortFunction,
datum1, datum2));
case SORTFUNC_REVCMP:
if (isNull1)
{
if (isNull2)
return 0;
return -1; /* NULL sorts before non-NULL */
}
if (isNull2)
return 1;
return -DatumGetInt32(myFunctionCall2(sortFunction,
datum1, datum2));
default:
elog(ERROR, "unrecognized SortFunctionKind: %d", (int) kind);
return 0; /* can't get here, but keep compiler quiet */
}
}
static void
copytup_heap(Tuplesortstate_mk *state, MKEntry *e, void *tup)
{
/*
* We expect the passed "tup" to be a TupleTableSlot, and form a
* MinimalTuple using the exported interface for that.
*/
TupleTableSlot *slot = (TupleTableSlot *) tup;
slot_getallattrs(slot);
e->ptr = (void *) memtuple_form_to(state->mt_bind,
slot_get_values(slot),
slot_get_isnull(slot),
NULL, NULL, false
);
state->totalTupleBytes += memtuple_get_size((MemTuple)e->ptr, NULL);
Assert(state->mt_bind);
}
/*
* Since MinimalTuple already has length in its first word, we don't need
* to write that separately.
*/
static long
writetup_heap(Tuplesortstate_mk *state, LogicalTape *lt, MKEntry *e)
{
uint32 tuplen = memtuple_get_size(e->ptr, NULL);
long ret = tuplen;
LogicalTapeWrite(state->tapeset, lt, e->ptr, tuplen);
if (state->randomAccess) /* need trailing length word? */
{
LogicalTapeWrite(state->tapeset, lt, (void *) &tuplen, sizeof(tuplen));
ret += sizeof(tuplen);
}
if (state->gpmon_pkt)
{
Gpmon_M_Incr(state->gpmon_pkt, GPMON_SORT_SPILLTUPLE);
Gpmon_M_Add(state->gpmon_pkt, GPMON_SORT_SPILLBYTE, tuplen);
Gpmon_M_Incr(state->gpmon_pkt, GPMON_SORT_CURRSPILLPASS_TUPLE);
Gpmon_M_Add(state->gpmon_pkt, GPMON_SORT_CURRSPILLPASS_BYTE, tuplen);
}
pfree(e->ptr);
e->ptr = NULL;
return ret;
}
static void
freetup_heap(MKEntry *e)
{
pfree(e->ptr);
e->ptr = NULL;
}
static void
readtup_heap(Tuplesortstate_mk *state, TuplesortPos_mk *pos, MKEntry *e, LogicalTape *lt, uint32 len)
{
uint32 tuplen;
size_t readSize;
Assert(is_under_sort_ctxt(state));
MemSet(e, 0, sizeof(MKEntry));
e->ptr = palloc(memtuple_size_from_uint32(len));
memtuple_set_mtlen((MemTuple)e->ptr, NULL, len);
Assert(lt);
readSize = LogicalTapeRead(state->tapeset, lt,
(void *) ((char *) e->ptr + sizeof(uint32)),
memtuple_size_from_uint32(len) - sizeof(uint32));
if (readSize != (size_t) (memtuple_size_from_uint32(len) - sizeof(uint32)))
elog(ERROR, "unexpected end of data");
if (state->randomAccess) /* need trailing length word? */
{
readSize = LogicalTapeRead(state->tapeset, lt, (void *)&tuplen, sizeof(tuplen));
if(readSize != sizeof(tuplen))
elog(ERROR, "unexpected end of data");
}
/* For shareinput on sort, the reader will not set mt_bind. In this case,
* we will not call compare.
*/
AssertImply(!state->mt_bind, state->status == TSS_SORTEDONTAPE);
}
static void
copytup_index(Tuplesortstate_mk *state, MKEntry *e, void *tup)
{
IndexTuple tuple = (IndexTuple) tup;
uint32 tuplen = IndexTupleSize(tuple);
/* copy the tuple into sort storage */
e->ptr = palloc(tuplen);
memcpy(e->ptr, tuple, tuplen);
state->totalTupleBytes += tuplen;
}
static long
writetup_index(Tuplesortstate_mk *state, LogicalTape *lt, MKEntry *e)
{
IndexTuple tuple = (IndexTuple) e->ptr;
uint32 tuplen = IndexTupleSize(tuple) + sizeof(tuplen);
long ret = tuplen;
LogicalTapeWrite(state->tapeset, lt, (void *) &tuplen, sizeof(tuplen));
LogicalTapeWrite(state->tapeset, lt, (void *) tuple, IndexTupleSize(tuple));
if (state->randomAccess) /* need trailing length word? */
{
LogicalTapeWrite(state->tapeset, lt, (void *) &tuplen, sizeof(tuplen));
ret += sizeof(tuplen);
}
pfree(tuple);
e->ptr = NULL;
return ret;
}
static void
freetup_index(MKEntry *e)
{
pfree(e->ptr);
e->ptr = NULL;
}
static void
readtup_index(Tuplesortstate_mk *state, TuplesortPos_mk *pos, MKEntry *e,
LogicalTape *lt, uint32 len)
{
uint32 tuplen = len - sizeof(uint32);
IndexTuple tuple = (IndexTuple) palloc(tuplen);
size_t readSize;
Assert(lt);
MemSet(e, 0, sizeof(MKEntry));
readSize = LogicalTapeRead(state->tapeset, lt, (void *)tuple, tuplen);
if(readSize != tuplen)
elog(ERROR, "unexpected end of data");
if (state->randomAccess) /* need trailing length word? */
{
readSize = LogicalTapeRead(state->tapeset, lt, (void *)&tuplen, sizeof(tuplen));
if (readSize != sizeof(tuplen))
elog(ERROR, "unexpected end of data");
}
e->ptr = (void *) tuple;
}
static void
copytup_datum(Tuplesortstate_mk *state, MKEntry *e, void *tup)
{
/* Not currently needed */
elog(ERROR, "copytup_datum() should not be called");
}
static long
writetup_datum(Tuplesortstate_mk *state, LogicalTape *lt, MKEntry *e)
{
void *waddr;
uint32 tuplen;
uint32 writtenlen;
bool needfree = false;
long ret;
if (mke_is_null(e))
{
waddr = NULL;
tuplen = 0;
}
else if (state->datumTypeByVal)
{
waddr = &e->d;
tuplen = sizeof(Datum);
}
else
{
waddr = DatumGetPointer(e->d);
tuplen = datumGetSize(e->d, false, state->datumTypeLen);
needfree = true;
Assert(tuplen != 0);
}
writtenlen = tuplen + sizeof(uint32);
ret = writtenlen;
LogicalTapeWrite(state->tapeset, lt, (void *) &writtenlen, sizeof(writtenlen));
LogicalTapeWrite(state->tapeset, lt, waddr, tuplen);
if (state->randomAccess) /* need trailing length word? */
{
LogicalTapeWrite(state->tapeset, lt, (void *) &writtenlen, sizeof(writtenlen));
ret += sizeof(writtenlen);
}
if (needfree)
pfree(waddr);
return ret;
}
/**
* No-op free: does nothing!
*/
static void
freetup_noop(MKEntry *e)
{
}
/**
* Free the datum if non-null, should only be used for by-reference datums
*/
static void
freetup_datum(MKEntry *e)
{
if (!mke_is_null(e))
{
void *waddr = DatumGetPointer(e->d);
pfree(waddr);
}
}
static void
readtup_datum(Tuplesortstate_mk *state, TuplesortPos_mk *pos, MKEntry *e,
LogicalTape *lt, uint32 len)
{
size_t readSize;
uint32 tuplen = len - sizeof(uint32);
Assert(lt);
MemSet(e, 0, sizeof(MKEntry));
if (tuplen == 0)
mke_set_null(e, state->nullfirst);
else if (state->datumTypeByVal)
{
mke_set_not_null(e);
Assert(tuplen == sizeof(Datum));
readSize = LogicalTapeRead(state->tapeset, lt, (void *)&e->d, tuplen);
if (readSize != tuplen)
elog(ERROR, "unexpected end of data");
}
else
{
void *raddr = palloc(tuplen);
readSize = LogicalTapeRead(state->tapeset, lt, raddr, tuplen);
if(readSize != tuplen)
elog(ERROR, "unexpected end of data");
mke_set_not_null(e);
e->d = PointerGetDatum(raddr);
}
if (state->randomAccess) /* need trailing length word? */
{
readSize = LogicalTapeRead(state->tapeset, lt, (void *)&tuplen, sizeof(tuplen));
if (readSize != sizeof(tuplen))
elog(ERROR, "unexpected end of data");
}
}
typedef struct refcnt_locale_str
{
int ref;
short xfrm_pos;
char isPrefixOnly;
char data[1];
} refcnt_locale_str;
typedef struct refcnt_locale_str_k
{
int ref;
short xfrm_pos;
char isPrefixOnly;
char data[16000];
} refcnt_locale_str_k;
/**
* Compare the datums for the given level. It is assumed that each entry has been prepared
* for the given level.
*/
int tupsort_compare_datum(MKEntry *v1, MKEntry *v2, MKLvContext *lvctxt, MKContext *context)
{
Assert(!mke_is_null(v1));
Assert(!mke_is_null(v2));
switch(lvctxt->lvtype)
{
case MKLV_TYPE_NONE:
return inlineApplySortFunction(&lvctxt->fmgrinfo, lvctxt->sortfnkind,
v1->d, false,
v2->d, false
);
case MKLV_TYPE_INT32:
{
int32 i1 = DatumGetInt32(v1->d);
int32 i2 = DatumGetInt32(v2->d);
int result = (i1 < i2) ? -1 : ((i1 == i2) ? 0 : 1);
return (lvctxt->sortfnkind == SORTFUNC_CMP) ? result : -result;
}
default:
return tupsort_compare_char(v1, v2, lvctxt, context);
}
Assert(!"Never reach here");
return 0;
}
void tupsort_cpfr(MKEntry *dst, MKEntry *src, MKLvContext *lvctxt)
{
Assert(dst);
if(mke_is_refc(dst))
{
Assert(dst->d != 0);
Assert(!mke_is_copied(dst));
tupsort_refcnt(DatumGetPointer(dst->d), -1);
}
if(mke_is_copied(dst))
{
Assert(dst->d != 0);
Assert(!mke_is_refc(dst));
pfree(DatumGetPointer(dst->d));
}
mke_clear_refc_copied(dst);
if(src)
{
*dst = *src;
if(!mke_is_null(src))
{
if(mke_is_refc(src))
tupsort_refcnt(DatumGetPointer(dst->d), 1);
else if(!lvctxt->typByVal)
{
Assert(src->d != 0);
dst->d = datumCopy(src->d, lvctxt->typByVal, lvctxt->typLen);
mke_set_copied(dst);
}
}
}
}
static void tupsort_refcnt(void *vp, int ref)
{
refcnt_locale_str *p = (refcnt_locale_str *) vp;
Assert(p && p->ref > 0);
Assert(ref == 1 || ref == -1);
if(ref == 1)
++p->ref;
else
{
if(--p->ref == 0)
pfree(p);
}
}
static int tupsort_compare_char(MKEntry *v1, MKEntry *v2, MKLvContext *lvctxt, MKContext *mkContext)
{
int result = 0;
refcnt_locale_str *p1 = (refcnt_locale_str *) DatumGetPointer(v1->d);
refcnt_locale_str *p2 = (refcnt_locale_str *) DatumGetPointer(v2->d);
Assert(!mke_is_null(v1));
Assert(!mke_is_null(v2));
Assert(!lc_collate_is_c());
Assert(lvctxt->sortfnkind == SORTFUNC_CMP || lvctxt->sortfnkind == SORTFUNC_REVCMP);
Assert(mkContext->fetchForPrep);
Assert(p1->ref > 0 && p2->ref > 0);
if(p1 == p2)
{
Assert(p1->ref >= 2);
result = 0;
}
else
{
result = strcmp(p1->data + p1->xfrm_pos, p2->data + p2->xfrm_pos);
if(result == 0)
{
if ( p1->isPrefixOnly || p2->isPrefixOnly)
{
/*
* only prefixes were equal so we must compare more of the strings
*
* do this by getting the true datum and calling the built-in comparison
* function
*/
Datum p1Original, p2Original;
bool p1IsNull, p2IsNull;
p1Original = (mkContext->fetchForPrep)(v1, mkContext, lvctxt, &p1IsNull);
p2Original = (mkContext->fetchForPrep)(v2, mkContext, lvctxt, &p2IsNull);
Assert(!p1IsNull); /* should not have been prepared if null */
Assert(!p2IsNull);
result = inlineApplySortFunction(&lvctxt->fmgrinfo, lvctxt->sortfnkind,
p1Original, false, p2Original, false );
}
else
{
/*
* See varstr_cmp for comment on comparing str with locale.
* Essentially, for some locale, strcoll may return eq even if
* original str are different.
*/
result = strcmp(p1->data, p2->data);
}
}
/* The values were equal -- de-dupe them */
if(result == 0)
{
if(p1->ref >= p2->ref)
{
v2->d = v1->d;
tupsort_refcnt(p1, 1);
tupsort_refcnt(p2, -1);
}
else
{
v1->d = v2->d;
tupsort_refcnt(p2, 1);
tupsort_refcnt(p1, -1);
}
}
}
return (lvctxt->sortfnkind == SORTFUNC_CMP) ? result : -result;
}
static int32 estimateMaxPrepareSizeForEntry(MKEntry *a, struct MKContext *mkContext)
{
int result = 0;
int lv;
Assert(mkContext->fetchForPrep != NULL);
for ( lv = 0; lv < mkContext->total_lv; lv++)
{
MKLvContext *level = mkContext->lvctxt + lv;
MKLvType levelType = level->lvtype;
if (levelType == MKLV_TYPE_CHAR ||
levelType == MKLV_TYPE_TEXT )
{
bool isnull;
Datum d = (mkContext->fetchForPrep)(a, mkContext, level, &isnull);
if ( ! isnull )
{
int amountThisDatum = estimatePrepareSpaceForChar(mkContext, a, d, levelType == MKLV_TYPE_CHAR);
if ( amountThisDatum > result )
result = amountThisDatum;
}
}
}
return result;
}
static Datum tupsort_fetch_datum_mtup(MKEntry *a, MKContext *mkctxt, MKLvContext *lvctxt, bool *isNullOut)
{
Datum d = memtuple_getattr(
(MemTuple) a->ptr,
mkctxt->mt_bind,
lvctxt->attno,
isNullOut
);
return d;
}
static Datum tupsort_fetch_datum_itup(MKEntry *a, MKContext *mkctxt, MKLvContext *lvctxt, bool *isNullOut)
{
Datum d = index_getattr(
(IndexTuple) a->ptr,
lvctxt->attno,
mkctxt->tupdesc,
isNullOut
);
return d;
}
void tupsort_prepare(MKEntry *a, MKContext *mkctxt, int lv)
{
MKLvContext *lvctxt = mkctxt->lvctxt + lv;
bool isnull;
Assert(mkctxt->fetchForPrep != NULL);
tupsort_cpfr(a, NULL, lvctxt);
a->d = (mkctxt->fetchForPrep)(a, mkctxt, lvctxt, &isnull);
if(!isnull)
mke_set_not_null(a);
else
mke_set_null(a, lvctxt->nullfirst);
if (lvctxt->lvtype == MKLV_TYPE_CHAR)
tupsort_prepare_char(a, true);
else if (lvctxt->lvtype == MKLV_TYPE_TEXT)
tupsort_prepare_char(a, false);
}
/* "True" length (not counting trailing blanks) of a BpChar */
static inline int bcTruelen(char *p, int len)
{
int i;
for (i = len - 1; i >= 0; i--)
{
if (p[i] != ' ')
break;
}
return i + 1;
}
/**
* should only be called for non-null Datum (caller must check the isnull flag from the fetch)
*/
static int32 estimatePrepareSpaceForChar(struct MKContext *mkContext, MKEntry *e, Datum d, bool isCHAR)
{
int len, transformedLength, retlen;
if(isCHAR)
{
char *p;
void *toFree;
varattrib_untoast_ptr_len(d, &p, &len, &toFree);
len = bcTruelen(p, len);
if ( toFree )
pfree(toFree);
}
else
{
/* since we don't need the data for checking the true length, just unpack enough to get the length
* this may avoid decompression
*/
len = varattrib_untoast_len(d);
}
/* figure out how much space transformed version will take */
transformedLength = len * mkContext->strxfrmScaleFactor + mkContext->strxfrmConstantFactor;
if ( len > STRXFRM_INPUT_LENGTH_LIMIT)
{
if ( transformedLength > STRXFRM_INPUT_LENGTH_LIMIT)
transformedLength = STRXFRM_INPUT_LENGTH_LIMIT;
/* we do not store the raw data for long input strings (in part because
* of compression producing a long input string from a much smaller datum) */
len = 0;
}
retlen = offsetof(refcnt_locale_str, data) + len + 1 + transformedLength + 1;
return retlen;
}
/**
* Prepare a character string by copying the datum out to a null-terminated string and
* then also keeping a strxfrmed copy of it.
*
* If the string is long then the copied out value is not kept and only the prefix of the strxfrm
* is kept.
*
* Note that this must be in sync with the estimation function.
*/
static void tupsort_prepare_char(MKEntry *a, bool isCHAR)
{
char *p;
void *tofree = NULL;
int len;
int lenToStore;
int transformedLenToStore;
int retlen;
refcnt_locale_str *ret;
refcnt_locale_str_k kstr;
int avail = sizeof(kstr.data);
bool storePrefixOnly;
Assert(!lc_collate_is_c());
if(mke_is_null(a))
return;
varattrib_untoast_ptr_len(a->d, &p, &len, &tofree);
if(isCHAR)
len = bcTruelen(p, len);
if ( len > STRXFRM_INPUT_LENGTH_LIMIT)
{
/* too long? store prefix of strxfrm only and DON'T store copy of original string */
storePrefixOnly = true;
lenToStore = 0;
}
else
{
storePrefixOnly = false;
lenToStore = len;
}
/* This assertion is true because avail is larger than
* STRXFRM_INPUT_LENGTH_LIMIT, which is the max of lenToStore ... */
Assert(lenToStore <= avail - 2);
Assert(lenToStore < 32768); /* so it will fit in a short (xfrm_pos field) */
Assert(STRXFRM_INPUT_LENGTH_LIMIT < 32768); /* so it will fit in a short (xfrm_pos field) */
kstr.ref = 1;
kstr.xfrm_pos = lenToStore+1;
memcpy(kstr.data, p, lenToStore);
kstr.data[lenToStore] = '\0';
avail -= lenToStore + 1;
Assert(avail > 0);
/*
* String transformation.
*/
if ( storePrefixOnly)
{
/* prefix only: we haven't copied from p into a null-terminated string so do that now
*/
char *possibleStr = kstr.data + kstr.xfrm_pos + STRXFRM_INPUT_LENGTH_LIMIT + 2;
char *str;
if ( avail >= STRXFRM_INPUT_LENGTH_LIMIT + 1 + len + 1 )
{
/* will segment this so first part is xformed string and next is the throw-away string
* to be passed to strxfrm */
str = possibleStr;
}
else
{
str = palloc(len + 1);
}
memcpy(str, p, len);
str[len] = '\0';
/*
* transform, but limit length of transformed string
*/
Assert(avail >= STRXFRM_INPUT_LENGTH_LIMIT + 1);
transformedLenToStore = (int) gp_strxfrm(kstr.data+kstr.xfrm_pos, str, STRXFRM_INPUT_LENGTH_LIMIT + 1);
if ( transformedLenToStore > STRXFRM_INPUT_LENGTH_LIMIT)
{
transformedLenToStore = STRXFRM_INPUT_LENGTH_LIMIT;
/* this is required for linux -- when there is not enough room then linux
* does NOT write the \0
*/
kstr.data[kstr.xfrm_pos + transformedLenToStore] = '\0';
}
if ( str != possibleStr )
pfree(str);
}
else transformedLenToStore = (int) gp_strxfrm(kstr.data+kstr.xfrm_pos, kstr.data, avail);
/*
* Copy or transform into the result as needed
*/
retlen = offsetof(refcnt_locale_str, data) + lenToStore + 1 + transformedLenToStore + 1;
ret = (refcnt_locale_str *) palloc(retlen);
if(transformedLenToStore < avail)
{
memcpy(ret, &kstr, retlen);
Assert(ret->ref == 1);
Assert(ret->data[ret->xfrm_pos - 1] == '\0');
Assert(ret->data[ret->xfrm_pos + transformedLenToStore] == '\0');
}
else
{
/* note that when avail (determined by refcnt_locale_str_k.data) is much larger than
* STRXFRM_INPUT_LENGTH_LIMIT then this code won't be hit.
*/
memcpy(ret, &kstr, offsetof(refcnt_locale_str, data) + lenToStore + 1);
avail = retlen - offsetof(refcnt_locale_str, data) - lenToStore - 1;
Assert(avail > transformedLenToStore);
Assert(ret->ref == 1);
Assert(ret->xfrm_pos == len + 1);
Assert(ret->data[len] == '\0');
transformedLenToStore = (int) gp_strxfrm(ret->data+ret->xfrm_pos, kstr.data, avail);
Assert(transformedLenToStore < avail);
Assert(ret->data[ret->xfrm_pos + transformedLenToStore] == '\0');
}
if(tofree)
pfree(tofree);
/* data string is length zero in this case (could just not even store it and have xfrm_pos == 0
* but that complicates code more)
*/
AssertImply(storePrefixOnly, ret->data[0] == '\0' );
/*
* finalize result
*/
ret->isPrefixOnly = storePrefixOnly ? 1 : 0;
a->d = PointerGetDatum(ret);
mke_set_refc(a);
}
/*
* tuplesort_inmem_limit_insert
* Adds a tuple for sorting when we are doing LIMIT sort and we (still) fit in memory
* If we're in here, it means we haven't exceeded memory yet.
* Three cases:
* - we're below LIMIT tuples. Then add to unsorted array, increase memory
* - we just hit LIMIT tuples. Switch to using a heap instead of array, so
* we only keep top LIMIT tuples.
* - we're above LIMIT tuples. Then add to heap, don't increase memory usage
*/
static void
tuplesort_inmem_limit_insert(Tuplesortstate_mk *state, MKEntry *entry)
{
Assert(state->mkctxt.limit > 0);
Assert(is_under_sort_ctxt(state));
Assert(!mke_is_empty(entry));
Assert(state->status == TSS_INITIAL);
if (state->mkheap == NULL)
{
Assert(state->entry_count < state->entry_allocsize);
/* We haven't created the heap yet.
* First add element to unsorted array so we don't lose it.
* Then, we have two cases:
* A. if we're under limit, we're done, iterate.
* B. We just hit the limit. Create heap. Add to heap.
*/
state->entries[state->entry_count++] = *entry;
state->numTuplesInMem++;
if (state->entry_count == state->mkctxt.limit)
{
/* B: just hit limit. Create heap from array */
state->mkheap = mkheap_from_array(state->entries,
state->entry_count,
state->entry_count,
&state->mkctxt);
state->entries = NULL;
state->entry_allocsize = 0;
state->entry_count = 0;
}
}
else
{
/* We already have the heap (and it fits in memory).
* Simply add to the heap, keeping only top LIMIT elements.
*/
Assert(state->mkheap && !mkheap_empty(state->mkheap));
Assert(state->mkheap->count == state->numTuplesInMem);
Assert(state->entry_count == 0);
(void) mkheap_putAndGet(state->mkheap, entry);
Assert(!mke_is_empty(entry));
Assert(entry->ptr);
pfree(entry->ptr);
entry->ptr = NULL;
}
}
/*
* tuplesort_inmem_nolimit_insert
* Adds a tuple for sorting when we are regular (no LIMIT) sort and we (still) fit in memory
*
* Simply add to the unsorted in-memory array. We'll either qsort or spill this later.
*/
static void
tuplesort_inmem_nolimit_insert(Tuplesortstate_mk *state, MKEntry *entry)
{
Assert(state->status == TSS_INITIAL);
Assert(state->mkctxt.limit == 0);
Assert(is_under_sort_ctxt(state));
Assert(!mke_is_empty(entry));
Assert(state->entry_count < state->entry_allocsize);
state->entries[state->entry_count] = *entry;
if ( state->mkctxt.fetchForPrep)
{
state->mkctxt.estimatedExtraForPrep += estimateMaxPrepareSizeForEntry(&state->entries[state->entry_count],
&state->mkctxt);
}
state->entry_count++;
state->numTuplesInMem++;
}
static void tuplesort_heap_insert(Tuplesortstate_mk *state, MKEntry *e)
{
int ins;
Assert(state->mkheap);
Assert(is_under_sort_ctxt(state));
ins = mkheap_putAndGet_run(state->mkheap, e, state->currentRun);
tupsort_cpfr(e, NULL, &state->mkctxt.lvctxt[mke_get_lv(e)]);
Assert(ins >= 0);
{
LogicalTape *lt = LogicalTapeSetGetTape(state->tapeset, state->tp_tapenum[state->destTape]);
WRITETUP(state, lt, e);
if(!mkheap_run_match(state->mkheap, state->currentRun))
{
markrunend(state, state->tp_tapenum[state->destTape]);
state->currentRun++;
state->tp_runs[state->destTape]++;
state->tp_dummy[state->destTape]--;
selectnewtape_mk(state);
}
}
}
static void tuplesort_limit_sort(Tuplesortstate_mk *state)
{
Assert(state->mkctxt.limit > 0);
Assert(is_under_sort_ctxt(state));
if(!state->mkheap)
{
Assert(state->entry_count <= state->mkctxt.limit);
mk_qsort(state->entries, state->entry_count, &state->mkctxt);
return;
}
else
{
int i;
#ifdef USE_ASSERT_CHECKING
mkheap_verify_heap(state->mkheap, 0);
#endif
state->entry_allocsize = mkheap_cnt(state->mkheap);
state->entries = palloc(state->entry_allocsize * sizeof (MKEntry));
/* Put these things to minimum */
for(i=0; i<state->entry_allocsize; ++i)
mke_set_comp_max(state->entries+i);
for(i=state->entry_allocsize-1; i>=0; --i)
mkheap_putAndGet(state->mkheap, state->entries+i);
Assert(mkheap_cnt(state->mkheap) == 0);
state->entry_count = state->entry_allocsize;
}
mkheap_destroy(state->mkheap);
state->mkheap = NULL;
}
void
tuplesort_set_gpmon_mk(Tuplesortstate_mk *state, gpmon_packet_t *gpmon_pkt, int *gpmon_tick)
{
state->gpmon_pkt = gpmon_pkt;
state->gpmon_sort_tick = gpmon_tick;
}
/*
* Write the metadata of a workfile set to disk
*/
void
tuplesort_write_spill_metadata_mk(Tuplesortstate_mk *state)
{
if (state->status == TSS_SORTEDINMEM)
{
/* No spill files, nothing to save */
return;
}
/* We don't know how to handle TSS_FINALMERGE yet */
Assert(state->status == TSS_SORTEDONTAPE);
tuplesort_flush_mk(state);
}
void tuplesort_set_spillfile_set_mk(Tuplesortstate_mk * state, workfile_set * sfs)
{
state->work_set = sfs;
}
void tuplesort_read_spill_metadata_mk(Tuplesortstate_mk *state)
{
Assert(state->work_set != NULL);
MemoryContext oldctxt = MemoryContextSwitchTo(state->sortcontext);
state->status = TSS_SORTEDONTAPE;
state->readtup = readtup_heap;
/* Open saved spill file set and metadata.
* Initialize logical tape set to point to the right blocks. */
state->tapeset_state_file = workfile_mgr_open_fileno(state->work_set, WORKFILE_NUM_MKSORT_METADATA);
ExecWorkFile *tape_file = workfile_mgr_open_fileno(state->work_set, WORKFILE_NUM_MKSORT_TAPESET);
state->tapeset = LoadLogicalTapeSetState(state->tapeset_state_file, tape_file);
state->currentRun = 0;
state->result_tape = LogicalTapeSetGetTape(state->tapeset, 0);
state->pos.eof_reached = false;
state->pos.markpos.tapepos.blkNum = 0;
state->pos.markpos.tapepos.offset = 0;
state->pos.markpos.mempos = 0;
state->pos.markpos_eof = false;
state->pos.cur_work_tape = NULL;
MemoryContextSwitchTo(oldctxt);
}
/* EOF */