diff --git a/src/backend/commands/explain.c b/src/backend/commands/explain.c index 949fefa23ae..c2fb7a088a2 100644 --- a/src/backend/commands/explain.c +++ b/src/backend/commands/explain.c @@ -102,6 +102,7 @@ static void show_tablesample(TableSampleClause *tsc, PlanState *planstate, List *ancestors, ExplainState *es); static void show_sort_info(SortState *sortstate, ExplainState *es); static void show_hash_info(HashState *hashstate, ExplainState *es); +static void show_hashagg_info(AggState *hashstate, ExplainState *es); static void show_tidbitmap_info(BitmapHeapScanState *planstate, ExplainState *es); static void show_instrumentation_count(const char *qlabel, int which, @@ -1844,6 +1845,8 @@ ExplainNode(PlanState *planstate, List *ancestors, case T_Agg: show_agg_keys(castNode(AggState, planstate), ancestors, es); show_upper_qual(plan->qual, "Filter", planstate, ancestors, es); + if (es->analyze) + show_hashagg_info((AggState *) planstate, es); if (plan->qual) show_instrumentation_count("Rows Removed by Filter", 1, planstate, es); @@ -2742,6 +2745,56 @@ show_hash_info(HashState *hashstate, ExplainState *es) } } +/* + * If EXPLAIN ANALYZE, show information on hash aggregate memory usage and + * batches. + */ +static void +show_hashagg_info(AggState *aggstate, ExplainState *es) +{ + Agg *agg = (Agg *)aggstate->ss.ps.plan; + long memPeakKb = (aggstate->hash_mem_peak + 1023) / 1024; + long diskKb = (aggstate->hash_disk_used + 1023) / 1024; + + + Assert(IsA(aggstate, AggState)); + + if (agg->aggstrategy != AGG_HASHED && + agg->aggstrategy != AGG_MIXED) + return; + + if (es->format == EXPLAIN_FORMAT_TEXT) + { + appendStringInfoSpaces(es->str, es->indent * 2); + appendStringInfo( + es->str, + "Memory Usage: %ldkB", + memPeakKb); + + if (aggstate->hash_batches_used > 0) + { + appendStringInfo( + es->str, + " Batches: %d Disk: %ldkB", + aggstate->hash_batches_used, diskKb); + } + + appendStringInfo( + es->str, + "\n"); + } + else + { + ExplainPropertyInteger("Peak Memory Usage", "kB", memPeakKb, es); + if (aggstate->hash_batches_used > 0) + { + ExplainPropertyInteger("HashAgg Batches", NULL, + aggstate->hash_batches_used, es); + ExplainPropertyInteger("Disk Usage", "kB", diskKb, es); + } + } +} + /* * If it's EXPLAIN ANALYZE, show exact/lossy pages for a BitmapHeapScan node */ diff --git a/src/backend/executor/execExpr.c b/src/backend/executor/execExpr.c index 8da2e2dcbba..fb3e81764ad 100644 --- a/src/backend/executor/execExpr.c +++ b/src/backend/executor/execExpr.c @@ -79,7 +79,8 @@ static void ExecInitCoerceToDomain(ExprEvalStep *scratch, CoerceToDomain *ctest, static void ExecBuildAggTransCall(ExprState *state, AggState *aggstate, ExprEvalStep *scratch, FunctionCallInfo fcinfo, AggStatePerTrans pertrans, - int transno, int setno, int setoff, bool ishash); + int transno, int setno, int setoff, bool ishash, + bool spilled); /* @@ -2927,7 +2928,7 @@ ExecInitCoerceToDomain(ExprEvalStep *scratch, CoerceToDomain *ctest, */ ExprState * ExecBuildAggTrans(AggState *aggstate, AggStatePerPhase phase, - bool doSort, bool doHash) + bool doSort, bool doHash, bool spilled) { ExprState *state = makeNode(ExprState); PlanState *parent = &aggstate->ss.ps; @@ -3160,7 +3161,8 @@ ExecBuildAggTrans(AggState *aggstate, AggStatePerPhase phase, for (setno = 0; setno < processGroupingSets; setno++) { ExecBuildAggTransCall(state, aggstate, &scratch, trans_fcinfo, - pertrans, transno, setno, setoff, false); + pertrans, transno, setno, setoff, false, + spilled); setoff++; } } @@ -3178,7 +3180,8 @@ ExecBuildAggTrans(AggState *aggstate, AggStatePerPhase phase, for (setno = 0; setno < numHashes; setno++) { ExecBuildAggTransCall(state, aggstate, &scratch, trans_fcinfo, - pertrans, transno, setno, setoff, true); + pertrans, transno, setno, setoff, true, + spilled); setoff++; } } @@ -3226,7 +3229,8 @@ static void ExecBuildAggTransCall(ExprState *state, AggState *aggstate, ExprEvalStep *scratch, FunctionCallInfo fcinfo, AggStatePerTrans pertrans, - int transno, int setno, int setoff, bool ishash) + int transno, int setno, int setoff, bool ishash, + bool spilled) { int adjust_init_jumpnull = -1; int adjust_strict_jumpnull = -1; @@ -3248,7 +3252,8 @@ ExecBuildAggTransCall(ExprState *state, AggState *aggstate, fcinfo->flinfo->fn_strict && pertrans->initValueIsNull) { - scratch->opcode = EEOP_AGG_INIT_TRANS; + scratch->opcode = spilled ? + EEOP_AGG_INIT_TRANS_SPILLED : EEOP_AGG_INIT_TRANS; scratch->d.agg_init_trans.aggstate = aggstate; scratch->d.agg_init_trans.pertrans = pertrans; scratch->d.agg_init_trans.setno = setno; @@ -3265,7 +3270,8 @@ ExecBuildAggTransCall(ExprState *state, AggState *aggstate, if (pertrans->numSortCols == 0 && fcinfo->flinfo->fn_strict) { - scratch->opcode = EEOP_AGG_STRICT_TRANS_CHECK; + scratch->opcode = spilled ? + EEOP_AGG_STRICT_TRANS_CHECK_SPILLED : EEOP_AGG_STRICT_TRANS_CHECK; scratch->d.agg_strict_trans_check.aggstate = aggstate; scratch->d.agg_strict_trans_check.setno = setno; scratch->d.agg_strict_trans_check.setoff = setoff; @@ -3283,9 +3289,11 @@ ExecBuildAggTransCall(ExprState *state, AggState *aggstate, /* invoke appropriate transition implementation */ if (pertrans->numSortCols == 0 && pertrans->transtypeByVal) - scratch->opcode = EEOP_AGG_PLAIN_TRANS_BYVAL; + scratch->opcode = spilled ? + EEOP_AGG_PLAIN_TRANS_BYVAL_SPILLED : EEOP_AGG_PLAIN_TRANS_BYVAL; else if (pertrans->numSortCols == 0) - scratch->opcode = EEOP_AGG_PLAIN_TRANS; + scratch->opcode = spilled ? + EEOP_AGG_PLAIN_TRANS_SPILLED : EEOP_AGG_PLAIN_TRANS; else if (pertrans->numInputs == 1) scratch->opcode = EEOP_AGG_ORDERED_TRANS_DATUM; else diff --git a/src/backend/executor/execExprInterp.c b/src/backend/executor/execExprInterp.c index 034970648f3..11ba8c09542 100644 --- a/src/backend/executor/execExprInterp.c +++ b/src/backend/executor/execExprInterp.c @@ -430,9 +430,13 @@ ExecInterpExpr(ExprState *state, ExprContext *econtext, bool *isnull) &&CASE_EEOP_AGG_STRICT_INPUT_CHECK_ARGS, &&CASE_EEOP_AGG_STRICT_INPUT_CHECK_NULLS, &&CASE_EEOP_AGG_INIT_TRANS, + &&CASE_EEOP_AGG_INIT_TRANS_SPILLED, &&CASE_EEOP_AGG_STRICT_TRANS_CHECK, + &&CASE_EEOP_AGG_STRICT_TRANS_CHECK_SPILLED, &&CASE_EEOP_AGG_PLAIN_TRANS_BYVAL, + &&CASE_EEOP_AGG_PLAIN_TRANS_BYVAL_SPILLED, &&CASE_EEOP_AGG_PLAIN_TRANS, + &&CASE_EEOP_AGG_PLAIN_TRANS_SPILLED, &&CASE_EEOP_AGG_ORDERED_TRANS_DATUM, &&CASE_EEOP_AGG_ORDERED_TRANS_TUPLE, &&CASE_EEOP_LAST @@ -1625,6 +1629,36 @@ ExecInterpExpr(ExprState *state, ExprContext *econtext, bool *isnull) EEO_NEXT(); } + EEO_CASE(EEOP_AGG_INIT_TRANS_SPILLED) + { + AggState *aggstate; + AggStatePerGroup pergroup; + AggStatePerGroup pergroup_allaggs; + + aggstate = op->d.agg_init_trans.aggstate; + pergroup_allaggs = aggstate->all_pergroups[op->d.agg_init_trans.setoff]; + + if (pergroup_allaggs == NULL) + EEO_NEXT(); + + pergroup = &pergroup_allaggs[op->d.agg_init_trans.transno]; + + /* If transValue has not yet been initialized, do so now. */ + if (pergroup->noTransValue) + { + AggStatePerTrans pertrans = op->d.agg_init_trans.pertrans; + + aggstate->curaggcontext = op->d.agg_init_trans.aggcontext; + aggstate->current_set = op->d.agg_init_trans.setno; + + ExecAggInitGroup(aggstate, pertrans, pergroup); + + /* copied trans value from input, done this round */ + EEO_JUMP(op->d.agg_init_trans.jumpnull); + } + + EEO_NEXT(); + } /* check that a strict aggregate's input isn't NULL */ EEO_CASE(EEOP_AGG_STRICT_TRANS_CHECK) @@ -1642,6 +1676,25 @@ ExecInterpExpr(ExprState *state, ExprContext *econtext, bool *isnull) EEO_NEXT(); } + EEO_CASE(EEOP_AGG_STRICT_TRANS_CHECK_SPILLED) + { + AggState *aggstate; + AggStatePerGroup pergroup; + AggStatePerGroup pergroup_allaggs; + + aggstate = op->d.agg_strict_trans_check.aggstate; + pergroup_allaggs = aggstate->all_pergroups[op->d.agg_strict_trans_check.setoff]; + + if (pergroup_allaggs == NULL) + EEO_NEXT(); + + pergroup = &pergroup_allaggs[op->d.agg_strict_trans_check.transno]; + + if (unlikely(pergroup->transValueIsNull)) + EEO_JUMP(op->d.agg_strict_trans_check.jumpnull); + + EEO_NEXT(); + } /* * Evaluate aggregate transition / combine function that has a @@ -1691,6 +1744,52 @@ ExecInterpExpr(ExprState *state, ExprContext *econtext, bool *isnull) EEO_NEXT(); } + EEO_CASE(EEOP_AGG_PLAIN_TRANS_BYVAL_SPILLED) + { + AggState *aggstate; + AggStatePerTrans pertrans; + AggStatePerGroup pergroup; + AggStatePerGroup pergroup_allaggs; + FunctionCallInfo fcinfo; + MemoryContext oldContext; + Datum newVal; + + aggstate = op->d.agg_trans.aggstate; + pertrans = op->d.agg_trans.pertrans; + + pergroup_allaggs = aggstate->all_pergroups[op->d.agg_trans.setoff]; + pergroup = &pergroup_allaggs[op->d.agg_trans.transno]; + + if (pergroup_allaggs == NULL) + EEO_NEXT(); + + Assert(pertrans->transtypeByVal); + + fcinfo = pertrans->transfn_fcinfo; + + /* cf. select_current_set() */ + aggstate->curaggcontext = op->d.agg_trans.aggcontext; + aggstate->current_set = op->d.agg_trans.setno; + + /* set up aggstate->curpertrans for AggGetAggref() */ + aggstate->curpertrans = pertrans; + + /* invoke transition function in per-tuple context */ + oldContext = MemoryContextSwitchTo(aggstate->tmpcontext->ecxt_per_tuple_memory); + + fcinfo->args[0].value = pergroup->transValue; + fcinfo->args[0].isnull = pergroup->transValueIsNull; + fcinfo->isnull = false; /* just in case transfn doesn't set it */ + + newVal = FunctionCallInvoke(fcinfo); + + pergroup->transValue = newVal; + pergroup->transValueIsNull = fcinfo->isnull; + + MemoryContextSwitchTo(oldContext); + + EEO_NEXT(); + } /* * Evaluate aggregate transition / combine function that has a @@ -1756,6 +1855,67 @@ ExecInterpExpr(ExprState *state, ExprContext *econtext, bool *isnull) EEO_NEXT(); } + EEO_CASE(EEOP_AGG_PLAIN_TRANS_SPILLED) + { + AggState *aggstate; + AggStatePerTrans pertrans; + AggStatePerGroup pergroup; + AggStatePerGroup pergroup_allaggs; + FunctionCallInfo fcinfo; + MemoryContext oldContext; + Datum newVal; + + aggstate = op->d.agg_trans.aggstate; + pertrans = op->d.agg_trans.pertrans; + + pergroup_allaggs = aggstate->all_pergroups[op->d.agg_trans.setoff]; + + if (pergroup_allaggs == NULL) + EEO_NEXT(); + + pergroup = &pergroup_allaggs[op->d.agg_trans.transno]; + + Assert(!pertrans->transtypeByVal); + + fcinfo = pertrans->transfn_fcinfo; + + /* cf. select_current_set() */ + aggstate->curaggcontext = op->d.agg_trans.aggcontext; + aggstate->current_set = op->d.agg_trans.setno; + + /* set up aggstate->curpertrans for AggGetAggref() */ + aggstate->curpertrans = pertrans; + + /* invoke transition function in per-tuple context */ + oldContext = MemoryContextSwitchTo(aggstate->tmpcontext->ecxt_per_tuple_memory); + + fcinfo->args[0].value = pergroup->transValue; + fcinfo->args[0].isnull = pergroup->transValueIsNull; + fcinfo->isnull = false; /* just in case transfn doesn't set it */ + + newVal = FunctionCallInvoke(fcinfo); + + /* + * For pass-by-ref datatype, must copy the new value into + * aggcontext and free the prior transValue. But if transfn + * returned a pointer to its first input, we don't need to do + * anything. Also, if transfn returned a pointer to a R/W + * expanded object that is already a child of the aggcontext, + * assume we can adopt that value without copying it. + */ + if (DatumGetPointer(newVal) != DatumGetPointer(pergroup->transValue)) + newVal = ExecAggTransReparent(aggstate, pertrans, + newVal, fcinfo->isnull, + pergroup->transValue, + pergroup->transValueIsNull); + + pergroup->transValue = newVal; + pergroup->transValueIsNull = fcinfo->isnull; + + MemoryContextSwitchTo(oldContext); + + EEO_NEXT(); + } /* process single-column ordered aggregate datum */ EEO_CASE(EEOP_AGG_ORDERED_TRANS_DATUM) diff --git a/src/backend/executor/execGrouping.c b/src/backend/executor/execGrouping.c index e361143094c..02dba3eac18 100644 --- a/src/backend/executor/execGrouping.c +++ b/src/backend/executor/execGrouping.c @@ -25,8 +25,9 @@ #include "utils/lsyscache.h" #include "utils/memutils.h" -static uint32 TupleHashTableHash(struct tuplehash_hash *tb, const MinimalTuple tuple); static int TupleHashTableMatch(struct tuplehash_hash *tb, const MinimalTuple tuple1, const MinimalTuple tuple2); +static TupleHashEntry LookupTupleHashEntry_internal( + TupleHashTable hashtable, TupleTableSlot *slot, bool *isnew, uint32 hash); /* * Define parameters for tuple hash table code generation. The interface is @@ -284,6 +285,17 @@ ResetTupleHashTable(TupleHashTable hashtable) tuplehash_reset(hashtable->hashtab); } +/* + * Destroy the hash table. Note that the tablecxt passed to + * BuildTupleHashTableExt() should also be reset, otherwise there will be + * leaks. + */ +void +DestroyTupleHashTable(TupleHashTable hashtable) +{ + tuplehash_destroy(hashtable->hashtab); +} + /* * Find or create a hashtable entry for the tuple group containing the * given tuple. The tuple must be the same type as the hashtable entries. @@ -300,10 +312,9 @@ TupleHashEntry LookupTupleHashEntry(TupleHashTable hashtable, TupleTableSlot *slot, bool *isnew) { - TupleHashEntryData *entry; - MemoryContext oldContext; - bool found; - MinimalTuple key; + TupleHashEntry entry; + MemoryContext oldContext; + uint32 hash; /* Need to run the hash functions in short-lived context */ oldContext = MemoryContextSwitchTo(hashtable->tempcxt); @@ -313,32 +324,29 @@ LookupTupleHashEntry(TupleHashTable hashtable, TupleTableSlot *slot, hashtable->in_hash_funcs = hashtable->tab_hash_funcs; hashtable->cur_eq_func = hashtable->tab_eq_func; - key = NULL; /* flag to reference inputslot */ + hash = TupleHashTableHash(hashtable->hashtab, NULL); + entry = LookupTupleHashEntry_internal(hashtable, slot, isnew, hash); - if (isnew) - { - entry = tuplehash_insert(hashtable->hashtab, key, &found); + MemoryContextSwitchTo(oldContext); - if (found) - { - /* found pre-existing entry */ - *isnew = false; - } - else - { - /* created new entry */ - *isnew = true; - /* zero caller data */ - entry->additional = NULL; - MemoryContextSwitchTo(hashtable->tablecxt); - /* Copy the first tuple into the table context */ - entry->firstTuple = ExecCopySlotMinimalTuple(slot); - } - } - else - { - entry = tuplehash_lookup(hashtable->hashtab, key); - } + return entry; +} + +/* + * A variant of LookupTupleHashEntry for callers that have already computed + * the hash value. + */ +TupleHashEntry +LookupTupleHashEntryHash(TupleHashTable hashtable, TupleTableSlot *slot, + bool *isnew, uint32 hash) +{ + TupleHashEntry entry; + MemoryContext oldContext; + + /* Need to run the hash functions in short-lived context */ + oldContext = MemoryContextSwitchTo(hashtable->tempcxt); + + entry = LookupTupleHashEntry_internal(hashtable, slot, isnew, hash); MemoryContextSwitchTo(oldContext); @@ -389,7 +397,7 @@ FindTupleHashEntry(TupleHashTable hashtable, TupleTableSlot *slot, * Also, the caller must select an appropriate memory context for running * the hash functions. (dynahash.c doesn't change CurrentMemoryContext.) */ -static uint32 +uint32 TupleHashTableHash(struct tuplehash_hash *tb, const MinimalTuple tuple) { TupleHashTable hashtable = (TupleHashTable) tb->private_data; @@ -450,6 +458,54 @@ TupleHashTableHash(struct tuplehash_hash *tb, const MinimalTuple tuple) return murmurhash32(hashkey); } +/* + * Does the work of LookupTupleHashEntry and LookupTupleHashEntryHash. Useful + * so that we can avoid switching the memory context multiple times for + * LookupTupleHashEntry. + */ +static TupleHashEntry +LookupTupleHashEntry_internal(TupleHashTable hashtable, TupleTableSlot *slot, + bool *isnew, uint32 hash) +{ + TupleHashEntryData *entry; + bool found; + MinimalTuple key; + + /* set up data needed by hash and match functions */ + hashtable->inputslot = slot; + hashtable->in_hash_funcs = hashtable->tab_hash_funcs; + hashtable->cur_eq_func = hashtable->tab_eq_func; + + key = NULL; /* flag to reference inputslot */ + + if (isnew) + { + entry = tuplehash_insert_hash(hashtable->hashtab, key, hash, &found); + + if (found) + { + /* found pre-existing entry */ + *isnew = false; + } + else + { + /* created new entry */ + *isnew = true; + /* zero caller data */ + entry->additional = NULL; + MemoryContextSwitchTo(hashtable->tablecxt); + /* Copy the first tuple into the table context */ + entry->firstTuple = ExecCopySlotMinimalTuple(slot); + } + } + else + { + entry = tuplehash_lookup_hash(hashtable->hashtab, key, hash); + } + + return entry; +} + /* * See whether two tuples (presumably of the same hash value) match */ diff --git a/src/backend/executor/nodeAgg.c b/src/backend/executor/nodeAgg.c index 6ee24eab3d2..bc5ab981e1d 100644 --- a/src/backend/executor/nodeAgg.c +++ b/src/backend/executor/nodeAgg.c @@ -194,6 +194,18 @@ * transition values. hashcontext is the single context created to support * all hash tables. * + * When the hash table memory exceeds work_mem, we advance the transition + * states only for groups already in the hash table. For tuples that would + * need to create a new hash table entries (and initialize new transition + * states), we spill them to disk to be processed later. The tuples are + * spilled in a partitioned manner, so that subsequent batches are smaller + * and less likely to exceed work_mem (if a batch does exceed work_mem, it + * must be spilled recursively). + * + * Note that it's possible for transition states to start small but then + * grow very large; for instance in the case of ARRAY_AGG. In such cases, + * it's still possible to significantly exceed work_mem. + * * Transition / Combine function invocation: * * For performance reasons transition functions, including combine @@ -229,15 +241,69 @@ #include "optimizer/optimizer.h" #include "parser/parse_agg.h" #include "parser/parse_coerce.h" +#include "storage/buffile.h" #include "utils/acl.h" #include "utils/builtins.h" #include "utils/datum.h" +#include "utils/dynahash.h" #include "utils/expandeddatum.h" #include "utils/lsyscache.h" #include "utils/memutils.h" #include "utils/syscache.h" #include "utils/tuplesort.h" +/* + * Control how many partitions are created when spilling HashAgg to + * disk. + * + * HASHAGG_PARTITION_FACTOR is multiplied by the estimated number of + * partitions needed such that each partition will fit in memory. The factor + * is set higher than one because there's not a high cost to having a few too + * many partitions, and it makes it less likely that a partition will need to + * be spilled recursively. Another benefit of having more, smaller partitions + * is that small hash tables may perform better than large ones due to memory + * caching effects. + * + * HASHAGG_PARTITION_MEM is the approximate amount of work_mem we should + * reserve for the partitions themselves (i.e. buffering of the files backing + * the partitions). This is sloppy, because we must reserve the memory before + * filling the hash table; but we choose the number of partitions at the time + * we need to spill. + * + * We also specify a min and max number of partitions per spill. Too few might + * mean a lot of wasted I/O from repeated spilling of the same tuples. Too + * many will result in lots of memory wasted buffering the spill files (and + * possibly pushing hidden costs to the OS for managing more files). + */ +#define HASHAGG_PARTITION_FACTOR 1.50 +#define HASHAGG_MIN_PARTITIONS 4 +#define HASHAGG_PARTITION_MEM (HASHAGG_MIN_PARTITIONS * BLCKSZ) + +/* + * Represents partitioned spill data for a single hashtable. + */ +typedef struct HashAggSpill +{ + int n_partitions; /* number of output partitions */ + int partition_bits; /* number of bits for partition mask + log2(n_partitions) parent partition bits */ + BufFile **partitions; /* output partition files */ + int64 *ntuples; /* number of tuples in each partition */ +} HashAggSpill; + +/* + * Represents work to be done for one pass of hash aggregation. Initially, + * only the input fields are set. If spilled to disk, also set the spill data. + */ +typedef struct HashAggBatch +{ + BufFile *input_file; /* input partition */ + int input_bits; /* number of bits for input partition mask */ + int64 input_tuples; /* number of tuples in this batch */ + int setno; /* grouping set */ + HashAggSpill spill; /* spill output */ +} HashAggBatch; + static void select_current_set(AggState *aggstate, int setno, bool is_hash); static void initialize_phase(AggState *aggstate, int newphase); static TupleTableSlot *fetch_input_tuple(AggState *aggstate); @@ -271,12 +337,35 @@ static void finalize_aggregates(AggState *aggstate, static TupleTableSlot *project_aggregates(AggState *aggstate); static Bitmapset *find_unaggregated_cols(AggState *aggstate); static bool find_unaggregated_cols_walker(Node *node, Bitmapset **colnos); -static void build_hash_table(AggState *aggstate); -static TupleHashEntryData *lookup_hash_entry(AggState *aggstate); +static void build_hash_table(AggState *aggstate, int setno, + int64 ngroups_estimate); +static void prepare_hash_slot(AggState *aggstate); +static void hash_recompile_expressions(AggState *aggstate); +static uint32 calculate_hash(AggState *aggstate); +static long hash_choose_num_buckets(AggState *aggstate, + long estimated_nbuckets, + Size memory); +static int hash_choose_num_spill_partitions(uint64 input_groups, + double hashentrysize); +static AggStatePerGroup lookup_hash_entry(AggState *aggstate, uint32 hash); static void lookup_hash_entries(AggState *aggstate); static TupleTableSlot *agg_retrieve_direct(AggState *aggstate); static void agg_fill_hash_table(AggState *aggstate); +static bool agg_refill_hash_table(AggState *aggstate); static TupleTableSlot *agg_retrieve_hash_table(AggState *aggstate); +static TupleTableSlot *agg_retrieve_hash_table_in_memory(AggState *aggstate); +static void hash_spill_init(HashAggSpill *spill, int input_bits, + uint64 input_tuples, double hashentrysize); +static Size hash_spill_tuple(HashAggSpill *spill, int input_bits, + TupleTableSlot *slot, uint32 hash); +static MinimalTuple hash_read_spilled(BufFile *file, uint32 *hashp); +static HashAggBatch *hash_batch_new(BufFile *input_file, int setno, + int64 input_tuples, int input_bits); +static void hash_finish_initial_spills(AggState *aggstate); +static void hash_spill_finish(AggState *aggstate, HashAggSpill *spill, + int setno, int input_bits); +static void hash_reset_spill(HashAggSpill *spill); +static void hash_reset_spills(AggState *aggstate); static Datum GetAggInitVal(Datum textInitVal, Oid transtype); static void build_pertrans_for_aggref(AggStatePerTrans pertrans, AggState *aggstate, EState *estate, @@ -1201,6 +1290,68 @@ project_aggregates(AggState *aggstate) return NULL; } +static bool +find_aggregated_cols_walker(Node *node, Bitmapset **colnos) +{ + if (node == NULL) + return false; + + if (IsA(node, Var)) + { + Var *var = (Var *) node; + + *colnos = bms_add_member(*colnos, var->varattno); + + return false; + } + return expression_tree_walker(node, find_aggregated_cols_walker, + (void *) colnos); +} + +/* + * find_aggregated_cols + * Construct a bitmapset of the column numbers of aggregated Vars + * appearing in our targetlist and qual (HAVING clause) + */ +static Bitmapset * +find_aggregated_cols(AggState *aggstate) +{ + Agg *node = (Agg *) aggstate->ss.ps.plan; + Bitmapset *colnos = NULL; + ListCell *temp; + + /* + * We only want the columns used by aggregations in the targetlist or qual + */ + if (node->plan.targetlist != NULL) + { + foreach(temp, (List *) node->plan.targetlist) + { + if (IsA(lfirst(temp), TargetEntry)) + { + Node *node = (Node *)((TargetEntry *)lfirst(temp))->expr; + if (IsA(node, Aggref) || IsA(node, GroupingFunc)) + find_aggregated_cols_walker(node, &colnos); + } + } + } + + if (node->plan.qual != NULL) + { + foreach(temp, (List *) node->plan.qual) + { + if (IsA(lfirst(temp), TargetEntry)) + { + Node *node = (Node *)((TargetEntry *)lfirst(temp))->expr; + if (IsA(node, Aggref) || IsA(node, GroupingFunc)) + find_aggregated_cols_walker(node, &colnos); + } + } + } + + return colnos; +} + /* * find_unaggregated_cols * Construct a bitmapset of the column numbers of un-aggregated Vars @@ -1254,46 +1405,84 @@ find_unaggregated_cols_walker(Node *node, Bitmapset **colnos) * for each entry. * * We have a separate hashtable and associated perhash data structure for each - * grouping set for which we're doing hashing. + * grouping set for which we're doing hashing. If setno is -1, build hash + * tables for all grouping sets. Otherwise, build only for the specified + * grouping set. * * The contents of the hash tables always live in the hashcontext's per-tuple * memory context (there is only one of these for all tables together, since * they are all reset at the same time). */ static void -build_hash_table(AggState *aggstate) +build_hash_table(AggState *aggstate, int setno, long ngroups_estimate) { - MemoryContext tmpmem = aggstate->tmpcontext->ecxt_per_tuple_memory; - Size additionalsize; - int i; + MemoryContext tmpmem = aggstate->tmpcontext->ecxt_per_tuple_memory; + Size additionalsize; + int i; Assert(aggstate->aggstrategy == AGG_HASHED || aggstate->aggstrategy == AGG_MIXED); + /* + * Used to make sure initial hash table allocation does not exceed + * work_mem. Note that the estimate does not include space for + * pass-by-reference transition data values, nor for the representative + * tuple of each group. + */ additionalsize = aggstate->numtrans * sizeof(AggStatePerGroupData); for (i = 0; i < aggstate->num_hashes; ++i) { AggStatePerHash perhash = &aggstate->perhash[i]; + int64 ngroups; + long nbuckets; + Size memory; Assert(perhash->aggnode->numGroups > 0); if (perhash->hashtable) - ResetTupleHashTable(perhash->hashtable); - else - perhash->hashtable = BuildTupleHashTableExt(&aggstate->ss.ps, - perhash->hashslot->tts_tupleDescriptor, - perhash->numCols, - perhash->hashGrpColIdxHash, - perhash->eqfuncoids, - perhash->hashfunctions, - perhash->aggnode->grpCollations, - perhash->aggnode->numGroups, - additionalsize, - aggstate->ss.ps.state->es_query_cxt, - aggstate->hashcontext->ecxt_per_tuple_memory, - tmpmem, - DO_AGGSPLIT_SKIPFINAL(aggstate->aggsplit)); + DestroyTupleHashTable(perhash->hashtable); + perhash->hashtable = NULL; + + /* + * If we are building a hash table for only a single grouping set, + * skip the others. + */ + if (setno >= 0 && setno != i) + continue; + + /* + * Use an estimate from execution time if we have it; otherwise fall + * back to the planner estimate. + */ + ngroups = ngroups_estimate > 0 ? + ngroups_estimate : perhash->aggnode->numGroups; + + /* divide memory by the number of hash tables we are initializing */ + memory = (long)work_mem * 1024L / + (setno >= 0 ? 1 : aggstate->num_hashes); + + /* choose reasonable number of buckets per hashtable */ + nbuckets = hash_choose_num_buckets(aggstate, ngroups, memory); + + perhash->hashtable = BuildTupleHashTableExt(&aggstate->ss.ps, + perhash->hashslot->tts_tupleDescriptor, + perhash->numCols, + perhash->hashGrpColIdxHash, + perhash->eqfuncoids, + perhash->hashfunctions, + perhash->aggnode->grpCollations, + nbuckets, + additionalsize, + aggstate->ss.ps.state->es_query_cxt, + aggstate->hashcontext->ecxt_per_tuple_memory, + tmpmem, + DO_AGGSPLIT_SKIPFINAL(aggstate->aggsplit)); } + + aggstate->hash_mem_current = MemoryContextMemAllocated( + aggstate->hashcontext->ecxt_per_tuple_memory, true); + aggstate->hash_ngroups_current = 0; + aggstate->hash_no_new_groups = false; } /* @@ -1325,6 +1514,7 @@ static void find_hash_columns(AggState *aggstate) { Bitmapset *base_colnos; + Bitmapset *aggregated_colnos; List *outerTlist = outerPlanState(aggstate)->plan->targetlist; int numHashes = aggstate->num_hashes; EState *estate = aggstate->ss.ps.state; @@ -1332,11 +1522,13 @@ find_hash_columns(AggState *aggstate) /* Find Vars that will be needed in tlist and qual */ base_colnos = find_unaggregated_cols(aggstate); + aggregated_colnos = find_aggregated_cols(aggstate); for (j = 0; j < numHashes; ++j) { AggStatePerHash perhash = &aggstate->perhash[j]; Bitmapset *colnos = bms_copy(base_colnos); + Bitmapset *allNeededColsInput; AttrNumber *grpColIdx = perhash->aggnode->grpColIdx; List *hashTlist = NIL; TupleDesc hashDesc; @@ -1383,6 +1575,19 @@ find_hash_columns(AggState *aggstate) for (i = 0; i < perhash->numCols; i++) colnos = bms_add_member(colnos, grpColIdx[i]); + /* + * Track the necessary columns from the input. This is important for + * spilling tuples so that we don't waste disk space with unneeded + * columns. + */ + allNeededColsInput = bms_union(colnos, aggregated_colnos); + perhash->numNeededColsInput = 0; + perhash->allNeededColsInput = palloc( + bms_num_members(allNeededColsInput) * sizeof(AttrNumber)); + + while ((i = bms_first_member(allNeededColsInput)) >= 0) + perhash->allNeededColsInput[perhash->numNeededColsInput++] = i; + /* * First build mapping for columns directly hashed. These are the * first, because they'll be accessed when computing hash values and @@ -1435,42 +1640,31 @@ find_hash_columns(AggState *aggstate) /* * Estimate per-hash-table-entry overhead for the planner. - * - * Note that the estimate does not include space for pass-by-reference - * transition data values, nor for the representative tuple of each group. - * Nor does this account of the target fill-factor and growth policy of the - * hash table. */ Size -hash_agg_entry_size(int numAggs) +hash_agg_entry_size(int numAggs, Size tupleWidth, Size transitionSpace) { - Size entrysize; - - /* This must match build_hash_table */ - entrysize = sizeof(TupleHashEntryData) + - numAggs * sizeof(AggStatePerGroupData); - entrysize = MAXALIGN(entrysize); - - return entrysize; + return + /* key */ + MAXALIGN(SizeofMinimalTupleHeader) + + MAXALIGN(tupleWidth) + + /* data */ + MAXALIGN(sizeof(TupleHashEntryData) + + numAggs * sizeof(AggStatePerGroupData)) + + transitionSpace; } /* - * Find or create a hashtable entry for the tuple group containing the current - * tuple (already set in tmpcontext's outertuple slot), in the current grouping - * set (which the caller must have selected - note that initialize_aggregate - * depends on this). - * - * When called, CurrentMemoryContext should be the per-query context. + * Extract the attributes that make up the grouping key into the + * hashslot. This is necessary to compute the hash of the grouping key. */ -static TupleHashEntryData * -lookup_hash_entry(AggState *aggstate) +static void +prepare_hash_slot(AggState *aggstate) { - TupleTableSlot *inputslot = aggstate->tmpcontext->ecxt_outertuple; - AggStatePerHash perhash = &aggstate->perhash[aggstate->current_set]; - TupleTableSlot *hashslot = perhash->hashslot; - TupleHashEntryData *entry; - bool isnew; - int i; + TupleTableSlot *inputslot = aggstate->tmpcontext->ecxt_outertuple; + AggStatePerHash perhash = &aggstate->perhash[aggstate->current_set]; + TupleTableSlot *hashslot = perhash->hashslot; + int i; /* transfer just the needed columns into hashslot */ slot_getsomeattrs(inputslot, perhash->largestGrpColIdx); @@ -1484,14 +1678,185 @@ lookup_hash_entry(AggState *aggstate) hashslot->tts_isnull[i] = inputslot->tts_isnull[varNumber]; } ExecStoreVirtualTuple(hashslot); +} + +/* + * Recompile the expressions for advancing aggregates while hashing. This is + * necessary for certain kinds of state changes that affect the resulting + * expression. For instance, changing aggstate->hash_spilled or + * aggstate->ss.ps.outerops require recompilation. + */ +static void +hash_recompile_expressions(AggState *aggstate) +{ + AggStatePerPhase phase; + + Assert(aggstate->aggstrategy == AGG_HASHED || + aggstate->aggstrategy == AGG_MIXED); + + if (aggstate->aggstrategy == AGG_HASHED) + phase = &aggstate->phases[0]; + else /* AGG_MIXED */ + phase = &aggstate->phases[1]; + + phase->evaltrans = ExecBuildAggTrans( + aggstate, phase, + aggstate->aggstrategy == AGG_MIXED ? true : false, /* dosort */ + true, /* dohash */ + aggstate->hash_spilled /* spilled */); +} + +/* + * Calculate the hash value for a tuple. It's useful to do this outside of the + * hash table so that we can reuse saved hash values rather than recomputing. + */ +static uint32 +calculate_hash(AggState *aggstate) +{ + AggStatePerHash perhash = &aggstate->perhash[aggstate->current_set]; + TupleHashTable hashtable = perhash->hashtable; + MemoryContext oldContext; + uint32 hash; + + /* set up data needed by hash and match functions */ + hashtable->inputslot = perhash->hashslot; + hashtable->in_hash_funcs = hashtable->tab_hash_funcs; + hashtable->cur_eq_func = hashtable->tab_eq_func; + + /* Need to run the hash functions in short-lived context */ + oldContext = MemoryContextSwitchTo(hashtable->tempcxt); + + hash = TupleHashTableHash(hashtable->hashtab, NULL); + + MemoryContextSwitchTo(oldContext); + + return hash; +} + +/* + * Choose a reasonable number of buckets for the initial hash table size. + */ +static long +hash_choose_num_buckets(AggState *aggstate, long ngroups, Size memory) +{ + long max_nbuckets; + int log2_ngroups; + long nbuckets; + + max_nbuckets = memory / aggstate->hashentrysize; + + /* + * Lowest power of two greater than ngroups, without exceeding + * max_nbuckets. + */ + for (log2_ngroups = 1, nbuckets = 2; + nbuckets < ngroups && nbuckets < max_nbuckets; + log2_ngroups++, nbuckets <<= 1); + + if (nbuckets > max_nbuckets && nbuckets > 2) + nbuckets >>= 1; + + return nbuckets; +} + +/* + * Determine the number of partitions to create when spilling. + */ +static int +hash_choose_num_spill_partitions(uint64 input_groups, double hashentrysize) +{ + Size mem_needed; + int partition_limit; + int npartitions; + + /* + * Avoid creating so many partitions that the memory requirements of the + * open partition files (estimated at BLCKSZ for buffering) are greater + * than 1/4 of work_mem. + */ + partition_limit = (work_mem * 1024L * 0.25) / BLCKSZ; + + /* pessimistically estimate that each input tuple creates a new group */ + mem_needed = HASHAGG_PARTITION_FACTOR * input_groups * hashentrysize; + + /* make enough partitions so that each one is likely to fit in memory */ + npartitions = 1 + (mem_needed / (work_mem * 1024L)); + + if (npartitions > partition_limit) + npartitions = partition_limit; + + if (npartitions < HASHAGG_MIN_PARTITIONS) + npartitions = HASHAGG_MIN_PARTITIONS; + if (npartitions > HASHAGG_MAX_PARTITIONS) + npartitions = HASHAGG_MAX_PARTITIONS; + + return npartitions; +} + +/* + * Find or create a hashtable entry for the tuple group containing the current + * tuple (already set in tmpcontext's outertuple slot), in the current grouping + * set (which the caller must have selected - note that initialize_aggregate + * depends on this). + * + * When called, CurrentMemoryContext should be the per-query context. + * + * If the hash table is at the memory limit, then only find existing hashtable + * entries; don't create new ones. If a tuple's group is not already present + * in the hash table for the current grouping set, return NULL and the caller + * will spill it to disk. + */ +static AggStatePerGroup +lookup_hash_entry(AggState *aggstate, uint32 hash) +{ + AggStatePerHash perhash = &aggstate->perhash[aggstate->current_set]; + TupleTableSlot *hashslot = perhash->hashslot; + TupleHashEntryData *entry; + bool isnew = false; + bool *p_isnew; + + /* if hash table already spilled, don't create new entries */ + p_isnew = aggstate->hash_no_new_groups ? NULL : &isnew; /* find or create the hashtable entry using the filtered tuple */ - entry = LookupTupleHashEntry(perhash->hashtable, hashslot, &isnew); + entry = LookupTupleHashEntryHash(perhash->hashtable, hashslot, p_isnew, + hash); + + if (entry == NULL) + return NULL; if (isnew) { - AggStatePerGroup pergroup; - int transno; + AggStatePerGroup pergroup; + int transno; + + aggstate->hash_ngroups_current++; + + aggstate->hash_mem_current = MemoryContextMemAllocated( + aggstate->hashcontext->ecxt_per_tuple_memory, true); + + if (aggstate->hash_mem_current > aggstate->hash_mem_peak) + aggstate->hash_mem_peak = aggstate->hash_mem_current; + + /* + * Check whether we need to spill. For small values of work_mem, the + * empty hash tables might exceed it; so don't spill unless there's at + * least one group in the hash table. + */ + if (aggstate->hash_ngroups_current > 0 && + (aggstate->hash_mem_current > aggstate->hash_mem_limit || + aggstate->hash_ngroups_current > aggstate->hash_ngroups_limit)) + { + aggstate->hash_no_new_groups = true; + if (!aggstate->hash_spilled) + { + aggstate->hash_spilled = true; + aggstate->hash_spills = palloc0( + sizeof(HashAggSpill) * aggstate->num_hashes); + + hash_recompile_expressions(aggstate); + } + } pergroup = (AggStatePerGroup) MemoryContextAlloc(perhash->hashtable->tablecxt, @@ -1511,7 +1876,7 @@ lookup_hash_entry(AggState *aggstate) } } - return entry; + return entry->additional; } /* @@ -1519,18 +1884,74 @@ lookup_hash_entry(AggState *aggstate) * returning an array of pergroup pointers suitable for advance_aggregates. * * Be aware that lookup_hash_entry can reset the tmpcontext. + * + * Some entries may be left NULL if we have reached the limit and have begun + * to spill. The same tuple will belong to different groups for each set, so + * may match a group already in memory for one set and match a group not in + * memory for another set. If we have begun to spill and a tuple doesn't match + * a group in memory for a particular set, it will be spilled. + * + * NB: It's possible to spill the same tuple for several different grouping + * sets. This may seem wasteful, but it's actually a trade-off: if we spill + * the tuple multiple times for multiple grouping sets, it can be partitioned + * for each grouping set, making the refilling of the hash table very + * efficient. */ static void lookup_hash_entries(AggState *aggstate) { - int numHashes = aggstate->num_hashes; AggStatePerGroup *pergroup = aggstate->hash_pergroup; int setno; - for (setno = 0; setno < numHashes; setno++) + for (setno = 0; setno < aggstate->num_hashes; setno++) { + uint32 hash; + select_current_set(aggstate, setno, true); - pergroup[setno] = lookup_hash_entry(aggstate)->additional; + prepare_hash_slot(aggstate); + hash = calculate_hash(aggstate); + pergroup[setno] = lookup_hash_entry(aggstate, hash); + + /* check to see if we need to spill the tuple for this grouping set */ + if (pergroup[setno] == NULL) + { + AggStatePerHash perhash = &aggstate->perhash[setno]; + TupleTableSlot *inputslot = aggstate->tmpcontext->ecxt_outertuple; + TupleTableSlot *spillslot = aggstate->hash_spill_slot; + HashAggSpill *spill = &aggstate->hash_spills[setno]; + int idx; + + if (spill->partitions == NULL) + hash_spill_init(spill, 0, perhash->aggnode->numGroups, + aggstate->hashentrysize); + + /* + * Copy only necessary attributes to spill slot before writing to + * disk. + */ + ExecClearTuple(spillslot); + memset(spillslot->tts_isnull, true, + spillslot->tts_tupleDescriptor->natts); + + /* deserialize needed attributes */ + if (perhash->numNeededColsInput > 0) + { + int maxNeededAttrIdx = perhash->numNeededColsInput - 1; + AttrNumber maxNeededAttr = + perhash->allNeededColsInput[maxNeededAttrIdx]; + slot_getsomeattrs(inputslot, maxNeededAttr); + } + + for (idx = 0; idx < perhash->numNeededColsInput; idx++) + { + AttrNumber att = perhash->allNeededColsInput[idx]; + spillslot->tts_values[att-1] = inputslot->tts_values[att-1]; + spillslot->tts_isnull[att-1] = inputslot->tts_isnull[att-1]; + } + + ExecStoreVirtualTuple(spillslot); + aggstate->hash_disk_used += hash_spill_tuple(spill, 0, spillslot, hash); + } } } @@ -1853,6 +2274,12 @@ agg_retrieve_direct(AggState *aggstate) if (TupIsNull(outerslot)) { /* no more outer-plan tuples available */ + + /* if we built hash tables, finalize any spills */ + if (aggstate->aggstrategy == AGG_MIXED && + aggstate->current_phase == 1) + hash_finish_initial_spills(aggstate); + if (hasGroupingSets) { aggstate->input_done = true; @@ -1955,6 +2382,9 @@ agg_fill_hash_table(AggState *aggstate) ResetExprContext(aggstate->tmpcontext); } + /* finalize spills, if any */ + hash_finish_initial_spills(aggstate); + aggstate->table_filled = true; /* Initialize to walk the first hash table */ select_current_set(aggstate, 0, true); @@ -1962,11 +2392,175 @@ agg_fill_hash_table(AggState *aggstate) &aggstate->perhash[0].hashiter); } +/* + * If any data was spilled during hash aggregation, reset the hash table and + * reprocess one batch of spilled data. After reprocessing a batch, the hash + * table will again contain data, ready to be consumed by + * agg_retrieve_hash_table_in_memory(). + * + * Should only be called after all in memory hash table entries have been + * consumed. + * + * Return false when input is exhausted and there's no more work to be done; + * otherwise return true. + */ +static bool +agg_refill_hash_table(AggState *aggstate) +{ + HashAggBatch *batch; + + if (aggstate->hash_batches == NIL) + return false; + + /* + * Each spill file contains spilled data for only a single grouping + * set. We want to ignore all others, which is done by setting the other + * pergroups to NULL. + */ + memset(aggstate->all_pergroups, 0, + sizeof(AggStatePerGroup) * + (aggstate->maxsets + aggstate->num_hashes)); + + batch = linitial(aggstate->hash_batches); + aggstate->hash_batches = list_delete_first(aggstate->hash_batches); + + /* + * Free memory and rebuild a single hash table for this batch's grouping + * set. + */ + ReScanExprContext(aggstate->hashcontext); + + /* estimate the number of groups to be the number of input tuples */ + build_hash_table(aggstate, batch->setno, batch->input_tuples); + + Assert(aggstate->current_phase == 0); + + if (aggstate->phase->aggstrategy == AGG_MIXED) + { + aggstate->current_phase = 1; + aggstate->phase = &aggstate->phases[aggstate->current_phase]; + } + + /* + * The first pass (agg_fill_hash_table) reads whatever kind of slot comes + * from the outer plan, and considers the slot fixed. But spilled tuples + * are always MinimalTuples, so if that's different from the outer plan we + * need to change it and recompile the aggregate expressions. + */ + if (aggstate->ss.ps.outerops != &TTSOpsMinimalTuple) + { + aggstate->ss.ps.outerops = &TTSOpsMinimalTuple; + hash_recompile_expressions(aggstate); + } + + for (;;) { + TupleTableSlot *slot = aggstate->hash_spill_slot; + MinimalTuple tuple; + uint32 hash; + + CHECK_FOR_INTERRUPTS(); + + tuple = hash_read_spilled(batch->input_file, &hash); + if (tuple == NULL) + break; + + ExecStoreMinimalTuple(tuple, slot, true); + aggstate->tmpcontext->ecxt_outertuple = slot; + + select_current_set(aggstate, batch->setno, true); + prepare_hash_slot(aggstate); + aggstate->hash_pergroup[batch->setno] = lookup_hash_entry(aggstate, hash); + + /* if there's no memory for a new group, spill */ + if (aggstate->hash_pergroup[batch->setno] == NULL) + { + if (batch->spill.partitions == NULL) + { + /* + * Estimate the number of groups for this batch as the total + * number of tuples in its input file. Although that's a worst + * case, it's not bad here for two reasons: (1) overestimating + * is better than underestimating; and (2) we've already + * scanned the relation once, so it's likely that we've + * already finalized many of the common values. + */ + hash_spill_init(&batch->spill, batch->input_bits, + batch->input_tuples, aggstate->hashentrysize); + } + + aggstate->hash_disk_used += hash_spill_tuple( + &batch->spill, batch->input_bits, slot, hash); + } + + /* Advance the aggregates (or combine functions) */ + advance_aggregates(aggstate); + + /* + * Reset per-input-tuple context after each tuple, but note that the + * hash lookups do this too + */ + ResetExprContext(aggstate->tmpcontext); + } + + BufFileClose(batch->input_file); + + aggstate->current_phase = 0; + aggstate->phase = &aggstate->phases[aggstate->current_phase]; + + /* update hashentrysize estimate based on contents */ + if (aggstate->hash_ngroups_current > 0) + { + aggstate->hashentrysize = (double)aggstate->hash_mem_current / + (double)aggstate->hash_ngroups_current; + } + + hash_spill_finish(aggstate, &batch->spill, batch->setno, + batch->input_bits); + + pfree(batch); + + /* Initialize to walk the first hash table */ + select_current_set(aggstate, 0, true); + ResetTupleHashIterator(aggstate->perhash[0].hashtable, + &aggstate->perhash[0].hashiter); + + return true; +} + /* * ExecAgg for hashed case: retrieving groups from hash table + * + * After exhausting in-memory tuples, also try refilling the hash table using + * previously-spilled tuples. Only returns NULL after all in-memory and + * spilled tuples are exhausted. */ static TupleTableSlot * agg_retrieve_hash_table(AggState *aggstate) +{ + TupleTableSlot *result = NULL; + + while (result == NULL) + { + result = agg_retrieve_hash_table_in_memory(aggstate); + if (result == NULL) + { + if (!agg_refill_hash_table(aggstate)) + { + aggstate->agg_done = true; + break; + } + } + } + + return result; +} + +/* + * Retrieve the groups from the in-memory hash tables without considering any + * spilled tuples. + */ +static TupleTableSlot * +agg_retrieve_hash_table_in_memory(AggState *aggstate) { ExprContext *econtext; AggStatePerAgg peragg; @@ -1995,7 +2589,7 @@ agg_retrieve_hash_table(AggState *aggstate) * We loop retrieving groups until we find one satisfying * aggstate->ss.ps.qual */ - while (!aggstate->agg_done) + for (;;) { TupleTableSlot *hashslot = perhash->hashslot; int i; @@ -2026,8 +2620,6 @@ agg_retrieve_hash_table(AggState *aggstate) } else { - /* No more hashtables, so done */ - aggstate->agg_done = true; return NULL; } } @@ -2084,6 +2676,281 @@ agg_retrieve_hash_table(AggState *aggstate) return NULL; } +/* + * hash_spill_init + * + * Called after we determined that spilling is necessary. Chooses the number + * of partitions to create, and initializes them. + */ +static void +hash_spill_init(HashAggSpill *spill, int input_bits, uint64 input_groups, + double hashentrysize) +{ + int npartitions; + int partition_bits; + + npartitions = hash_choose_num_spill_partitions(input_groups, + hashentrysize); + partition_bits = my_log2(npartitions); + + /* make sure that we don't exhaust the hash bits */ + if (partition_bits + input_bits >= 32) + partition_bits = 32 - input_bits; + + /* number of partitions will be a power of two */ + npartitions = 1L << partition_bits; + + spill->partition_bits = partition_bits; + spill->n_partitions = npartitions; + spill->partitions = palloc0(sizeof(BufFile *) * npartitions); + spill->ntuples = palloc0(sizeof(int64) * npartitions); +} + +/* + * hash_spill_tuple + * + * No room for new groups in the hash table. Save for later in the appropriate + * partition spill file. + */ +static Size +hash_spill_tuple(HashAggSpill *spill, int input_bits, TupleTableSlot *slot, + uint32 hash) +{ + int partition; + MinimalTuple tuple; + BufFile *file; + int written; + int total_written = 0; + bool shouldFree; + + Assert(spill->partitions != NULL); + + /* + * When spilling tuples from the input, the slot will be virtual + * (containing only the needed attributes and the rest as NULL), and we + * need to materialize the minimal tuple. When spilling tuples + * recursively, the slot will hold a minimal tuple already. + */ + tuple = ExecFetchSlotMinimalTuple(slot, &shouldFree); + + if (spill->partition_bits == 0) + partition = 0; + else + partition = (hash << input_bits) >> + (32 - spill->partition_bits); + + spill->ntuples[partition]++; + + if (spill->partitions[partition] == NULL) + spill->partitions[partition] = BufFileCreateTemp(false); + file = spill->partitions[partition]; + + written = BufFileWrite(file, (void *) &hash, sizeof(uint32)); + if (written != sizeof(uint32)) + ereport(ERROR, + (errcode_for_file_access(), + errmsg("could not write to HashAgg temporary file: %m"))); + total_written += written; + + written = BufFileWrite(file, (void *) tuple, tuple->t_len); + if (written != tuple->t_len) + ereport(ERROR, + (errcode_for_file_access(), + errmsg("could not write to HashAgg temporary file: %m"))); + total_written += written; + + if (shouldFree) + pfree(tuple); + + return total_written; +} + +/* + * read_spilled_tuple + * read the next tuple from a batch file. Return NULL if no more. + */ +static MinimalTuple +hash_read_spilled(BufFile *file, uint32 *hashp) +{ + MinimalTuple tuple; + uint32 t_len; + size_t nread; + uint32 hash; + + nread = BufFileRead(file, &hash, sizeof(uint32)); + if (nread == 0) + return NULL; + if (nread != sizeof(uint32)) + ereport(ERROR, + (errcode_for_file_access(), + errmsg("could not read from HashAgg temporary file: %m"))); + if (hashp != NULL) + *hashp = hash; + + nread = BufFileRead(file, &t_len, sizeof(t_len)); + if (nread != sizeof(uint32)) + ereport(ERROR, + (errcode_for_file_access(), + errmsg("could not read from HashAgg temporary file: %m"))); + + tuple = (MinimalTuple) palloc(t_len); + tuple->t_len = t_len; + + nread = BufFileRead(file, (void *)((char *)tuple + sizeof(uint32)), + t_len - sizeof(uint32)); + if (nread != t_len - sizeof(uint32)) + ereport(ERROR, + (errcode_for_file_access(), + errmsg("could not read from HashAgg temporary file: %m"))); + + return tuple; +} + +/* + * new_hashagg_batch + * + * Construct a HashAggBatch item, which represents one iteration of HashAgg to + * be done. Should be called in the aggregate's memory context. + */ +static HashAggBatch * +hash_batch_new(BufFile *input_file, int setno, int64 input_tuples, + int input_bits) +{ + HashAggBatch *batch = palloc0(sizeof(HashAggBatch)); + + batch->input_file = input_file; + batch->input_bits = input_bits; + batch->input_tuples = input_tuples; + batch->setno = setno; + + /* batch->spill will be set only after spilling this batch */ + + return batch; +} + +/* + * hash_finish_initial_spills + * + * After a HashAggBatch has been processed, it may have spilled tuples to + * disk. If so, turn the spilled partitions into new batches that must later + * be executed. + */ +static void +hash_finish_initial_spills(AggState *aggstate) +{ + int setno; + + if (aggstate->hash_spills == NULL) + return; + + /* update hashentrysize estimate based on contents */ + Assert(aggstate->hash_ngroups_current > 0); + aggstate->hashentrysize = (double)aggstate->hash_mem_current / + (double)aggstate->hash_ngroups_current; + + for (setno = 0; setno < aggstate->num_hashes; setno++) + hash_spill_finish(aggstate, &aggstate->hash_spills[setno], setno, 0); + + pfree(aggstate->hash_spills); + aggstate->hash_spills = NULL; +} + +/* + * hash_spill_finish + * + * Transform spill files into new batches. + */ +static void +hash_spill_finish(AggState *aggstate, HashAggSpill *spill, int setno, int input_bits) +{ + int i; + + if (spill->n_partitions == 0) + return; /* didn't spill */ + + for (i = 0; i < spill->n_partitions; i++) + { + BufFile *file = spill->partitions[i]; + MemoryContext oldContext; + HashAggBatch *new_batch; + + /* partition is empty */ + if (file == NULL) + continue; + + /* rewind file for reading */ + if (BufFileSeek(file, 0, 0L, SEEK_SET)) + ereport(ERROR, + (errcode_for_file_access(), + errmsg("could not rewind HashAgg temporary file: %m"))); + + oldContext = MemoryContextSwitchTo(aggstate->ss.ps.state->es_query_cxt); + new_batch = hash_batch_new(file, setno, spill->ntuples[i], + spill->partition_bits + input_bits); + aggstate->hash_batches = lappend(aggstate->hash_batches, new_batch); + aggstate->hash_batches_used++; + MemoryContextSwitchTo(oldContext); + } + + pfree(spill->ntuples); + pfree(spill->partitions); +} + +/* + * Clear a HashAggSpill, free its memory, and close its files. + */ +static void +hash_reset_spill(HashAggSpill *spill) +{ + int i; + for (i = 0; i < spill->n_partitions; i++) + { + BufFile *file = spill->partitions[i]; + + if (file != NULL) + BufFileClose(file); + } + if (spill->ntuples != NULL) + pfree(spill->ntuples); + if (spill->partitions != NULL) + pfree(spill->partitions); +} + +/* + * Find and reset all active HashAggSpills. + */ +static void +hash_reset_spills(AggState *aggstate) +{ + ListCell *lc; + + if (aggstate->hash_spills != NULL) + { + int setno; + + for (setno = 0; setno < aggstate->num_hashes; setno++) + hash_reset_spill(&aggstate->hash_spills[setno]); + + pfree(aggstate->hash_spills); + aggstate->hash_spills = NULL; + } + + foreach(lc, aggstate->hash_batches) + { + HashAggBatch *batch = (HashAggBatch*) lfirst(lc); + if (batch->input_file != NULL) + { + BufFileClose(batch->input_file); + batch->input_file = NULL; + } + hash_reset_spill(&batch->spill); + pfree(batch); + } + list_free(aggstate->hash_batches); + aggstate->hash_batches = NIL; +} + + /* ----------------- * ExecInitAgg * @@ -2268,6 +3135,10 @@ ExecInitAgg(Agg *node, EState *estate, int eflags) aggstate->ss.ps.outeropsfixed = false; } + if (use_hashing) + aggstate->hash_spill_slot = ExecInitExtraTupleSlot(estate, scanDesc, + &TTSOpsMinimalTuple); + /* * Initialize result type, slot and projection. */ @@ -2493,11 +3364,41 @@ ExecInitAgg(Agg *node, EState *estate, int eflags) */ if (use_hashing) { + Plan *outerplan = outerPlan(node); + /* this is an array of pointers, not structures */ aggstate->hash_pergroup = pergroups; + aggstate->hashentrysize = hash_agg_entry_size( + aggstate->numtrans, outerplan->plan_width, node->transitionSpace); + + /* + * Initialize the thresholds at which we stop creating new hash entries + * and start spilling. + */ + if (hashagg_mem_overflow) + aggstate->hash_mem_limit = SIZE_MAX; + else if (work_mem * 1024L > HASHAGG_PARTITION_MEM * 2) + aggstate->hash_mem_limit = + work_mem * 1024L - HASHAGG_PARTITION_MEM; + else + aggstate->hash_mem_limit = work_mem * 1024L; + + /* + * Set a separate limit on the maximum number of groups to + * create. This is important for aggregates where the initial state + * size is small, but aggtransspace is large. + */ + if (hashagg_mem_overflow) + aggstate->hash_ngroups_limit = LONG_MAX; + else if (aggstate->hash_mem_limit > aggstate->hashentrysize) + aggstate->hash_ngroups_limit = + aggstate->hash_mem_limit / aggstate->hashentrysize; + else + aggstate->hash_ngroups_limit = 1; + find_hash_columns(aggstate); - build_hash_table(aggstate); + build_hash_table(aggstate, -1, 0); aggstate->table_filled = false; } @@ -2903,7 +3804,7 @@ ExecInitAgg(Agg *node, EState *estate, int eflags) else Assert(false); - phase->evaltrans = ExecBuildAggTrans(aggstate, phase, dosort, dohash); + phase->evaltrans = ExecBuildAggTrans(aggstate, phase, dosort, dohash, false); } @@ -3398,6 +4299,8 @@ ExecEndAgg(AggState *node) if (node->sort_out) tuplesort_end(node->sort_out); + hash_reset_spills(node); + for (transno = 0; transno < node->numtrans; transno++) { AggStatePerTrans pertrans = &node->pertrans[transno]; @@ -3453,12 +4356,13 @@ ExecReScanAgg(AggState *node) return; /* - * If we do have the hash table, and the subplan does not have any - * parameter changes, and none of our own parameter changes affect - * input expressions of the aggregated functions, then we can just - * rescan the existing hash table; no need to build it again. + * If we do have the hash table, and it never spilled, and the subplan + * does not have any parameter changes, and none of our own parameter + * changes affect input expressions of the aggregated functions, then + * we can just rescan the existing hash table; no need to build it + * again. */ - if (outerPlan->chgParam == NULL && + if (outerPlan->chgParam == NULL && !node->hash_spilled && !bms_overlap(node->ss.ps.chgParam, aggnode->aggParams)) { ResetTupleHashIterator(node->perhash[0].hashtable, @@ -3515,9 +4419,21 @@ ExecReScanAgg(AggState *node) */ if (node->aggstrategy == AGG_HASHED || node->aggstrategy == AGG_MIXED) { + hash_reset_spills(node); + + node->hash_spilled = false; + node->hash_no_new_groups = false; + node->hash_mem_current = 0; + node->hash_ngroups_current = 0; + + /* reset stats */ + node->hash_mem_peak = 0; + node->hash_disk_used = 0; + node->hash_batches_used = 0; + ReScanExprContext(node->hashcontext); /* Rebuild an empty hash table */ - build_hash_table(node); + build_hash_table(node, -1, 0); node->table_filled = false; /* iterator will be reset when the table is filled */ } diff --git a/src/backend/jit/llvm/llvmjit_expr.c b/src/backend/jit/llvm/llvmjit_expr.c index ffd887c71aa..93517d03819 100644 --- a/src/backend/jit/llvm/llvmjit_expr.c +++ b/src/backend/jit/llvm/llvmjit_expr.c @@ -2082,6 +2082,7 @@ llvm_compile_expr(ExprState *state) } case EEOP_AGG_INIT_TRANS: + case EEOP_AGG_INIT_TRANS_SPILLED: { AggState *aggstate; AggStatePerTrans pertrans; @@ -2092,6 +2093,7 @@ llvm_compile_expr(ExprState *state) LLVMValueRef v_allpergroupsp; LLVMValueRef v_pergroupp; + LLVMValueRef v_pergroup_allaggs; LLVMValueRef v_setoff, v_transno; @@ -2119,11 +2121,32 @@ llvm_compile_expr(ExprState *state) "aggstate.all_pergroups"); v_setoff = l_int32_const(op->d.agg_init_trans.setoff); v_transno = l_int32_const(op->d.agg_init_trans.transno); - v_pergroupp = - LLVMBuildGEP(b, - l_load_gep1(b, v_allpergroupsp, v_setoff, ""), - &v_transno, 1, ""); + v_pergroup_allaggs = l_load_gep1(b, v_allpergroupsp, v_setoff, ""); + /* + * When no tuples at all have spilled, we avoid adding this + * extra branch. But after some tuples have spilled, this + * branch is necessary, so we recompile the expression + * using a new opcode. + */ + if (opcode == EEOP_AGG_INIT_TRANS_SPILLED) + { + LLVMBasicBlockRef b_check_notransvalue = l_bb_before_v( + opblocks[i + 1], "op.%d.check_notransvalue", i); + + LLVMBuildCondBr( + b, + LLVMBuildICmp(b, LLVMIntEQ, + LLVMBuildPtrToInt( + b, v_pergroup_allaggs, TypeSizeT, ""), + l_sizet_const(0), ""), + opblocks[i + 1], + b_check_notransvalue); + + LLVMPositionBuilderAtEnd(b, b_check_notransvalue); + } + + v_pergroupp = LLVMBuildGEP(b, v_pergroup_allaggs, &v_transno, 1, ""); v_notransvalue = l_load_struct_gep(b, v_pergroupp, FIELDNO_AGGSTATEPERGROUPDATA_NOTRANSVALUE, @@ -2180,6 +2203,7 @@ llvm_compile_expr(ExprState *state) } case EEOP_AGG_STRICT_TRANS_CHECK: + case EEOP_AGG_STRICT_TRANS_CHECK_SPILLED: { AggState *aggstate; LLVMValueRef v_setoff, @@ -2190,6 +2214,7 @@ llvm_compile_expr(ExprState *state) LLVMValueRef v_transnull; LLVMValueRef v_pergroupp; + LLVMValueRef v_pergroup_allaggs; int jumpnull = op->d.agg_strict_trans_check.jumpnull; @@ -2209,11 +2234,32 @@ llvm_compile_expr(ExprState *state) l_int32_const(op->d.agg_strict_trans_check.setoff); v_transno = l_int32_const(op->d.agg_strict_trans_check.transno); - v_pergroupp = - LLVMBuildGEP(b, - l_load_gep1(b, v_allpergroupsp, v_setoff, ""), - &v_transno, 1, ""); + v_pergroup_allaggs = l_load_gep1(b, v_allpergroupsp, v_setoff, ""); + + /* + * When no tuples at all have spilled, we avoid adding this + * extra branch. But after some tuples have spilled, this + * branch is necessary, so we recompile the expression + * using a new opcode. + */ + if (opcode == EEOP_AGG_STRICT_TRANS_CHECK_SPILLED) + { + LLVMBasicBlockRef b_check_transnull = l_bb_before_v( + opblocks[i + 1], "op.%d.check_transnull", i); + + LLVMBuildCondBr( + b, + LLVMBuildICmp(b, LLVMIntEQ, + LLVMBuildPtrToInt(b, v_pergroup_allaggs, + TypeSizeT, ""), + l_sizet_const(0), ""), + opblocks[jumpnull], + b_check_transnull); + + LLVMPositionBuilderAtEnd(b, b_check_transnull); + } + v_pergroupp = LLVMBuildGEP(b, v_pergroup_allaggs, &v_transno, 1, ""); v_transnull = l_load_struct_gep(b, v_pergroupp, FIELDNO_AGGSTATEPERGROUPDATA_TRANSVALUEISNULL, @@ -2229,7 +2275,9 @@ llvm_compile_expr(ExprState *state) } case EEOP_AGG_PLAIN_TRANS_BYVAL: + case EEOP_AGG_PLAIN_TRANS_BYVAL_SPILLED: case EEOP_AGG_PLAIN_TRANS: + case EEOP_AGG_PLAIN_TRANS_SPILLED: { AggState *aggstate; AggStatePerTrans pertrans; @@ -2255,6 +2303,7 @@ llvm_compile_expr(ExprState *state) LLVMValueRef v_pertransp; LLVMValueRef v_pergroupp; + LLVMValueRef v_pergroup_allaggs; LLVMValueRef v_retval; @@ -2282,10 +2331,33 @@ llvm_compile_expr(ExprState *state) "aggstate.all_pergroups"); v_setoff = l_int32_const(op->d.agg_trans.setoff); v_transno = l_int32_const(op->d.agg_trans.transno); - v_pergroupp = - LLVMBuildGEP(b, - l_load_gep1(b, v_allpergroupsp, v_setoff, ""), - &v_transno, 1, ""); + v_pergroup_allaggs = l_load_gep1(b, v_allpergroupsp, v_setoff, ""); + + /* + * When no tuples at all have spilled, we avoid adding this + * extra branch. But after some tuples have spilled, this + * branch is necessary, so we recompile the expression + * using a new opcode. + */ + if (opcode == EEOP_AGG_PLAIN_TRANS_BYVAL_SPILLED || + opcode == EEOP_AGG_PLAIN_TRANS_SPILLED) + { + LLVMBasicBlockRef b_advance_transval = l_bb_before_v( + opblocks[i + 1], "op.%d.advance_transval", i); + + LLVMBuildCondBr( + b, + LLVMBuildICmp(b, LLVMIntEQ, + LLVMBuildPtrToInt(b, v_pergroup_allaggs, + TypeSizeT, ""), + l_sizet_const(0), ""), + opblocks[i + 1], + b_advance_transval); + + LLVMPositionBuilderAtEnd(b, b_advance_transval); + } + + v_pergroupp = LLVMBuildGEP(b, v_pergroup_allaggs, &v_transno, 1, ""); v_fcinfo = l_ptr_const(fcinfo, l_ptr(StructFunctionCallInfoData)); diff --git a/src/backend/optimizer/path/costsize.c b/src/backend/optimizer/path/costsize.c index 3f0d2899635..1b3ea3321c6 100644 --- a/src/backend/optimizer/path/costsize.c +++ b/src/backend/optimizer/path/costsize.c @@ -77,6 +77,7 @@ #include "access/htup_details.h" #include "access/tsmapi.h" #include "executor/executor.h" +#include "executor/nodeAgg.h" #include "executor/nodeHash.h" #include "miscadmin.h" #include "nodes/makefuncs.h" @@ -2154,7 +2155,7 @@ cost_agg(Path *path, PlannerInfo *root, int numGroupCols, double numGroups, List *quals, Cost input_startup_cost, Cost input_total_cost, - double input_tuples) + double input_tuples, double input_width) { double output_tuples; Cost startup_cost; @@ -2220,20 +2221,69 @@ cost_agg(Path *path, PlannerInfo *root, total_cost += aggcosts->finalCost.per_tuple * numGroups; total_cost += cpu_tuple_cost * numGroups; output_tuples = numGroups; + + /* + * We don't need to compute the disk costs of hash aggregation here, + * because the planner does not choose hash aggregation for grouping + * sets that it doesn't expect to fit in memory. + */ } else { + double hashentrysize = hash_agg_entry_size( + aggcosts->numAggs, input_width, aggcosts->transitionSpace); + double nbatches = + (numGroups * hashentrysize) / (work_mem * 1024L); + double pages_written = 0.0; + double pages_read = 0.0; + /* must be AGG_HASHED */ startup_cost = input_total_cost; if (!enable_hashagg) startup_cost += disable_cost; startup_cost += aggcosts->transCost.startup; startup_cost += aggcosts->transCost.per_tuple * input_tuples; + /* cost of computing hash value */ startup_cost += (cpu_operator_cost * numGroupCols) * input_tuples; startup_cost += aggcosts->finalCost.startup; + + /* + * Add the disk costs of hash aggregation that spills to disk. + * + * Groups that go into the hash table stay in memory until finalized, + * so spilling and reprocessing tuples doesn't incur additional + * invocations of transCost or finalCost. Furthermore, the computed + * hash value is stored with the spilled tuples, so we don't incur + * extra invocations of the hash function. + * + * The disk cost depends on the depth of recursion; each level + * requiring one additional write and then read of a tuple. Writes are + * random and reads are sequential, so we assume 1/2 random and half + * sequential. + * + * Hash Agg begins returning tuples after the first batch is + * complete. Accrue writes (spilled tuples) to startup_cost and reads + * only to total_cost. This is not perfect; it penalizes startup_cost + * in the case of recursive spills. Also, transCost is entirely + * counted in startup_cost; but some of that cost could be counted + * only against total_cost. + */ + if (!hashagg_mem_overflow && nbatches > 1.0) + { + double depth; + double pages; + + pages = relation_byte_size(input_tuples, input_width) / BLCKSZ; + depth = ceil( log(nbatches - 1) / log(HASHAGG_MAX_PARTITIONS) ); + pages_written = pages_read = pages * depth; + startup_cost += pages_written * random_page_cost; + } + total_cost = startup_cost; total_cost += aggcosts->finalCost.per_tuple * numGroups; + /* cost of retrieving from hash table */ total_cost += cpu_tuple_cost * numGroups; + total_cost += pages_read * seq_page_cost; output_tuples = numGroups; } diff --git a/src/backend/optimizer/plan/createplan.c b/src/backend/optimizer/plan/createplan.c index 8c8b4f8ed69..465b933f2ec 100644 --- a/src/backend/optimizer/plan/createplan.c +++ b/src/backend/optimizer/plan/createplan.c @@ -1644,6 +1644,7 @@ create_unique_plan(PlannerInfo *root, UniquePath *best_path, int flags) NIL, NIL, best_path->path.rows, + 0, subplan); } else @@ -2096,6 +2097,7 @@ create_agg_plan(PlannerInfo *root, AggPath *best_path) NIL, NIL, best_path->numGroups, + best_path->transitionSpace, subplan); copy_generic_path_info(&plan->plan, (Path *) best_path); @@ -2257,6 +2259,7 @@ create_groupingsets_plan(PlannerInfo *root, GroupingSetsPath *best_path) rollup->gsets, NIL, rollup->numGroups, + best_path->transitionSpace, sort_plan); /* @@ -2295,6 +2298,7 @@ create_groupingsets_plan(PlannerInfo *root, GroupingSetsPath *best_path) rollup->gsets, chain, rollup->numGroups, + best_path->transitionSpace, subplan); /* Copy cost data from Path to Plan */ @@ -6194,8 +6198,8 @@ Agg * make_agg(List *tlist, List *qual, AggStrategy aggstrategy, AggSplit aggsplit, int numGroupCols, AttrNumber *grpColIdx, Oid *grpOperators, Oid *grpCollations, - List *groupingSets, List *chain, - double dNumGroups, Plan *lefttree) + List *groupingSets, List *chain, double dNumGroups, + int32 transitionSpace, Plan *lefttree) { Agg *node = makeNode(Agg); Plan *plan = &node->plan; @@ -6211,6 +6215,7 @@ make_agg(List *tlist, List *qual, node->grpOperators = grpOperators; node->grpCollations = grpCollations; node->numGroups = numGroups; + node->transitionSpace = transitionSpace; node->aggParams = NULL; /* SS_finalize_plan() will fill this */ node->groupingSets = groupingSets; node->chain = chain; diff --git a/src/backend/optimizer/plan/planner.c b/src/backend/optimizer/plan/planner.c index cb54b15507b..a5686d822b3 100644 --- a/src/backend/optimizer/plan/planner.c +++ b/src/backend/optimizer/plan/planner.c @@ -4867,13 +4867,8 @@ create_distinct_paths(PlannerInfo *root, allow_hash = false; /* policy-based decision not to hash */ else { - Size hashentrysize; - - /* Estimate per-hash-entry space at tuple width... */ - hashentrysize = MAXALIGN(cheapest_input_path->pathtarget->width) + - MAXALIGN(SizeofMinimalTupleHeader); - /* plus the per-hash-entry overhead */ - hashentrysize += hash_agg_entry_size(0); + Size hashentrysize = hash_agg_entry_size( + 0, cheapest_input_path->pathtarget->width, 0); /* Allow hashing only if hashtable is predicted to fit in work_mem */ allow_hash = (hashentrysize * numDistinctRows <= work_mem * 1024L); @@ -6533,7 +6528,8 @@ add_paths_to_grouping_rel(PlannerInfo *root, RelOptInfo *input_rel, * were unable to sort above, then we'd better generate a Path, so * that we at least have one. */ - if (hashaggtablesize < work_mem * 1024L || + if (enable_hashagg_spill || + hashaggtablesize < work_mem * 1024L || grouped_rel->pathlist == NIL) { /* @@ -6566,7 +6562,8 @@ add_paths_to_grouping_rel(PlannerInfo *root, RelOptInfo *input_rel, agg_final_costs, dNumGroups); - if (hashaggtablesize < work_mem * 1024L) + if (enable_hashagg_spill || + hashaggtablesize < work_mem * 1024L) add_path(grouped_rel, (Path *) create_agg_path(root, grouped_rel, @@ -6835,7 +6832,7 @@ create_partial_grouping_paths(PlannerInfo *root, * Tentatively produce a partial HashAgg Path, depending on if it * looks as if the hash table will fit in work_mem. */ - if (hashaggtablesize < work_mem * 1024L && + if ((enable_hashagg_spill || hashaggtablesize < work_mem * 1024L) && cheapest_total_path != NULL) { add_path(partially_grouped_rel, (Path *) @@ -6862,7 +6859,7 @@ create_partial_grouping_paths(PlannerInfo *root, dNumPartialPartialGroups); /* Do the same for partial paths. */ - if (hashaggtablesize < work_mem * 1024L && + if ((enable_hashagg_spill || hashaggtablesize < work_mem * 1024L) && cheapest_partial_path != NULL) { add_partial_path(partially_grouped_rel, (Path *) diff --git a/src/backend/optimizer/prep/prepunion.c b/src/backend/optimizer/prep/prepunion.c index b01c9bbae7d..5f8fc50f8d3 100644 --- a/src/backend/optimizer/prep/prepunion.c +++ b/src/backend/optimizer/prep/prepunion.c @@ -1072,7 +1072,7 @@ choose_hashed_setop(PlannerInfo *root, List *groupClauses, numGroupCols, dNumGroups, NIL, input_path->startup_cost, input_path->total_cost, - input_path->rows); + input_path->rows, input_path->pathtarget->width); /* * Now for the sorted case. Note that the input is *always* unsorted, diff --git a/src/backend/optimizer/util/pathnode.c b/src/backend/optimizer/util/pathnode.c index 60c93ee7c59..1cb4fed1f81 100644 --- a/src/backend/optimizer/util/pathnode.c +++ b/src/backend/optimizer/util/pathnode.c @@ -1704,7 +1704,8 @@ create_unique_path(PlannerInfo *root, RelOptInfo *rel, Path *subpath, NIL, subpath->startup_cost, subpath->total_cost, - rel->rows); + rel->rows, + subpath->pathtarget->width); } if (sjinfo->semi_can_btree && sjinfo->semi_can_hash) @@ -2949,6 +2950,7 @@ create_agg_path(PlannerInfo *root, pathnode->aggstrategy = aggstrategy; pathnode->aggsplit = aggsplit; pathnode->numGroups = numGroups; + pathnode->transitionSpace = aggcosts ? aggcosts->transitionSpace : 0; pathnode->groupClause = groupClause; pathnode->qual = qual; @@ -2957,7 +2959,7 @@ create_agg_path(PlannerInfo *root, list_length(groupClause), numGroups, qual, subpath->startup_cost, subpath->total_cost, - subpath->rows); + subpath->rows, subpath->pathtarget->width); /* add tlist eval cost for each output row */ pathnode->path.startup_cost += target->cost.startup; @@ -3036,6 +3038,7 @@ create_groupingsets_path(PlannerInfo *root, pathnode->aggstrategy = aggstrategy; pathnode->rollups = rollups; pathnode->qual = having_qual; + pathnode->transitionSpace = agg_costs ? agg_costs->transitionSpace : 0; Assert(rollups != NIL); Assert(aggstrategy != AGG_PLAIN || list_length(rollups) == 1); @@ -3067,7 +3070,8 @@ create_groupingsets_path(PlannerInfo *root, having_qual, subpath->startup_cost, subpath->total_cost, - subpath->rows); + subpath->rows, + subpath->pathtarget->width); is_first = false; if (!rollup->is_hashed) is_first_sort = false; @@ -3090,7 +3094,8 @@ create_groupingsets_path(PlannerInfo *root, rollup->numGroups, having_qual, 0.0, 0.0, - subpath->rows); + subpath->rows, + subpath->pathtarget->width); if (!rollup->is_hashed) is_first_sort = false; } @@ -3115,7 +3120,8 @@ create_groupingsets_path(PlannerInfo *root, having_qual, sort_path.startup_cost, sort_path.total_cost, - sort_path.rows); + sort_path.rows, + subpath->pathtarget->width); } pathnode->path.total_cost += agg_path.total_cost; diff --git a/src/backend/utils/adt/selfuncs.c b/src/backend/utils/adt/selfuncs.c index ff02b5aafab..45c715385c7 100644 --- a/src/backend/utils/adt/selfuncs.c +++ b/src/backend/utils/adt/selfuncs.c @@ -3526,16 +3526,8 @@ double estimate_hashagg_tablesize(Path *path, const AggClauseCosts *agg_costs, double dNumGroups) { - Size hashentrysize; - - /* Estimate per-hash-entry space at tuple width... */ - hashentrysize = MAXALIGN(path->pathtarget->width) + - MAXALIGN(SizeofMinimalTupleHeader); - - /* plus space for pass-by-ref transition values... */ - hashentrysize += agg_costs->transitionSpace; - /* plus the per-hash-entry overhead */ - hashentrysize += hash_agg_entry_size(agg_costs->numAggs); + Size hashentrysize = hash_agg_entry_size( + agg_costs->numAggs, path->pathtarget->width, agg_costs->transitionSpace); /* * Note that this disregards the effect of fill-factor and growth policy diff --git a/src/include/executor/execExpr.h b/src/include/executor/execExpr.h index d21dbead0a2..e50a7ad6712 100644 --- a/src/include/executor/execExpr.h +++ b/src/include/executor/execExpr.h @@ -226,9 +226,13 @@ typedef enum ExprEvalOp EEOP_AGG_STRICT_INPUT_CHECK_ARGS, EEOP_AGG_STRICT_INPUT_CHECK_NULLS, EEOP_AGG_INIT_TRANS, + EEOP_AGG_INIT_TRANS_SPILLED, EEOP_AGG_STRICT_TRANS_CHECK, + EEOP_AGG_STRICT_TRANS_CHECK_SPILLED, EEOP_AGG_PLAIN_TRANS_BYVAL, + EEOP_AGG_PLAIN_TRANS_BYVAL_SPILLED, EEOP_AGG_PLAIN_TRANS, + EEOP_AGG_PLAIN_TRANS_SPILLED, EEOP_AGG_ORDERED_TRANS_DATUM, EEOP_AGG_ORDERED_TRANS_TUPLE, diff --git a/src/include/executor/executor.h b/src/include/executor/executor.h index 6298c7c8cad..e8d88f2ce26 100644 --- a/src/include/executor/executor.h +++ b/src/include/executor/executor.h @@ -140,11 +140,17 @@ extern TupleHashTable BuildTupleHashTableExt(PlanState *parent, extern TupleHashEntry LookupTupleHashEntry(TupleHashTable hashtable, TupleTableSlot *slot, bool *isnew); +extern TupleHashEntry LookupTupleHashEntryHash(TupleHashTable hashtable, + TupleTableSlot *slot, + bool *isnew, uint32 hash); extern TupleHashEntry FindTupleHashEntry(TupleHashTable hashtable, TupleTableSlot *slot, ExprState *eqcomp, FmgrInfo *hashfunctions); +extern uint32 TupleHashTableHash(struct tuplehash_hash *tb, + const MinimalTuple tuple); extern void ResetTupleHashTable(TupleHashTable hashtable); +extern void DestroyTupleHashTable(TupleHashTable hashtable); /* * prototypes from functions in execJunk.c @@ -250,7 +256,7 @@ extern ExprState *ExecInitQual(List *qual, PlanState *parent); extern ExprState *ExecInitCheck(List *qual, PlanState *parent); extern List *ExecInitExprList(List *nodes, PlanState *parent); extern ExprState *ExecBuildAggTrans(AggState *aggstate, struct AggStatePerPhaseData *phase, - bool doSort, bool doHash); + bool doSort, bool doHash, bool spilled); extern ExprState *ExecBuildGroupingEqual(TupleDesc ldesc, TupleDesc rdesc, const TupleTableSlotOps *lops, const TupleTableSlotOps *rops, int numCols, diff --git a/src/include/executor/nodeAgg.h b/src/include/executor/nodeAgg.h index 68c9e5f5400..29bbb9b0d09 100644 --- a/src/include/executor/nodeAgg.h +++ b/src/include/executor/nodeAgg.h @@ -302,13 +302,17 @@ typedef struct AggStatePerHashData AttrNumber *hashGrpColIdxInput; /* hash col indices in input slot */ AttrNumber *hashGrpColIdxHash; /* indices in hash table tuples */ Agg *aggnode; /* original Agg node, for numGroups etc. */ + int numNeededColsInput; /* number of columns needed from input */ + AttrNumber *allNeededColsInput; /* all columns needed from input */ } AggStatePerHashData; +#define HASHAGG_MAX_PARTITIONS 256 extern AggState *ExecInitAgg(Agg *node, EState *estate, int eflags); extern void ExecEndAgg(AggState *node); extern void ExecReScanAgg(AggState *node); -extern Size hash_agg_entry_size(int numAggs); +extern Size hash_agg_entry_size(int numAggs, Size tupleWidth, + Size transitionSpace); #endif /* NODEAGG_H */ diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h index 0c2a77aaf8d..8d4a36a3538 100644 --- a/src/include/nodes/execnodes.h +++ b/src/include/nodes/execnodes.h @@ -2066,13 +2066,30 @@ typedef struct AggState HeapTuple grp_firstTuple; /* copy of first tuple of current group */ /* these fields are used in AGG_HASHED and AGG_MIXED modes: */ bool table_filled; /* hash table filled yet? */ - int num_hashes; + int num_hashes; /* number of hash tables active at once */ + bool hash_spilled; /* any hash table ever spilled? */ + double hashentrysize; /* estimate revised during execution */ + struct HashAggSpill *hash_spills; /* HashAggSpill for each hash table, + exists only during first pass if spilled */ + TupleTableSlot *hash_spill_slot; /* slot for reading from spill files */ + bool hash_no_new_groups; /* we hit a limit during the current batch + and we must not create new groups */ + Size hash_mem_current; /* current hash table memory usage */ + Size hash_mem_limit; /* limit before spilling hash table */ + Size hash_mem_peak; /* peak hash table memory usage */ + long hash_ngroups_current; /* number of groups currently in + memory in all hash tables */ + long hash_ngroups_limit; /* limit before spilling hash table */ + uint64 hash_disk_used; /* bytes of disk space used */ + int hash_batches_used; /* batches used during entire execution */ + List *hash_batches; /* hash batches remaining to be processed */ + AggStatePerHash perhash; /* array of per-hashtable data */ AggStatePerGroup *hash_pergroup; /* grouping set indexed array of * per-group pointers */ /* support for evaluation of agg input expressions: */ -#define FIELDNO_AGGSTATE_ALL_PERGROUPS 34 +#define FIELDNO_AGGSTATE_ALL_PERGROUPS 47 AggStatePerGroup *all_pergroups; /* array of first ->pergroups, than * ->hash_pergroup */ ProjectionInfo *combinedproj; /* projection machinery */ diff --git a/src/include/nodes/pathnodes.h b/src/include/nodes/pathnodes.h index 31b631cfe0f..625a8aecb77 100644 --- a/src/include/nodes/pathnodes.h +++ b/src/include/nodes/pathnodes.h @@ -1663,6 +1663,7 @@ typedef struct AggPath AggStrategy aggstrategy; /* basic strategy, see nodes.h */ AggSplit aggsplit; /* agg-splitting mode, see nodes.h */ double numGroups; /* estimated number of groups in input */ + int32 transitionSpace; /* estimated transition state size */ List *groupClause; /* a list of SortGroupClause's */ List *qual; /* quals (HAVING quals), if any */ } AggPath; @@ -1700,6 +1701,7 @@ typedef struct GroupingSetsPath AggStrategy aggstrategy; /* basic strategy */ List *rollups; /* list of RollupData */ List *qual; /* quals (HAVING quals), if any */ + int32 transitionSpace; /* estimated transition state size */ } GroupingSetsPath; /* diff --git a/src/include/nodes/plannodes.h b/src/include/nodes/plannodes.h index 477b4da192c..ea3e0a643ec 100644 --- a/src/include/nodes/plannodes.h +++ b/src/include/nodes/plannodes.h @@ -813,6 +813,7 @@ typedef struct Agg Oid *grpOperators; /* equality operators to compare with */ Oid *grpCollations; long numGroups; /* estimated number of groups in input */ + int32 transitionSpace; /* estimated transition state size */ Bitmapset *aggParams; /* IDs of Params used in Aggref inputs */ /* Note: planner provides numGroups & aggParams only in HASHED/MIXED case */ List *groupingSets; /* grouping sets to use */ diff --git a/src/include/optimizer/cost.h b/src/include/optimizer/cost.h index b72e2d08290..fa6ad5e5857 100644 --- a/src/include/optimizer/cost.h +++ b/src/include/optimizer/cost.h @@ -115,7 +115,7 @@ extern void cost_agg(Path *path, PlannerInfo *root, int numGroupCols, double numGroups, List *quals, Cost input_startup_cost, Cost input_total_cost, - double input_tuples); + double input_tuples, double input_width); extern void cost_windowagg(Path *path, PlannerInfo *root, List *windowFuncs, int numPartCols, int numOrderCols, Cost input_startup_cost, Cost input_total_cost, diff --git a/src/include/optimizer/planmain.h b/src/include/optimizer/planmain.h index e7aaddd50d6..e20a66404ba 100644 --- a/src/include/optimizer/planmain.h +++ b/src/include/optimizer/planmain.h @@ -54,8 +54,8 @@ extern Sort *make_sort_from_sortclauses(List *sortcls, Plan *lefttree); extern Agg *make_agg(List *tlist, List *qual, AggStrategy aggstrategy, AggSplit aggsplit, int numGroupCols, AttrNumber *grpColIdx, Oid *grpOperators, Oid *grpCollations, - List *groupingSets, List *chain, - double dNumGroups, Plan *lefttree); + List *groupingSets, List *chain, double dNumGroups, + int32 transitionSpace, Plan *lefttree); extern Limit *make_limit(Plan *lefttree, Node *limitOffset, Node *limitCount); /* diff --git a/src/test/regress/expected/aggregates.out b/src/test/regress/expected/aggregates.out index d091ae4c6e4..92e5dbad77e 100644 --- a/src/test/regress/expected/aggregates.out +++ b/src/test/regress/expected/aggregates.out @@ -2331,3 +2331,95 @@ explain (costs off) -> Seq Scan on onek (8 rows) +-- +-- Compare results between plans using sorting and plans using hash +-- aggregation. Force spilling in both cases by setting work_mem low. +-- +set work_mem='64kB'; +-- Produce results with sorting. +set enable_hashagg = false; +set jit_above_cost = 0; +explain (costs off) +select g%100000 as c1, sum(g::numeric) as c2, count(*) as c3 + from generate_series(0, 199999) g + group by g%100000; + QUERY PLAN +------------------------------------------------ + GroupAggregate + Group Key: ((g % 100000)) + -> Sort + Sort Key: ((g % 100000)) + -> Function Scan on generate_series g +(5 rows) + +create table agg_group_1 as +select g%100000 as c1, sum(g::numeric) as c2, count(*) as c3 + from generate_series(0, 199999) g + group by g%100000; +set jit_above_cost to default; +create table agg_group_2 as +select (g/2)::numeric as c1, sum(7::int4) as c2, count(*) as c3 + from generate_series(0, 1999) g + group by g/2; +create table agg_group_3 as +select (g/2)::numeric as c1, array_agg(g::numeric) as c2, count(*) as c3 + from generate_series(0, 1999) g + group by g/2; +-- Produce results with hash aggregation +set enable_hashagg = true; +set enable_sort = false; +set jit_above_cost = 0; +explain (costs off) +select g%100000 as c1, sum(g::numeric) as c2, count(*) as c3 + from generate_series(0, 199999) g + group by g%100000; + QUERY PLAN +------------------------------------------ + HashAggregate + Group Key: (g % 100000) + -> Function Scan on generate_series g +(3 rows) + +create table agg_hash_1 as +select g%100000 as c1, sum(g::numeric) as c2, count(*) as c3 + from generate_series(0, 199999) g + group by g%100000; +set jit_above_cost to default; +create table agg_hash_2 as +select (g/2)::numeric as c1, sum(7::int4) as c2, count(*) as c3 + from generate_series(0, 1999) g + group by g/2; +create table agg_hash_3 as +select (g/2)::numeric as c1, array_agg(g::numeric) as c2, count(*) as c3 + from generate_series(0, 1999) g + group by g/2; +set enable_sort = true; +set work_mem to default; +-- Compare group aggregation results to hash aggregation results +(select * from agg_hash_1 except select * from agg_group_1) + union all +(select * from agg_group_1 except select * from agg_hash_1); + c1 | c2 | c3 +----+----+---- +(0 rows) + +(select * from agg_hash_2 except select * from agg_group_2) + union all +(select * from agg_group_2 except select * from agg_hash_2); + c1 | c2 | c3 +----+----+---- +(0 rows) + +(select * from agg_hash_3 except select * from agg_group_3) + union all +(select * from agg_group_3 except select * from agg_hash_3); + c1 | c2 | c3 +----+----+---- +(0 rows) + +drop table agg_group_1; +drop table agg_group_2; +drop table agg_group_3; +drop table agg_hash_1; +drop table agg_hash_2; +drop table agg_hash_3; diff --git a/src/test/regress/expected/groupingsets.out b/src/test/regress/expected/groupingsets.out index c1f802c88a7..767f60a96c7 100644 --- a/src/test/regress/expected/groupingsets.out +++ b/src/test/regress/expected/groupingsets.out @@ -1633,4 +1633,127 @@ select v||'a', case when grouping(v||'a') = 1 then 1 else 0 end, count(*) | 1 | 2 (4 rows) +-- +-- Compare results between plans using sorting and plans using hash +-- aggregation. Force spilling in both cases by setting work_mem low. +-- +SET work_mem='64kB'; +-- Produce results with sorting. +set enable_hashagg = false; +set jit_above_cost = 0; +explain (costs off) +select g1000, g100, g10, sum(g::numeric), count(*), max(g::text) from + (select g%1000 as g1000, g%100 as g100, g%10 as g10, g + from generate_series(0,199999) g) s +group by cube (g1000,g100,g10); + QUERY PLAN +--------------------------------------------------------------- + GroupAggregate + Group Key: ((g.g % 1000)), ((g.g % 100)), ((g.g % 10)) + Group Key: ((g.g % 1000)), ((g.g % 100)) + Group Key: ((g.g % 1000)) + Group Key: () + Sort Key: ((g.g % 100)), ((g.g % 10)) + Group Key: ((g.g % 100)), ((g.g % 10)) + Group Key: ((g.g % 100)) + Sort Key: ((g.g % 10)), ((g.g % 1000)) + Group Key: ((g.g % 10)), ((g.g % 1000)) + Group Key: ((g.g % 10)) + -> Sort + Sort Key: ((g.g % 1000)), ((g.g % 100)), ((g.g % 10)) + -> Function Scan on generate_series g +(14 rows) + +create table gs_group_1 as +select g1000, g100, g10, sum(g::numeric), count(*), max(g::text) from + (select g%1000 as g1000, g%100 as g100, g%10 as g10, g + from generate_series(0,199999) g) s +group by cube (g1000,g100,g10); +set jit_above_cost to default; +create table gs_group_2 as +select g1000, g100, g10, sum(g::numeric), count(*), max(g::text) from + (select g/20 as g1000, g/200 as g100, g/2000 as g10, g + from generate_series(0,19999) g) s +group by cube (g1000,g100,g10); +create table gs_group_3 as +select g100, g10, array_agg(g) as a, count(*) as c, max(g::text) as m from + (select g/200 as g100, g/2000 as g10, g + from generate_series(0,19999) g) s +group by grouping sets (g100,g10); +-- Produce results with hash aggregation. +set enable_hashagg = true; +set enable_sort = false; +set work_mem='64kB'; +set jit_above_cost = 0; +explain (costs off) +select g1000, g100, g10, sum(g::numeric), count(*), max(g::text) from + (select g%1000 as g1000, g%100 as g100, g%10 as g10, g + from generate_series(0,199999) g) s +group by cube (g1000,g100,g10); + QUERY PLAN +--------------------------------------------------------------- + GroupAggregate + Group Key: ((g.g % 1000)), ((g.g % 100)), ((g.g % 10)) + Group Key: ((g.g % 1000)), ((g.g % 100)) + Group Key: ((g.g % 1000)) + Group Key: () + Sort Key: ((g.g % 100)), ((g.g % 10)) + Group Key: ((g.g % 100)), ((g.g % 10)) + Group Key: ((g.g % 100)) + Sort Key: ((g.g % 10)), ((g.g % 1000)) + Group Key: ((g.g % 10)), ((g.g % 1000)) + Group Key: ((g.g % 10)) + -> Sort + Sort Key: ((g.g % 1000)), ((g.g % 100)), ((g.g % 10)) + -> Function Scan on generate_series g +(14 rows) + +create table gs_hash_1 as +select g1000, g100, g10, sum(g::numeric), count(*), max(g::text) from + (select g%1000 as g1000, g%100 as g100, g%10 as g10, g + from generate_series(0,199999) g) s +group by cube (g1000,g100,g10); +set jit_above_cost to default; +create table gs_hash_2 as +select g1000, g100, g10, sum(g::numeric), count(*), max(g::text) from + (select g/20 as g1000, g/200 as g100, g/2000 as g10, g + from generate_series(0,19999) g) s +group by cube (g1000,g100,g10); +create table gs_hash_3 as +select g100, g10, array_agg(g) as a, count(*) as c, max(g::text) as m from + (select g/200 as g100, g/2000 as g10, g + from generate_series(0,19999) g) s +group by grouping sets (g100,g10); +set enable_sort = true; +set work_mem to default; +-- Compare results +(select * from gs_hash_1 except select * from gs_group_1) + union all +(select * from gs_group_1 except select * from gs_hash_1); + g1000 | g100 | g10 | sum | count | max +-------+------+-----+-----+-------+----- +(0 rows) + +(select * from gs_hash_2 except select * from gs_group_2) + union all +(select * from gs_group_2 except select * from gs_hash_2); + g1000 | g100 | g10 | sum | count | max +-------+------+-----+-----+-------+----- +(0 rows) + +(select g100,g10,unnest(a),c,m from gs_hash_3 except + select g100,g10,unnest(a),c,m from gs_group_3) + union all +(select g100,g10,unnest(a),c,m from gs_group_3 except + select g100,g10,unnest(a),c,m from gs_hash_3); + g100 | g10 | unnest | c | m +------+-----+--------+---+--- +(0 rows) + +drop table gs_group_1; +drop table gs_group_2; +drop table gs_group_3; +drop table gs_hash_1; +drop table gs_hash_2; +drop table gs_hash_3; -- end diff --git a/src/test/regress/expected/select_distinct.out b/src/test/regress/expected/select_distinct.out index f3696c6d1de..11c6f50fbfa 100644 --- a/src/test/regress/expected/select_distinct.out +++ b/src/test/regress/expected/select_distinct.out @@ -148,6 +148,68 @@ SELECT count(*) FROM 4 (1 row) +-- +-- Compare results between plans using sorting and plans using hash +-- aggregation. Force spilling in both cases by setting work_mem low. +-- +SET work_mem='64kB'; +-- Produce results with sorting. +SET enable_hashagg=FALSE; +SET jit_above_cost=0; +EXPLAIN (costs off) +SELECT DISTINCT g%1000 FROM generate_series(0,9999) g; + QUERY PLAN +------------------------------------------------ + Unique + -> Sort + Sort Key: ((g % 1000)) + -> Function Scan on generate_series g +(4 rows) + +CREATE TABLE distinct_group_1 AS +SELECT DISTINCT g%1000 FROM generate_series(0,9999) g; +SET jit_above_cost TO DEFAULT; +CREATE TABLE distinct_group_2 AS +SELECT DISTINCT (g%1000)::text FROM generate_series(0,9999) g; +SET enable_hashagg=TRUE; +-- Produce results with hash aggregation. +SET enable_sort=FALSE; +SET jit_above_cost=0; +EXPLAIN (costs off) +SELECT DISTINCT g%1000 FROM generate_series(0,9999) g; + QUERY PLAN +------------------------------------------ + HashAggregate + Group Key: (g % 1000) + -> Function Scan on generate_series g +(3 rows) + +CREATE TABLE distinct_hash_1 AS +SELECT DISTINCT g%1000 FROM generate_series(0,9999) g; +SET jit_above_cost TO DEFAULT; +CREATE TABLE distinct_hash_2 AS +SELECT DISTINCT (g%1000)::text FROM generate_series(0,9999) g; +SET enable_sort=TRUE; +SET work_mem TO DEFAULT; +-- Compare results +(SELECT * FROM distinct_hash_1 EXCEPT SELECT * FROM distinct_group_1) + UNION ALL +(SELECT * FROM distinct_group_1 EXCEPT SELECT * FROM distinct_hash_1); + ?column? +---------- +(0 rows) + +(SELECT * FROM distinct_hash_1 EXCEPT SELECT * FROM distinct_group_1) + UNION ALL +(SELECT * FROM distinct_group_1 EXCEPT SELECT * FROM distinct_hash_1); + ?column? +---------- +(0 rows) + +DROP TABLE distinct_hash_1; +DROP TABLE distinct_hash_2; +DROP TABLE distinct_group_1; +DROP TABLE distinct_group_2; -- -- Also, some tests of IS DISTINCT FROM, which doesn't quite deserve its -- very own regression file. diff --git a/src/test/regress/sql/aggregates.sql b/src/test/regress/sql/aggregates.sql index 17fb256aec5..bcd336c5812 100644 --- a/src/test/regress/sql/aggregates.sql +++ b/src/test/regress/sql/aggregates.sql @@ -1017,3 +1017,91 @@ select v||'a', case when v||'a' = 'aa' then 1 else 0 end, count(*) explain (costs off) select 1 from tenk1 where (hundred, thousand) in (select twothousand, twothousand from onek); + +-- +-- Compare results between plans using sorting and plans using hash +-- aggregation. Force spilling in both cases by setting work_mem low. +-- + +set work_mem='64kB'; + +-- Produce results with sorting. + +set enable_hashagg = false; + +set jit_above_cost = 0; + +explain (costs off) +select g%100000 as c1, sum(g::numeric) as c2, count(*) as c3 + from generate_series(0, 199999) g + group by g%100000; + +create table agg_group_1 as +select g%100000 as c1, sum(g::numeric) as c2, count(*) as c3 + from generate_series(0, 199999) g + group by g%100000; + +set jit_above_cost to default; + +create table agg_group_2 as +select (g/2)::numeric as c1, sum(7::int4) as c2, count(*) as c3 + from generate_series(0, 1999) g + group by g/2; + +create table agg_group_3 as +select (g/2)::numeric as c1, array_agg(g::numeric) as c2, count(*) as c3 + from generate_series(0, 1999) g + group by g/2; + +-- Produce results with hash aggregation + +set enable_hashagg = true; +set enable_sort = false; + +set jit_above_cost = 0; + +explain (costs off) +select g%100000 as c1, sum(g::numeric) as c2, count(*) as c3 + from generate_series(0, 199999) g + group by g%100000; + +create table agg_hash_1 as +select g%100000 as c1, sum(g::numeric) as c2, count(*) as c3 + from generate_series(0, 199999) g + group by g%100000; + +set jit_above_cost to default; + +create table agg_hash_2 as +select (g/2)::numeric as c1, sum(7::int4) as c2, count(*) as c3 + from generate_series(0, 1999) g + group by g/2; + +create table agg_hash_3 as +select (g/2)::numeric as c1, array_agg(g::numeric) as c2, count(*) as c3 + from generate_series(0, 1999) g + group by g/2; + +set enable_sort = true; +set work_mem to default; + +-- Compare group aggregation results to hash aggregation results + +(select * from agg_hash_1 except select * from agg_group_1) + union all +(select * from agg_group_1 except select * from agg_hash_1); + +(select * from agg_hash_2 except select * from agg_group_2) + union all +(select * from agg_group_2 except select * from agg_hash_2); + +(select * from agg_hash_3 except select * from agg_group_3) + union all +(select * from agg_group_3 except select * from agg_hash_3); + +drop table agg_group_1; +drop table agg_group_2; +drop table agg_group_3; +drop table agg_hash_1; +drop table agg_hash_2; +drop table agg_hash_3; diff --git a/src/test/regress/sql/groupingsets.sql b/src/test/regress/sql/groupingsets.sql index 95ac3fb52f6..bf8bce6ed31 100644 --- a/src/test/regress/sql/groupingsets.sql +++ b/src/test/regress/sql/groupingsets.sql @@ -441,4 +441,103 @@ select v||'a', case when grouping(v||'a') = 1 then 1 else 0 end, count(*) from unnest(array[1,1], array['a','b']) u(i,v) group by rollup(i, v||'a') order by 1,3; +-- +-- Compare results between plans using sorting and plans using hash +-- aggregation. Force spilling in both cases by setting work_mem low. +-- + +SET work_mem='64kB'; + +-- Produce results with sorting. + +set enable_hashagg = false; + +set jit_above_cost = 0; + +explain (costs off) +select g1000, g100, g10, sum(g::numeric), count(*), max(g::text) from + (select g%1000 as g1000, g%100 as g100, g%10 as g10, g + from generate_series(0,199999) g) s +group by cube (g1000,g100,g10); + +create table gs_group_1 as +select g1000, g100, g10, sum(g::numeric), count(*), max(g::text) from + (select g%1000 as g1000, g%100 as g100, g%10 as g10, g + from generate_series(0,199999) g) s +group by cube (g1000,g100,g10); + +set jit_above_cost to default; + +create table gs_group_2 as +select g1000, g100, g10, sum(g::numeric), count(*), max(g::text) from + (select g/20 as g1000, g/200 as g100, g/2000 as g10, g + from generate_series(0,19999) g) s +group by cube (g1000,g100,g10); + +create table gs_group_3 as +select g100, g10, array_agg(g) as a, count(*) as c, max(g::text) as m from + (select g/200 as g100, g/2000 as g10, g + from generate_series(0,19999) g) s +group by grouping sets (g100,g10); + +-- Produce results with hash aggregation. + +set enable_hashagg = true; +set enable_sort = false; +set work_mem='64kB'; + +set jit_above_cost = 0; + +explain (costs off) +select g1000, g100, g10, sum(g::numeric), count(*), max(g::text) from + (select g%1000 as g1000, g%100 as g100, g%10 as g10, g + from generate_series(0,199999) g) s +group by cube (g1000,g100,g10); + +create table gs_hash_1 as +select g1000, g100, g10, sum(g::numeric), count(*), max(g::text) from + (select g%1000 as g1000, g%100 as g100, g%10 as g10, g + from generate_series(0,199999) g) s +group by cube (g1000,g100,g10); + +set jit_above_cost to default; + +create table gs_hash_2 as +select g1000, g100, g10, sum(g::numeric), count(*), max(g::text) from + (select g/20 as g1000, g/200 as g100, g/2000 as g10, g + from generate_series(0,19999) g) s +group by cube (g1000,g100,g10); + +create table gs_hash_3 as +select g100, g10, array_agg(g) as a, count(*) as c, max(g::text) as m from + (select g/200 as g100, g/2000 as g10, g + from generate_series(0,19999) g) s +group by grouping sets (g100,g10); + +set enable_sort = true; +set work_mem to default; + +-- Compare results + +(select * from gs_hash_1 except select * from gs_group_1) + union all +(select * from gs_group_1 except select * from gs_hash_1); + +(select * from gs_hash_2 except select * from gs_group_2) + union all +(select * from gs_group_2 except select * from gs_hash_2); + +(select g100,g10,unnest(a),c,m from gs_hash_3 except + select g100,g10,unnest(a),c,m from gs_group_3) + union all +(select g100,g10,unnest(a),c,m from gs_group_3 except + select g100,g10,unnest(a),c,m from gs_hash_3); + +drop table gs_group_1; +drop table gs_group_2; +drop table gs_group_3; +drop table gs_hash_1; +drop table gs_hash_2; +drop table gs_hash_3; + -- end diff --git a/src/test/regress/sql/select_distinct.sql b/src/test/regress/sql/select_distinct.sql index a605e86449e..33102744ebf 100644 --- a/src/test/regress/sql/select_distinct.sql +++ b/src/test/regress/sql/select_distinct.sql @@ -45,6 +45,68 @@ SELECT count(*) FROM SELECT count(*) FROM (SELECT DISTINCT two, four, two FROM tenk1) ss; +-- +-- Compare results between plans using sorting and plans using hash +-- aggregation. Force spilling in both cases by setting work_mem low. +-- + +SET work_mem='64kB'; + +-- Produce results with sorting. + +SET enable_hashagg=FALSE; + +SET jit_above_cost=0; + +EXPLAIN (costs off) +SELECT DISTINCT g%1000 FROM generate_series(0,9999) g; + +CREATE TABLE distinct_group_1 AS +SELECT DISTINCT g%1000 FROM generate_series(0,9999) g; + +SET jit_above_cost TO DEFAULT; + +CREATE TABLE distinct_group_2 AS +SELECT DISTINCT (g%1000)::text FROM generate_series(0,9999) g; + +SET enable_hashagg=TRUE; + +-- Produce results with hash aggregation. + +SET enable_sort=FALSE; + +SET jit_above_cost=0; + +EXPLAIN (costs off) +SELECT DISTINCT g%1000 FROM generate_series(0,9999) g; + +CREATE TABLE distinct_hash_1 AS +SELECT DISTINCT g%1000 FROM generate_series(0,9999) g; + +SET jit_above_cost TO DEFAULT; + +CREATE TABLE distinct_hash_2 AS +SELECT DISTINCT (g%1000)::text FROM generate_series(0,9999) g; + +SET enable_sort=TRUE; + +SET work_mem TO DEFAULT; + +-- Compare results + +(SELECT * FROM distinct_hash_1 EXCEPT SELECT * FROM distinct_group_1) + UNION ALL +(SELECT * FROM distinct_group_1 EXCEPT SELECT * FROM distinct_hash_1); + +(SELECT * FROM distinct_hash_1 EXCEPT SELECT * FROM distinct_group_1) + UNION ALL +(SELECT * FROM distinct_group_1 EXCEPT SELECT * FROM distinct_hash_1); + +DROP TABLE distinct_hash_1; +DROP TABLE distinct_hash_2; +DROP TABLE distinct_group_1; +DROP TABLE distinct_group_2; + -- -- Also, some tests of IS DISTINCT FROM, which doesn't quite deserve its -- very own regression file.