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

Fix transition tables for partition/inheritance.

Browse Source 
We disallow row-level triggers with transition tables on child tables.
Transition tables for triggers on the parent table contain only those
columns present in the parent.  (We can't mix tuple formats in a
single transition table.)

Patch by Thomas Munro

Discussion: https://postgr.es/m/CA%2BTgmoZzTBBAsEUh4MazAN7ga%3D8SsMC-Knp-6cetts9yNZUCcg%40mail.gmail.com
This commit is contained in:
Andrew Gierth
2017-06-28 18:55:03 +01:00
parent 99255d73c0
commit 501ed02cf6
13 changed files with 1143 additions and 110 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
src/backend/catalog/pg_inherits.c
View File

@@ -273,6 +273,30 @@ has_subclass(Oid relationId)
return result;
}
/*
* has_superclass - does this relation inherit from another? The caller
* should hold a lock on the given relation so that it can't be concurrently
* added to or removed from an inheritance hierarchy.
*/
bool
has_superclass(Oid relationId)
{
Relation catalog;
SysScanDesc scan;
ScanKeyData skey;
bool result;
catalog = heap_open(InheritsRelationId, AccessShareLock);
ScanKeyInit(&skey, Anum_pg_inherits_inhrelid, BTEqualStrategyNumber,
F_OIDEQ, ObjectIdGetDatum(relationId));
scan = systable_beginscan(catalog, InheritsRelidSeqnoIndexId, true,
NULL, 1, &skey);
result = HeapTupleIsValid(systable_getnext(scan));
systable_endscan(scan);
heap_close(catalog, AccessShareLock);
return result;
}
/*
* Given two type OIDs, determine whether the first is a complex type

70
src/backend/commands/copy.c
View File

@@ -171,6 +171,8 @@ typedef struct CopyStateData
ResultRelInfo *partitions; /* Per partition result relation */
TupleConversionMap **partition_tupconv_maps;
TupleTableSlot *partition_tuple_slot;
TransitionCaptureState *transition_capture;
TupleConversionMap **transition_tupconv_maps;
/*
* These variables are used to reduce overhead in textual COPY FROM.
@@ -1436,6 +1438,36 @@ BeginCopy(ParseState *pstate,
cstate->num_partitions = num_partitions;
cstate->partition_tupconv_maps = partition_tupconv_maps;
cstate->partition_tuple_slot = partition_tuple_slot;
/*
* If there are any triggers with transition tables on the named
* relation, we need to be prepared to capture transition tuples
* from child relations too.
*/
cstate->transition_capture =
MakeTransitionCaptureState(rel->trigdesc);
/*
* If we are capturing transition tuples, they may need to be
* converted from partition format back to partitioned table
* format (this is only ever necessary if a BEFORE trigger
* modifies the tuple).
*/
if (cstate->transition_capture != NULL)
{
int i;
cstate->transition_tupconv_maps = (TupleConversionMap **)
palloc0(sizeof(TupleConversionMap *) *
cstate->num_partitions);
for (i = 0; i < cstate->num_partitions; ++i)
{
cstate->transition_tupconv_maps[i] =
convert_tuples_by_name(RelationGetDescr(cstate->partitions[i].ri_RelationDesc),
RelationGetDescr(rel),
gettext_noop("could not convert row type"));
}
}
}
}
else
@@ -2591,6 +2623,35 @@ CopyFrom(CopyState cstate)
*/
estate->es_result_relation_info = resultRelInfo;
/*
* If we're capturing transition tuples, we might need to convert
* from the partition rowtype to parent rowtype.
*/
if (cstate->transition_capture != NULL)
{
if (resultRelInfo->ri_TrigDesc &&
(resultRelInfo->ri_TrigDesc->trig_insert_before_row ||
resultRelInfo->ri_TrigDesc->trig_insert_instead_row))
{
/*
* If there are any BEFORE or INSTEAD triggers on the
* partition, we'll have to be ready to convert their
* result back to tuplestore format.
*/
cstate->transition_capture->tcs_original_insert_tuple = NULL;
cstate->transition_capture->tcs_map =
cstate->transition_tupconv_maps[leaf_part_index];
}
else
{
/*
* Otherwise, just remember the original unconverted
* tuple, to avoid a needless round trip conversion.
*/
cstate->transition_capture->tcs_original_insert_tuple = tuple;
cstate->transition_capture->tcs_map = NULL;
}
}
/*
* We might need to convert from the parent rowtype to the
* partition rowtype.
@@ -2703,7 +2764,7 @@ CopyFrom(CopyState cstate)
/* AFTER ROW INSERT Triggers */
ExecARInsertTriggers(estate, resultRelInfo, tuple,
recheckIndexes);
recheckIndexes, cstate->transition_capture);
list_free(recheckIndexes);
}
@@ -2856,7 +2917,7 @@ CopyFromInsertBatch(CopyState cstate, EState *estate, CommandId mycid,
estate, false, NULL, NIL);
ExecARInsertTriggers(estate, resultRelInfo,
bufferedTuples[i],
recheckIndexes);
recheckIndexes, NULL);
list_free(recheckIndexes);
}
}
@@ -2866,14 +2927,15 @@ CopyFromInsertBatch(CopyState cstate, EState *estate, CommandId mycid,
* anyway.
*/
else if (resultRelInfo->ri_TrigDesc != NULL &&
resultRelInfo->ri_TrigDesc->trig_insert_after_row)
(resultRelInfo->ri_TrigDesc->trig_insert_after_row ||
resultRelInfo->ri_TrigDesc->trig_insert_new_table))
{
for (i = 0; i < nBufferedTuples; i++)
{
cstate->cur_lineno = firstBufferedLineNo + i;
ExecARInsertTriggers(estate, resultRelInfo,
bufferedTuples[i],
NIL);
NIL, NULL);
}
}

