database/postgres
database/postgres
1
0
Fork 0
mirror of https://github.com/postgres/postgres.git synced 2025-05-03 22:24:49 +03:00
Code

Fix per-tuple memory leak in partition tuple routing

Browse Source 
Some operations were being done in a longer-lived memory context,
causing intra-query leaks.  It's not noticeable unless you're doing a
large COPY, but if you are, it eats enough memory to cause a problem.

Co-authored-by: Kohei KaiGai <kaigai@heterodb.com>
Co-authored-by: Amit Langote <Langote_Amit_f8@lab.ntt.co.jp>
Co-authored-by: Álvaro Herrera <alvherre@alvh.no-ip.org>
Discussion: https://postgr.es/m/CAOP8fzYtVFWZADq4c=KoTAqgDrHWfng+AnEPEZccyxqxPVbbWQ@mail.gmail.com
This commit is contained in:
Alvaro Herrera 2018-08-01 15:06:47 -04:00
parent e9bbfe6088
commit 34295b87fb
1 changed files with 38 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

52
src/backend/executor/execPartition.c
View File

@ -193,9 +193,15 @@ ExecFindPartition(ResultRelInfo *resultRelInfo, PartitionDispatch *pd,
Datum values[PARTITION_MAX_KEYS]; Datum values[PARTITION_MAX_KEYS];
bool isnull[PARTITION_MAX_KEYS]; bool isnull[PARTITION_MAX_KEYS];
Relation rel; Relation rel;
PartitionDispatch parent; PartitionDispatch dispatch;
ExprContext *ecxt = GetPerTupleExprContext(estate); ExprContext *ecxt = GetPerTupleExprContext(estate);
TupleTableSlot *ecxt_scantuple_old = ecxt->ecxt_scantuple; TupleTableSlot *ecxt_scantuple_old = ecxt->ecxt_scantuple;
TupleTableSlot *myslot = NULL;
MemoryContext oldcxt;
HeapTuple tuple;
/* use per-tuple context here to avoid leaking memory */
oldcxt = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
/* /*
* First check the root table's partition constraint, if any. No point in * First check the root table's partition constraint, if any. No point in
@ -205,26 +211,24 @@ ExecFindPartition(ResultRelInfo *resultRelInfo, PartitionDispatch *pd,
ExecPartitionCheck(resultRelInfo, slot, estate, true); ExecPartitionCheck(resultRelInfo, slot, estate, true);
/* start with the root partitioned table */ /* start with the root partitioned table */
parent = pd[0]; tuple = ExecFetchSlotTuple(slot);
dispatch = pd[0];
while (true) while (true)
{ {
PartitionDesc partdesc; PartitionDesc partdesc;
TupleTableSlot *myslot = parent->tupslot; TupleConversionMap *map = dispatch->tupmap;
TupleConversionMap *map = parent->tupmap;
int cur_index = -1; int cur_index = -1;
rel = parent->reldesc; rel = dispatch->reldesc;
partdesc = RelationGetPartitionDesc(rel); partdesc = RelationGetPartitionDesc(rel);
/* /*
* Convert the tuple to this parent's layout so that we can do certain * Convert the tuple to this parent's layout, if different from the
* things we do below. * current relation.
*/ */
myslot = dispatch->tupslot;
if (myslot != NULL && map != NULL) if (myslot != NULL && map != NULL)
{ {
HeapTuple tuple = ExecFetchSlotTuple(slot);
ExecClearTuple(myslot);
tuple = do_convert_tuple(tuple, map); tuple = do_convert_tuple(tuple, map);
ExecStoreTuple(tuple, myslot, InvalidBuffer, true); ExecStoreTuple(tuple, myslot, InvalidBuffer, true);
slot = myslot; slot = myslot;
@ -239,7 +243,7 @@ ExecFindPartition(ResultRelInfo *resultRelInfo, PartitionDispatch *pd,
* So update ecxt_scantuple accordingly. * So update ecxt_scantuple accordingly.
*/ */
ecxt->ecxt_scantuple = slot; ecxt->ecxt_scantuple = slot;
FormPartitionKeyDatum(parent, slot, estate, values, isnull); FormPartitionKeyDatum(dispatch, slot, estate, values, isnull);
/* /*
* Nothing for get_partition_for_tuple() to do if there are no * Nothing for get_partition_for_tuple() to do if there are no
@ -263,14 +267,32 @@ ExecFindPartition(ResultRelInfo *resultRelInfo, PartitionDispatch *pd,
result = -1; result = -1;
break; break;
} }
else if (parent->indexes[cur_index] >= 0) else if (dispatch->indexes[cur_index] >= 0)
{ {
result = parent->indexes[cur_index]; result = dispatch->indexes[cur_index];
/* success! */
break; break;
} }
else else
parent = pd[-parent->indexes[cur_index]]; {
/* move down one level */
dispatch = pd[-dispatch->indexes[cur_index]];
/*
* Release the dedicated slot, if it was used. Create a copy of
* the tuple first, for the next iteration.
*/
if (slot == myslot)
{
tuple = ExecCopySlotTuple(myslot);
ExecClearTuple(myslot);
} }
}
}
/* Release the tuple in the lowest parent's dedicated slot. */
if (slot == myslot)
ExecClearTuple(myslot);
/* A partition was not found. */ /* A partition was not found. */
if (result < 0) if (result < 0)
@ -287,7 +309,9 @@ ExecFindPartition(ResultRelInfo *resultRelInfo, PartitionDispatch *pd,
val_desc ? errdetail("Partition key of the failing row contains %s.", val_desc) : 0)); val_desc ? errdetail("Partition key of the failing row contains %s.", val_desc) : 0));
} }
MemoryContextSwitchTo(oldcxt);
ecxt->ecxt_scantuple = ecxt_scantuple_old; ecxt->ecxt_scantuple = ecxt_scantuple_old;
return result; return result;
} }