database/postgres
1
0
Fork 0
mirror of https://github.com/postgres/postgres.git synced 2025-06-27 23:21:58 +03:00
Code Activity

Fix handling of pending inserts in nodeModifyTable.c.

Browse Source 
Commit b663a4136, which allowed FDWs to INSERT rows in bulk, added to
nodeModifyTable.c code to flush pending inserts to the foreign-table
result relation(s) before completing processing of the ModifyTable node,
but the code failed to take into account the case where the INSERT query
has modifying CTEs, leading to incorrect results.

Also, that commit failed to flush pending inserts before firing BEFORE
ROW triggers so that rows are visible to such triggers.

In that commit we scanned through EState's
es_tuple_routing_result_relations or es_opened_result_relations list to
find the foreign-table result relations to which pending inserts are
flushed, but that would be inefficient in some cases.  So to fix, 1) add
a List member to EState to record the insert-pending result relations,
and 2) modify nodeModifyTable.c so that it adds the foreign-table result
relation to the list in ExecInsert() if appropriate, and flushes pending
inserts properly using the list where needed.

While here, fix a copy-and-pasteo in a comment in ExecBatchInsert(),
which was added by that commit.

Back-patch to v14 where that commit appeared.

Discussion: https://postgr.es/m/CAPmGK16qutyCmyJJzgQOhfBq%3DNoGDqTB6O0QBZTihrbqre%2BoxA%40mail.gmail.com
This commit is contained in:
Etsuro Fujita
2022-11-25 17:45:00 +09:00
parent 9e492d6b69
commit ffbb7e65a8
7 changed files with 302 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

96
src/backend/executor/nodeModifyTable.c
View File