28
src/backend/commands/tablecmds.c
View File

@@ -10933,6 +10933,7 @@ ATExecAddInherit(Relation child_rel, RangeVar *parent, LOCKMODE lockmode)
Relation parent_rel;
List *children;
ObjectAddress address;
const char *trigger_name;
/*
* A self-exclusive lock is needed here. See the similar case in
@@ -11014,6 +11015,19 @@ ATExecAddInherit(Relation child_rel, RangeVar *parent, LOCKMODE lockmode)
RelationGetRelationName(child_rel),
RelationGetRelationName(parent_rel))));
/*
* If child_rel has row-level triggers with transition tables, we
* currently don't allow it to become an inheritance child. See also
* prohibitions in ATExecAttachPartition() and CreateTrigger().
*/
trigger_name = FindTriggerIncompatibleWithInheritance(child_rel->trigdesc);
if (trigger_name != NULL)
ereport(ERROR,
(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
errmsg("trigger \"%s\" prevents table \"%s\" from becoming an inheritance child",
trigger_name, RelationGetRelationName(child_rel)),
errdetail("ROW triggers with transition tables are not supported in inheritance hierarchies")));
/* OK to create inheritance */
CreateInheritance(child_rel, parent_rel);
@@ -13418,6 +13432,7 @@ ATExecAttachPartition(List **wqueue, Relation rel, PartitionCmd *cmd)
TupleDesc tupleDesc;
bool skip_validate = false;
ObjectAddress address;
const char *trigger_name;
attachRel = heap_openrv(cmd->name, AccessExclusiveLock);
@@ -13547,6 +13562,19 @@ ATExecAttachPartition(List **wqueue, Relation rel, PartitionCmd *cmd)
errdetail("New partition should contain only the columns present in parent.")));
}
/*
* If child_rel has row-level triggers with transition tables, we
* currently don't allow it to become a partition. See also prohibitions
* in ATExecAddInherit() and CreateTrigger().
*/
trigger_name = FindTriggerIncompatibleWithInheritance(attachRel->trigdesc);
if (trigger_name != NULL)
ereport(ERROR,
(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
errmsg("trigger \"%s\" prevents table \"%s\" from becoming a partition",
trigger_name, RelationGetRelationName(attachRel)),
errdetail("ROW triggers with transition tables are not supported on partitions")));
/* OK to create inheritance. Rest of the checks performed there */
CreateInheritance(attachRel, rel);

218
src/backend/commands/trigger.c
View File

