agora inbox for [email protected]help / color / mirror / Atom feed
[PATCH 16/18] doc review for logical decoding of prepared xacts 9+ messages / 2 participants [nested] [flat]
* [PATCH 16/18] doc review for logical decoding of prepared xacts @ 2021-01-24 03:03 Justin Pryzby <[email protected]> 0 siblings, 0 replies; 9+ messages in thread From: Justin Pryzby @ 2021-01-24 03:03 UTC (permalink / raw) 0aa8a01d04c8fe200b7a106878eebc3d0af9105c --- doc/src/sgml/logicaldecoding.sgml | 30 +++++++++++++++--------------- 1 file changed, 15 insertions(+), 15 deletions(-) diff --git a/doc/src/sgml/logicaldecoding.sgml b/doc/src/sgml/logicaldecoding.sgml index b854f2ccfc..71e9f36b8e 100644 --- a/doc/src/sgml/logicaldecoding.sgml +++ b/doc/src/sgml/logicaldecoding.sgml @@ -791,9 +791,9 @@ typedef void (*LogicalDecodeMessageCB) (struct LogicalDecodingContext *ctx, <para> The optional <function>filter_prepare_cb</function> callback is called to determine whether data that is part of the current - two-phase commit transaction should be considered for decode - at this prepare stage or as a regular one-phase transaction at - <command>COMMIT PREPARED</command> time later. To signal that + two-phase commit transaction should be considered for decoding + at this prepare stage or later as a regular one-phase transaction at + <command>COMMIT PREPARED</command> time. To signal that decoding should be skipped, return <literal>true</literal>; <literal>false</literal> otherwise. When the callback is not defined, <literal>false</literal> is assumed (i.e. nothing is @@ -820,11 +820,11 @@ typedef bool (*LogicalDecodeFilterPrepareCB) (struct LogicalDecodingContext *ctx The required <function>begin_prepare_cb</function> callback is called whenever the start of a prepared transaction has been decoded. The <parameter>gid</parameter> field, which is part of the - <parameter>txn</parameter> parameter can be used in this callback to - check if the plugin has already received this prepare in which case it + <parameter>txn</parameter> parameter, can be used in this callback to + check if the plugin has already received this PREPARE in which case it can skip the remaining changes of the transaction. This can only happen - if the user restarts the decoding after receiving the prepare for a - transaction but before receiving the commit prepared say because of some + if the user restarts the decoding after receiving the PREPARE for a + transaction but before receiving the COMMIT PREPARED, say because of some error. <programlisting> typedef void (*LogicalDecodeBeginPrepareCB) (struct LogicalDecodingContext *ctx, @@ -842,7 +842,7 @@ typedef bool (*LogicalDecodeFilterPrepareCB) (struct LogicalDecodingContext *ctx decoded. The <function>change_cb</function> callback for all modified rows will have been called before this, if there have been any modified rows. The <parameter>gid</parameter> field, which is part of the - <parameter>txn</parameter> parameter can be used in this callback. + <parameter>txn</parameter> parameter, can be used in this callback. <programlisting> typedef void (*LogicalDecodePrepareCB) (struct LogicalDecodingContext *ctx, ReorderBufferTXN *txn, @@ -856,9 +856,9 @@ typedef bool (*LogicalDecodeFilterPrepareCB) (struct LogicalDecodingContext *ctx <para> The required <function>commit_prepared_cb</function> callback is called - whenever a transaction commit prepared has been decoded. The + whenever a transaction COMMIT PREPARED has been decoded. The <parameter>gid</parameter> field, which is part of the - <parameter>txn</parameter> parameter can be used in this callback. + <parameter>txn</parameter> parameter, can be used in this callback. <programlisting> typedef void (*LogicalDecodeCommitPreparedCB) (struct LogicalDecodingContext *ctx, ReorderBufferTXN *txn, @@ -872,15 +872,15 @@ typedef bool (*LogicalDecodeFilterPrepareCB) (struct LogicalDecodingContext *ctx <para> The required <function>rollback_prepared_cb</function> callback is called - whenever a transaction rollback prepared has been decoded. The + whenever a transaction ROLLBACK PREPARED has been decoded. The <parameter>gid</parameter> field, which is part of the - <parameter>txn</parameter> parameter can be used in this callback. The + <parameter>txn</parameter> parameter, can be used in this callback. The parameters <parameter>prepare_end_lsn</parameter> and <parameter>prepare_time</parameter> can be used to check if the plugin - has received this prepare transaction in which case it can apply the + has received this PREPARE TRANSACTION in which case it can apply the rollback, otherwise, it can skip the rollback operation. The <parameter>gid</parameter> alone is not sufficient because the downstream - node can have prepared transaction with same identifier. + node can have a prepared transaction with same identifier. <programlisting> typedef void (*LogicalDecodeRollbackPreparedCB) (struct LogicalDecodingContext *ctx, ReorderBufferTXN *txn, @@ -1122,7 +1122,7 @@ OutputPluginWrite(ctx, true); the <function>stream_commit_cb</function> callback (or possibly aborted using the <function>stream_abort_cb</function> callback). If two-phase commits are supported, the transaction can be prepared using the - <function>stream_prepare_cb</function> callback, commit prepared using the + <function>stream_prepare_cb</function> callback, COMMIT PREPARED using the <function>commit_prepared_cb</function> callback or aborted using the <function>rollback_prepared_cb</function>. </para> -- 2.17.0 --lc9FT7cWel8HagAv Content-Type: text/x-diff; charset=us-ascii Content-Disposition: attachment; filename="0017-an-old-and-deleted-has-happened.patch" ^ permalink raw reply [nested|flat] 9+ messages in thread
* [PATCH 2/3] Allocate hash batches related BufFile in a dedicated context @ 2023-03-27 13:54 Jehan-Guillaume de Rorthais <[email protected]> 0 siblings, 0 replies; 9+ messages in thread From: Jehan-Guillaume de Rorthais @ 2023-03-27 13:54 UTC (permalink / raw) --- src/backend/executor/nodeHash.c | 43 ++++++++++++++++------- src/backend/executor/nodeHashjoin.c | 22 ++++++++---- src/backend/utils/sort/sharedtuplestore.c | 8 +++++ src/include/executor/hashjoin.h | 30 ++++++++++------ src/include/executor/nodeHashjoin.h | 2 +- 5 files changed, 75 insertions(+), 30 deletions(-) diff --git a/src/backend/executor/nodeHash.c b/src/backend/executor/nodeHash.c index 5fd1c5553b..e529f50a3e 100644 --- a/src/backend/executor/nodeHash.c +++ b/src/backend/executor/nodeHash.c @@ -484,7 +484,7 @@ ExecHashTableCreate(HashState *state, List *hashOperators, List *hashCollations, * * The hashtable control block is just palloc'd from the executor's * per-query memory context. Everything else should be kept inside the - * subsidiary hashCxt or batchCxt. + * subsidiary hashCxt, batchCxt or spillCxt. */ hashtable = palloc_object(HashJoinTableData); hashtable->nbuckets = nbuckets; @@ -538,6 +538,10 @@ ExecHashTableCreate(HashState *state, List *hashOperators, List *hashCollations, "HashBatchContext", ALLOCSET_DEFAULT_SIZES); + hashtable->spillCxt = AllocSetContextCreate(hashtable->hashCxt, + "HashSpillContext", + ALLOCSET_DEFAULT_SIZES); + /* Allocate data that will live for the life of the hashjoin */ oldcxt = MemoryContextSwitchTo(hashtable->hashCxt); @@ -570,12 +574,19 @@ ExecHashTableCreate(HashState *state, List *hashOperators, List *hashCollations, if (nbatch > 1 && hashtable->parallel_state == NULL) { + MemoryContext oldctx; + /* * allocate and initialize the file arrays in hashCxt (not needed for * parallel case which uses shared tuplestores instead of raw files) */ + oldctx = MemoryContextSwitchTo(hashtable->spillCxt); + hashtable->innerBatchFile = palloc0_array(BufFile *, nbatch); hashtable->outerBatchFile = palloc0_array(BufFile *, nbatch); + + MemoryContextSwitchTo(oldctx); + /* The files will not be opened until needed... */ /* ... but make sure we have temp tablespaces established for them */ PrepareTempTablespaces(); @@ -913,7 +924,6 @@ ExecHashIncreaseNumBatches(HashJoinTable hashtable) int oldnbatch = hashtable->nbatch; int curbatch = hashtable->curbatch; int nbatch; - MemoryContext oldcxt; long ninmemory; long nfreed; HashMemoryChunk oldchunks; @@ -934,13 +944,16 @@ ExecHashIncreaseNumBatches(HashJoinTable hashtable) hashtable, nbatch, hashtable->spaceUsed); #endif - oldcxt = MemoryContextSwitchTo(hashtable->hashCxt); - if (hashtable->innerBatchFile == NULL) { + MemoryContext oldcxt = MemoryContextSwitchTo(hashtable->spillCxt); + /* we had no file arrays before */ hashtable->innerBatchFile = palloc0_array(BufFile *, nbatch); hashtable->outerBatchFile = palloc0_array(BufFile *, nbatch); + + MemoryContextSwitchTo(oldcxt); + /* time to establish the temp tablespaces, too */ PrepareTempTablespaces(); } @@ -951,8 +964,6 @@ ExecHashIncreaseNumBatches(HashJoinTable hashtable) hashtable->outerBatchFile = repalloc0_array(hashtable->outerBatchFile, BufFile *, oldnbatch, nbatch); } - MemoryContextSwitchTo(oldcxt); - hashtable->nbatch = nbatch; /* @@ -1024,7 +1035,8 @@ ExecHashIncreaseNumBatches(HashJoinTable hashtable) Assert(batchno > curbatch); ExecHashJoinSaveTuple(HJTUPLE_MINTUPLE(hashTuple), hashTuple->hashvalue, - &hashtable->innerBatchFile[batchno]); + &hashtable->innerBatchFile[batchno], + hashtable->spillCxt); hashtable->spaceUsed -= hashTupleSize; nfreed++; @@ -1683,7 +1695,8 @@ ExecHashTableInsert(HashJoinTable hashtable, Assert(batchno > hashtable->curbatch); ExecHashJoinSaveTuple(tuple, hashvalue, - &hashtable->innerBatchFile[batchno]); + &hashtable->innerBatchFile[batchno], + hashtable->spillCxt); } if (shouldFree) @@ -2664,7 +2677,8 @@ ExecHashRemoveNextSkewBucket(HashJoinTable hashtable) /* Put the tuple into a temp file for later batches */ Assert(batchno > hashtable->curbatch); ExecHashJoinSaveTuple(tuple, hashvalue, - &hashtable->innerBatchFile[batchno]); + &hashtable->innerBatchFile[batchno], + hashtable->spillCxt); pfree(hashTuple); hashtable->spaceUsed -= tupleSize; hashtable->spaceUsedSkew -= tupleSize; @@ -3093,8 +3107,11 @@ ExecParallelHashJoinSetUpBatches(HashJoinTable hashtable, int nbatch) pstate->nbatch = nbatch; batches = dsa_get_address(hashtable->area, pstate->batches); - /* Use hash join memory context. */ - oldcxt = MemoryContextSwitchTo(hashtable->hashCxt); + /* + * Use hash join spill memory context to allocate accessors and their + * buffers + */ + oldcxt = MemoryContextSwitchTo(hashtable->spillCxt); /* Allocate this backend's accessor array. */ hashtable->nbatch = nbatch; @@ -3196,8 +3213,8 @@ ExecParallelHashEnsureBatchAccessors(HashJoinTable hashtable) */ Assert(DsaPointerIsValid(pstate->batches)); - /* Use hash join memory context. */ - oldcxt = MemoryContextSwitchTo(hashtable->hashCxt); + /* Use hash join spill memory context to allocate accessors */ + oldcxt = MemoryContextSwitchTo(hashtable->spillCxt); /* Allocate this backend's accessor array. */ hashtable->nbatch = pstate->nbatch; diff --git a/src/backend/executor/nodeHashjoin.c b/src/backend/executor/nodeHashjoin.c index a39164c15d..8618e0df40 100644 --- a/src/backend/executor/nodeHashjoin.c +++ b/src/backend/executor/nodeHashjoin.c @@ -491,7 +491,8 @@ ExecHashJoinImpl(PlanState *pstate, bool parallel) Assert(parallel_state == NULL); Assert(batchno > hashtable->curbatch); ExecHashJoinSaveTuple(mintuple, hashvalue, - &hashtable->outerBatchFile[batchno]); + &hashtable->outerBatchFile[batchno], + hashtable->spillCxt); if (shouldFree) heap_free_minimal_tuple(mintuple); @@ -1313,21 +1314,30 @@ ExecParallelHashJoinNewBatch(HashJoinState *hjstate) * The data recorded in the file for each tuple is its hash value, * then the tuple in MinimalTuple format. * - * Note: it is important always to call this in the regular executor - * context, not in a shorter-lived context; else the temp file buffers - * will get messed up. + * If this is the first write to the batch file, this function first + * create it. The associated BufFile buffer is allocated in the given + * context. It is important to always give the HashSpillContext + * context. First to avoid a shorter-lived context, else the temp file + * buffers will get messed up. Second, for a better accounting of the + * spilling memory consumption. + * */ void ExecHashJoinSaveTuple(MinimalTuple tuple, uint32 hashvalue, - BufFile **fileptr) + BufFile **fileptr, MemoryContext cxt) { BufFile *file = *fileptr; if (file == NULL) { - /* First write to this batch file, so open it. */ + MemoryContext oldctx; + + oldctx = MemoryContextSwitchTo(cxt); + file = BufFileCreateTemp(false); *fileptr = file; + + MemoryContextSwitchTo(oldctx); } BufFileWrite(file, &hashvalue, sizeof(uint32)); diff --git a/src/backend/utils/sort/sharedtuplestore.c b/src/backend/utils/sort/sharedtuplestore.c index 0831249159..236be65f22 100644 --- a/src/backend/utils/sort/sharedtuplestore.c +++ b/src/backend/utils/sort/sharedtuplestore.c @@ -308,11 +308,15 @@ sts_puttuple(SharedTuplestoreAccessor *accessor, void *meta_data, { SharedTuplestoreParticipant *participant; char name[MAXPGPATH]; + MemoryContext oldcxt; /* Create one. Only this backend will write into it. */ sts_filename(name, accessor, accessor->participant); + + oldcxt = MemoryContextSwitchTo(accessor->context); accessor->write_file = BufFileCreateFileSet(&accessor->fileset->fs, name); + MemoryContextSwitchTo(oldcxt); /* Set up the shared state for this backend's file. */ participant = &accessor->sts->participants[accessor->participant]; @@ -527,11 +531,15 @@ sts_parallel_scan_next(SharedTuplestoreAccessor *accessor, void *meta_data) if (accessor->read_file == NULL) { char name[MAXPGPATH]; + MemoryContext oldcxt; sts_filename(name, accessor, accessor->read_participant); + + oldcxt = MemoryContextSwitchTo(accessor->context); accessor->read_file = BufFileOpenFileSet(&accessor->fileset->fs, name, O_RDONLY, false); + MemoryContextSwitchTo(oldcxt); } /* Seek and load the chunk header. */ diff --git a/src/include/executor/hashjoin.h b/src/include/executor/hashjoin.h index 8ee59d2c71..09113f0192 100644 --- a/src/include/executor/hashjoin.h +++ b/src/include/executor/hashjoin.h @@ -23,22 +23,31 @@ /* ---------------------------------------------------------------- * hash-join hash table structures * - * Each active hashjoin has a HashJoinTable control block, which is - * palloc'd in the executor's per-query context. All other storage needed - * for the hashjoin is kept in private memory contexts, two for each hashjoin. - * This makes it easy and fast to release the storage when we don't need it - * anymore. (Exception: data associated with the temp files lives in the - * per-query context too, since we always call buffile.c in that context.) + * Each active hashjoin has a HashJoinTable structure, which is + * palloc'd in the executor's per-query context. Other storages needed for + * each hashjoin is kept in child contexts, three for each hashjoin: + * - HashTableContext (hashCxt): the parent hash table storage context + * - HashSpillContext (spillCxt): storage for temp files buffers + * - HashBatchContext (batchCxt): storage for a batch in serial hash join * * The hashtable contexts are made children of the per-query context, ensuring * that they will be discarded at end of statement even if the join is * aborted early by an error. (Likewise, any temporary files we make will * be cleaned up by the virtual file manager in event of an error.) * - * Storage that should live through the entire join is allocated from the - * "hashCxt", while storage that is only wanted for the current batch is - * allocated in the "batchCxt". By resetting the batchCxt at the end of - * each batch, we free all the per-batch storage reliably and without tedium. + * The "hashCxt" context makes it easy and fast to release the storage when we + * don't need it anymore. Storage that should live through the entire join is + * allocated from the "hashCxt" (mainly hash's meta datas). + * + * Data associated with temp files lives in the "spillCxt" context which lives + * during the entire join as temp files might need to survives batches. These + * files are explicitly destroyed by calling BufFileClose() when the code is + * done with them. The aim of this context is to help accounting the memory + * allocations dedicated to temp files and their buffers. + * + * Finaly, storage that is only wanted for the current batch is allocated in + * the "batchCxt". By resetting the batchCxt at the end of each batch, we free + * all the per-batch storage reliably and without tedium. * * During first scan of inner relation, we get its tuples from executor. * If nbatch > 1 then tuples that don't belong in first batch get saved @@ -350,6 +359,7 @@ typedef struct HashJoinTableData MemoryContext hashCxt; /* context for whole-hash-join storage */ MemoryContext batchCxt; /* context for this-batch-only storage */ + MemoryContext spillCxt; /* context for spilling to temp files */ /* used for dense allocation of tuples (into linked chunks) */ HashMemoryChunk chunks; /* one list for the whole batch */ diff --git a/src/include/executor/nodeHashjoin.h b/src/include/executor/nodeHashjoin.h index d367070883..958df42240 100644 --- a/src/include/executor/nodeHashjoin.h +++ b/src/include/executor/nodeHashjoin.h @@ -29,6 +29,6 @@ extern void ExecHashJoinInitializeWorker(HashJoinState *state, ParallelWorkerContext *pwcxt); extern void ExecHashJoinSaveTuple(MinimalTuple tuple, uint32 hashvalue, - BufFile **fileptr); + BufFile **fileptr, MemoryContext spillcxt); #endif /* NODEHASHJOIN_H */ -- 2.40.1 --MP_/WHuN2qnG.U=I.v0AxCf+Cxe-- ^ permalink raw reply [nested|flat] 9+ messages in thread
* [PATCH 2/3] Allocate hash batches related BufFile in a dedicated context @ 2023-03-27 13:54 Jehan-Guillaume de Rorthais <[email protected]> 0 siblings, 0 replies; 9+ messages in thread From: Jehan-Guillaume de Rorthais @ 2023-03-27 13:54 UTC (permalink / raw) --- src/backend/executor/nodeHash.c | 43 ++++++++++++++++------- src/backend/executor/nodeHashjoin.c | 18 +++++++--- src/backend/utils/sort/sharedtuplestore.c | 8 +++++ src/include/executor/hashjoin.h | 15 ++++++-- src/include/executor/nodeHashjoin.h | 2 +- 5 files changed, 64 insertions(+), 22 deletions(-) diff --git a/src/backend/executor/nodeHash.c b/src/backend/executor/nodeHash.c index 5fd1c5553b..a4fbf29301 100644 --- a/src/backend/executor/nodeHash.c +++ b/src/backend/executor/nodeHash.c @@ -484,7 +484,7 @@ ExecHashTableCreate(HashState *state, List *hashOperators, List *hashCollations, * * The hashtable control block is just palloc'd from the executor's * per-query memory context. Everything else should be kept inside the - * subsidiary hashCxt or batchCxt. + * subsidiary hashCxt, batchCxt or fileCxt. */ hashtable = palloc_object(HashJoinTableData); hashtable->nbuckets = nbuckets; @@ -538,6 +538,10 @@ ExecHashTableCreate(HashState *state, List *hashOperators, List *hashCollations, "HashBatchContext", ALLOCSET_DEFAULT_SIZES); + hashtable->fileCxt = AllocSetContextCreate(hashtable->hashCxt, + "HashBatchFiles", + ALLOCSET_DEFAULT_SIZES); + /* Allocate data that will live for the life of the hashjoin */ oldcxt = MemoryContextSwitchTo(hashtable->hashCxt); @@ -570,15 +574,21 @@ ExecHashTableCreate(HashState *state, List *hashOperators, List *hashCollations, if (nbatch > 1 && hashtable->parallel_state == NULL) { + MemoryContext oldctx; + /* * allocate and initialize the file arrays in hashCxt (not needed for * parallel case which uses shared tuplestores instead of raw files) */ + oldctx = MemoryContextSwitchTo(hashtable->fileCxt); + hashtable->innerBatchFile = palloc0_array(BufFile *, nbatch); hashtable->outerBatchFile = palloc0_array(BufFile *, nbatch); /* The files will not be opened until needed... */ /* ... but make sure we have temp tablespaces established for them */ PrepareTempTablespaces(); + + MemoryContextSwitchTo(oldctx); } MemoryContextSwitchTo(oldcxt); @@ -913,7 +923,6 @@ ExecHashIncreaseNumBatches(HashJoinTable hashtable) int oldnbatch = hashtable->nbatch; int curbatch = hashtable->curbatch; int nbatch; - MemoryContext oldcxt; long ninmemory; long nfreed; HashMemoryChunk oldchunks; @@ -934,13 +943,16 @@ ExecHashIncreaseNumBatches(HashJoinTable hashtable) hashtable, nbatch, hashtable->spaceUsed); #endif - oldcxt = MemoryContextSwitchTo(hashtable->hashCxt); - if (hashtable->innerBatchFile == NULL) { + MemoryContext oldcxt = MemoryContextSwitchTo(hashtable->fileCxt); + /* we had no file arrays before */ hashtable->innerBatchFile = palloc0_array(BufFile *, nbatch); hashtable->outerBatchFile = palloc0_array(BufFile *, nbatch); + + MemoryContextSwitchTo(oldcxt); + /* time to establish the temp tablespaces, too */ PrepareTempTablespaces(); } @@ -951,8 +963,6 @@ ExecHashIncreaseNumBatches(HashJoinTable hashtable) hashtable->outerBatchFile = repalloc0_array(hashtable->outerBatchFile, BufFile *, oldnbatch, nbatch); } - MemoryContextSwitchTo(oldcxt); - hashtable->nbatch = nbatch; /* @@ -1022,9 +1032,11 @@ ExecHashIncreaseNumBatches(HashJoinTable hashtable) { /* dump it out */ Assert(batchno > curbatch); + ExecHashJoinSaveTuple(HJTUPLE_MINTUPLE(hashTuple), hashTuple->hashvalue, - &hashtable->innerBatchFile[batchno]); + &hashtable->innerBatchFile[batchno], + hashtable->fileCxt); hashtable->spaceUsed -= hashTupleSize; nfreed++; @@ -1681,9 +1693,11 @@ ExecHashTableInsert(HashJoinTable hashtable, * put the tuple into a temp file for later batches */ Assert(batchno > hashtable->curbatch); + ExecHashJoinSaveTuple(tuple, hashvalue, - &hashtable->innerBatchFile[batchno]); + &hashtable->innerBatchFile[batchno], + hashtable->fileCxt); } if (shouldFree) @@ -2663,8 +2677,11 @@ ExecHashRemoveNextSkewBucket(HashJoinTable hashtable) { /* Put the tuple into a temp file for later batches */ Assert(batchno > hashtable->curbatch); + ExecHashJoinSaveTuple(tuple, hashvalue, - &hashtable->innerBatchFile[batchno]); + &hashtable->innerBatchFile[batchno], + hashtable->fileCxt); + pfree(hashTuple); hashtable->spaceUsed -= tupleSize; hashtable->spaceUsedSkew -= tupleSize; @@ -3093,8 +3110,8 @@ ExecParallelHashJoinSetUpBatches(HashJoinTable hashtable, int nbatch) pstate->nbatch = nbatch; batches = dsa_get_address(hashtable->area, pstate->batches); - /* Use hash join memory context. */ - oldcxt = MemoryContextSwitchTo(hashtable->hashCxt); + /* Use hash join file memory context. */ + oldcxt = MemoryContextSwitchTo(hashtable->fileCxt); /* Allocate this backend's accessor array. */ hashtable->nbatch = nbatch; @@ -3196,8 +3213,8 @@ ExecParallelHashEnsureBatchAccessors(HashJoinTable hashtable) */ Assert(DsaPointerIsValid(pstate->batches)); - /* Use hash join memory context. */ - oldcxt = MemoryContextSwitchTo(hashtable->hashCxt); + /* Use hash join file memory context. */ + oldcxt = MemoryContextSwitchTo(hashtable->fileCxt); /* Allocate this backend's accessor array. */ hashtable->nbatch = pstate->nbatch; diff --git a/src/backend/executor/nodeHashjoin.c b/src/backend/executor/nodeHashjoin.c index 920d1831c2..ac72fbfbb6 100644 --- a/src/backend/executor/nodeHashjoin.c +++ b/src/backend/executor/nodeHashjoin.c @@ -485,8 +485,10 @@ ExecHashJoinImpl(PlanState *pstate, bool parallel) */ Assert(parallel_state == NULL); Assert(batchno > hashtable->curbatch); + ExecHashJoinSaveTuple(mintuple, hashvalue, - &hashtable->outerBatchFile[batchno]); + &hashtable->outerBatchFile[batchno], + hashtable->fileCxt); if (shouldFree) heap_free_minimal_tuple(mintuple); @@ -1308,21 +1310,27 @@ ExecParallelHashJoinNewBatch(HashJoinState *hjstate) * The data recorded in the file for each tuple is its hash value, * then the tuple in MinimalTuple format. * - * Note: it is important always to call this in the regular executor - * context, not in a shorter-lived context; else the temp file buffers - * will get messed up. + * Note: it is important always to call this in the HashBatchFiles context, + * not in a shorter-lived context; else the temp file buffers will get messed + * up. */ void ExecHashJoinSaveTuple(MinimalTuple tuple, uint32 hashvalue, - BufFile **fileptr) + BufFile **fileptr, MemoryContext filecxt) { BufFile *file = *fileptr; if (file == NULL) { + MemoryContext oldctx; + + oldctx = MemoryContextSwitchTo(filecxt); + /* First write to this batch file, so open it. */ file = BufFileCreateTemp(false); *fileptr = file; + + MemoryContextSwitchTo(oldctx); } BufFileWrite(file, &hashvalue, sizeof(uint32)); diff --git a/src/backend/utils/sort/sharedtuplestore.c b/src/backend/utils/sort/sharedtuplestore.c index 0831249159..236be65f22 100644 --- a/src/backend/utils/sort/sharedtuplestore.c +++ b/src/backend/utils/sort/sharedtuplestore.c @@ -308,11 +308,15 @@ sts_puttuple(SharedTuplestoreAccessor *accessor, void *meta_data, { SharedTuplestoreParticipant *participant; char name[MAXPGPATH]; + MemoryContext oldcxt; /* Create one. Only this backend will write into it. */ sts_filename(name, accessor, accessor->participant); + + oldcxt = MemoryContextSwitchTo(accessor->context); accessor->write_file = BufFileCreateFileSet(&accessor->fileset->fs, name); + MemoryContextSwitchTo(oldcxt); /* Set up the shared state for this backend's file. */ participant = &accessor->sts->participants[accessor->participant]; @@ -527,11 +531,15 @@ sts_parallel_scan_next(SharedTuplestoreAccessor *accessor, void *meta_data) if (accessor->read_file == NULL) { char name[MAXPGPATH]; + MemoryContext oldcxt; sts_filename(name, accessor, accessor->read_participant); + + oldcxt = MemoryContextSwitchTo(accessor->context); accessor->read_file = BufFileOpenFileSet(&accessor->fileset->fs, name, O_RDONLY, false); + MemoryContextSwitchTo(oldcxt); } /* Seek and load the chunk header. */ diff --git a/src/include/executor/hashjoin.h b/src/include/executor/hashjoin.h index 8ee59d2c71..74867c3e40 100644 --- a/src/include/executor/hashjoin.h +++ b/src/include/executor/hashjoin.h @@ -25,10 +25,14 @@ * * Each active hashjoin has a HashJoinTable control block, which is * palloc'd in the executor's per-query context. All other storage needed - * for the hashjoin is kept in private memory contexts, two for each hashjoin. + * for the hashjoin is kept in private memory contexts, three for each + * hashjoin: + * - HashTableContext (hashCxt): the control block associated to the hash table + * - HashBatchContext (batchCxt): storages for batches + * - HashBatchFiles (fileCxt): storage for temp files buffers + * * This makes it easy and fast to release the storage when we don't need it - * anymore. (Exception: data associated with the temp files lives in the - * per-query context too, since we always call buffile.c in that context.) + * anymore. * * The hashtable contexts are made children of the per-query context, ensuring * that they will be discarded at end of statement even if the join is @@ -39,6 +43,10 @@ * "hashCxt", while storage that is only wanted for the current batch is * allocated in the "batchCxt". By resetting the batchCxt at the end of * each batch, we free all the per-batch storage reliably and without tedium. + * Note that data associated with the temp files lives in the "fileCxt" context + * which lives during the entire join as temp files might need to survives + * batches. These files are explicitly destroyed by calling BufFileClose() + * when the code is done with them. * * During first scan of inner relation, we get its tuples from executor. * If nbatch > 1 then tuples that don't belong in first batch get saved @@ -350,6 +358,7 @@ typedef struct HashJoinTableData MemoryContext hashCxt; /* context for whole-hash-join storage */ MemoryContext batchCxt; /* context for this-batch-only storage */ + MemoryContext fileCxt; /* context for the BufFile related storage */ /* used for dense allocation of tuples (into linked chunks) */ HashMemoryChunk chunks; /* one list for the whole batch */ diff --git a/src/include/executor/nodeHashjoin.h b/src/include/executor/nodeHashjoin.h index d367070883..a8f9ae1989 100644 --- a/src/include/executor/nodeHashjoin.h +++ b/src/include/executor/nodeHashjoin.h @@ -29,6 +29,6 @@ extern void ExecHashJoinInitializeWorker(HashJoinState *state, ParallelWorkerContext *pwcxt); extern void ExecHashJoinSaveTuple(MinimalTuple tuple, uint32 hashvalue, - BufFile **fileptr); + BufFile **fileptr, MemoryContext filecxt); #endif /* NODEHASHJOIN_H */ -- 2.39.2 --MP_/w5PdZ2hoFAvEnG3QNPoMNWk-- ^ permalink raw reply [nested|flat] 9+ messages in thread
* [PATCH 2/3] Allocate hash batches related BufFile in a dedicated context @ 2023-03-27 13:54 Jehan-Guillaume de Rorthais <[email protected]> 0 siblings, 0 replies; 9+ messages in thread From: Jehan-Guillaume de Rorthais @ 2023-03-27 13:54 UTC (permalink / raw) --- src/backend/executor/nodeHash.c | 43 ++++++++++++++++------- src/backend/executor/nodeHashjoin.c | 24 +++++++++---- src/backend/utils/sort/sharedtuplestore.c | 8 +++++ src/include/executor/hashjoin.h | 30 ++++++++++------ src/include/executor/nodeHashjoin.h | 2 +- 5 files changed, 76 insertions(+), 31 deletions(-) diff --git a/src/backend/executor/nodeHash.c b/src/backend/executor/nodeHash.c index 5fd1c5553b..e529f50a3e 100644 --- a/src/backend/executor/nodeHash.c +++ b/src/backend/executor/nodeHash.c @@ -484,7 +484,7 @@ ExecHashTableCreate(HashState *state, List *hashOperators, List *hashCollations, * * The hashtable control block is just palloc'd from the executor's * per-query memory context. Everything else should be kept inside the - * subsidiary hashCxt or batchCxt. + * subsidiary hashCxt, batchCxt or spillCxt. */ hashtable = palloc_object(HashJoinTableData); hashtable->nbuckets = nbuckets; @@ -538,6 +538,10 @@ ExecHashTableCreate(HashState *state, List *hashOperators, List *hashCollations, "HashBatchContext", ALLOCSET_DEFAULT_SIZES); + hashtable->spillCxt = AllocSetContextCreate(hashtable->hashCxt, + "HashSpillContext", + ALLOCSET_DEFAULT_SIZES); + /* Allocate data that will live for the life of the hashjoin */ oldcxt = MemoryContextSwitchTo(hashtable->hashCxt); @@ -570,12 +574,19 @@ ExecHashTableCreate(HashState *state, List *hashOperators, List *hashCollations, if (nbatch > 1 && hashtable->parallel_state == NULL) { + MemoryContext oldctx; + /* * allocate and initialize the file arrays in hashCxt (not needed for * parallel case which uses shared tuplestores instead of raw files) */ + oldctx = MemoryContextSwitchTo(hashtable->spillCxt); + hashtable->innerBatchFile = palloc0_array(BufFile *, nbatch); hashtable->outerBatchFile = palloc0_array(BufFile *, nbatch); + + MemoryContextSwitchTo(oldctx); + /* The files will not be opened until needed... */ /* ... but make sure we have temp tablespaces established for them */ PrepareTempTablespaces(); @@ -913,7 +924,6 @@ ExecHashIncreaseNumBatches(HashJoinTable hashtable) int oldnbatch = hashtable->nbatch; int curbatch = hashtable->curbatch; int nbatch; - MemoryContext oldcxt; long ninmemory; long nfreed; HashMemoryChunk oldchunks; @@ -934,13 +944,16 @@ ExecHashIncreaseNumBatches(HashJoinTable hashtable) hashtable, nbatch, hashtable->spaceUsed); #endif - oldcxt = MemoryContextSwitchTo(hashtable->hashCxt); - if (hashtable->innerBatchFile == NULL) { + MemoryContext oldcxt = MemoryContextSwitchTo(hashtable->spillCxt); + /* we had no file arrays before */ hashtable->innerBatchFile = palloc0_array(BufFile *, nbatch); hashtable->outerBatchFile = palloc0_array(BufFile *, nbatch); + + MemoryContextSwitchTo(oldcxt); + /* time to establish the temp tablespaces, too */ PrepareTempTablespaces(); } @@ -951,8 +964,6 @@ ExecHashIncreaseNumBatches(HashJoinTable hashtable) hashtable->outerBatchFile = repalloc0_array(hashtable->outerBatchFile, BufFile *, oldnbatch, nbatch); } - MemoryContextSwitchTo(oldcxt); - hashtable->nbatch = nbatch; /* @@ -1024,7 +1035,8 @@ ExecHashIncreaseNumBatches(HashJoinTable hashtable) Assert(batchno > curbatch); ExecHashJoinSaveTuple(HJTUPLE_MINTUPLE(hashTuple), hashTuple->hashvalue, - &hashtable->innerBatchFile[batchno]); + &hashtable->innerBatchFile[batchno], + hashtable->spillCxt); hashtable->spaceUsed -= hashTupleSize; nfreed++; @@ -1683,7 +1695,8 @@ ExecHashTableInsert(HashJoinTable hashtable, Assert(batchno > hashtable->curbatch); ExecHashJoinSaveTuple(tuple, hashvalue, - &hashtable->innerBatchFile[batchno]); + &hashtable->innerBatchFile[batchno], + hashtable->spillCxt); } if (shouldFree) @@ -2664,7 +2677,8 @@ ExecHashRemoveNextSkewBucket(HashJoinTable hashtable) /* Put the tuple into a temp file for later batches */ Assert(batchno > hashtable->curbatch); ExecHashJoinSaveTuple(tuple, hashvalue, - &hashtable->innerBatchFile[batchno]); + &hashtable->innerBatchFile[batchno], + hashtable->spillCxt); pfree(hashTuple); hashtable->spaceUsed -= tupleSize; hashtable->spaceUsedSkew -= tupleSize; @@ -3093,8 +3107,11 @@ ExecParallelHashJoinSetUpBatches(HashJoinTable hashtable, int nbatch) pstate->nbatch = nbatch; batches = dsa_get_address(hashtable->area, pstate->batches); - /* Use hash join memory context. */ - oldcxt = MemoryContextSwitchTo(hashtable->hashCxt); + /* + * Use hash join spill memory context to allocate accessors and their + * buffers + */ + oldcxt = MemoryContextSwitchTo(hashtable->spillCxt); /* Allocate this backend's accessor array. */ hashtable->nbatch = nbatch; @@ -3196,8 +3213,8 @@ ExecParallelHashEnsureBatchAccessors(HashJoinTable hashtable) */ Assert(DsaPointerIsValid(pstate->batches)); - /* Use hash join memory context. */ - oldcxt = MemoryContextSwitchTo(hashtable->hashCxt); + /* Use hash join spill memory context to allocate accessors */ + oldcxt = MemoryContextSwitchTo(hashtable->spillCxt); /* Allocate this backend's accessor array. */ hashtable->nbatch = pstate->nbatch; diff --git a/src/backend/executor/nodeHashjoin.c b/src/backend/executor/nodeHashjoin.c index 920d1831c2..3b903cc905 100644 --- a/src/backend/executor/nodeHashjoin.c +++ b/src/backend/executor/nodeHashjoin.c @@ -486,7 +486,8 @@ ExecHashJoinImpl(PlanState *pstate, bool parallel) Assert(parallel_state == NULL); Assert(batchno > hashtable->curbatch); ExecHashJoinSaveTuple(mintuple, hashvalue, - &hashtable->outerBatchFile[batchno]); + &hashtable->outerBatchFile[batchno], + hashtable->spillCxt); if (shouldFree) heap_free_minimal_tuple(mintuple); @@ -1307,22 +1308,31 @@ ExecParallelHashJoinNewBatch(HashJoinState *hjstate) * * The data recorded in the file for each tuple is its hash value, * then the tuple in MinimalTuple format. - * - * Note: it is important always to call this in the regular executor - * context, not in a shorter-lived context; else the temp file buffers - * will get messed up. */ void ExecHashJoinSaveTuple(MinimalTuple tuple, uint32 hashvalue, - BufFile **fileptr) + BufFile **fileptr, MemoryContext spillcxt) { BufFile *file = *fileptr; if (file == NULL) { - /* First write to this batch file, so open it. */ + MemoryContext oldctx; + + /* + * First write to this batch file, so open it. + * + * It is important to always create BufFile'd temp files in the + * HashSpillContext context. First to avoid a shorter-lived context, + * else the temp file buffers will get messed up. Second, for a better + * accounting of the spilling memory consumption. + */ + oldctx = MemoryContextSwitchTo(spillcxt); + file = BufFileCreateTemp(false); *fileptr = file; + + MemoryContextSwitchTo(oldctx); } BufFileWrite(file, &hashvalue, sizeof(uint32)); diff --git a/src/backend/utils/sort/sharedtuplestore.c b/src/backend/utils/sort/sharedtuplestore.c index 0831249159..236be65f22 100644 --- a/src/backend/utils/sort/sharedtuplestore.c +++ b/src/backend/utils/sort/sharedtuplestore.c @@ -308,11 +308,15 @@ sts_puttuple(SharedTuplestoreAccessor *accessor, void *meta_data, { SharedTuplestoreParticipant *participant; char name[MAXPGPATH]; + MemoryContext oldcxt; /* Create one. Only this backend will write into it. */ sts_filename(name, accessor, accessor->participant); + + oldcxt = MemoryContextSwitchTo(accessor->context); accessor->write_file = BufFileCreateFileSet(&accessor->fileset->fs, name); + MemoryContextSwitchTo(oldcxt); /* Set up the shared state for this backend's file. */ participant = &accessor->sts->participants[accessor->participant]; @@ -527,11 +531,15 @@ sts_parallel_scan_next(SharedTuplestoreAccessor *accessor, void *meta_data) if (accessor->read_file == NULL) { char name[MAXPGPATH]; + MemoryContext oldcxt; sts_filename(name, accessor, accessor->read_participant); + + oldcxt = MemoryContextSwitchTo(accessor->context); accessor->read_file = BufFileOpenFileSet(&accessor->fileset->fs, name, O_RDONLY, false); + MemoryContextSwitchTo(oldcxt); } /* Seek and load the chunk header. */ diff --git a/src/include/executor/hashjoin.h b/src/include/executor/hashjoin.h index 8ee59d2c71..09113f0192 100644 --- a/src/include/executor/hashjoin.h +++ b/src/include/executor/hashjoin.h @@ -23,22 +23,31 @@ /* ---------------------------------------------------------------- * hash-join hash table structures * - * Each active hashjoin has a HashJoinTable control block, which is - * palloc'd in the executor's per-query context. All other storage needed - * for the hashjoin is kept in private memory contexts, two for each hashjoin. - * This makes it easy and fast to release the storage when we don't need it - * anymore. (Exception: data associated with the temp files lives in the - * per-query context too, since we always call buffile.c in that context.) + * Each active hashjoin has a HashJoinTable structure, which is + * palloc'd in the executor's per-query context. Other storages needed for + * each hashjoin is kept in child contexts, three for each hashjoin: + * - HashTableContext (hashCxt): the parent hash table storage context + * - HashSpillContext (spillCxt): storage for temp files buffers + * - HashBatchContext (batchCxt): storage for a batch in serial hash join * * The hashtable contexts are made children of the per-query context, ensuring * that they will be discarded at end of statement even if the join is * aborted early by an error. (Likewise, any temporary files we make will * be cleaned up by the virtual file manager in event of an error.) * - * Storage that should live through the entire join is allocated from the - * "hashCxt", while storage that is only wanted for the current batch is - * allocated in the "batchCxt". By resetting the batchCxt at the end of - * each batch, we free all the per-batch storage reliably and without tedium. + * The "hashCxt" context makes it easy and fast to release the storage when we + * don't need it anymore. Storage that should live through the entire join is + * allocated from the "hashCxt" (mainly hash's meta datas). + * + * Data associated with temp files lives in the "spillCxt" context which lives + * during the entire join as temp files might need to survives batches. These + * files are explicitly destroyed by calling BufFileClose() when the code is + * done with them. The aim of this context is to help accounting the memory + * allocations dedicated to temp files and their buffers. + * + * Finaly, storage that is only wanted for the current batch is allocated in + * the "batchCxt". By resetting the batchCxt at the end of each batch, we free + * all the per-batch storage reliably and without tedium. * * During first scan of inner relation, we get its tuples from executor. * If nbatch > 1 then tuples that don't belong in first batch get saved @@ -350,6 +359,7 @@ typedef struct HashJoinTableData MemoryContext hashCxt; /* context for whole-hash-join storage */ MemoryContext batchCxt; /* context for this-batch-only storage */ + MemoryContext spillCxt; /* context for spilling to temp files */ /* used for dense allocation of tuples (into linked chunks) */ HashMemoryChunk chunks; /* one list for the whole batch */ diff --git a/src/include/executor/nodeHashjoin.h b/src/include/executor/nodeHashjoin.h index d367070883..958df42240 100644 --- a/src/include/executor/nodeHashjoin.h +++ b/src/include/executor/nodeHashjoin.h @@ -29,6 +29,6 @@ extern void ExecHashJoinInitializeWorker(HashJoinState *state, ParallelWorkerContext *pwcxt); extern void ExecHashJoinSaveTuple(MinimalTuple tuple, uint32 hashvalue, - BufFile **fileptr); + BufFile **fileptr, MemoryContext spillcxt); #endif /* NODEHASHJOIN_H */ -- 2.39.2 --MP_/=afXY/P8/zXs58RNNJ5YG6H-- ^ permalink raw reply [nested|flat] 9+ messages in thread
* [PATCH] Allocate hash batches related BufFile in a dedicated context @ 2023-03-27 13:54 Jehan-Guillaume de Rorthais <[email protected]> 0 siblings, 0 replies; 9+ messages in thread From: Jehan-Guillaume de Rorthais @ 2023-03-27 13:54 UTC (permalink / raw) --- src/backend/executor/nodeHash.c | 27 ++++++++++++++++++++++----- src/backend/executor/nodeHashjoin.c | 18 +++++++++++++----- src/include/executor/hashjoin.h | 15 ++++++++++++--- src/include/executor/nodeHashjoin.h | 2 +- 4 files changed, 48 insertions(+), 14 deletions(-) diff --git a/src/backend/executor/nodeHash.c b/src/backend/executor/nodeHash.c index 748c9b0024..0c0d5b4a3c 100644 --- a/src/backend/executor/nodeHash.c +++ b/src/backend/executor/nodeHash.c @@ -484,7 +484,7 @@ ExecHashTableCreate(HashState *state, List *hashOperators, List *hashCollations, * * The hashtable control block is just palloc'd from the executor's * per-query memory context. Everything else should be kept inside the - * subsidiary hashCxt or batchCxt. + * subsidiary hashCxt, batchCxt or fileCxt. */ hashtable = palloc_object(HashJoinTableData); hashtable->nbuckets = nbuckets; @@ -538,6 +538,10 @@ ExecHashTableCreate(HashState *state, List *hashOperators, List *hashCollations, "HashBatchContext", ALLOCSET_DEFAULT_SIZES); + hashtable->fileCxt = AllocSetContextCreate(hashtable->hashCxt, + "HashBatchFiles", + ALLOCSET_DEFAULT_SIZES); + /* Allocate data that will live for the life of the hashjoin */ oldcxt = MemoryContextSwitchTo(hashtable->hashCxt); @@ -570,15 +574,21 @@ ExecHashTableCreate(HashState *state, List *hashOperators, List *hashCollations, if (nbatch > 1 && hashtable->parallel_state == NULL) { + MemoryContext oldctx; + /* * allocate and initialize the file arrays in hashCxt (not needed for * parallel case which uses shared tuplestores instead of raw files) */ + oldctx = MemoryContextSwitchTo(hashtable->fileCxt); + hashtable->innerBatchFile = palloc0_array(BufFile *, nbatch); hashtable->outerBatchFile = palloc0_array(BufFile *, nbatch); /* The files will not be opened until needed... */ /* ... but make sure we have temp tablespaces established for them */ PrepareTempTablespaces(); + + MemoryContextSwitchTo(oldctx); } MemoryContextSwitchTo(oldcxt); @@ -934,7 +944,7 @@ ExecHashIncreaseNumBatches(HashJoinTable hashtable) hashtable, nbatch, hashtable->spaceUsed); #endif - oldcxt = MemoryContextSwitchTo(hashtable->hashCxt); + oldcxt = MemoryContextSwitchTo(hashtable->fileCxt); if (hashtable->innerBatchFile == NULL) { @@ -1022,9 +1032,11 @@ ExecHashIncreaseNumBatches(HashJoinTable hashtable) { /* dump it out */ Assert(batchno > curbatch); + ExecHashJoinSaveTuple(HJTUPLE_MINTUPLE(hashTuple), hashTuple->hashvalue, - &hashtable->innerBatchFile[batchno]); + &hashtable->innerBatchFile[batchno], + hashtable->fileCxt); hashtable->spaceUsed -= hashTupleSize; nfreed++; @@ -1681,9 +1693,11 @@ ExecHashTableInsert(HashJoinTable hashtable, * put the tuple into a temp file for later batches */ Assert(batchno > hashtable->curbatch); + ExecHashJoinSaveTuple(tuple, hashvalue, - &hashtable->innerBatchFile[batchno]); + &hashtable->innerBatchFile[batchno], + hashtable->fileCxt); } if (shouldFree) @@ -2534,8 +2548,11 @@ ExecHashRemoveNextSkewBucket(HashJoinTable hashtable) { /* Put the tuple into a temp file for later batches */ Assert(batchno > hashtable->curbatch); + ExecHashJoinSaveTuple(tuple, hashvalue, - &hashtable->innerBatchFile[batchno]); + &hashtable->innerBatchFile[batchno], + hashtable->fileCxt); + pfree(hashTuple); hashtable->spaceUsed -= tupleSize; hashtable->spaceUsedSkew -= tupleSize; diff --git a/src/backend/executor/nodeHashjoin.c b/src/backend/executor/nodeHashjoin.c index f189fb4d28..6055abde49 100644 --- a/src/backend/executor/nodeHashjoin.c +++ b/src/backend/executor/nodeHashjoin.c @@ -432,8 +432,10 @@ ExecHashJoinImpl(PlanState *pstate, bool parallel) */ Assert(parallel_state == NULL); Assert(batchno > hashtable->curbatch); + ExecHashJoinSaveTuple(mintuple, hashvalue, - &hashtable->outerBatchFile[batchno]); + &hashtable->outerBatchFile[batchno], + hashtable->fileCxt); if (shouldFree) heap_free_minimal_tuple(mintuple); @@ -1234,21 +1236,27 @@ ExecParallelHashJoinNewBatch(HashJoinState *hjstate) * The data recorded in the file for each tuple is its hash value, * then the tuple in MinimalTuple format. * - * Note: it is important always to call this in the regular executor - * context, not in a shorter-lived context; else the temp file buffers - * will get messed up. + * Note: it is important always to call this in the HashBatchFiles context, + * not in a shorter-lived context; else the temp file buffers will get messed + * up. */ void ExecHashJoinSaveTuple(MinimalTuple tuple, uint32 hashvalue, - BufFile **fileptr) + BufFile **fileptr, MemoryContext filecxt) { BufFile *file = *fileptr; if (file == NULL) { + MemoryContext oldctx; + + oldctx = MemoryContextSwitchTo(filecxt); + /* First write to this batch file, so open it. */ file = BufFileCreateTemp(false); *fileptr = file; + + MemoryContextSwitchTo(oldctx); } BufFileWrite(file, &hashvalue, sizeof(uint32)); diff --git a/src/include/executor/hashjoin.h b/src/include/executor/hashjoin.h index acb7592ca0..d759235d7f 100644 --- a/src/include/executor/hashjoin.h +++ b/src/include/executor/hashjoin.h @@ -25,10 +25,14 @@ * * Each active hashjoin has a HashJoinTable control block, which is * palloc'd in the executor's per-query context. All other storage needed - * for the hashjoin is kept in private memory contexts, two for each hashjoin. + * for the hashjoin is kept in private memory contexts, three for each + * hashjoin: + * - HashTableContext (hashCxt): the control block associated to the hash table + * - HashBatchContext (batchCxt): storages for batches + * - HashBatchFiles (fileCxt): storage for temp files buffers + * * This makes it easy and fast to release the storage when we don't need it - * anymore. (Exception: data associated with the temp files lives in the - * per-query context too, since we always call buffile.c in that context.) + * anymore. * * The hashtable contexts are made children of the per-query context, ensuring * that they will be discarded at end of statement even if the join is @@ -39,6 +43,10 @@ * "hashCxt", while storage that is only wanted for the current batch is * allocated in the "batchCxt". By resetting the batchCxt at the end of * each batch, we free all the per-batch storage reliably and without tedium. + * Note that data associated with the temp files lives in the "fileCxt" context + * which lives during the entire join as temp files might need to survives + * batches. These files are explicitly destroyed by calling BufFileClose() + * when the code is done with them. * * During first scan of inner relation, we get its tuples from executor. * If nbatch > 1 then tuples that don't belong in first batch get saved @@ -348,6 +356,7 @@ typedef struct HashJoinTableData MemoryContext hashCxt; /* context for whole-hash-join storage */ MemoryContext batchCxt; /* context for this-batch-only storage */ + MemoryContext fileCxt; /* context for the BufFile related storage */ /* used for dense allocation of tuples (into linked chunks) */ HashMemoryChunk chunks; /* one list for the whole batch */ diff --git a/src/include/executor/nodeHashjoin.h b/src/include/executor/nodeHashjoin.h index d367070883..a8f9ae1989 100644 --- a/src/include/executor/nodeHashjoin.h +++ b/src/include/executor/nodeHashjoin.h @@ -29,6 +29,6 @@ extern void ExecHashJoinInitializeWorker(HashJoinState *state, ParallelWorkerContext *pwcxt); extern void ExecHashJoinSaveTuple(MinimalTuple tuple, uint32 hashvalue, - BufFile **fileptr); + BufFile **fileptr, MemoryContext filecxt); #endif /* NODEHASHJOIN_H */ -- 2.39.2 --MP_/HNeUeUP7WKSzkiBqcPpwVN+-- ^ permalink raw reply [nested|flat] 9+ messages in thread
* [PATCH 2/4] Allocate hash batches related BufFile in a dedicated context @ 2023-03-27 13:54 Jehan-Guillaume de Rorthais <[email protected]> 0 siblings, 0 replies; 9+ messages in thread From: Jehan-Guillaume de Rorthais @ 2023-03-27 13:54 UTC (permalink / raw) --- src/backend/executor/nodeHash.c | 35 +++++++++++++++++++++-------- src/backend/executor/nodeHashjoin.c | 18 ++++++++++----- src/include/executor/hashjoin.h | 15 ++++++++++--- src/include/executor/nodeHashjoin.h | 2 +- 4 files changed, 52 insertions(+), 18 deletions(-) diff --git a/src/backend/executor/nodeHash.c b/src/backend/executor/nodeHash.c index a45bd3a315..4544296391 100644 --- a/src/backend/executor/nodeHash.c +++ b/src/backend/executor/nodeHash.c @@ -484,7 +484,7 @@ ExecHashTableCreate(HashState *state, List *hashOperators, List *hashCollations, * * The hashtable control block is just palloc'd from the executor's * per-query memory context. Everything else should be kept inside the - * subsidiary hashCxt or batchCxt. + * subsidiary hashCxt, batchCxt or fileCxt. */ hashtable = palloc_object(HashJoinTableData); hashtable->nbuckets = nbuckets; @@ -538,6 +538,10 @@ ExecHashTableCreate(HashState *state, List *hashOperators, List *hashCollations, "HashBatchContext", ALLOCSET_DEFAULT_SIZES); + hashtable->fileCxt = AllocSetContextCreate(hashtable->hashCxt, + "HashBatchFiles", + ALLOCSET_DEFAULT_SIZES); + /* Allocate data that will live for the life of the hashjoin */ oldcxt = MemoryContextSwitchTo(hashtable->hashCxt); @@ -570,15 +574,21 @@ ExecHashTableCreate(HashState *state, List *hashOperators, List *hashCollations, if (nbatch > 1 && hashtable->parallel_state == NULL) { + MemoryContext oldctx; + /* * allocate and initialize the file arrays in hashCxt (not needed for * parallel case which uses shared tuplestores instead of raw files) */ + oldctx = MemoryContextSwitchTo(hashtable->fileCxt); + hashtable->innerBatchFile = palloc0_array(BufFile *, nbatch); hashtable->outerBatchFile = palloc0_array(BufFile *, nbatch); /* The files will not be opened until needed... */ /* ... but make sure we have temp tablespaces established for them */ PrepareTempTablespaces(); + + MemoryContextSwitchTo(oldctx); } MemoryContextSwitchTo(oldcxt); @@ -913,7 +923,6 @@ ExecHashIncreaseNumBatches(HashJoinTable hashtable) int oldnbatch = hashtable->nbatch; int curbatch = hashtable->curbatch; int nbatch; - MemoryContext oldcxt; long ninmemory; long nfreed; HashMemoryChunk oldchunks; @@ -934,13 +943,16 @@ ExecHashIncreaseNumBatches(HashJoinTable hashtable) hashtable, nbatch, hashtable->spaceUsed); #endif - oldcxt = MemoryContextSwitchTo(hashtable->hashCxt); - if (hashtable->innerBatchFile == NULL) { + MemoryContext oldcxt = MemoryContextSwitchTo(hashtable->fileCxt); + /* we had no file arrays before */ hashtable->innerBatchFile = palloc0_array(BufFile *, nbatch); hashtable->outerBatchFile = palloc0_array(BufFile *, nbatch); + + MemoryContextSwitchTo(oldcxt); + /* time to establish the temp tablespaces, too */ PrepareTempTablespaces(); } @@ -951,8 +963,6 @@ ExecHashIncreaseNumBatches(HashJoinTable hashtable) hashtable->outerBatchFile = repalloc0_array(hashtable->outerBatchFile, BufFile *, oldnbatch, nbatch); } - MemoryContextSwitchTo(oldcxt); - hashtable->nbatch = nbatch; /* @@ -1022,9 +1032,11 @@ ExecHashIncreaseNumBatches(HashJoinTable hashtable) { /* dump it out */ Assert(batchno > curbatch); + ExecHashJoinSaveTuple(HJTUPLE_MINTUPLE(hashTuple), hashTuple->hashvalue, - &hashtable->innerBatchFile[batchno]); + &hashtable->innerBatchFile[batchno], + hashtable->fileCxt); hashtable->spaceUsed -= hashTupleSize; nfreed++; @@ -1681,9 +1693,11 @@ ExecHashTableInsert(HashJoinTable hashtable, * put the tuple into a temp file for later batches */ Assert(batchno > hashtable->curbatch); + ExecHashJoinSaveTuple(tuple, hashvalue, - &hashtable->innerBatchFile[batchno]); + &hashtable->innerBatchFile[batchno], + hashtable->fileCxt); } if (shouldFree) @@ -2663,8 +2677,11 @@ ExecHashRemoveNextSkewBucket(HashJoinTable hashtable) { /* Put the tuple into a temp file for later batches */ Assert(batchno > hashtable->curbatch); + ExecHashJoinSaveTuple(tuple, hashvalue, - &hashtable->innerBatchFile[batchno]); + &hashtable->innerBatchFile[batchno], + hashtable->fileCxt); + pfree(hashTuple); hashtable->spaceUsed -= tupleSize; hashtable->spaceUsedSkew -= tupleSize; diff --git a/src/backend/executor/nodeHashjoin.c b/src/backend/executor/nodeHashjoin.c index 5454afbff2..5bc7f814c6 100644 --- a/src/backend/executor/nodeHashjoin.c +++ b/src/backend/executor/nodeHashjoin.c @@ -485,8 +485,10 @@ ExecHashJoinImpl(PlanState *pstate, bool parallel) */ Assert(parallel_state == NULL); Assert(batchno > hashtable->curbatch); + ExecHashJoinSaveTuple(mintuple, hashvalue, - &hashtable->outerBatchFile[batchno]); + &hashtable->outerBatchFile[batchno], + hashtable->fileCxt); if (shouldFree) heap_free_minimal_tuple(mintuple); @@ -1297,21 +1299,27 @@ ExecParallelHashJoinNewBatch(HashJoinState *hjstate) * The data recorded in the file for each tuple is its hash value, * then the tuple in MinimalTuple format. * - * Note: it is important always to call this in the regular executor - * context, not in a shorter-lived context; else the temp file buffers - * will get messed up. + * Note: it is important always to call this in the HashBatchFiles context, + * not in a shorter-lived context; else the temp file buffers will get messed + * up. */ void ExecHashJoinSaveTuple(MinimalTuple tuple, uint32 hashvalue, - BufFile **fileptr) + BufFile **fileptr, MemoryContext filecxt) { BufFile *file = *fileptr; if (file == NULL) { + MemoryContext oldctx; + + oldctx = MemoryContextSwitchTo(filecxt); + /* First write to this batch file, so open it. */ file = BufFileCreateTemp(false); *fileptr = file; + + MemoryContextSwitchTo(oldctx); } BufFileWrite(file, &hashvalue, sizeof(uint32)); diff --git a/src/include/executor/hashjoin.h b/src/include/executor/hashjoin.h index 8ee59d2c71..74867c3e40 100644 --- a/src/include/executor/hashjoin.h +++ b/src/include/executor/hashjoin.h @@ -25,10 +25,14 @@ * * Each active hashjoin has a HashJoinTable control block, which is * palloc'd in the executor's per-query context. All other storage needed - * for the hashjoin is kept in private memory contexts, two for each hashjoin. + * for the hashjoin is kept in private memory contexts, three for each + * hashjoin: + * - HashTableContext (hashCxt): the control block associated to the hash table + * - HashBatchContext (batchCxt): storages for batches + * - HashBatchFiles (fileCxt): storage for temp files buffers + * * This makes it easy and fast to release the storage when we don't need it - * anymore. (Exception: data associated with the temp files lives in the - * per-query context too, since we always call buffile.c in that context.) + * anymore. * * The hashtable contexts are made children of the per-query context, ensuring * that they will be discarded at end of statement even if the join is @@ -39,6 +43,10 @@ * "hashCxt", while storage that is only wanted for the current batch is * allocated in the "batchCxt". By resetting the batchCxt at the end of * each batch, we free all the per-batch storage reliably and without tedium. + * Note that data associated with the temp files lives in the "fileCxt" context + * which lives during the entire join as temp files might need to survives + * batches. These files are explicitly destroyed by calling BufFileClose() + * when the code is done with them. * * During first scan of inner relation, we get its tuples from executor. * If nbatch > 1 then tuples that don't belong in first batch get saved @@ -350,6 +358,7 @@ typedef struct HashJoinTableData MemoryContext hashCxt; /* context for whole-hash-join storage */ MemoryContext batchCxt; /* context for this-batch-only storage */ + MemoryContext fileCxt; /* context for the BufFile related storage */ /* used for dense allocation of tuples (into linked chunks) */ HashMemoryChunk chunks; /* one list for the whole batch */ diff --git a/src/include/executor/nodeHashjoin.h b/src/include/executor/nodeHashjoin.h index d367070883..a8f9ae1989 100644 --- a/src/include/executor/nodeHashjoin.h +++ b/src/include/executor/nodeHashjoin.h @@ -29,6 +29,6 @@ extern void ExecHashJoinInitializeWorker(HashJoinState *state, ParallelWorkerContext *pwcxt); extern void ExecHashJoinSaveTuple(MinimalTuple tuple, uint32 hashvalue, - BufFile **fileptr); + BufFile **fileptr, MemoryContext filecxt); #endif /* NODEHASHJOIN_H */ -- 2.39.2 --MP_/xtmrMo8rz+yGL.lLMmD=jly Content-Type: text/x-patch Content-Transfer-Encoding: 7bit Content-Disposition: attachment; filename=0003-v4-Add-some-debug-and-metrics.patch ^ permalink raw reply [nested|flat] 9+ messages in thread
* [PATCH] Allocate hash batches related BufFile in a dedicated context @ 2023-03-27 13:54 Jehan-Guillaume de Rorthais <[email protected]> 0 siblings, 0 replies; 9+ messages in thread From: Jehan-Guillaume de Rorthais @ 2023-03-27 13:54 UTC (permalink / raw) --- src/backend/executor/nodeHash.c | 39 +++++++++++++++++++++++++++-- src/backend/executor/nodeHashjoin.c | 14 ++++++++--- src/include/executor/hashjoin.h | 1 + 3 files changed, 49 insertions(+), 5 deletions(-) diff --git a/src/backend/executor/nodeHash.c b/src/backend/executor/nodeHash.c index 748c9b0024..3f1ae9755e 100644 --- a/src/backend/executor/nodeHash.c +++ b/src/backend/executor/nodeHash.c @@ -484,7 +484,7 @@ ExecHashTableCreate(HashState *state, List *hashOperators, List *hashCollations, * * The hashtable control block is just palloc'd from the executor's * per-query memory context. Everything else should be kept inside the - * subsidiary hashCxt or batchCxt. + * subsidiary hashCxt, batchCxt or fileCxt. */ hashtable = palloc_object(HashJoinTableData); hashtable->nbuckets = nbuckets; @@ -538,6 +538,10 @@ ExecHashTableCreate(HashState *state, List *hashOperators, List *hashCollations, "HashBatchContext", ALLOCSET_DEFAULT_SIZES); + hashtable->fileCxt = AllocSetContextCreate(CurrentMemoryContext, + "HashBatchFiles", + ALLOCSET_DEFAULT_SIZES); + /* Allocate data that will live for the life of the hashjoin */ oldcxt = MemoryContextSwitchTo(hashtable->hashCxt); @@ -570,15 +574,21 @@ ExecHashTableCreate(HashState *state, List *hashOperators, List *hashCollations, if (nbatch > 1 && hashtable->parallel_state == NULL) { + MemoryContext oldctx; + /* * allocate and initialize the file arrays in hashCxt (not needed for * parallel case which uses shared tuplestores instead of raw files) */ + oldctx = MemoryContextSwitchTo(hashtable->fileCxt); + hashtable->innerBatchFile = palloc0_array(BufFile *, nbatch); hashtable->outerBatchFile = palloc0_array(BufFile *, nbatch); /* The files will not be opened until needed... */ /* ... but make sure we have temp tablespaces established for them */ PrepareTempTablespaces(); + + MemoryContextSwitchTo(oldctx); } MemoryContextSwitchTo(oldcxt); @@ -934,7 +944,7 @@ ExecHashIncreaseNumBatches(HashJoinTable hashtable) hashtable, nbatch, hashtable->spaceUsed); #endif - oldcxt = MemoryContextSwitchTo(hashtable->hashCxt); + oldcxt = MemoryContextSwitchTo(hashtable->fileCxt); if (hashtable->innerBatchFile == NULL) { @@ -1020,12 +1030,19 @@ ExecHashIncreaseNumBatches(HashJoinTable hashtable) } else { + MemoryContext oldctx; + /* dump it out */ Assert(batchno > curbatch); + + oldctx = MemoryContextSwitchTo(hashtable->fileCxt); + ExecHashJoinSaveTuple(HJTUPLE_MINTUPLE(hashTuple), hashTuple->hashvalue, &hashtable->innerBatchFile[batchno]); + MemoryContextSwitchTo(oldctx); + hashtable->spaceUsed -= hashTupleSize; nfreed++; } @@ -1042,6 +1059,9 @@ ExecHashIncreaseNumBatches(HashJoinTable hashtable) oldchunks = nextchunk; } + if (hashtable->fileCxt->mem_allocated > hashtable->spaceAllowed) + elog(WARNING, "Growing number of hash batch is exhausting memory"); + #ifdef HJDEBUG printf("Hashjoin %p: freed %ld of %ld tuples, space now %zu\n", hashtable, nfreed, ninmemory, hashtable->spaceUsed); @@ -1677,13 +1697,20 @@ ExecHashTableInsert(HashJoinTable hashtable, } else { + MemoryContext oldctx; + /* * put the tuple into a temp file for later batches */ Assert(batchno > hashtable->curbatch); + + oldctx = MemoryContextSwitchTo(hashtable->fileCxt); + ExecHashJoinSaveTuple(tuple, hashvalue, &hashtable->innerBatchFile[batchno]); + + MemoryContextSwitchTo(oldctx); } if (shouldFree) @@ -2532,10 +2559,18 @@ ExecHashRemoveNextSkewBucket(HashJoinTable hashtable) } else { + MemoryContext oldctx; + /* Put the tuple into a temp file for later batches */ Assert(batchno > hashtable->curbatch); + + oldctx = MemoryContextSwitchTo(hashtable->fileCxt); + ExecHashJoinSaveTuple(tuple, hashvalue, &hashtable->innerBatchFile[batchno]); + + MemoryContextSwitchTo(oldctx); + pfree(hashTuple); hashtable->spaceUsed -= tupleSize; hashtable->spaceUsedSkew -= tupleSize; diff --git a/src/backend/executor/nodeHashjoin.c b/src/backend/executor/nodeHashjoin.c index f189fb4d28..ba1f27c2c4 100644 --- a/src/backend/executor/nodeHashjoin.c +++ b/src/backend/executor/nodeHashjoin.c @@ -422,6 +422,7 @@ ExecHashJoinImpl(PlanState *pstate, bool parallel) if (batchno != hashtable->curbatch && node->hj_CurSkewBucketNo == INVALID_SKEW_BUCKET_NO) { + MemoryContext oldctx; bool shouldFree; MinimalTuple mintuple = ExecFetchSlotMinimalTuple(outerTupleSlot, &shouldFree); @@ -432,9 +433,14 @@ ExecHashJoinImpl(PlanState *pstate, bool parallel) */ Assert(parallel_state == NULL); Assert(batchno > hashtable->curbatch); + + oldctx = MemoryContextSwitchTo(hashtable->fileCxt); + ExecHashJoinSaveTuple(mintuple, hashvalue, &hashtable->outerBatchFile[batchno]); + MemoryContextSwitchTo(oldctx); + if (shouldFree) heap_free_minimal_tuple(mintuple); @@ -1234,9 +1240,9 @@ ExecParallelHashJoinNewBatch(HashJoinState *hjstate) * The data recorded in the file for each tuple is its hash value, * then the tuple in MinimalTuple format. * - * Note: it is important always to call this in the regular executor - * context, not in a shorter-lived context; else the temp file buffers - * will get messed up. + * Note: it is important always to call this in the HashBatchFiles context, + * not in a shorter-lived context; else the temp file buffers will get messed + * up. */ void ExecHashJoinSaveTuple(MinimalTuple tuple, uint32 hashvalue, @@ -1246,6 +1252,8 @@ ExecHashJoinSaveTuple(MinimalTuple tuple, uint32 hashvalue, if (file == NULL) { + Assert(strcmp(CurrentMemoryContext->name, "HashBatchFiles") == 0); + /* First write to this batch file, so open it. */ file = BufFileCreateTemp(false); *fileptr = file; diff --git a/src/include/executor/hashjoin.h b/src/include/executor/hashjoin.h index acb7592ca0..d36beb7229 100644 --- a/src/include/executor/hashjoin.h +++ b/src/include/executor/hashjoin.h @@ -348,6 +348,7 @@ typedef struct HashJoinTableData MemoryContext hashCxt; /* context for whole-hash-join storage */ MemoryContext batchCxt; /* context for this-batch-only storage */ + MemoryContext fileCxt; /* context for the BufFile related storage */ /* used for dense allocation of tuples (into linked chunks) */ HashMemoryChunk chunks; /* one list for the whole batch */ -- 2.39.2 --MP_/GVa.HActNIsT4j_JhNzVDcp-- ^ permalink raw reply [nested|flat] 9+ messages in thread
* [PATCH] Allocate hash batches related BufFile in a dedicated context @ 2023-03-27 13:54 Jehan-Guillaume de Rorthais <[email protected]> 0 siblings, 0 replies; 9+ messages in thread From: Jehan-Guillaume de Rorthais @ 2023-03-27 13:54 UTC (permalink / raw) --- src/backend/executor/nodeHash.c | 43 +++++++++++++++++++++++++---- src/backend/executor/nodeHashjoin.c | 18 ++++++++---- src/include/executor/hashjoin.h | 15 ++++++++-- src/include/executor/nodeHashjoin.h | 2 +- 4 files changed, 64 insertions(+), 14 deletions(-) diff --git a/src/backend/executor/nodeHash.c b/src/backend/executor/nodeHash.c index 748c9b0024..3da83ac22a 100644 --- a/src/backend/executor/nodeHash.c +++ b/src/backend/executor/nodeHash.c @@ -484,7 +484,7 @@ ExecHashTableCreate(HashState *state, List *hashOperators, List *hashCollations, * * The hashtable control block is just palloc'd from the executor's * per-query memory context. Everything else should be kept inside the - * subsidiary hashCxt or batchCxt. + * subsidiary hashCxt, batchCxt or fileCxt. */ hashtable = palloc_object(HashJoinTableData); hashtable->nbuckets = nbuckets; @@ -538,6 +538,10 @@ ExecHashTableCreate(HashState *state, List *hashOperators, List *hashCollations, "HashBatchContext", ALLOCSET_DEFAULT_SIZES); + hashtable->fileCxt = AllocSetContextCreate(CurrentMemoryContext, + "HashBatchFiles", + ALLOCSET_DEFAULT_SIZES); + /* Allocate data that will live for the life of the hashjoin */ oldcxt = MemoryContextSwitchTo(hashtable->hashCxt); @@ -570,15 +574,21 @@ ExecHashTableCreate(HashState *state, List *hashOperators, List *hashCollations, if (nbatch > 1 && hashtable->parallel_state == NULL) { + MemoryContext oldctx; + /* * allocate and initialize the file arrays in hashCxt (not needed for * parallel case which uses shared tuplestores instead of raw files) */ + oldctx = MemoryContextSwitchTo(hashtable->fileCxt); + hashtable->innerBatchFile = palloc0_array(BufFile *, nbatch); hashtable->outerBatchFile = palloc0_array(BufFile *, nbatch); /* The files will not be opened until needed... */ /* ... but make sure we have temp tablespaces established for them */ PrepareTempTablespaces(); + + MemoryContextSwitchTo(oldctx); } MemoryContextSwitchTo(oldcxt); @@ -929,12 +939,18 @@ ExecHashIncreaseNumBatches(HashJoinTable hashtable) nbatch = oldnbatch * 2; Assert(nbatch > 1); + elog(WARNING, "increasing number of batches from %d to %d", oldnbatch, nbatch); + + elog(LOG, "ExecHashIncreaseNumBatches ======= context stats start ======="); + MemoryContextStats(TopMemoryContext); + + #ifdef HJDEBUG printf("Hashjoin %p: increasing nbatch to %d because space = %zu\n", hashtable, nbatch, hashtable->spaceUsed); #endif - oldcxt = MemoryContextSwitchTo(hashtable->hashCxt); + oldcxt = MemoryContextSwitchTo(hashtable->fileCxt); if (hashtable->innerBatchFile == NULL) { @@ -1022,9 +1038,11 @@ ExecHashIncreaseNumBatches(HashJoinTable hashtable) { /* dump it out */ Assert(batchno > curbatch); + ExecHashJoinSaveTuple(HJTUPLE_MINTUPLE(hashTuple), hashTuple->hashvalue, - &hashtable->innerBatchFile[batchno]); + &hashtable->innerBatchFile[batchno], + hashtable->fileCxt); hashtable->spaceUsed -= hashTupleSize; nfreed++; @@ -1042,6 +1060,13 @@ ExecHashIncreaseNumBatches(HashJoinTable hashtable) oldchunks = nextchunk; } + if (hashtable->fileCxt->mem_allocated > hashtable->spaceAllowed) + elog(NOTICE, + "Growing number of hash batch to %d is exhausting allowed memory (%ld > %ld)", + nbatch, + hashtable->fileCxt->mem_allocated, + hashtable->spaceAllowed); + #ifdef HJDEBUG printf("Hashjoin %p: freed %ld of %ld tuples, space now %zu\n", hashtable, nfreed, ninmemory, hashtable->spaceUsed); @@ -1063,6 +1088,9 @@ ExecHashIncreaseNumBatches(HashJoinTable hashtable) hashtable); #endif } + + elog(LOG, "ExecHashIncreaseNumBatches ======= context stats end ======="); + MemoryContextStats(TopMemoryContext); } /* @@ -1681,9 +1709,11 @@ ExecHashTableInsert(HashJoinTable hashtable, * put the tuple into a temp file for later batches */ Assert(batchno > hashtable->curbatch); + ExecHashJoinSaveTuple(tuple, hashvalue, - &hashtable->innerBatchFile[batchno]); + &hashtable->innerBatchFile[batchno], + hashtable->fileCxt); } if (shouldFree) @@ -2534,8 +2564,11 @@ ExecHashRemoveNextSkewBucket(HashJoinTable hashtable) { /* Put the tuple into a temp file for later batches */ Assert(batchno > hashtable->curbatch); + ExecHashJoinSaveTuple(tuple, hashvalue, - &hashtable->innerBatchFile[batchno]); + &hashtable->innerBatchFile[batchno], + hashtable->fileCxt); + pfree(hashTuple); hashtable->spaceUsed -= tupleSize; hashtable->spaceUsedSkew -= tupleSize; diff --git a/src/backend/executor/nodeHashjoin.c b/src/backend/executor/nodeHashjoin.c index f189fb4d28..6055abde49 100644 --- a/src/backend/executor/nodeHashjoin.c +++ b/src/backend/executor/nodeHashjoin.c @@ -432,8 +432,10 @@ ExecHashJoinImpl(PlanState *pstate, bool parallel) */ Assert(parallel_state == NULL); Assert(batchno > hashtable->curbatch); + ExecHashJoinSaveTuple(mintuple, hashvalue, - &hashtable->outerBatchFile[batchno]); + &hashtable->outerBatchFile[batchno], + hashtable->fileCxt); if (shouldFree) heap_free_minimal_tuple(mintuple); @@ -1234,21 +1236,27 @@ ExecParallelHashJoinNewBatch(HashJoinState *hjstate) * The data recorded in the file for each tuple is its hash value, * then the tuple in MinimalTuple format. * - * Note: it is important always to call this in the regular executor - * context, not in a shorter-lived context; else the temp file buffers - * will get messed up. + * Note: it is important always to call this in the HashBatchFiles context, + * not in a shorter-lived context; else the temp file buffers will get messed + * up. */ void ExecHashJoinSaveTuple(MinimalTuple tuple, uint32 hashvalue, - BufFile **fileptr) + BufFile **fileptr, MemoryContext filecxt) { BufFile *file = *fileptr; if (file == NULL) { + MemoryContext oldctx; + + oldctx = MemoryContextSwitchTo(filecxt); + /* First write to this batch file, so open it. */ file = BufFileCreateTemp(false); *fileptr = file; + + MemoryContextSwitchTo(oldctx); } BufFileWrite(file, &hashvalue, sizeof(uint32)); diff --git a/src/include/executor/hashjoin.h b/src/include/executor/hashjoin.h index acb7592ca0..d759235d7f 100644 --- a/src/include/executor/hashjoin.h +++ b/src/include/executor/hashjoin.h @@ -25,10 +25,14 @@ * * Each active hashjoin has a HashJoinTable control block, which is * palloc'd in the executor's per-query context. All other storage needed - * for the hashjoin is kept in private memory contexts, two for each hashjoin. + * for the hashjoin is kept in private memory contexts, three for each + * hashjoin: + * - HashTableContext (hashCxt): the control block associated to the hash table + * - HashBatchContext (batchCxt): storages for batches + * - HashBatchFiles (fileCxt): storage for temp files buffers + * * This makes it easy and fast to release the storage when we don't need it - * anymore. (Exception: data associated with the temp files lives in the - * per-query context too, since we always call buffile.c in that context.) + * anymore. * * The hashtable contexts are made children of the per-query context, ensuring * that they will be discarded at end of statement even if the join is @@ -39,6 +43,10 @@ * "hashCxt", while storage that is only wanted for the current batch is * allocated in the "batchCxt". By resetting the batchCxt at the end of * each batch, we free all the per-batch storage reliably and without tedium. + * Note that data associated with the temp files lives in the "fileCxt" context + * which lives during the entire join as temp files might need to survives + * batches. These files are explicitly destroyed by calling BufFileClose() + * when the code is done with them. * * During first scan of inner relation, we get its tuples from executor. * If nbatch > 1 then tuples that don't belong in first batch get saved @@ -348,6 +356,7 @@ typedef struct HashJoinTableData MemoryContext hashCxt; /* context for whole-hash-join storage */ MemoryContext batchCxt; /* context for this-batch-only storage */ + MemoryContext fileCxt; /* context for the BufFile related storage */ /* used for dense allocation of tuples (into linked chunks) */ HashMemoryChunk chunks; /* one list for the whole batch */ diff --git a/src/include/executor/nodeHashjoin.h b/src/include/executor/nodeHashjoin.h index d367070883..a8f9ae1989 100644 --- a/src/include/executor/nodeHashjoin.h +++ b/src/include/executor/nodeHashjoin.h @@ -29,6 +29,6 @@ extern void ExecHashJoinInitializeWorker(HashJoinState *state, ParallelWorkerContext *pwcxt); extern void ExecHashJoinSaveTuple(MinimalTuple tuple, uint32 hashvalue, - BufFile **fileptr); + BufFile **fileptr, MemoryContext filecxt); #endif /* NODEHASHJOIN_H */ -- 2.39.2 --MP_/cg+MWpd=CPm0_9SBk5uZwGx-- ^ permalink raw reply [nested|flat] 9+ messages in thread
* [PATCH v8 2/3] Allocate hash batches related BufFile in a dedicated context @ 2023-05-16 13:42 Jehan-Guillaume de Rorthais <[email protected]> 0 siblings, 0 replies; 9+ messages in thread From: Jehan-Guillaume de Rorthais @ 2023-05-16 13:42 UTC (permalink / raw) --- src/backend/executor/nodeHash.c | 43 ++++++++++++++++------- src/backend/executor/nodeHashjoin.c | 22 ++++++++---- src/backend/utils/sort/sharedtuplestore.c | 8 +++++ src/include/executor/hashjoin.h | 29 ++++++++++----- src/include/executor/nodeHashjoin.h | 2 +- 5 files changed, 75 insertions(+), 29 deletions(-) diff --git a/src/backend/executor/nodeHash.c b/src/backend/executor/nodeHash.c index 5fd1c5553b..e529f50a3e 100644 --- a/src/backend/executor/nodeHash.c +++ b/src/backend/executor/nodeHash.c @@ -484,7 +484,7 @@ ExecHashTableCreate(HashState *state, List *hashOperators, List *hashCollations, * * The hashtable control block is just palloc'd from the executor's * per-query memory context. Everything else should be kept inside the - * subsidiary hashCxt or batchCxt. + * subsidiary hashCxt, batchCxt or spillCxt. */ hashtable = palloc_object(HashJoinTableData); hashtable->nbuckets = nbuckets; @@ -538,6 +538,10 @@ ExecHashTableCreate(HashState *state, List *hashOperators, List *hashCollations, "HashBatchContext", ALLOCSET_DEFAULT_SIZES); + hashtable->spillCxt = AllocSetContextCreate(hashtable->hashCxt, + "HashSpillContext", + ALLOCSET_DEFAULT_SIZES); + /* Allocate data that will live for the life of the hashjoin */ oldcxt = MemoryContextSwitchTo(hashtable->hashCxt); @@ -570,12 +574,19 @@ ExecHashTableCreate(HashState *state, List *hashOperators, List *hashCollations, if (nbatch > 1 && hashtable->parallel_state == NULL) { + MemoryContext oldctx; + /* * allocate and initialize the file arrays in hashCxt (not needed for * parallel case which uses shared tuplestores instead of raw files) */ + oldctx = MemoryContextSwitchTo(hashtable->spillCxt); + hashtable->innerBatchFile = palloc0_array(BufFile *, nbatch); hashtable->outerBatchFile = palloc0_array(BufFile *, nbatch); + + MemoryContextSwitchTo(oldctx); + /* The files will not be opened until needed... */ /* ... but make sure we have temp tablespaces established for them */ PrepareTempTablespaces(); @@ -913,7 +924,6 @@ ExecHashIncreaseNumBatches(HashJoinTable hashtable) int oldnbatch = hashtable->nbatch; int curbatch = hashtable->curbatch; int nbatch; - MemoryContext oldcxt; long ninmemory; long nfreed; HashMemoryChunk oldchunks; @@ -934,13 +944,16 @@ ExecHashIncreaseNumBatches(HashJoinTable hashtable) hashtable, nbatch, hashtable->spaceUsed); #endif - oldcxt = MemoryContextSwitchTo(hashtable->hashCxt); - if (hashtable->innerBatchFile == NULL) { + MemoryContext oldcxt = MemoryContextSwitchTo(hashtable->spillCxt); + /* we had no file arrays before */ hashtable->innerBatchFile = palloc0_array(BufFile *, nbatch); hashtable->outerBatchFile = palloc0_array(BufFile *, nbatch); + + MemoryContextSwitchTo(oldcxt); + /* time to establish the temp tablespaces, too */ PrepareTempTablespaces(); } @@ -951,8 +964,6 @@ ExecHashIncreaseNumBatches(HashJoinTable hashtable) hashtable->outerBatchFile = repalloc0_array(hashtable->outerBatchFile, BufFile *, oldnbatch, nbatch); } - MemoryContextSwitchTo(oldcxt); - hashtable->nbatch = nbatch; /* @@ -1024,7 +1035,8 @@ ExecHashIncreaseNumBatches(HashJoinTable hashtable) Assert(batchno > curbatch); ExecHashJoinSaveTuple(HJTUPLE_MINTUPLE(hashTuple), hashTuple->hashvalue, - &hashtable->innerBatchFile[batchno]); + &hashtable->innerBatchFile[batchno], + hashtable->spillCxt); hashtable->spaceUsed -= hashTupleSize; nfreed++; @@ -1683,7 +1695,8 @@ ExecHashTableInsert(HashJoinTable hashtable, Assert(batchno > hashtable->curbatch); ExecHashJoinSaveTuple(tuple, hashvalue, - &hashtable->innerBatchFile[batchno]); + &hashtable->innerBatchFile[batchno], + hashtable->spillCxt); } if (shouldFree) @@ -2664,7 +2677,8 @@ ExecHashRemoveNextSkewBucket(HashJoinTable hashtable) /* Put the tuple into a temp file for later batches */ Assert(batchno > hashtable->curbatch); ExecHashJoinSaveTuple(tuple, hashvalue, - &hashtable->innerBatchFile[batchno]); + &hashtable->innerBatchFile[batchno], + hashtable->spillCxt); pfree(hashTuple); hashtable->spaceUsed -= tupleSize; hashtable->spaceUsedSkew -= tupleSize; @@ -3093,8 +3107,11 @@ ExecParallelHashJoinSetUpBatches(HashJoinTable hashtable, int nbatch) pstate->nbatch = nbatch; batches = dsa_get_address(hashtable->area, pstate->batches); - /* Use hash join memory context. */ - oldcxt = MemoryContextSwitchTo(hashtable->hashCxt); + /* + * Use hash join spill memory context to allocate accessors and their + * buffers + */ + oldcxt = MemoryContextSwitchTo(hashtable->spillCxt); /* Allocate this backend's accessor array. */ hashtable->nbatch = nbatch; @@ -3196,8 +3213,8 @@ ExecParallelHashEnsureBatchAccessors(HashJoinTable hashtable) */ Assert(DsaPointerIsValid(pstate->batches)); - /* Use hash join memory context. */ - oldcxt = MemoryContextSwitchTo(hashtable->hashCxt); + /* Use hash join spill memory context to allocate accessors */ + oldcxt = MemoryContextSwitchTo(hashtable->spillCxt); /* Allocate this backend's accessor array. */ hashtable->nbatch = pstate->nbatch; diff --git a/src/backend/executor/nodeHashjoin.c b/src/backend/executor/nodeHashjoin.c index 93a78f6f74..e9e58e9647 100644 --- a/src/backend/executor/nodeHashjoin.c +++ b/src/backend/executor/nodeHashjoin.c @@ -491,7 +491,8 @@ ExecHashJoinImpl(PlanState *pstate, bool parallel) Assert(parallel_state == NULL); Assert(batchno > hashtable->curbatch); ExecHashJoinSaveTuple(mintuple, hashvalue, - &hashtable->outerBatchFile[batchno]); + &hashtable->outerBatchFile[batchno], + hashtable->spillCxt); if (shouldFree) heap_free_minimal_tuple(mintuple); @@ -1313,21 +1314,30 @@ ExecParallelHashJoinNewBatch(HashJoinState *hjstate) * The data recorded in the file for each tuple is its hash value, * then the tuple in MinimalTuple format. * - * Note: it is important always to call this in the regular executor - * context, not in a shorter-lived context; else the temp file buffers - * will get messed up. + * If this is the first write to the batch file, this function first + * create it. The associated BufFile buffer is allocated in the given + * context. It is important to always give the HashSpillContext + * context. First to avoid a shorter-lived context, else the temp file + * buffers will get messed up. Second, for a better accounting of the + * spilling memory consumption. + * */ void ExecHashJoinSaveTuple(MinimalTuple tuple, uint32 hashvalue, - BufFile **fileptr) + BufFile **fileptr, MemoryContext cxt) { BufFile *file = *fileptr; if (file == NULL) { - /* First write to this batch file, so open it. */ + MemoryContext oldctx; + + oldctx = MemoryContextSwitchTo(cxt); + file = BufFileCreateTemp(false); *fileptr = file; + + MemoryContextSwitchTo(oldctx); } BufFileWrite(file, &hashvalue, sizeof(uint32)); diff --git a/src/backend/utils/sort/sharedtuplestore.c b/src/backend/utils/sort/sharedtuplestore.c index 0831249159..236be65f22 100644 --- a/src/backend/utils/sort/sharedtuplestore.c +++ b/src/backend/utils/sort/sharedtuplestore.c @@ -308,11 +308,15 @@ sts_puttuple(SharedTuplestoreAccessor *accessor, void *meta_data, { SharedTuplestoreParticipant *participant; char name[MAXPGPATH]; + MemoryContext oldcxt; /* Create one. Only this backend will write into it. */ sts_filename(name, accessor, accessor->participant); + + oldcxt = MemoryContextSwitchTo(accessor->context); accessor->write_file = BufFileCreateFileSet(&accessor->fileset->fs, name); + MemoryContextSwitchTo(oldcxt); /* Set up the shared state for this backend's file. */ participant = &accessor->sts->participants[accessor->participant]; @@ -527,11 +531,15 @@ sts_parallel_scan_next(SharedTuplestoreAccessor *accessor, void *meta_data) if (accessor->read_file == NULL) { char name[MAXPGPATH]; + MemoryContext oldcxt; sts_filename(name, accessor, accessor->read_participant); + + oldcxt = MemoryContextSwitchTo(accessor->context); accessor->read_file = BufFileOpenFileSet(&accessor->fileset->fs, name, O_RDONLY, false); + MemoryContextSwitchTo(oldcxt); } /* Seek and load the chunk header. */ diff --git a/src/include/executor/hashjoin.h b/src/include/executor/hashjoin.h index 8ee59d2c71..ac27222d18 100644 --- a/src/include/executor/hashjoin.h +++ b/src/include/executor/hashjoin.h @@ -23,12 +23,12 @@ /* ---------------------------------------------------------------- * hash-join hash table structures * - * Each active hashjoin has a HashJoinTable control block, which is - * palloc'd in the executor's per-query context. All other storage needed - * for the hashjoin is kept in private memory contexts, two for each hashjoin. - * This makes it easy and fast to release the storage when we don't need it - * anymore. (Exception: data associated with the temp files lives in the - * per-query context too, since we always call buffile.c in that context.) + * Each active hashjoin has a HashJoinTable structure, which is + * palloc'd in the executor's per-query context. Other storages needed for + * each hashjoin is kept in child contexts, three for each hashjoin: + * - HashTableContext (hashCxt): the parent hash table storage context + * - HashSpillContext (spillCxt): storage for temp files buffers + * - HashBatchContext (batchCxt): storage for a batch in serial hash join * * The hashtable contexts are made children of the per-query context, ensuring * that they will be discarded at end of statement even if the join is @@ -36,9 +36,19 @@ * be cleaned up by the virtual file manager in event of an error.) * * Storage that should live through the entire join is allocated from the - * "hashCxt", while storage that is only wanted for the current batch is - * allocated in the "batchCxt". By resetting the batchCxt at the end of - * each batch, we free all the per-batch storage reliably and without tedium. + * "hashCxt" (mainly hash's meta datas). Also, the "hashCxt" context is the + * parent of "spillCxt" and "batchCxt". It makes it easy and fast to release + * the storage when we don't need it anymore. + * + * Data associated with temp files lives in the "spillCxt" context which lives + * during the entire join as temp files might need to survives batches. These + * files are explicitly destroyed by calling BufFileClose() when the code is + * done with them. The aim of this context is to help accounting the memory + * allocations dedicated to temp files and their buffers. + * + * Finaly, storage that is only wanted for the current batch is allocated in + * the "batchCxt". By resetting the batchCxt at the end of each batch, we free + * all the per-batch storage reliably and without tedium. * * During first scan of inner relation, we get its tuples from executor. * If nbatch > 1 then tuples that don't belong in first batch get saved @@ -350,6 +360,7 @@ typedef struct HashJoinTableData MemoryContext hashCxt; /* context for whole-hash-join storage */ MemoryContext batchCxt; /* context for this-batch-only storage */ + MemoryContext spillCxt; /* context for spilling to temp files */ /* used for dense allocation of tuples (into linked chunks) */ HashMemoryChunk chunks; /* one list for the whole batch */ diff --git a/src/include/executor/nodeHashjoin.h b/src/include/executor/nodeHashjoin.h index d367070883..fb33994366 100644 --- a/src/include/executor/nodeHashjoin.h +++ b/src/include/executor/nodeHashjoin.h @@ -29,6 +29,6 @@ extern void ExecHashJoinInitializeWorker(HashJoinState *state, ParallelWorkerContext *pwcxt); extern void ExecHashJoinSaveTuple(MinimalTuple tuple, uint32 hashvalue, - BufFile **fileptr); + BufFile **fileptr, MemoryContext cxt); #endif /* NODEHASHJOIN_H */ -- 2.40.1 --MP_/KUKoyOpWagGRmSskvJu5/tn-- ^ permalink raw reply [nested|flat] 9+ messages in thread
end of thread, other threads:[~2023-05-16 13:42 UTC | newest] Thread overview: 9+ messages (download: mbox mbox.gz follow: Atom feed) -- links below jump to the message on this page -- 2021-01-24 03:03 [PATCH 16/18] doc review for logical decoding of prepared xacts Justin Pryzby <[email protected]> 2023-03-27 13:54 [PATCH 2/3] Allocate hash batches related BufFile in a dedicated context Jehan-Guillaume de Rorthais <[email protected]> 2023-03-27 13:54 [PATCH 2/4] Allocate hash batches related BufFile in a dedicated context Jehan-Guillaume de Rorthais <[email protected]> 2023-03-27 13:54 [PATCH] Allocate hash batches related BufFile in a dedicated context Jehan-Guillaume de Rorthais <[email protected]> 2023-03-27 13:54 [PATCH] Allocate hash batches related BufFile in a dedicated context Jehan-Guillaume de Rorthais <[email protected]> 2023-03-27 13:54 [PATCH 2/3] Allocate hash batches related BufFile in a dedicated context Jehan-Guillaume de Rorthais <[email protected]> 2023-03-27 13:54 [PATCH 2/3] Allocate hash batches related BufFile in a dedicated context Jehan-Guillaume de Rorthais <[email protected]> 2023-03-27 13:54 [PATCH] Allocate hash batches related BufFile in a dedicated context Jehan-Guillaume de Rorthais <[email protected]> 2023-05-16 13:42 [PATCH v8 2/3] Allocate hash batches related BufFile in a dedicated context Jehan-Guillaume de Rorthais <[email protected]>
This inbox is served by agora; see mirroring instructions for how to clone and mirror all data and code used for this inbox