@ -142,6 +142,7 @@ static void ExecBatchInsert(ModifyTableState *mtstate,
int numSlots,
EState *estate,
bool canSetTag);
static void ExecPendingInserts(EState *estate);
static void ExecCrossPartitionUpdateForeignKey(ModifyTableContext *context,
ResultRelInfo *sourcePartInfo,
ResultRelInfo *destPartInfo,
@ -761,6 +762,10 @@ ExecInsert(ModifyTableContext *context,
if (resultRelInfo->ri_TrigDesc &&
resultRelInfo->ri_TrigDesc->trig_insert_before_row)
{
/* Flush any pending inserts, so rows are visible to the triggers */
if (estate->es_insert_pending_result_relations != NIL)
ExecPendingInserts(estate);
if (!ExecBRInsertTriggers(estate, resultRelInfo, slot))
return NULL; /* "do nothing" */
}
@ -794,6 +799,8 @@ ExecInsert(ModifyTableContext *context,
*/
if (resultRelInfo->ri_BatchSize > 1)
{
bool flushed = false;
/*
* When we've reached the desired batch size, perform the
* insertion.
@ -806,6 +813,7 @@ ExecInsert(ModifyTableContext *context,
resultRelInfo->ri_NumSlots,
estate, canSetTag);
resultRelInfo->ri_NumSlots = 0;
flushed = true;
}
oldContext = MemoryContextSwitchTo(estate->es_query_cxt);
@ -848,6 +856,24 @@ ExecInsert(ModifyTableContext *context,
ExecCopySlot(resultRelInfo->ri_PlanSlots[resultRelInfo->ri_NumSlots],
planSlot);
/*
* If these are the first tuples stored in the buffers, add the
* target rel to the es_insert_pending_result_relations list,
* except in the case where flushing was done above, in which case
* the target rel would already have been added to the list, so no
* need to do this.
*/
if (resultRelInfo->ri_NumSlots == 0 && !flushed)
{
Assert(!list_member_ptr(estate->es_insert_pending_result_relations,
resultRelInfo));
estate->es_insert_pending_result_relations =
lappend(estate->es_insert_pending_result_relations,
resultRelInfo);
}
Assert(list_member_ptr(estate->es_insert_pending_result_relations,
resultRelInfo));
resultRelInfo->ri_NumSlots++;
MemoryContextSwitchTo(oldContext);
@ -1166,9 +1192,8 @@ ExecBatchInsert(ModifyTableState *mtstate,
slot = rslots[i];
/*
* AFTER ROW Triggers or RETURNING expressions might reference the
* tableoid column, so (re-)initialize tts_tableOid before evaluating
* them.
* AFTER ROW Triggers might reference the tableoid column, so
* (re-)initialize tts_tableOid before evaluating them.
*/
slot->tts_tableOid = RelationGetRelid(resultRelInfo->ri_RelationDesc);
@ -1188,6 +1213,32 @@ ExecBatchInsert(ModifyTableState *mtstate,
estate->es_processed += numInserted;
}
/*
* ExecPendingInserts -- flushes all pending inserts to the foreign tables
*/
static void
ExecPendingInserts(EState *estate)
{
ListCell *lc;
foreach(lc, estate->es_insert_pending_result_relations)
{
ResultRelInfo *resultRelInfo = (ResultRelInfo *) lfirst(lc);
ModifyTableState *mtstate = resultRelInfo->ri_ModifyTableState;
Assert(mtstate);
ExecBatchInsert(mtstate, resultRelInfo,
resultRelInfo->ri_Slots,
resultRelInfo->ri_PlanSlots,
resultRelInfo->ri_NumSlots,
estate, mtstate->canSetTag);
resultRelInfo->ri_NumSlots = 0;
}
list_free(estate->es_insert_pending_result_relations);
estate->es_insert_pending_result_relations = NIL;
}
/*
* ExecDeletePrologue -- subroutine for ExecDelete
*
@ -1203,9 +1254,15 @@ ExecDeletePrologue(ModifyTableContext *context, ResultRelInfo *resultRelInfo,
/* BEFORE ROW DELETE triggers */
if (resultRelInfo->ri_TrigDesc &&
resultRelInfo->ri_TrigDesc->trig_delete_before_row)
{
/* Flush any pending inserts, so rows are visible to the triggers */
if (context->estate->es_insert_pending_result_relations != NIL)
ExecPendingInserts(context->estate);
return ExecBRDeleteTriggers(context->estate, context->epqstate,
resultRelInfo, tupleid, oldtuple,
epqreturnslot);
}
return true;
}
@ -1780,9 +1837,15 @@ ExecUpdatePrologue(ModifyTableContext *context, ResultRelInfo *resultRelInfo,
/* BEFORE ROW UPDATE triggers */
if (resultRelInfo->ri_TrigDesc &&
resultRelInfo->ri_TrigDesc->trig_update_before_row)
{
/* Flush any pending inserts, so rows are visible to the triggers */
if (context->estate->es_insert_pending_result_relations != NIL)
ExecPendingInserts(context->estate);
return ExecBRUpdateTriggers(context->estate, context->epqstate,
resultRelInfo, tupleid, oldtuple, slot,
&context->tmfd);
}
return true;
}
@ -3452,9 +3515,6 @@ ExecModifyTable(PlanState *pstate)
HeapTupleData oldtupdata;
HeapTuple oldtuple;
ItemPointer tupleid;
PartitionTupleRouting *proute = node->mt_partition_tuple_routing;
List *relinfos = NIL;
ListCell *lc;
CHECK_FOR_INTERRUPTS();
@ -3756,21 +3816,8 @@ ExecModifyTable(PlanState *pstate)
/*
* Insert remaining tuples for batch insert.
*/
if (proute)
relinfos = estate->es_tuple_routing_result_relations;
else
relinfos = estate->es_opened_result_relations;
foreach(lc, relinfos)
{
resultRelInfo = lfirst(lc);
if (resultRelInfo->ri_NumSlots > 0)
ExecBatchInsert(node, resultRelInfo,
resultRelInfo->ri_Slots,
resultRelInfo->ri_PlanSlots,
resultRelInfo->ri_NumSlots,
estate, node->canSetTag);
}
if (estate->es_insert_pending_result_relations != NIL)
ExecPendingInserts(estate);
/*
* We're done, but fire AFTER STATEMENT triggers before exiting.
@ -4295,6 +4342,13 @@ ExecInitModifyTable(ModifyTable *node, EState *estate, int eflags)
}
else
resultRelInfo->ri_BatchSize = 1;
/*
* If doing batch insert, setup back-link so we can easily find the
* mtstate again.
*/
if (resultRelInfo->ri_BatchSize > 1)
resultRelInfo->ri_ModifyTableState = mtstate;
}
/*