From f0b907168d75eb007095e74ca43a4c87b333cbd2 Mon Sep 17 00:00:00 2001 From: Tomas Vondra Date: Sun, 5 Jan 2025 21:24:23 +0100 Subject: [PATCH v20250208 1/3] Balance memory usage with hashjoin batch explosion The basic logic to pick the number of hashjoin batches is concerned only with the in-memory hash table, adding batches to keep the hash table within the limit defined by work_mem and hash_mem_multiplier. It ignores the memory needed by the batch files, but with enough batches this may be a substantial amount of memory, easily orders of magnitude more than the hash table. We've seen reports of hash joins with hundreds of thousands or millions of batch files, consuming gigabytes of memory, and triggering OOM. These cases are not too common, but it's clearly possible to hit them. This patch improves the situation by rebalancing how the memory is distributed between the hash table and batch files, to minimize the total memory consumption. Whenever we need to increase the capacity of the hash node, we can do that by either doubling the number of batches or doubling the size of the in-memory hash table. The outcome is the same, allowing the hash node to handle a relation twice the size. But the memory requirements may be substantially different, depending on the current hashjoin parameters (for low nbatch values it's better to add batches, for high nbatch values it's better to allow a larger hash table). It may seem a bit strange, as it clearly allows exceeding the memory limit specified by the GUC parameters. But it has always been like this, except that the code assumed adding batches is free. The patch just makes this visible and explicit. Increasing the hashtable memory limit may also help to prevent the batch explosion in the first place. Given enough hash collisions or duplicate hashes it's easy to get a batch that can't be split, resulting in a cycle of quickly doubling the number of batches. Allowing the hashtable to get larger may stop this. --- src/backend/executor/nodeHash.c | 146 ++++++++++++++++++++++++++++ src/backend/utils/misc/guc_tables.c | 11 +++ src/include/executor/hashjoin.h | 2 + 3 files changed, 159 insertions(+) diff --git a/src/backend/executor/nodeHash.c b/src/backend/executor/nodeHash.c index 6f8a379e3b9..580269367c3 100644 --- a/src/backend/executor/nodeHash.c +++ b/src/backend/executor/nodeHash.c @@ -80,6 +80,8 @@ static bool ExecParallelHashTuplePrealloc(HashJoinTable hashtable, static void ExecParallelHashMergeCounters(HashJoinTable hashtable); static void ExecParallelHashCloseBatchAccessors(HashJoinTable hashtable); +/* enable adaptive adjustment of hashtable size */ +bool enable_hashjoin_adjust = false; /* ---------------------------------------------------------------- * ExecHash @@ -848,6 +850,105 @@ ExecChooseHashTableSize(double ntuples, int tupwidth, bool useskew, nbatch = pg_nextpower2_32(Max(2, minbatch)); } + /* + * Optimize the total amount of memory consumed by the hash node. + * + * The nbatch calculation above focuses on the size of the in-memory hash + * table, ignoring the memory used by batch files. But that can be a lot + * of memory - each batch file has a BLCKSZ buffer, and we may need two + * files per batch (inner and outer side). So with enough batches this can + * be significantly more memory than the hashtable itself, and it grows + * quickly as we're adding more batches. + * + * It might seem cleaner to adjust the calculation above, to consider + * memory for both the in-memory hashtable and the batch files, and ensure + * it fits into hash_table_bytes. That is, look for a nbatch value so that + * + * (inner_rel_bytes / nbatch) + (2 * nbatch * BLCKSZ) <= hash_table_bytes + * + * But that has a flaw - for sufficiently large inner_rel_bytes value it + * may not have a solution (either the hash table is too large or it + * requires too many batches). So instead we merely try to minimize the + * impact, and use as little memory as possible, instead of strictly + * enforcing the memory limit. (But we haven't really enforced it before + * either, as we simply ignored the batch files.) + * + * The basic observation is that given an inner relation of a given size, + * we may divide it in arbitrary number of batches, which determines the + * memory consumption per the already mentioned formula: + * + * (inner_rel_bytes / nbatch) + (2 * nbatch * BLCKSZ) + * + * That is, we can reduce the number of batches to (nbatch/2), at the + * cost of doubling the size of the in-memory hash table. But these two + * terms work in opposite ways - size of the in-memory part decreases + * with nbatch, while the batch file memory grows very quickly. Initially + * the memory usage is dominated by in-memory hash table (for nbatch=0), + * then at some point the batch files start to consume more memory. + * + * If you combine these two, the memory consumption (for a fixed size of + * the inner relation) has a u-shape, with a minimum at some nbatch value. + * Our goal is to look for this minimum. We do that by calculating memory + * usage for (nbatch/2), and accepting it if it's lower than current. + * + * This means we're only ever reducing nbatch values, we'll never increase + * it (as we're not considering nbatch*2). We could counsider that too, + * depending on which part of the [nbatch,work_mem] table we're in. And + * for cases with high work_mem values, we would find that adding batches + * reduces memory usage. But the hashtable size is what we consider when + * calculating the initial nbatch value, and if it's dominating the memory + * usage, if means we're not exceeding the expected memory limit (at least + * not significantly). There is little risk of OOM or memory overruns. Our + * goal is not to minimize the memory usage, but to enforce the limit set + * by the user. Minimizing the memory usage would result in spilling many + * more batch files, which does not seem great for performance. So we only + * ever reduce nbatch, never increase it. + * + * While growing the hashtable, we also adjust the number of buckets, to + * not have more than one tuple per bucket. We can only do this during + * the initial sizing - once we start building the hash, we can't add + * buckets, due to how ExecHashGetBucketAndBatch() calculates batchno + * and bucketno from the hash. Increasing the nbucket value would move + * the batchno part in a way that could result in the batchno going + * backwards, but that violates the expectation that splitting a batch + * moves the tuples only to "future" batches. + * + * So after the initial sizing (here in ExecChooseHashTableSize), the + * number of buckets is effectively fixed. ExecHashGetBucketAndBatch + * could calculate batchno/bucketno in a different way, but that's + * left as a separate improvement. To some extent this is a preexisting + * issue - if we set growEnabled=false, this allows the hashtable to + * exceed the memory limit too, and we don't adjust the bucket count. + * However, that likely happens due to duplicate values and/or hash + * collisions, so it's not clear if increasing the bucket count would + * actually spread the tuples through the buckets. It would help with + * skewed data sets, when we may disable the growth early, and then + * add more tuples with distinct hash values. + */ + while (nbatch > 0) + { + /* how much memory would we use with half the batches? */ + size_t space = hash_table_bytes * 2 + (nbatch * BLCKSZ); + size_t current = hash_table_bytes + (2 * nbatch * BLCKSZ); + + /* Is the adaptive behavior enabled? */ + if (!enable_hashjoin_adjust) + break; + + /* If the memory usage does not decrease, we have the optimum. */ + if (current < space) + break; + + /* + * It's better to use half the batches, so do that and adjust the + * nbucket in the opposite direction, and the allowance. + */ + nbatch /= 2; + nbuckets *= 2; + + *space_allowed = *space_allowed * 2; + } + Assert(nbuckets > 0); Assert(nbatch > 0); @@ -890,6 +991,47 @@ ExecHashTableDestroy(HashJoinTable hashtable) pfree(hashtable); } +/* + * Consider adjusting the allowed hash table size, depending on the number + * of batches, to minimize the overall memory usage (for both the hashtable + * and batch files). + * + * Returns true if we chose to increase the batch size (and thus we don't + * need to add batches), and false if we should increase nbatch. + * + * XXX Note that while we're adjusting the size of the hash table, we're not + * adjusting the (optimal) number of buckets. We can't change that once we + * start building the hash, due to how ExecHashGetBucketAndBatch splits the + * hash into batchno/bucketno. + */ +static bool +ExecHashIncreaseBatchSize(HashJoinTable hashtable) +{ + /* + * How much memory would doubling nbatch use? Each batch may require + * two buffered files (inner/outer), with a BLCKSZ buffer. + */ + size_t batchSpace = (hashtable->nbatch * 2 * BLCKSZ); + + /* Do nothing if the adaptive behavior is disabled. */ + if (!enable_hashjoin_adjust) + return false; + + /* + * Compare the new space needed for doubling nbatch and for enlarging the + * in-memory hash table. If doubling the hash table needs less memory, + * just do that. Otherwise, continue with doubling the nbatch. + * + */ + if (hashtable->spaceAllowed <= batchSpace) + { + hashtable->spaceAllowed *= 2; + return true; + } + + return false; +} + /* * ExecHashIncreaseNumBatches * increase the original number of batches in order to reduce @@ -913,6 +1055,10 @@ ExecHashIncreaseNumBatches(HashJoinTable hashtable) if (oldnbatch > Min(INT_MAX / 2, MaxAllocSize / (sizeof(void *) * 2))) return; + /* consider increasing size of the in-memory hash table instead */ + if (ExecHashIncreaseBatchSize(hashtable)) + return; + nbatch = oldnbatch * 2; Assert(nbatch > 1); diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c index ce7534d4d23..6ef37c51783 100644 --- a/src/backend/utils/misc/guc_tables.c +++ b/src/backend/utils/misc/guc_tables.c @@ -46,6 +46,7 @@ #include "commands/vacuum.h" #include "common/file_utils.h" #include "common/scram-common.h" +#include "executor/hashjoin.h" #include "jit/jit.h" #include "libpq/auth.h" #include "libpq/libpq.h" @@ -900,6 +901,16 @@ struct config_bool ConfigureNamesBool[] = true, NULL, NULL, NULL }, + { + {"enable_hashjoin_adjust", PGC_USERSET, QUERY_TUNING_METHOD, + gettext_noop("Enables adjusting hashtable size to minimize memory usage."), + NULL, + GUC_EXPLAIN + }, + &enable_hashjoin_adjust, + false, + NULL, NULL, NULL + }, { {"enable_gathermerge", PGC_USERSET, QUERY_TUNING_METHOD, gettext_noop("Enables the planner's use of gather merge plans."), diff --git a/src/include/executor/hashjoin.h b/src/include/executor/hashjoin.h index ecff4842fd3..0eb1da8f66a 100644 --- a/src/include/executor/hashjoin.h +++ b/src/include/executor/hashjoin.h @@ -71,6 +71,8 @@ * ---------------------------------------------------------------- */ +extern PGDLLIMPORT bool enable_hashjoin_adjust; + /* these are in nodes/execnodes.h: */ /* typedef struct HashJoinTupleData *HashJoinTuple; */ /* typedef struct HashJoinTableData *HashJoinTable; */ -- 2.47.1