@@ -24,6 +24,7 @@
#include "catalog/objectaccess.h"
#include "catalog/pg_constraint.h"
#include "catalog/pg_constraint_fn.h"
#include "catalog/pg_inherits_fn.h"
#include "catalog/pg_proc.h"
#include "catalog/pg_trigger.h"
#include "catalog/pg_type.h"
@@ -96,7 +97,8 @@ static HeapTuple ExecCallTriggerFunc(TriggerData *trigdata,
static void AfterTriggerSaveEvent(EState *estate, ResultRelInfo *relinfo,
int event, bool row_trigger,
HeapTuple oldtup, HeapTuple newtup,
List *recheckIndexes, Bitmapset *modifiedCols);
List *recheckIndexes, Bitmapset *modifiedCols,
TransitionCaptureState *transition_capture);
static void AfterTriggerEnlargeQueryState(void);
@@ -354,13 +356,6 @@ CreateTrigger(CreateTrigStmt *stmt, const char *queryString,
* adjustments will be needed below.
*/
if (rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
ereport(ERROR,
(errcode(ERRCODE_WRONG_OBJECT_TYPE),
errmsg("\"%s\" is a partitioned table",
RelationGetRelationName(rel)),
errdetail("Triggers on partitioned tables cannot have transition tables.")));
if (rel->rd_rel->relkind == RELKIND_FOREIGN_TABLE)
ereport(ERROR,
(errcode(ERRCODE_WRONG_OBJECT_TYPE),
@@ -375,6 +370,27 @@ CreateTrigger(CreateTrigStmt *stmt, const char *queryString,
RelationGetRelationName(rel)),
errdetail("Triggers on views cannot have transition tables.")));
/*
* We currently don't allow row-level triggers with transition
* tables on partition or inheritance children. Such triggers
* would somehow need to see tuples converted to the format of the
* table they're attached to, and it's not clear which subset of
* tuples each child should see. See also the prohibitions in
* ATExecAttachPartition() and ATExecAddInherit().
*/
if (TRIGGER_FOR_ROW(tgtype) && has_superclass(rel->rd_id))
{
/* Use appropriate error message. */
if (rel->rd_rel->relispartition)
ereport(ERROR,
(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
errmsg("ROW triggers with transition tables are not supported on partitions")));
else
ereport(ERROR,
(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
errmsg("ROW triggers with transition tables are not supported on inheritance children")));
}
if (stmt->timing != TRIGGER_TYPE_AFTER)
ereport(ERROR,
(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
@@ -2028,6 +2044,64 @@ equalTriggerDescs(TriggerDesc *trigdesc1, TriggerDesc *trigdesc2)
}
#endif /* NOT_USED */
/*
* Check if there is a row-level trigger with transition tables that prevents
* a table from becoming an inheritance child or partition. Return the name
* of the first such incompatible trigger, or NULL if there is none.
*/
const char *
FindTriggerIncompatibleWithInheritance(TriggerDesc *trigdesc)
{
if (trigdesc != NULL)
{
int i;
for (i = 0; i < trigdesc->numtriggers; ++i)
{
Trigger *trigger = &trigdesc->triggers[i];
if (trigger->tgoldtable != NULL || trigger->tgnewtable != NULL)
return trigger->tgname;
}
}
return NULL;
}
/*
* Make a TransitionCaptureState object from a given TriggerDesc. The
* resulting object holds the flags which control whether transition tuples
* are collected when tables are modified. This allows us to use the flags
* from a parent table to control the collection of transition tuples from
* child tables.
*
* If there are no triggers with transition tables configured for 'trigdesc',
* then return NULL.
*
* The resulting object can be passed to the ExecAR* functions. The caller
* should set tcs_map or tcs_original_insert_tuple as appropriate when dealing
* with child tables.
*/
TransitionCaptureState *
MakeTransitionCaptureState(TriggerDesc *trigdesc)
{
TransitionCaptureState *state = NULL;
if (trigdesc != NULL &&
(trigdesc->trig_delete_old_table || trigdesc->trig_update_old_table ||
trigdesc->trig_update_new_table || trigdesc->trig_insert_new_table))
{
state = (TransitionCaptureState *)
palloc0(sizeof(TransitionCaptureState));
state->tcs_delete_old_table = trigdesc->trig_delete_old_table;
state->tcs_update_old_table = trigdesc->trig_update_old_table;
state->tcs_update_new_table = trigdesc->trig_update_new_table;
state->tcs_insert_new_table = trigdesc->trig_insert_new_table;
}
return state;
}
/*
* Call a trigger function.
*
@@ -2192,7 +2266,7 @@ ExecASInsertTriggers(EState *estate, ResultRelInfo *relinfo)
if (trigdesc && trigdesc->trig_insert_after_statement)
AfterTriggerSaveEvent(estate, relinfo, TRIGGER_EVENT_INSERT,
false, NULL, NULL, NIL, NULL);
false, NULL, NULL, NIL, NULL, NULL);
}
TupleTableSlot *
@@ -2263,14 +2337,18 @@ ExecBRInsertTriggers(EState *estate, ResultRelInfo *relinfo,
void
ExecARInsertTriggers(EState *estate, ResultRelInfo *relinfo,
HeapTuple trigtuple, List *recheckIndexes)
HeapTuple trigtuple, List *recheckIndexes,
TransitionCaptureState *transition_capture)
{
TriggerDesc *trigdesc = relinfo->ri_TrigDesc;
if (trigdesc &&
(trigdesc->trig_insert_after_row || trigdesc->trig_insert_new_table))
if ((trigdesc && trigdesc->trig_insert_after_row) ||
(trigdesc && !transition_capture && trigdesc->trig_insert_new_table) ||
(transition_capture && transition_capture->tcs_insert_new_table))
AfterTriggerSaveEvent(estate, relinfo, TRIGGER_EVENT_INSERT,
true, NULL, trigtuple, recheckIndexes, NULL);
true, NULL, trigtuple,
recheckIndexes, NULL,
transition_capture);
}
TupleTableSlot *
@@ -2398,7 +2476,7 @@ ExecASDeleteTriggers(EState *estate, ResultRelInfo *relinfo)
if (trigdesc && trigdesc->trig_delete_after_statement)
AfterTriggerSaveEvent(estate, relinfo, TRIGGER_EVENT_DELETE,
false, NULL, NULL, NIL, NULL);
false, NULL, NULL, NIL, NULL, NULL);
}
bool
@@ -2473,12 +2551,14 @@ ExecBRDeleteTriggers(EState *estate, EPQState *epqstate,
void
ExecARDeleteTriggers(EState *estate, ResultRelInfo *relinfo,
ItemPointer tupleid,
HeapTuple fdw_trigtuple)
HeapTuple fdw_trigtuple,
TransitionCaptureState *transition_capture)
{
TriggerDesc *trigdesc = relinfo->ri_TrigDesc;
if (trigdesc &&
(trigdesc->trig_delete_after_row || trigdesc->trig_delete_old_table))
if ((trigdesc && trigdesc->trig_delete_after_row) ||
(trigdesc && !transition_capture && trigdesc->trig_delete_old_table) ||
(transition_capture && transition_capture->tcs_delete_old_table))
{
HeapTuple trigtuple;
@@ -2494,7 +2574,8 @@ ExecARDeleteTriggers(EState *estate, ResultRelInfo *relinfo,
trigtuple = fdw_trigtuple;
AfterTriggerSaveEvent(estate, relinfo, TRIGGER_EVENT_DELETE,
true, trigtuple, NULL, NIL, NULL);
true, trigtuple, NULL, NIL, NULL,
transition_capture);
if (trigtuple != fdw_trigtuple)
heap_freetuple(trigtuple);
}
@@ -2610,7 +2691,8 @@ ExecASUpdateTriggers(EState *estate, ResultRelInfo *relinfo)
if (trigdesc && trigdesc->trig_update_after_statement)
AfterTriggerSaveEvent(estate, relinfo, TRIGGER_EVENT_UPDATE,
false, NULL, NULL, NIL,
GetUpdatedColumns(relinfo, estate));
GetUpdatedColumns(relinfo, estate),
NULL);
}
TupleTableSlot *
@@ -2735,12 +2817,18 @@ ExecARUpdateTriggers(EState *estate, ResultRelInfo *relinfo,
ItemPointer tupleid,
HeapTuple fdw_trigtuple,
HeapTuple newtuple,
List *recheckIndexes)
List *recheckIndexes,
TransitionCaptureState *transition_capture)
{
TriggerDesc *trigdesc = relinfo->ri_TrigDesc;
if (trigdesc && (trigdesc->trig_update_after_row ||
trigdesc->trig_update_old_table || trigdesc->trig_update_new_table))
if ((trigdesc && trigdesc->trig_update_after_row) ||
(trigdesc && !transition_capture &&
(trigdesc->trig_update_old_table ||
trigdesc->trig_update_new_table)) ||
(transition_capture &&
(transition_capture->tcs_update_old_table ||
transition_capture->tcs_update_new_table)))
{
HeapTuple trigtuple;
@@ -2757,7 +2845,8 @@ ExecARUpdateTriggers(EState *estate, ResultRelInfo *relinfo,
AfterTriggerSaveEvent(estate, relinfo, TRIGGER_EVENT_UPDATE,
true, trigtuple, newtuple, recheckIndexes,
GetUpdatedColumns(relinfo, estate));
GetUpdatedColumns(relinfo, estate),
transition_capture);
if (trigtuple != fdw_trigtuple)
heap_freetuple(trigtuple);
}
@@ -2888,7 +2977,7 @@ ExecASTruncateTriggers(EState *estate, ResultRelInfo *relinfo)
if (trigdesc && trigdesc->trig_truncate_after_statement)
AfterTriggerSaveEvent(estate, relinfo, TRIGGER_EVENT_TRUNCATE,
false, NULL, NULL, NIL, NULL);
false, NULL, NULL, NIL, NULL, NULL);
}
@@ -5090,7 +5179,8 @@ static void
AfterTriggerSaveEvent(EState *estate, ResultRelInfo *relinfo,
int event, bool row_trigger,
HeapTuple oldtup, HeapTuple newtup,
List *recheckIndexes, Bitmapset *modifiedCols)
List *recheckIndexes, Bitmapset *modifiedCols,
TransitionCaptureState *transition_capture)
{
Relation rel = relinfo->ri_RelationDesc;
TriggerDesc *trigdesc = relinfo->ri_TrigDesc;
@@ -5120,10 +5210,49 @@ AfterTriggerSaveEvent(EState *estate, ResultRelInfo *relinfo,
*/
if (row_trigger)
{
if ((event == TRIGGER_EVENT_DELETE &&
trigdesc->trig_delete_old_table) ||
(event == TRIGGER_EVENT_UPDATE &&
trigdesc->trig_update_old_table))
HeapTuple original_insert_tuple = NULL;
TupleConversionMap *map = NULL;
bool delete_old_table = false;
bool update_old_table = false;
bool update_new_table = false;
bool insert_new_table = false;
if (transition_capture != NULL)
{
/*
* A TransitionCaptureState object was provided to tell us which
* tuples to capture based on a parent table named in a DML
* statement. We may be dealing with a child table with an
* incompatible TupleDescriptor, in which case we'll need a map to
* convert them. As a small optimization, we may receive the
* original tuple from an insertion into a partitioned table to
* avoid a wasteful parent->child->parent round trip.
*/
delete_old_table = transition_capture->tcs_delete_old_table;
update_old_table = transition_capture->tcs_update_old_table;
update_new_table = transition_capture->tcs_update_new_table;
insert_new_table = transition_capture->tcs_insert_new_table;
map = transition_capture->tcs_map;
original_insert_tuple =
transition_capture->tcs_original_insert_tuple;
}
else if (trigdesc != NULL)
{
/*
* Check if we need to capture transition tuples for triggers
* defined on this relation directly. This case is useful for
* cases like execReplication.c which don't set up a
* TriggerCaptureState because they don't know how to work with
* partitions.
*/
delete_old_table = trigdesc->trig_delete_old_table;
update_old_table = trigdesc->trig_update_old_table;
update_new_table = trigdesc->trig_update_new_table;
insert_new_table = trigdesc->trig_insert_new_table;
}
if ((event == TRIGGER_EVENT_DELETE && delete_old_table) ||
(event == TRIGGER_EVENT_UPDATE && update_old_table))
{
Tuplestorestate *old_tuplestore;
@@ -5131,12 +5260,18 @@ AfterTriggerSaveEvent(EState *estate, ResultRelInfo *relinfo,
old_tuplestore =
GetTriggerTransitionTuplestore
(afterTriggers.old_tuplestores);
tuplestore_puttuple(old_tuplestore, oldtup);
if (map != NULL)
{
HeapTuple converted = do_convert_tuple(oldtup, map);
tuplestore_puttuple(old_tuplestore, converted);
pfree(converted);
}
else
tuplestore_puttuple(old_tuplestore, oldtup);
}
if ((event == TRIGGER_EVENT_INSERT &&
trigdesc->trig_insert_new_table) ||
(event == TRIGGER_EVENT_UPDATE &&
trigdesc->trig_update_new_table))
if ((event == TRIGGER_EVENT_INSERT && insert_new_table) ||
(event == TRIGGER_EVENT_UPDATE && update_new_table))
{
Tuplestorestate *new_tuplestore;
@@ -5144,11 +5279,22 @@ AfterTriggerSaveEvent(EState *estate, ResultRelInfo *relinfo,
new_tuplestore =
GetTriggerTransitionTuplestore
(afterTriggers.new_tuplestores);
tuplestore_puttuple(new_tuplestore, newtup);
if (original_insert_tuple != NULL)
tuplestore_puttuple(new_tuplestore, original_insert_tuple);
else if (map != NULL)
{
HeapTuple converted = do_convert_tuple(newtup, map);
tuplestore_puttuple(new_tuplestore, converted);
pfree(converted);
}
else
tuplestore_puttuple(new_tuplestore, newtup);
}
/* If transition tables are the only reason we're here, return. */
if ((event == TRIGGER_EVENT_DELETE && !trigdesc->trig_delete_after_row) ||
if (trigdesc == NULL ||
(event == TRIGGER_EVENT_DELETE && !trigdesc->trig_delete_after_row) ||
(event == TRIGGER_EVENT_INSERT && !trigdesc->trig_insert_after_row) ||
(event == TRIGGER_EVENT_UPDATE && !trigdesc->trig_update_after_row))
return;

2
src/backend/executor/execMain.c
View File

@@ -3198,7 +3198,7 @@ EvalPlanQualEnd(EPQState *epqstate)
* 'tup_conv_maps' receives an array of TupleConversionMap objects with one
* entry for every leaf partition (required to convert input tuple based
* on the root table's rowtype to a leaf partition's rowtype after tuple
* routing is done
* routing is done)
* 'partition_tuple_slot' receives a standalone TupleTableSlot to be used
* to manipulate any given leaf partition's rowtype after that partition
* is chosen by tuple-routing.

6
src/backend/executor/execReplication.c
View File

@@ -417,7 +417,7 @@ ExecSimpleRelationInsert(EState *estate, TupleTableSlot *slot)
/* AFTER ROW INSERT Triggers */
ExecARInsertTriggers(estate, resultRelInfo, tuple,
recheckIndexes);
recheckIndexes, NULL);
list_free(recheckIndexes);
}
@@ -479,7 +479,7 @@ ExecSimpleRelationUpdate(EState *estate, EPQState *epqstate,
/* AFTER ROW UPDATE Triggers */
ExecARUpdateTriggers(estate, resultRelInfo,
&searchslot->tts_tuple->t_self,
NULL, tuple, recheckIndexes);
NULL, tuple, recheckIndexes, NULL);
list_free(recheckIndexes);
}
@@ -522,7 +522,7 @@ ExecSimpleRelationDelete(EState *estate, EPQState *epqstate,
/* AFTER ROW DELETE Triggers */
ExecARDeleteTriggers(estate, resultRelInfo,
&searchslot->tts_tuple->t_self, NULL);
&searchslot->tts_tuple->t_self, NULL, NULL);
list_free(recheckIndexes);
}

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

@@ -313,6 +313,36 @@ ExecInsert(ModifyTableState *mtstate,
/* For ExecInsertIndexTuples() to work on the partition's indexes */
estate->es_result_relation_info = resultRelInfo;
/*
* If we're capturing transition tuples, we might need to convert from
* the partition rowtype to parent rowtype.
*/
if (mtstate->mt_transition_capture != NULL)
{
if (resultRelInfo->ri_TrigDesc &&
(resultRelInfo->ri_TrigDesc->trig_insert_before_row ||
resultRelInfo->ri_TrigDesc->trig_insert_instead_row))
{
/*
* If there are any BEFORE or INSTEAD triggers on the
* partition, we'll have to be ready to convert their result
* back to tuplestore format.
*/
mtstate->mt_transition_capture->tcs_original_insert_tuple = NULL;
mtstate->mt_transition_capture->tcs_map =
mtstate->mt_transition_tupconv_maps[leaf_part_index];
}
else
{
/*
* Otherwise, just remember the original unconverted tuple, to
* avoid a needless round trip conversion.
*/
mtstate->mt_transition_capture->tcs_original_insert_tuple = tuple;
mtstate->mt_transition_capture->tcs_map = NULL;
}
}
/*
* We might need to convert from the parent rowtype to the partition
* rowtype.
@@ -588,7 +618,8 @@ ExecInsert(ModifyTableState *mtstate,
}
/* AFTER ROW INSERT Triggers */
ExecARInsertTriggers(estate, resultRelInfo, tuple, recheckIndexes);
ExecARInsertTriggers(estate, resultRelInfo, tuple, recheckIndexes,
mtstate->mt_transition_capture);
list_free(recheckIndexes);
@@ -636,7 +667,8 @@ ExecInsert(ModifyTableState *mtstate,
* ----------------------------------------------------------------
*/
static TupleTableSlot *
ExecDelete(ItemPointer tupleid,
ExecDelete(ModifyTableState *mtstate,
ItemPointer tupleid,
HeapTuple oldtuple,
TupleTableSlot *planSlot,
EPQState *epqstate,
@@ -813,7 +845,8 @@ ldelete:;
(estate->es_processed)++;
/* AFTER ROW DELETE Triggers */
ExecARDeleteTriggers(estate, resultRelInfo, tupleid, oldtuple);
ExecARDeleteTriggers(estate, resultRelInfo, tupleid, oldtuple,
mtstate->mt_transition_capture);
/* Process RETURNING if present */
if (resultRelInfo->ri_projectReturning)
@@ -894,7 +927,8 @@ ldelete:;
* ----------------------------------------------------------------
*/
static TupleTableSlot *
ExecUpdate(ItemPointer tupleid,
ExecUpdate(ModifyTableState *mtstate,
ItemPointer tupleid,
HeapTuple oldtuple,
TupleTableSlot *slot,
TupleTableSlot *planSlot,
@@ -1122,7 +1156,8 @@ lreplace:;
/* AFTER ROW UPDATE Triggers */
ExecARUpdateTriggers(estate, resultRelInfo, tupleid, oldtuple, tuple,
recheckIndexes);
recheckIndexes,
mtstate->mt_transition_capture);
list_free(recheckIndexes);
@@ -1329,7 +1364,7 @@ ExecOnConflictUpdate(ModifyTableState *mtstate,
*/
/* Execute UPDATE with projection */
*returning = ExecUpdate(&tuple.t_self, NULL,
*returning = ExecUpdate(mtstate, &tuple.t_self, NULL,
mtstate->mt_conflproj, planSlot,
&mtstate->mt_epqstate, mtstate->ps.state,
canSetTag);
@@ -1376,20 +1411,31 @@ fireBSTriggers(ModifyTableState *node)
}
/*
* Process AFTER EACH STATEMENT triggers
* Return the ResultRelInfo for which we will fire AFTER STATEMENT triggers.
* This is also the relation into whose tuple format all captured transition
* tuples must be converted.
*/
static void
fireASTriggers(ModifyTableState *node)
static ResultRelInfo *
getASTriggerResultRelInfo(ModifyTableState *node)
{
ResultRelInfo *resultRelInfo = node->resultRelInfo;
/*
* If the node modifies a partitioned table, we must fire its triggers.
* Note that in that case, node->resultRelInfo points to the first leaf
* partition, not the root table.
*/
if (node->rootResultRelInfo != NULL)
resultRelInfo = node->rootResultRelInfo;
return node->rootResultRelInfo;
else
return node->resultRelInfo;
}
/*
* Process AFTER EACH STATEMENT triggers
*/
static void
fireASTriggers(ModifyTableState *node)
{
ResultRelInfo *resultRelInfo = getASTriggerResultRelInfo(node);
switch (node->operation)
{
@@ -1411,6 +1457,72 @@ fireASTriggers(ModifyTableState *node)
}
}
/*
* Set up the state needed for collecting transition tuples for AFTER
* triggers.
*/
static void
ExecSetupTransitionCaptureState(ModifyTableState *mtstate, EState *estate)
{
ResultRelInfo *targetRelInfo = getASTriggerResultRelInfo(mtstate);
int i;
/* Check for transition tables on the directly targeted relation. */
mtstate->mt_transition_capture =
MakeTransitionCaptureState(targetRelInfo->ri_TrigDesc);
/*
* If we found that we need to collect transition tuples then we may also
* need tuple conversion maps for any children that have TupleDescs that
* aren't compatible with the tuplestores.
*/
if (mtstate->mt_transition_capture != NULL)
{
ResultRelInfo *resultRelInfos;
int numResultRelInfos;
/* Find the set of partitions so that we can find their TupleDescs. */
if (mtstate->mt_partition_dispatch_info != NULL)
{
/*
* For INSERT via partitioned table, so we need TupleDescs based
* on the partition routing table.
*/
resultRelInfos = mtstate->mt_partitions;
numResultRelInfos = mtstate->mt_num_partitions;
}
else
{
/* Otherwise we need the ResultRelInfo for each subplan. */
resultRelInfos = mtstate->resultRelInfo;
numResultRelInfos = mtstate->mt_nplans;
}
/*
* Build array of conversion maps from each child's TupleDesc to the
* one used in the tuplestore. The map pointers may be NULL when no
* conversion is necessary, which is hopefully a common case for
* partitions.
*/
mtstate->mt_transition_tupconv_maps = (TupleConversionMap **)
palloc0(sizeof(TupleConversionMap *) * numResultRelInfos);
for (i = 0; i < numResultRelInfos; ++i)
{
mtstate->mt_transition_tupconv_maps[i] =
convert_tuples_by_name(RelationGetDescr(resultRelInfos[i].ri_RelationDesc),
RelationGetDescr(targetRelInfo->ri_RelationDesc),
gettext_noop("could not convert row type"));
}
/*
* Install the conversion map for the first plan for UPDATE and DELETE
* operations. It will be advanced each time we switch to the next
* plan. (INSERT operations set it every time.)
*/
mtstate->mt_transition_capture->tcs_map =
mtstate->mt_transition_tupconv_maps[0];
}
}
/* ----------------------------------------------------------------
* ExecModifyTable
@@ -1509,6 +1621,13 @@ ExecModifyTable(ModifyTableState *node)
estate->es_result_relation_info = resultRelInfo;
EvalPlanQualSetPlan(&node->mt_epqstate, subplanstate->plan,
node->mt_arowmarks[node->mt_whichplan]);
if (node->mt_transition_capture != NULL)
{
/* Prepare to convert transition tuples from this child. */
Assert(node->mt_transition_tupconv_maps != NULL);
node->mt_transition_capture->tcs_map =
node->mt_transition_tupconv_maps[node->mt_whichplan];
}
continue;
}
else
@@ -1618,11 +1737,11 @@ ExecModifyTable(ModifyTableState *node)
estate, node->canSetTag);
break;
case CMD_UPDATE:
slot = ExecUpdate(tupleid, oldtuple, slot, planSlot,
slot = ExecUpdate(node, tupleid, oldtuple, slot, planSlot,
&node->mt_epqstate, estate, node->canSetTag);
break;
case CMD_DELETE:
slot = ExecDelete(tupleid, oldtuple, planSlot,
slot = ExecDelete(node, tupleid, oldtuple, planSlot,
&node->mt_epqstate, estate, node->canSetTag);
break;
default:
@@ -1804,6 +1923,9 @@ ExecInitModifyTable(ModifyTable *node, EState *estate, int eflags)
mtstate->mt_partition_tuple_slot = partition_tuple_slot;
}
/* Build state for collecting transition tuples */
ExecSetupTransitionCaptureState(mtstate, estate);
/*
* Initialize any WITH CHECK OPTION constraints if needed.
*/