This is an automated email from the ASF dual-hosted git repository. tuhaihe pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/cloudberry.git
commit 392978a8b1bc79a5b1916d7284ee2abcd932e5b4 Author: liushengsong <[email protected]> AuthorDate: Mon May 11 11:24:11 2026 +0800 Support distribution key update in MERGE for partitioned tables Previously, MERGE ... WHEN MATCHED THEN UPDATE SET <dist_key> = ... was rejected with "cannot update column in merge with distributed column". This commit adds support by extending the SplitMerge node to handle distribution key updates via DELETE + INSERT routing. Key changes: Planner (preptlist.c, createplan.c, setrefs.c, cdbpath.c): - Detect distribution key modification in MERGE UPDATE actions and set merge_need_split_update flag - Add all target table columns to subplan targetlist so SplitMerge can project complete rows for INSERT - Expand UPDATE action targetlists to include all columns (not just SET columns) using expand_insert_targetlist - Build SplitMerge targetlist in N+M+1 format: N target table columns + M subplan columns + 1 DMLAction column - Always use root table action lists (not per-partition adjusted lists) to ensure hashAttnos match root attribute numbers - Add set_splitmerge_tlist_references for proper OUTER_VAR conversion Executor (nodeSplitMerge.c, nodeModifyTable.c): - SplitMerge splits MATCHED UPDATE into DELETE + INSERT tuple pair, each routed to the correct segment via hash computation - NOT MATCHED rows get PASSTHROUGH action for normal ExecMerge processing - ModifyTable handles DMLAction-tagged tuples from SplitMerge output - Support lazy partition routing for CMD_MERGE DML_INSERT Refactoring (nodeSplitMerge.c): - Extract computeTargetSegment(), SwitchResultRelForPartition(), BuildRootUpdateTupleDesc() helper functions - Define SPLITMERGE_ACTION_PASSTHROUGH constant - Remove dead code and consolidate duplicated logic --- .../src/test/regress/expected/privileges.out | 2 +- .../test/regress/expected/privileges_optimizer.out | 2 +- src/backend/cdb/cdbpath.c | 39 +- src/backend/executor/nodeModifyTable.c | 79 ++- src/backend/executor/nodeSplitMerge.c | 538 +++++++++++++-------- src/backend/nodes/copyfuncs.funcs.c | 2 + src/backend/nodes/outfuncs_common.c | 2 + src/backend/nodes/readfast.c | 2 + src/backend/optimizer/plan/createplan.c | 144 +++++- src/backend/optimizer/plan/setrefs.c | 142 +++++- src/backend/optimizer/prep/preptlist.c | 52 +- src/include/nodes/execnodes.h | 7 + src/include/nodes/pathnodes.h | 3 + src/include/nodes/plannodes.h | 2 + src/include/optimizer/prep.h | 5 + src/test/regress/expected/merge.out | 255 +++++++++- src/test/regress/expected/privileges.out | 2 +- src/test/regress/expected/triggers.out | 2 +- src/test/regress/sql/merge.sql | 152 ++++++ 19 files changed, 1203 insertions(+), 229 deletions(-) diff --git a/contrib/pax_storage/src/test/regress/expected/privileges.out b/contrib/pax_storage/src/test/regress/expected/privileges.out index 8b13a388124..65d52afd6d8 100644 --- a/contrib/pax_storage/src/test/regress/expected/privileges.out +++ b/contrib/pax_storage/src/test/regress/expected/privileges.out @@ -1020,7 +1020,7 @@ WHEN MATCHED THEN UPDATE SET b = s.b, a = t.a + 1 WHEN NOT MATCHED THEN INSERT VALUES (a, b); -ERROR: cannot update column in merge with distributed column +ERROR: permission denied for table mtarget -- fail (no SELECT on t.b) MERGE INTO mtarget t USING msource s ON t.a = s.a WHEN MATCHED AND t.b IS NOT NULL THEN diff --git a/contrib/pax_storage/src/test/regress/expected/privileges_optimizer.out b/contrib/pax_storage/src/test/regress/expected/privileges_optimizer.out index a06cac9a351..34647031e5a 100644 --- a/contrib/pax_storage/src/test/regress/expected/privileges_optimizer.out +++ b/contrib/pax_storage/src/test/regress/expected/privileges_optimizer.out @@ -1020,7 +1020,7 @@ WHEN MATCHED THEN UPDATE SET b = s.b, a = t.a + 1 WHEN NOT MATCHED THEN INSERT VALUES (a, b); -ERROR: cannot update column in merge with distributed column +ERROR: permission denied for table mtarget -- fail (no SELECT on t.b) MERGE INTO mtarget t USING msource s ON t.a = s.a WHEN MATCHED AND t.b IS NOT NULL THEN diff --git a/src/backend/cdb/cdbpath.c b/src/backend/cdb/cdbpath.c index 1f835283dcc..cc703b6e1a7 100644 --- a/src/backend/cdb/cdbpath.c +++ b/src/backend/cdb/cdbpath.c @@ -75,7 +75,7 @@ static bool try_redistribute(PlannerInfo *root, CdbpathMfjRel *g, static SplitUpdatePath *make_splitupdate_path(PlannerInfo *root, Path *subpath, Index rti); -static SplitMergePath *make_split_merge_path(PlannerInfo *root, Path *subpath, List* resultRelations, List *mergeActionLists); +static SplitMergePath *make_split_merge_path(PlannerInfo *root, Path *subpath, List* resultRelations, List *mergeActionLists, bool hasSplitUpdate); static bool can_elide_explicit_motion(PlannerInfo *root, Index rti, Path *subpath, GpPolicy *policy); /* @@ -2798,19 +2798,22 @@ create_motion_path_for_merge(PlannerInfo *root, List *resultRelations, GpPolicy { /* * If merge contain CMD_INSERT, we need split merge to let new - * insert tuple redistributed to correct segment. otherwise, we - * create motion as the same as update/delete in create_motion_path_for_upddel + * insert tuple redistributed to correct segment. If merge has + * UPDATE that modifies distribution key, we also need split merge + * to handle the DELETE+INSERT split. */ foreach(l, mergeActionLists) { List *mergeActionList = lfirst(l); - foreach(lc, mergeActionList) + foreach(lc, mergeActionList) { MergeAction *action = lfirst(lc); if (action->commandType == CMD_INSERT) need_split_merge = true; } } + if (root->merge_need_split_update) + need_split_merge = true; if (need_split_merge) { @@ -2820,7 +2823,7 @@ create_motion_path_for_merge(PlannerInfo *root, List *resultRelations, GpPolicy rel = build_simple_rel(root, linitial_int(resultRelations), NULL /*parent*/); targetLocus = cdbpathlocus_from_baserel(root, rel, 0); - subpath = (Path *) make_split_merge_path(root, subpath, resultRelations, mergeActionLists); + subpath = (Path *) make_split_merge_path(root, subpath, resultRelations, mergeActionLists, root->merge_need_split_update); subpath = cdbpath_create_explicit_motion_path(root, subpath, targetLocus); @@ -2923,14 +2926,35 @@ turn_volatile_seggen_to_singleqe(PlannerInfo *root, Path *path, Node *node) } static SplitMergePath * -make_split_merge_path(PlannerInfo *root, Path *subpath, List *resultRelations, List *mergeActionLists) +make_split_merge_path(PlannerInfo *root, Path *subpath, List *resultRelations, List *mergeActionLists, bool hasSplitUpdate) { PathTarget *splitMergePathTarget; SplitMergePath *splitmergepath; splitMergePathTarget = copy_pathtarget(subpath->pathtarget); - /* populate information generated above into splitupdate node */ + /* + * Same restriction as SplitUpdate: updating a distribution key + * is not allowed when the target relation has update triggers. + */ + if (hasSplitUpdate) + { + RangeTblEntry *rte = rt_fetch(root->parse->resultRelation, + root->parse->rtable); + if (has_update_triggers(rte->relid, true)) + ereport(ERROR, + (errcode(ERRCODE_GP_FEATURE_NOT_YET), + errmsg("UPDATE on distributed key column not allowed on relation with update triggers"))); + } + + /* When hasSplitUpdate, add DMLAction column for split UPDATE handling */ + if (hasSplitUpdate) + { + DMLActionExpr *actionExpr = makeNode(DMLActionExpr); + add_column_to_pathtarget(splitMergePathTarget, (Expr *) actionExpr, 0); + } + + /* populate information generated above into splitmerge node */ splitmergepath = makeNode(SplitMergePath); splitmergepath->path.pathtype = T_SplitMerge; splitmergepath->path.parent = subpath->parent; @@ -2947,6 +2971,7 @@ make_split_merge_path(PlannerInfo *root, Path *subpath, List *resultRelations, L splitmergepath->subpath = subpath; splitmergepath->resultRelations = resultRelations; splitmergepath->mergeActionLists = mergeActionLists; + splitmergepath->hasSplitUpdate = hasSplitUpdate; return splitmergepath; } diff --git a/src/backend/executor/nodeModifyTable.c b/src/backend/executor/nodeModifyTable.c index e7b19f959c6..4671b2bd892 100644 --- a/src/backend/executor/nodeModifyTable.c +++ b/src/backend/executor/nodeModifyTable.c @@ -1095,7 +1095,7 @@ ExecInsert(ModifyTableContext *context, */ if (mtstate->operation == CMD_UPDATE) wco_kind = WCO_RLS_UPDATE_CHECK; - else if (mtstate->operation == CMD_MERGE) + else if (mtstate->operation == CMD_MERGE && context->relaction != NULL) wco_kind = (context->relaction->mas_action->commandType == CMD_UPDATE) ? WCO_RLS_UPDATE_CHECK : WCO_RLS_INSERT_CHECK; else @@ -4376,7 +4376,77 @@ ExecModifyTable(PlanState *pstate) break; case CMD_MERGE: - slot = ExecMerge(&context, resultRelInfo, tupleid, node->canSetTag); + if (action == (int) DML_INSERT) + { + /* + * Split merge INSERT: extract non-junk columns (positions + * 1..N) from the plan slot into a properly-typed insert + * slot. We can't use ExecGetInsertNewTuple because the + * SplitMerge's ExecInitMergeTupleSlots may have set + * ri_projectNewInfoValid without building the INSERT + * projection. + */ + ResultRelInfo *saved = resultRelInfo; + TupleTableSlot *insertSlot; + int natts; + + resultRelInfo = node->rootResultRelInfo; + natts = RelationGetNumberOfAttributes(resultRelInfo->ri_RelationDesc); + + /* + * Lazily set up partition tuple routing for split-update + * MERGE INSERT on partitioned tables. + */ + if (resultRelInfo->ri_RelationDesc->rd_rel->relkind == + RELKIND_PARTITIONED_TABLE && + node->mt_partition_tuple_routing == NULL) + { + node->mt_partition_tuple_routing = + ExecSetupPartitionTupleRouting(estate, + resultRelInfo->ri_RelationDesc); + } + + insertSlot = resultRelInfo->ri_newTupleSlot; + if (insertSlot == NULL) + { + insertSlot = table_slot_create(resultRelInfo->ri_RelationDesc, + &estate->es_tupleTable); + resultRelInfo->ri_newTupleSlot = insertSlot; + } + + ExecClearTuple(insertSlot); + slot_getallattrs(context.planSlot); + memcpy(insertSlot->tts_values, context.planSlot->tts_values, + natts * sizeof(Datum)); + memcpy(insertSlot->tts_isnull, context.planSlot->tts_isnull, + natts * sizeof(bool)); + ExecStoreVirtualTuple(insertSlot); + + /* Set relaction to NULL to avoid ExecInsert dereferencing it */ + context.relaction = NULL; + + slot = ExecInsert(&context, resultRelInfo, insertSlot, + node->canSetTag, NULL, NULL, + true /* splitUpdate */); + resultRelInfo = saved; + } + else if (action == (int) DML_DELETE) + { + /* + * Split merge DELETE: delete the old tuple on this segment. + */ + slot = ExecDelete(&context, resultRelInfo, tupleid, oldtuple, segid, + false, /* processReturning */ + false, /* changingPart */ + node->canSetTag, + NULL, NULL, NULL, + true /* splitUpdate */); + } + else + { + /* Normal MERGE processing (no split or pass-through) */ + slot = ExecMerge(&context, resultRelInfo, tupleid, node->canSetTag); + } break; default: @@ -4665,6 +4735,11 @@ ExecInitModifyTable(ModifyTable *node, EState *estate, int eflags) mtstate->mt_action_attno = ExecFindJunkAttributeInTlist(subplan->targetlist, "DMLAction"); } + else if (operation == CMD_MERGE) + { + mtstate->mt_action_attno = + ExecFindJunkAttributeInTlist(subplan->targetlist, "DMLAction"); + } } /* * Do additional per-result-relation initialization. diff --git a/src/backend/executor/nodeSplitMerge.c b/src/backend/executor/nodeSplitMerge.c index 09b5a0d4b97..5916a661dba 100644 --- a/src/backend/executor/nodeSplitMerge.c +++ b/src/backend/executor/nodeSplitMerge.c @@ -24,8 +24,14 @@ #include "executor/instrument.h" #include "executor/nodeSplitMerge.h" +#include "nodes/nodeFuncs.h" #include "utils/memutils.h" +/* + * Action value for rows that should be processed by ModifyTable's + * normal ExecMerge path (NOT MATCHED or MATCHED pass-through). + */ +#define SPLITMERGE_ACTION_PASSTHROUGH (-1) typedef struct MTTargetRelLookup { @@ -68,122 +74,142 @@ evalHashKey(SplitMergeState *node, Datum *values, bool *isnulls) return target_seg; } +/* + * Compute target segment ID from the given slot's values. + * Returns 0 if no hash is configured (DISTRIBUTED RANDOMLY). + */ +static int32 +computeTargetSegment(SplitMergeState *node, TupleTableSlot *slot) +{ + SplitMerge *plannode = (SplitMerge *) node->ps.plan; + + if (node->cdbhash) + return evalHashKey(node, slot->tts_values, slot->tts_isnull); + else + return cdbhashrandomseg(plannode->numHashSegments); +} + +/* + * Build a tuple in the N+M+1 format for hasSplitUpdate. + * + * The output slot layout is: + * [0..N-1] target table columns (from projSlot, or NULL) + * [N..N+M-1] subplan columns (from inputSlot) + * [N+M] DMLAction + * + * N = node->subplan_offset, M = inputSlot column count. + */ +static void +BuildSplitMergeTuple(SplitMergeState *node, TupleTableSlot *outSlot, + TupleTableSlot *inputSlot, TupleTableSlot *projSlot, + int dmlAction, int32 segid) +{ + int offset = node->subplan_offset; + int natts_input = inputSlot->tts_tupleDescriptor->natts; + int natts_out = outSlot->tts_tupleDescriptor->natts; + + ExecClearTuple(outSlot); + + memset(outSlot->tts_values, 0, natts_out * sizeof(Datum)); + memset(outSlot->tts_isnull, true, natts_out * sizeof(bool)); + + /* Positions 0..N-1: projected target table values (if provided) */ + if (projSlot) + { + int natts_proj = projSlot->tts_tupleDescriptor->natts; + slot_getallattrs(projSlot); + memcpy(outSlot->tts_values, projSlot->tts_values, + natts_proj * sizeof(Datum)); + memcpy(outSlot->tts_isnull, projSlot->tts_isnull, + natts_proj * sizeof(bool)); + } + /* Positions N..N+M-1: subplan columns */ + slot_getallattrs(inputSlot); + memcpy(outSlot->tts_values + offset, inputSlot->tts_values, + natts_input * sizeof(Datum)); + memcpy(outSlot->tts_isnull + offset, inputSlot->tts_isnull, + natts_input * sizeof(bool)); + /* gp_segment_id within subplan region */ + outSlot->tts_values[offset + node->segid_attno - 1] = Int32GetDatum(segid); + outSlot->tts_isnull[offset + node->segid_attno - 1] = false; + + /* DMLAction at the end */ + outSlot->tts_values[node->action_attno - 1] = Int32GetDatum(dmlAction); + outSlot->tts_isnull[node->action_attno - 1] = false; + + ExecStoreVirtualTuple(outSlot); +} + +/* + * MergeTupleTableSlot + * + * Handle a NOT MATCHED row: evaluate WHEN NOT MATCHED actions, + * project INSERT values, and compute target segment for routing. + */ static TupleTableSlot * -MergeTupleTableSlot(TupleTableSlot *slot, SplitMerge *plannode, SplitMergeState *node, ResultRelInfo *resultRelInfo) +MergeTupleTableSlot(TupleTableSlot *slot, SplitMerge *plannode, + SplitMergeState *node, ResultRelInfo *resultRelInfo) { ExprContext *econtext = node->ps.ps_ExprContext; - - List *actionStates = NIL; ListCell *l; TupleTableSlot *newslot = NULL; + int32 target_seg = 0; - /* - * For INSERT actions, the root relation's merge action is OK since the - * INSERT's targetlist and the WHEN conditions can only refer to the - * source relation and hence it does not matter which result relation we - * work with. - * - * XXX does this mean that we can avoid creating copies of actionStates on - * partitioned tables, for not-matched actions? - */ - actionStates = resultRelInfo->ri_notMatchedMergeAction; - - /* - * Make source tuple available to ExecQual and ExecProject. We don't need - * the target tuple, since the WHEN quals and targetlist can't refer to - * the target columns. - */ econtext->ecxt_scantuple = NULL; econtext->ecxt_innertuple = slot; econtext->ecxt_outertuple = NULL; - foreach(l, actionStates) + /* Evaluate NOT MATCHED actions to find INSERT projection */ + foreach(l, resultRelInfo->ri_notMatchedMergeAction) { MergeActionState *action = (MergeActionState *) lfirst(l); - CmdType commandType = action->mas_action->commandType; - /* - * Test condition, if any. - * - * In the absence of any condition, we perform the action - * unconditionally (no need to check separately since ExecQual() will - * return true if there are no conditions to evaluate). - */ if (!ExecQual(action->mas_whenqual, econtext)) continue; - /* Perform stated action */ - switch (commandType) - { - case CMD_INSERT: - - /* - * Project the tuple. In case of a partitioned table, the - * projection was already built to use the root's descriptor, - * so we don't need to map the tuple here. - */ - newslot = ExecProject(action->mas_proj); - - break; - case CMD_NOTHING: - /* Do nothing */ - break; - default: - elog(ERROR, "unknown action in MERGE WHEN NOT MATCHED clause"); - } + if (action->mas_action->commandType == CMD_INSERT) + newslot = ExecProject(action->mas_proj); + /* else CMD_NOTHING: do nothing */ - /* - * We've activated one of the WHEN clauses, so we don't search - * further. This is required behaviour, not an optimization. - */ - break; + break; /* only first matching action */ } + /* Compute target segment for INSERT, or 0 for DO NOTHING */ if (newslot) - { - /* Compute segment ID for the new row */ - int32 target_seg; - - if (node->cdbhash) - target_seg = evalHashKey(node, newslot->tts_values, newslot->tts_isnull); - else - target_seg = cdbhashrandomseg(plannode->numHashSegments); + target_seg = computeTargetSegment(node, newslot); - slot->tts_values[node->segid_attno - 1] = Int32GetDatum(target_seg); - slot->tts_isnull[node->segid_attno - 1] = false; - } - else + /* Build output in the appropriate format */ + if (plannode->hasSplitUpdate) { - /* - * No newslot generated means that insert action will not be triggered. - * So we just redistributed tuple to any segment, like segment 0. - */ - slot->tts_values[node->segid_attno - 1] = Int32GetDatum(0); - slot->tts_isnull[node->segid_attno - 1] = false; + BuildSplitMergeTuple(node, node->ps.ps_ResultTupleSlot, + slot, NULL, SPLITMERGE_ACTION_PASSTHROUGH, + target_seg); + return node->ps.ps_ResultTupleSlot; } + /* Non-hasSplitUpdate: modify slot in-place */ + slot->tts_values[node->segid_attno - 1] = Int32GetDatum(target_seg); + slot->tts_isnull[node->segid_attno - 1] = false; return slot; } /* * ExecLookupResultRelByOid - * If the table with given OID is among the result relations to be - * updated by the given ModifyTable node, return its ResultRelInfo. + * If the table with given OID is among the result relations to be + * updated by the given SplitMerge node, return its ResultRelInfo. * * If not found, return NULL if missing_ok, else raise error. * - * If update_cache is true, then upon successful lookup, update the node's - * one-element cache. ONLY ExecModifyTable may pass true for this. + * If update_cache is true, update the node's one-element cache. */ static ResultRelInfo * MergeExecLookupResultRelByOid(SplitMergeState *node, Oid resultoid, - bool missing_ok, bool update_cache) + bool missing_ok, bool update_cache) { if (node->mt_resultOidHash) { - /* Use the pre-built hash table to locate the rel */ MTTargetRelLookup *mtlookup; mtlookup = (MTTargetRelLookup *) @@ -200,7 +226,6 @@ MergeExecLookupResultRelByOid(SplitMergeState *node, Oid resultoid, } else { - /* With few target rels, just search the ResultRelInfo array */ for (int ndx = 0; ndx < node->nrel; ndx++) { ResultRelInfo *rInfo = node->resultRelInfo + ndx; @@ -222,64 +247,163 @@ MergeExecLookupResultRelByOid(SplitMergeState *node, Oid resultoid, return NULL; } -/** - * Splits every TupleTableSlot into two TupleTableSlots: DELETE and INSERT. +/* + * SwitchResultRelForPartition + * + * For partitioned tables, look up the correct ResultRelInfo based on tableoid + * from the slot. Updates the cached result relation if it changes. + */ +static ResultRelInfo * +SwitchResultRelForPartition(SplitMergeState *node, TupleTableSlot *slot, + ResultRelInfo *resultRelInfo) +{ + bool isNull; + Datum d; + Oid resultoid; + + if (!AttributeNumberIsValid(node->mt_resultOidAttno)) + return resultRelInfo; + + d = ExecGetJunkAttribute(slot, node->mt_resultOidAttno, &isNull); + Assert(!isNull); + resultoid = DatumGetObjectId(d); + + if (resultoid != node->mt_lastResultOid) + resultRelInfo = MergeExecLookupResultRelByOid(node, resultoid, + false, true); + return resultRelInfo; +} + +/* + * MergeMatchedSplitUpdate + * + * Handle a MATCHED row when hasSplitUpdate is true. + * + * For UPDATE: splits into DELETE + INSERT tuples with DMLAction markers. + * Returns DELETE first; INSERT is saved for the next call. + * For DELETE: emits a single DELETE tuple. + * For DO NOTHING / no match: emits a pass-through tuple. + */ +static TupleTableSlot * +MergeMatchedSplitUpdate(TupleTableSlot *slot, SplitMerge *plannode, + SplitMergeState *node, ResultRelInfo *resultRelInfo) +{ + ExprContext *econtext = node->ps.ps_ExprContext; + ListCell *l; + int32 old_segid; + + econtext->ecxt_scantuple = slot; + econtext->ecxt_innertuple = slot; + econtext->ecxt_outertuple = NULL; + + old_segid = DatumGetInt32(slot->tts_values[node->segid_attno - 1]); + + foreach(l, resultRelInfo->ri_matchedMergeAction) + { + MergeActionState *action = (MergeActionState *) lfirst(l); + CmdType commandType = action->mas_action->commandType; + + if (!ExecQual(action->mas_whenqual, econtext)) + continue; + + switch (commandType) + { + case CMD_UPDATE: + { + TupleTableSlot *newslot; + int32 new_segid; + + newslot = ExecProject(action->mas_proj); + new_segid = computeTargetSegment(node, newslot); + + /* DELETE tuple: routed to old segment */ + BuildSplitMergeTuple(node, node->deleteTuple, slot, NULL, + (int) DML_DELETE, old_segid); + + /* INSERT tuple: projected new values, routed to new segment */ + BuildSplitMergeTuple(node, node->insertTuple, slot, newslot, + (int) DML_INSERT, new_segid); + + /* Return DELETE first, INSERT on next call */ + node->processInsert = true; + return node->deleteTuple; + } + + case CMD_DELETE: + BuildSplitMergeTuple(node, node->ps.ps_ResultTupleSlot, + slot, NULL, (int) DML_DELETE, old_segid); + return node->ps.ps_ResultTupleSlot; + + case CMD_NOTHING: + break; /* fall through to pass-through below */ + + default: + elog(ERROR, "unknown action in MERGE WHEN MATCHED clause"); + } + + break; /* only first matching action */ + } + + /* No UPDATE/DELETE action matched - pass-through */ + BuildSplitMergeTuple(node, node->ps.ps_ResultTupleSlot, slot, NULL, + SPLITMERGE_ACTION_PASSTHROUGH, old_segid); + return node->ps.ps_ResultTupleSlot; +} + +/* + * ExecSplitMerge + * + * Main entry point. For each input tuple from the JOIN: + * - NOT MATCHED: compute target segment for INSERT routing + * - MATCHED + hasSplitUpdate: split UPDATE into DELETE + INSERT + * - MATCHED (no split): pass through */ static TupleTableSlot * ExecSplitMerge(PlanState *pstate) { SplitMergeState *node = castNode(SplitMergeState, pstate); - PlanState *outerNode = outerPlanState(node); + PlanState *outerNode = outerPlanState(node); SplitMerge *plannode = (SplitMerge *) node->ps.plan; ResultRelInfo *resultRelInfo = node->resultRelInfo + node->mt_lastResultIndex; Datum datum; bool isNull; - Oid resultoid; - - TupleTableSlot *slot = NULL; - TupleTableSlot *result = NULL; + TupleTableSlot *slot; Assert(outerNode != NULL); - slot = ExecProcNode(outerNode); + /* Return pending INSERT tuple from a previous split UPDATE */ + if (node->processInsert) + { + node->processInsert = false; + return node->insertTuple; + } + slot = ExecProcNode(outerNode); if (TupIsNull(slot)) - { return NULL; - } datum = ExecGetJunkAttribute(slot, resultRelInfo->ri_RowIdAttNo, &isNull); - /* ctid is NULL means that not matched, then check the insert action */ if (isNull) - result = MergeTupleTableSlot(slot, plannode, node, resultRelInfo); - else { - /* if partion table must switch resultRelInfo */ - if (AttributeNumberIsValid(node->mt_resultOidAttno)) - { - datum = ExecGetJunkAttribute(slot, node->mt_resultOidAttno, &isNull); - Assert(!isNull); - resultoid = DatumGetObjectId(datum); - if (resultoid != node->mt_lastResultOid) - resultRelInfo = MergeExecLookupResultRelByOid(node, resultoid, - false, true); - } - result = slot; + /* NOT MATCHED: compute target segment for INSERT routing */ + return MergeTupleTableSlot(slot, plannode, node, resultRelInfo); } - return result; -} + /* MATCHED: switch to correct partition if needed */ + resultRelInfo = SwitchResultRelForPartition(node, slot, resultRelInfo); + if (plannode->hasSplitUpdate) + return MergeMatchedSplitUpdate(slot, plannode, node, resultRelInfo); + /* No split update: pass through */ + return slot; +} /* * Initializes the tuple slots in a ResultRelInfo for any MERGE action. - * - * We mark 'projectNewInfoValid' even though the projections themselves - * are not initialized here. */ static void ExecInitMergeTupleSlots(SplitMergeState *mtstate, @@ -297,9 +421,51 @@ ExecInitMergeTupleSlots(SplitMergeState *mtstate, &estate->es_tupleTable); resultRelInfo->ri_projectNewInfoValid = true; } + /* - * Init SplitMerge Node. A memory context is created to hold Split Tuples. - * */ + * Build a TupleDesc for the root table's column layout from the plan's + * non-junk target list entries. Used for UPDATE projections so the result + * matches the SplitMerge output's first N columns regardless of which + * child partition is being updated. + */ +static TupleDesc +BuildRootUpdateTupleDesc(List *targetlist) +{ + TupleDesc desc; + ListCell *lc; + int nnonjunk = 0; + int col = 0; + + foreach(lc, targetlist) + { + TargetEntry *tle = (TargetEntry *) lfirst(lc); + if (!tle->resjunk) + nnonjunk++; + } + + if (nnonjunk == 0) + return NULL; + + desc = CreateTemplateTupleDesc(nnonjunk); + foreach(lc, targetlist) + { + TargetEntry *tle = (TargetEntry *) lfirst(lc); + if (!tle->resjunk) + { + col++; + TupleDescInitEntry(desc, col, tle->resname, + exprType((Node *) tle->expr), + exprTypmod((Node *) tle->expr), 0); + TupleDescInitEntryCollation(desc, col, + exprCollation((Node *) tle->expr)); + } + } + return desc; +} + +/* + * ExecInitSplitMerge + */ SplitMergeState* ExecInitSplitMerge(SplitMerge *node, EState *estate, int eflags) { @@ -307,62 +473,64 @@ ExecInitSplitMerge(SplitMerge *node, EState *estate, int eflags) ResultRelInfo *resultRelInfo; ExprContext *econtext; ListCell *lc; - int i; - + int i; + Plan *outerPlan = outerPlan(node); - /* Check for unsupported flags */ Assert(!(eflags & (EXEC_FLAG_BACKWARD | EXEC_FLAG_MARK | EXEC_FLAG_REWIND))); splitmergestate = makeNode(SplitMergeState); - splitmergestate->ps.plan = (Plan *)node; + splitmergestate->ps.plan = (Plan *) node; splitmergestate->ps.state = estate; splitmergestate->ps.ExecProcNode = ExecSplitMerge; + splitmergestate->processInsert = false; - /* - * then initialize outer plan - */ - Plan *outerPlan = outerPlan(node); outerPlanState(splitmergestate) = ExecInitNode(outerPlan, estate, eflags); - ExecAssignExprContext(estate, &splitmergestate->ps); econtext = splitmergestate->ps.ps_ExprContext; + /* Initialize result relations */ splitmergestate->nrel = list_length(node->resultRelations); - splitmergestate->resultRelInfo = (ResultRelInfo *)palloc(splitmergestate->nrel * sizeof(ResultRelInfo)); + splitmergestate->resultRelInfo = (ResultRelInfo *) + palloc(splitmergestate->nrel * sizeof(ResultRelInfo)); resultRelInfo = splitmergestate->resultRelInfo; - i = 0; foreach(lc, node->resultRelations) { Index resultRelation = lfirst_int(lc); - - ExecInitResultRelation(estate, resultRelInfo, resultRelation); - resultRelInfo->ri_RowIdAttNo = ExecFindJunkAttributeInTlist(outerPlan->targetlist, "ctid"); + ExecInitResultRelation(estate, resultRelInfo, resultRelation); + resultRelInfo->ri_RowIdAttNo = + ExecFindJunkAttributeInTlist(outerPlan->targetlist, "ctid"); if (!AttributeNumberIsValid(resultRelInfo->ri_RowIdAttNo)) elog(ERROR, "could not find junk ctid column"); - resultRelInfo++; - i++; } splitmergestate->mt_lastResultIndex = 0; splitmergestate->mt_lastResultOid = InvalidOid; + /* Build root-table-format slot for UPDATE projections (hasSplitUpdate only) */ + TupleTableSlot *rootUpdateSlot = NULL; + TupleDesc rootUpdateDesc = NULL; + if (node->hasSplitUpdate) + { + rootUpdateDesc = BuildRootUpdateTupleDesc(node->plan.targetlist); + if (rootUpdateDesc) + rootUpdateSlot = ExecInitExtraTupleSlot(estate, rootUpdateDesc, + &TTSOpsVirtual); + } + /* Initialize merge action states and projections */ i = 0; foreach(lc, node->mergeActionLists) { List *mergeActionList = lfirst(lc); - TupleDesc relationDesc; ListCell *l; resultRelInfo = splitmergestate->resultRelInfo + i; i++; - relationDesc = RelationGetDescr(resultRelInfo->ri_RelationDesc); - /* initialize slots for MERGE fetches from this rel */ if (unlikely(!resultRelInfo->ri_projectNewInfoValid)) ExecInitMergeTupleSlots(splitmergestate, resultRelInfo); @@ -370,24 +538,13 @@ ExecInitSplitMerge(SplitMerge *node, EState *estate, int eflags) { MergeAction *action = (MergeAction *) lfirst(l); MergeActionState *action_state; - TupleTableSlot *tgtslot; - TupleDesc tgtdesc; List **list; - /* - * Build action merge state for this rel. (For partitions, - * equivalent code exists in ExecInitPartitionInfo.) - */ action_state = makeNode(MergeActionState); action_state->mas_action = action; action_state->mas_whenqual = ExecInitQual((List *) action->qual, &splitmergestate->ps); - /* - * We create two lists - one for WHEN MATCHED actions and one for - * WHEN NOT MATCHED actions - and stick the MergeActionState into - * the appropriate list. - */ if (action_state->mas_action->matched) list = &resultRelInfo->ri_matchedMergeAction; else @@ -397,29 +554,22 @@ ExecInitSplitMerge(SplitMerge *node, EState *estate, int eflags) switch (action->commandType) { case CMD_INSERT: - - /* - * If the MERGE targets a partitioned table, any INSERT - * actions must be routed through it, not the child - * relations. Initialize the routing struct and the root - * table's "new" tuple slot for that, if not already done. - * The projection we prepare, for all relations, uses the - * root relation descriptor, and targets the plan's root - * slot. (This is consistent with the fact that we - * checked the plan output to match the root relation, - * above.) - */ - /* not partitioned? use the stock relation and slot */ - tgtslot = resultRelInfo->ri_newTupleSlot; - tgtdesc = RelationGetDescr(resultRelInfo->ri_RelationDesc); - action_state->mas_proj = ExecBuildProjectionInfo(action->targetList, econtext, - tgtslot, + resultRelInfo->ri_newTupleSlot, &splitmergestate->ps, - tgtdesc); + RelationGetDescr(resultRelInfo->ri_RelationDesc)); break; case CMD_UPDATE: + if (node->hasSplitUpdate && rootUpdateSlot != NULL) + { + action_state->mas_proj = + ExecBuildProjectionInfo(action->targetList, econtext, + rootUpdateSlot, + &splitmergestate->ps, + rootUpdateDesc); + } + break; case CMD_DELETE: case CMD_NOTHING: break; @@ -430,55 +580,68 @@ ExecInitSplitMerge(SplitMerge *node, EState *estate, int eflags) } } - /* - * Look up the positions of the gp_segment_id in the subplan's target - * list, and in the result. - */ + /* Look up junk attribute positions in subplan output */ splitmergestate->segid_attno = ExecFindJunkAttributeInTlist(outerPlan->targetlist, "gp_segment_id"); - splitmergestate->mt_resultOidAttno = ExecFindJunkAttributeInTlist(outerPlan->targetlist, "tableoid"); - - Assert(AttributeNumberIsValid(splitmergestate->mt_resultOidAttno) || splitmergestate->nrel == 1); - /* - * DML nodes do not project. - */ + Assert(AttributeNumberIsValid(splitmergestate->mt_resultOidAttno) || + splitmergestate->nrel == 1); + + /* Initialize hasSplitUpdate-specific state */ + if (node->hasSplitUpdate) + { + splitmergestate->action_attno = + ExecFindJunkAttributeInTlist(node->plan.targetlist, "DMLAction"); + Assert(AttributeNumberIsValid(splitmergestate->action_attno)); + + /* subplan_offset = N = total output columns - subplan columns - DMLAction */ + splitmergestate->subplan_offset = + list_length(node->plan.targetlist) - + list_length(outerPlan->targetlist) - 1; + Assert(splitmergestate->subplan_offset > 0); + + /* Dedicated slots for split DELETE + INSERT tuple pair */ + { + TupleDesc tupDesc = ExecTypeFromTL(node->plan.targetlist); + splitmergestate->deleteTuple = + ExecInitExtraTupleSlot(estate, tupDesc, &TTSOpsVirtual); + splitmergestate->insertTuple = + ExecInitExtraTupleSlot(estate, tupDesc, &TTSOpsVirtual); + } + } + else + { + splitmergestate->action_attno = InvalidAttrNumber; + splitmergestate->subplan_offset = 0; + } + ExecInitResultTupleSlotTL(&splitmergestate->ps, &TTSOpsVirtual); splitmergestate->ps.ps_ProjInfo = NULL; - /* - * Initialize for computing hash key - */ + /* Initialize hash for computing target segment */ if (node->numHashAttrs > 0) { splitmergestate->cdbhash = makeCdbHash(node->numHashSegments, - node->numHashAttrs, - node->hashFuncs); + node->numHashAttrs, + node->hashFuncs); } if (estate->es_instrument && (estate->es_instrument & INSTRUMENT_CDB)) - { splitmergestate->ps.cdbexplainbuf = makeStringInfo(); - } return splitmergestate; } -/* Release Resources Requested by SplitMerge node. */ +/* Release resources requested by SplitMerge node. */ void ExecEndSplitMerge(SplitMergeState *node) { - for (int i = 0; i < node->nrel; i++) { ResultRelInfo *resultRelInfo = node->resultRelInfo + i; - /* - * Cleanup the initialized batch slots. This only matters for FDWs - * with batching, but the other cases will have ri_NumSlotsInitialized - * == 0. - */ + for (int j = 0; j < resultRelInfo->ri_NumSlotsInitialized; j++) { ExecDropSingleTupleTableSlot(resultRelInfo->ri_Slots[j]); @@ -486,17 +649,14 @@ ExecEndSplitMerge(SplitMergeState *node) } } - /* - * Free the exprcontext - */ ExecFreeExprContext(&node->ps); - - - /* - * clean out the tuple table - */ + if (node->ps.ps_ResultTupleSlot) ExecClearTuple(node->ps.ps_ResultTupleSlot); + if (node->insertTuple) + ExecClearTuple(node->insertTuple); + if (node->deleteTuple) + ExecClearTuple(node->deleteTuple); + ExecEndNode(outerPlanState(node)); } - diff --git a/src/backend/nodes/copyfuncs.funcs.c b/src/backend/nodes/copyfuncs.funcs.c index e774ba2a549..6f312d17574 100644 --- a/src/backend/nodes/copyfuncs.funcs.c +++ b/src/backend/nodes/copyfuncs.funcs.c @@ -6475,6 +6475,8 @@ _copySplitMerge(const SplitMerge *from) COPY_SCALAR_FIELD(numHashSegments); COPY_NODE_FIELD(resultRelations); COPY_NODE_FIELD(mergeActionLists); + COPY_SCALAR_FIELD(hasSplitUpdate); + COPY_SCALAR_FIELD(rootResultRelation); return newnode; } diff --git a/src/backend/nodes/outfuncs_common.c b/src/backend/nodes/outfuncs_common.c index a4ba4c77e98..c518e38db0d 100644 --- a/src/backend/nodes/outfuncs_common.c +++ b/src/backend/nodes/outfuncs_common.c @@ -463,6 +463,8 @@ _outSplitMerge(StringInfo str, const SplitMerge *node) WRITE_OID_ARRAY(hashFuncs, node->numHashAttrs); WRITE_NODE_FIELD(resultRelations); WRITE_NODE_FIELD(mergeActionLists); + WRITE_BOOL_FIELD(hasSplitUpdate); + WRITE_UINT_FIELD(rootResultRelation); _outPlanInfo(str, (Plan *) node); } diff --git a/src/backend/nodes/readfast.c b/src/backend/nodes/readfast.c index 3222f8187ab..ff3cb5eaddf 100644 --- a/src/backend/nodes/readfast.c +++ b/src/backend/nodes/readfast.c @@ -1167,6 +1167,8 @@ _readSplitMerge(void) READ_NODE_FIELD(resultRelations); READ_NODE_FIELD(mergeActionLists); + READ_BOOL_FIELD(hasSplitUpdate); + READ_UINT_FIELD(rootResultRelation); ReadCommonPlan(&local_node->plan); diff --git a/src/backend/optimizer/plan/createplan.c b/src/backend/optimizer/plan/createplan.c index ad0e41ae065..585c1385068 100644 --- a/src/backend/optimizer/plan/createplan.c +++ b/src/backend/optimizer/plan/createplan.c @@ -3188,7 +3188,7 @@ create_modifytable_plan(PlannerInfo *root, ModifyTablePath *best_path) * root->processed_tlist. The code to create the Split Update node * takes care to label junk columns correctly, instead. */ - if (!best_path->splitUpdate) + if (!best_path->splitUpdate && !root->merge_need_split_update) apply_tlist_labeling(subplan->targetlist, root->processed_tlist); } @@ -3669,6 +3669,7 @@ create_splitmerge_plan(PlannerInfo *root, SplitMergePath *path) int lastresno; Oid *hashFuncs; int i; + int nrels; // RelOptInfo *relOptInfo = root->simple_rel_array[linitial_int(path->resultRelations)]; @@ -3703,20 +3704,92 @@ create_splitmerge_plan(PlannerInfo *root, SplitMergePath *path) copy_generic_path_info(&splitmerge->plan, (Path *) path); - lc = list_head(subplan->targetlist); lastresno = 0; - /* Copy all attributes. */ - for (; lc != NULL; lc = lnext(subplan->targetlist, lc)) + if (path->hasSplitUpdate) { - TargetEntry *tle = (TargetEntry *) lfirst(lc); - TargetEntry *newtle; + /* + * When hasSplitUpdate, build a targetlist with target table columns + * as non-junk entries (positions 1..N), followed by all subplan + * entries as junk, plus a DMLAction junk column. + * + * This allows ModifyTable to use ExecGetInsertNewTuple to extract + * the target table columns for DML_INSERT tuples. + */ + int natts = resultDesc->natts; - newtle = makeTargetEntry(tle->expr, - ++lastresno, - tle->resname, - tle->resjunk); - splitmerge->plan.targetlist = lappend(splitmerge->plan.targetlist, newtle); + /* + * First: target table columns as non-junk. + * Use Var nodes (not Const) so that ExecInitInsertProjection + * on the Motion output can extract actual values at runtime. + * These Vars use a dummy varno (0) that set_splitmerge_tlist_references + * will convert to OUTER_VAR with varattno = tle->resno. + */ + for (i = 0; i < natts; i++) + { + Form_pg_attribute attr = &resultDesc->attrs[i]; + TargetEntry *tle; + + if (attr->attisdropped) + { + Const *nullConst = makeConst(INT4OID, -1, InvalidOid, + sizeof(int32), (Datum) 0, + true, true); + tle = makeTargetEntry((Expr *) nullConst, + ++lastresno, + pstrdup(NameStr(attr->attname)), + false); + } + else + { + Var *var = makeVar(0, /* dummy varno, fixed in setrefs */ + (AttrNumber) (i + 1), + attr->atttypid, + attr->atttypmod, + attr->attcollation, + 0); + tle = makeTargetEntry((Expr *) var, + ++lastresno, + pstrdup(NameStr(attr->attname)), + false); + } + splitmerge->plan.targetlist = lappend(splitmerge->plan.targetlist, tle); + } + + /* Then: all subplan entries as junk */ + foreach(lc, subplan->targetlist) + { + TargetEntry *tle = (TargetEntry *) lfirst(lc); + TargetEntry *newtle; + + newtle = makeTargetEntry(tle->expr, + ++lastresno, + tle->resname, + true); /* mark as junk */ + splitmerge->plan.targetlist = lappend(splitmerge->plan.targetlist, newtle); + } + + /* Finally: DMLAction junk column */ + splitmerge->plan.targetlist = lappend(splitmerge->plan.targetlist, + makeTargetEntry((Expr *) makeNode(DMLActionExpr), + ++lastresno, + "DMLAction", + true)); + } + else + { + /* Without hasSplitUpdate: copy subplan targetlist as-is */ + foreach(lc, subplan->targetlist) + { + TargetEntry *tle = (TargetEntry *) lfirst(lc); + TargetEntry *newtle; + + newtle = makeTargetEntry(tle->expr, + ++lastresno, + tle->resname, + tle->resjunk); + splitmerge->plan.targetlist = lappend(splitmerge->plan.targetlist, newtle); + } } /* Look up the right hash functions for the hash expressions */ @@ -3737,6 +3810,50 @@ create_splitmerge_plan(PlannerInfo *root, SplitMergePath *path) splitmerge->hashFuncs = hashFuncs; splitmerge->numHashSegments = cdbpolicy->numsegments; + splitmerge->resultRelations = path->resultRelations; + splitmerge->hasSplitUpdate = path->hasSplitUpdate; + splitmerge->rootResultRelation = relOptInfo->relid; + + /* + * Always use the root table's action lists (not per-partition adjusted + * lists). SplitMerge's hashAttnos use root table attribute numbers + * (from cdbpolicy->attrs), so the INSERT projection must produce values + * in root table column order for evalHashKey to read the correct columns. + * Per-partition lists from adjust_appendrel_attrs_multilevel reorder + * resnos to match child partition layout, which would cause hash + * computation on wrong columns for partitions with reordered columns. + */ + nrels = list_length(path->resultRelations); + splitmerge->mergeActionLists = NIL; + for (i = 0; i < nrels; i++) + splitmerge->mergeActionLists = lappend(splitmerge->mergeActionLists, + copyObject(root->parse->mergeActionList)); + + /* + * For split-update MERGE, expand UPDATE action targetlists to include + * all target table columns (not just the SET columns). SplitMerge needs + * complete rows to project INSERT tuples. Uses root table RTI for Vars + * so they match the subplan output. + */ + if (path->hasSplitUpdate) + { + ListCell *lca; + foreach(lca, splitmerge->mergeActionLists) + { + List *actionList = lfirst(lca); + ListCell *lc2; + foreach(lc2, actionList) + { + MergeAction *action = (MergeAction *) lfirst(lc2); + if (action->commandType == CMD_UPDATE) + action->targetList = expand_insert_targetlist(root, + action->targetList, + resultRel, + relOptInfo->relid); + } + } + } + relation_close(resultRel, NoLock); /* @@ -3745,9 +3862,6 @@ create_splitmerge_plan(PlannerInfo *root, SplitMergePath *path) */ root->numMotions++; - splitmerge->mergeActionLists = path->mergeActionLists; - splitmerge->resultRelations = path->resultRelations; - return (Plan *) splitmerge; } @@ -8908,7 +9022,7 @@ cdbpathtoplan_create_motion_plan(PlannerInfo *root, * the DMLActionExpr column, so we cannot apply the * labeling here even if we wanted. */ - if (!IsA(subplan, SplitUpdate)) + if (!IsA(subplan, SplitUpdate) && !IsA(subplan, SplitMerge)) apply_tlist_labeling(subplan->targetlist, root->processed_tlist); segmentid_tle = find_junk_tle(subplan->targetlist, "gp_segment_id"); diff --git a/src/backend/optimizer/plan/setrefs.c b/src/backend/optimizer/plan/setrefs.c index 67d03390c61..99c5b1372a6 100644 --- a/src/backend/optimizer/plan/setrefs.c +++ b/src/backend/optimizer/plan/setrefs.c @@ -186,6 +186,7 @@ static Node *convert_combining_aggrefs(Node *node, void *context); static Node *convert_deduplicated_aggrefs(Node *node, void *context); static void set_dummy_tlist_references(Plan *plan, int rtoffset); static void set_splitupdate_tlist_references(Plan *plan, int rtoffset); +static void set_splitmerge_tlist_references(Plan *plan, int rtoffset); static indexed_tlist *build_tlist_index(List *tlist); static Var *search_indexed_tlist_for_var(Var *var, indexed_tlist *itlist, @@ -1630,8 +1631,61 @@ set_plan_refs(PlannerInfo *root, Plan *plan, int rtoffset) set_splitupdate_tlist_references(plan, rtoffset); break; case T_SplitMerge: - /* mergeActionLists will be process in T_ModifyTable */ - set_dummy_tlist_references(plan, rtoffset); + { + SplitMerge *sm = (SplitMerge *) plan; + Plan *childplan = plan->lefttree; + + set_splitmerge_tlist_references(plan, rtoffset); + + /* + * Fix SplitMerge's own mergeActionLists to reference the + * child plan's output positions. This is separate from + * ModifyTable's mergeActionLists which reference the + * Motion output. + */ + if (sm->mergeActionLists != NIL && childplan != NULL) + { + indexed_tlist *itlist; + ListCell *lca; + + itlist = build_tlist_index(childplan->targetlist); + + foreach(lca, sm->mergeActionLists) + { + List *mergeActionList = lfirst(lca); + ListCell *lc2; + + /* + * SplitMerge always uses root table action lists, + * so action Vars reference root table attributes. + * Use root table RTI as acceptable_rel. + */ + Index acceptable_rel = sm->rootResultRelation; + + foreach(lc2, mergeActionList) + { + MergeAction *action = (MergeAction *) lfirst(lc2); + + action->targetList = fix_join_expr(root, + action->targetList, + NULL, itlist, + acceptable_rel, + rtoffset, + NRM_EQUAL, + NUM_EXEC_TLIST(plan)); + + action->qual = (Node *) fix_join_expr(root, + (List *) action->qual, + NULL, itlist, + acceptable_rel, + rtoffset, + NRM_EQUAL, + NUM_EXEC_QUAL(plan)); + } + } + pfree(itlist); + } + } break; default: elog(ERROR, "unrecognized node type: %d", @@ -3185,7 +3239,91 @@ set_splitupdate_tlist_references(Plan *plan, int rtoffset) /* We don't touch plan->qual here */ } +/* + * set_splitmerge_tlist_references + * Like set_splitupdate_tlist_references, but handles the case where + * hasSplitUpdate prepends N target-table columns before the child plan's + * columns. Non-junk entries (positions 1..N) become OUTER_VAR references + * to themselves (tle->resno), allowing the Motion node to pass them + * through and ModifyTable to extract them via projection. + * Junk entries (positions N+1..N+M) become OUTER_VAR references to + * the child plan's output positions (tracked via child_resno). + * Const entries (dropped columns) and DMLActionExpr are kept as-is. + */ +static void +set_splitmerge_tlist_references(Plan *plan, int rtoffset) +{ + List *output_targetlist; + ListCell *l; + int child_resno = 0; + output_targetlist = NIL; + foreach(l, plan->targetlist) + { + TargetEntry *tle = (TargetEntry *) lfirst(l); + Var *oldvar = (Var *) tle->expr; + Var *newvar; + + if (IsA(tle->expr, DMLActionExpr)) + { + output_targetlist = lappend(output_targetlist, tle); + continue; + } + else if (IsA(tle->expr, Const)) + { + /* Dropped column placeholder - keep as Const */ + output_targetlist = lappend(output_targetlist, tle); + continue; + } + + if (!tle->resjunk) + { + /* + * Non-junk entry (target table column, position 1..N). + * Create OUTER_VAR reference to tle->resno (self-referencing). + * At runtime, SplitMerge fills these with actual INSERT values. + * The Motion node passes them through, and ModifyTable extracts them. + */ + newvar = makeVar(OUTER_VAR, + tle->resno, + exprType((Node *) oldvar), + exprTypmod((Node *) oldvar), + exprCollation((Node *) oldvar), + 0); + newvar->varnosyn = 0; + newvar->varattnosyn = 0; + } + else + { + /* + * Junk entry (subplan column, position N+1..N+M). + * Create OUTER_VAR reference to the child plan's output position. + */ + child_resno++; + newvar = makeVar(OUTER_VAR, + child_resno, + exprType((Node *) oldvar), + exprTypmod((Node *) oldvar), + exprCollation((Node *) oldvar), + 0); + if (IsA(oldvar, Var)) + { + newvar->varnosyn = oldvar->varnosyn + rtoffset; + newvar->varattnosyn = oldvar->varattnosyn; + } + else + { + newvar->varnosyn = 0; + newvar->varattnosyn = 0; + } + } + + tle = flatCopyTargetEntry(tle); + tle->expr = (Expr *) newvar; + output_targetlist = lappend(output_targetlist, tle); + } + plan->targetlist = output_targetlist; +} /* * build_tlist_index --- build an index data structure for a child tlist diff --git a/src/backend/optimizer/prep/preptlist.c b/src/backend/optimizer/prep/preptlist.c index ca2dd95f284..5625cae3854 100644 --- a/src/backend/optimizer/prep/preptlist.c +++ b/src/backend/optimizer/prep/preptlist.c @@ -60,7 +60,7 @@ static List *supplement_simply_updatable_targetlist(PlannerInfo *root, List *range_table, List *tlist); -static List *expand_insert_targetlist(PlannerInfo *root, List *tlist, Relation rel, Index split_update_result_relation); +List *expand_insert_targetlist(PlannerInfo *root, List *tlist, Relation rel, Index split_update_result_relation); @@ -143,15 +143,48 @@ preprocess_targetlist(PlannerInfo *root) } else if (command_type == CMD_MERGE) { - /* update distributed column in merge is not supported now */ + /* Check if any MERGE UPDATE action modifies distribution key columns */ foreach(lc, parse->mergeActionList) { MergeAction *action = lfirst(lc); - if(action->commandType == CMD_UPDATE) + if (action->commandType == CMD_UPDATE && + check_splitupdate(action->targetList, result_relation, target_relation)) { - if(check_splitupdate(action->targetList, result_relation, target_relation)) - ereport(ERROR, (errcode(ERRCODE_GP_FEATURE_NOT_YET), - errmsg("cannot update column in merge with distributed column"))); + root->merge_need_split_update = true; + break; + } + } + + /* + * When merge_need_split_update, the SplitMerge node needs all target + * table columns in the subplan output so that the UPDATE projection + * can read unchanged columns from the old row. Add all non-dropped + * user columns of the target table to the subplan targetlist. + */ + if (root->merge_need_split_update) + { + int natts = RelationGetNumberOfAttributes(target_relation); + + for (int attno = 1; attno <= natts; attno++) + { + Form_pg_attribute att = TupleDescAttr(target_relation->rd_att, + attno - 1); + Var *var; + + if (att->attisdropped) + continue; + + var = makeVar(result_relation, attno, + att->atttypid, att->atttypmod, + att->attcollation, 0); + + if (tlist_member((Expr *) var, tlist)) + continue; /* already present */ + + tlist = lappend(tlist, + makeTargetEntry((Expr *) var, + list_length(tlist) + 1, + NULL, true)); } } } @@ -225,8 +258,9 @@ preprocess_targetlist(PlannerInfo *root) Var *var = (Var *) lfirst(l2); TargetEntry *tle; - if (IsA(var, Var) && var->varno == result_relation) - continue; /* don't need it */ + if (IsA(var, Var) && var->varno == result_relation && + !root->merge_need_split_update) + continue; /* don't need it unless split update */ if (tlist_member((Expr *) var, tlist)) continue; /* already got it */ @@ -411,7 +445,7 @@ extract_update_targetlist_colnos(List *tlist, bool reorder_resno) * Once upon a time we also did more or less this with UPDATE targetlists, * but now this code is only applied to INSERT targetlists. */ -static List * +List * expand_insert_targetlist(PlannerInfo *root, List *tlist, Relation rel, Index split_update_result_relation) { List *new_tlist = NIL; diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h index d6b96ebe88e..ce0c00347be 100644 --- a/src/include/nodes/execnodes.h +++ b/src/include/nodes/execnodes.h @@ -3413,6 +3413,13 @@ typedef struct SplitMergeState AttrNumber mt_resultOidAttno; + /* Fields for split update in MERGE */ + TupleTableSlot *insertTuple; /* pending INSERT tuple for split update */ + TupleTableSlot *deleteTuple; /* DELETE tuple for split update */ + bool processInsert; /* true = next call returns insertTuple */ + AttrNumber action_attno; /* attribute number of DMLAction column in output */ + int subplan_offset; /* number of target table columns prepended to output */ + } SplitMergeState; /* diff --git a/src/include/nodes/pathnodes.h b/src/include/nodes/pathnodes.h index 9face599e2a..d5556d66c4b 100644 --- a/src/include/nodes/pathnodes.h +++ b/src/include/nodes/pathnodes.h @@ -663,6 +663,8 @@ struct PlannerInfo int upd_del_replicated_table; bool is_split_update; /* true if UPDATE that modifies * distribution key columns */ + bool merge_need_split_update; /* true if MERGE has UPDATE that + * modifies distribution key columns */ bool is_correlated_subplan; /* true for correlated subqueries nested within subplans */ int numPureOrderedAggs; /* CDB: number that use ORDER BY/WITHIN GROUP, not counting DISTINCT */ @@ -2808,6 +2810,7 @@ typedef struct SplitMergePath List *resultRelations; List *mergeActionLists; /* per-target-table lists of actions for * MERGE */ + bool hasSplitUpdate; /* true if UPDATE modifies distribution key */ } SplitMergePath; /* diff --git a/src/include/nodes/plannodes.h b/src/include/nodes/plannodes.h index 00ebe7f493e..f759a976ec6 100644 --- a/src/include/nodes/plannodes.h +++ b/src/include/nodes/plannodes.h @@ -1956,6 +1956,8 @@ typedef struct SplitMerge List *mergeActionLists; /* per-target-table lists of actions for * MERGE */ + bool hasSplitUpdate; /* true if UPDATE modifies distribution key */ + Index rootResultRelation; /* root table RTI for partitioned tables */ } SplitMerge; /* diff --git a/src/include/optimizer/prep.h b/src/include/optimizer/prep.h index 0526d7c0526..dcc247f6256 100644 --- a/src/include/optimizer/prep.h +++ b/src/include/optimizer/prep.h @@ -18,6 +18,7 @@ #include "nodes/pathnodes.h" #include "nodes/plannodes.h" +#include "utils/relcache.h" /* @@ -44,6 +45,10 @@ extern void preprocess_targetlist(PlannerInfo *root); extern List *extract_update_targetlist_colnos(List *tlist, bool reorder_resno); +extern List *expand_insert_targetlist(PlannerInfo *root, List *tlist, + Relation rel, + Index split_update_result_relation); + extern PlanRowMark *get_plan_rowmark(List *rowmarks, Index rtindex); /* diff --git a/src/test/regress/expected/merge.out b/src/test/regress/expected/merge.out index d6568e4b271..18a8a6a3eb0 100644 --- a/src/test/regress/expected/merge.out +++ b/src/test/regress/expected/merge.out @@ -1847,7 +1847,7 @@ MERGE INTO pa_target t ON t.tid = s.sid AND t.tid IN (1,2,3,4) WHEN MATCHED THEN UPDATE SET tid = tid - 1; -ERROR: cannot update column in merge with distributed column +ERROR: new row violates row-level security policy for table "pa_target" ROLLBACK; DROP TABLE pa_source; DROP TABLE pa_target CASCADE; @@ -2233,6 +2233,259 @@ DROP TABLE test1; DROP TABLE target, target2; DROP TABLE source, source2; DROP FUNCTION merge_trigfunc(); +-- Test MERGE with distribution key updates (split-update) +-- These tests verify that MERGE ... WHEN MATCHED THEN UPDATE SET dist_col = ... +-- works correctly by using the SplitMerge mechanism (DELETE old + INSERT new). +-- Basic: update distribution key with constant +CREATE TABLE merge_dist_t (id int, val text) DISTRIBUTED BY (id); +CREATE TABLE merge_dist_s (id int, val text) DISTRIBUTED BY (id); +INSERT INTO merge_dist_t VALUES (1, 'old'); +INSERT INTO merge_dist_s VALUES (1, 'new'); +-- Check segment before merge +SELECT gp_segment_id, * FROM merge_dist_t ORDER BY id; + gp_segment_id | id | val +---------------+----+----- + 1 | 1 | old +(1 row) + +MERGE INTO merge_dist_t t USING merge_dist_s s ON t.id = s.id +WHEN MATCHED THEN UPDATE SET id = 101, val = s.val; +-- Check segment after merge: should match direct INSERT of 101 +SELECT gp_segment_id, * FROM merge_dist_t ORDER BY id; + gp_segment_id | id | val +---------------+-----+----- + 0 | 101 | new +(1 row) + +INSERT INTO merge_dist_t VALUES (101, 'direct'); +SELECT gp_segment_id, * FROM merge_dist_t WHERE id = 101 ORDER BY val; + gp_segment_id | id | val +---------------+-----+-------- + 0 | 101 | direct + 0 | 101 | new +(2 rows) + +DELETE FROM merge_dist_t WHERE val = 'direct'; +-- EXPLAIN VERBOSE: SplitMerge plan for simple table dist key update +EXPLAIN (VERBOSE, COSTS OFF) +MERGE INTO merge_dist_t t USING merge_dist_s s ON t.id = s.id +WHEN MATCHED THEN UPDATE SET id = t.id + 100, val = s.val +WHEN NOT MATCHED THEN INSERT VALUES (s.id, s.val); + QUERY PLAN +------------------------------------------------------------------------------------------------------------------- + Merge on public.merge_dist_t t + -> Explicit Redistribute Motion 3:3 (slice1; segments: 3) + Output: t.id, t.val, t.id, t.val, t.ctid, t.gp_segment_id, s.val, s.id, s.ctid, (DMLAction) + -> Split Merge + Output: t.id, t.val, t.id, t.val, t.ctid, t.gp_segment_id, s.val, s.id, s.ctid, DMLAction + -> Hash Left Join + Output: t.id, t.val, t.ctid, t.gp_segment_id, s.val, s.id, s.ctid + Hash Cond: (s.id = t.id) + -> Seq Scan on public.merge_dist_s s + Output: s.val, s.id, s.ctid + -> Hash + Output: t.id, t.val, t.ctid, t.gp_segment_id + -> Seq Scan on public.merge_dist_t t + Output: t.id, t.val, t.ctid, t.gp_segment_id + Optimizer: Postgres query optimizer +(15 rows) + +-- Update distribution key with expression referencing old row +TRUNCATE merge_dist_t, merge_dist_s; +INSERT INTO merge_dist_t VALUES (1, 'a'), (2, 'b'); +INSERT INTO merge_dist_s VALUES (1, 'x'), (2, 'y'); +MERGE INTO merge_dist_t t USING merge_dist_s s ON t.id = s.id +WHEN MATCHED THEN UPDATE SET id = t.id + 100, val = s.val; +SELECT * FROM merge_dist_t ORDER BY id; + id | val +-----+----- + 101 | x + 102 | y +(2 rows) + +-- Mixed: MATCHED update dist key + NOT MATCHED insert +TRUNCATE merge_dist_t, merge_dist_s; +INSERT INTO merge_dist_t VALUES (1, 'old'); +INSERT INTO merge_dist_s VALUES (1, 'new'), (2, 'two'); +MERGE INTO merge_dist_t t USING merge_dist_s s ON t.id = s.id +WHEN MATCHED THEN UPDATE SET id = t.id + 100, val = s.val +WHEN NOT MATCHED THEN INSERT VALUES (s.id, s.val); +SELECT * FROM merge_dist_t ORDER BY id; + id | val +-----+----- + 2 | two + 101 | new +(2 rows) + +-- Multi-column distribution key: only update one dist column +CREATE TABLE merge_dist_mc (a int, b int, val text) DISTRIBUTED BY (a, b); +CREATE TABLE merge_dist_mc_s (a int, b int, val text) DISTRIBUTED BY (a); +INSERT INTO merge_dist_mc VALUES (1, 1, 'old'); +INSERT INTO merge_dist_mc_s VALUES (1, 1, 'new'); +MERGE INTO merge_dist_mc t USING merge_dist_mc_s s ON t.a = s.a AND t.b = s.b +WHEN MATCHED THEN UPDATE SET a = t.a + 100, val = s.val; +SELECT * FROM merge_dist_mc ORDER BY a; + a | b | val +-----+---+----- + 101 | 1 | new +(1 row) + +-- Conditional WHEN: first clause updates dist key, second does not +TRUNCATE merge_dist_t, merge_dist_s; +INSERT INTO merge_dist_t VALUES (1, 'one'), (2, 'two'); +INSERT INTO merge_dist_s VALUES (1, 'new1'), (2, 'new2'); +MERGE INTO merge_dist_t t USING merge_dist_s s ON t.id = s.id +WHEN MATCHED AND t.id = 1 THEN UPDATE SET id = t.id + 100, val = s.val +WHEN MATCHED THEN UPDATE SET val = s.val; +SELECT * FROM merge_dist_t ORDER BY id; + id | val +-----+------ + 2 | new2 + 101 | new1 +(2 rows) + +-- Normal MERGE still works alongside split-update tests +TRUNCATE merge_dist_t, merge_dist_s; +INSERT INTO merge_dist_t VALUES (1, 'old'); +INSERT INTO merge_dist_s VALUES (1, 'updated'), (2, 'inserted'); +MERGE INTO merge_dist_t t USING merge_dist_s s ON t.id = s.id +WHEN MATCHED THEN UPDATE SET val = s.val +WHEN NOT MATCHED THEN INSERT VALUES (s.id, s.val); +SELECT * FROM merge_dist_t ORDER BY id; + id | val +----+---------- + 1 | updated + 2 | inserted +(2 rows) + +DROP TABLE merge_dist_t, merge_dist_s; +DROP TABLE merge_dist_mc, merge_dist_mc_s; +-- Partitioned table: MERGE with distribution key update +CREATE TABLE merge_dist_pa_s (sid integer, delta float) DISTRIBUTED BY (sid); +INSERT INTO merge_dist_pa_s VALUES (1, 5), (2, 10), (4, 30); +CREATE TABLE merge_dist_pa_t (tid integer, balance float, val text) + PARTITION BY LIST (tid) DISTRIBUTED BY (tid); +CREATE TABLE merge_dist_pa_p1 (tid integer, balance float, val text) DISTRIBUTED BY (tid); +CREATE TABLE merge_dist_pa_p2 (balance float, tid integer, val text) DISTRIBUTED BY (tid); +CREATE TABLE merge_dist_pa_p4 (extraid text, tid integer, balance float, val text) DISTRIBUTED BY (tid); +ALTER TABLE merge_dist_pa_p4 DROP COLUMN extraid; +ALTER TABLE merge_dist_pa_t ATTACH PARTITION merge_dist_pa_p1 FOR VALUES IN (0, 1); +ALTER TABLE merge_dist_pa_t ATTACH PARTITION merge_dist_pa_p2 FOR VALUES IN (2, 3); +ALTER TABLE merge_dist_pa_t ATTACH PARTITION merge_dist_pa_p4 FOR VALUES IN (4, 5); +INSERT INTO merge_dist_pa_t VALUES (1, 100, 'p1'), (2, 200, 'p2'), (4, 400, 'p4'); +-- EXPLAIN VERBOSE: SplitMerge plan for partitioned table dist key update +EXPLAIN (VERBOSE, COSTS OFF) +MERGE INTO merge_dist_pa_t t USING merge_dist_pa_s s ON t.tid = s.sid +WHEN MATCHED THEN UPDATE SET tid = tid - 1 +WHEN NOT MATCHED THEN INSERT VALUES (s.sid, s.delta, 'new'); + QUERY PLAN +--------------------------------------------------------------------------------------------------------------------------------------------------------- + Merge on public.merge_dist_pa_t t + Merge on public.merge_dist_pa_p1 t_1 + Merge on public.merge_dist_pa_p2 t_2 + Merge on public.merge_dist_pa_p4 t_3 + -> Explicit Redistribute Motion 3:3 (slice1; segments: 3) + Output: t.tid, t.balance, t.val, t.tid, t.balance, t.val, s.sid, s.delta, s.ctid, t.tableoid, t.ctid, t.gp_segment_id, (DMLAction) + -> Split Merge + Output: t.tid, t.balance, t.val, t.tid, t.balance, t.val, s.sid, s.delta, s.ctid, t.tableoid, t.ctid, t.gp_segment_id, DMLAction + -> Hash Right Join + Output: t.tid, t.balance, t.val, s.sid, s.delta, s.ctid, t.tableoid, t.ctid, t.gp_segment_id + Hash Cond: (t.tid = s.sid) + -> Append + Partition Selectors: $1 + -> Seq Scan on public.merge_dist_pa_p1 t_1 + Output: t_1.tid, t_1.balance, t_1.val, t_1.tableoid, t_1.ctid, t_1.gp_segment_id + -> Seq Scan on public.merge_dist_pa_p2 t_2 + Output: t_2.tid, t_2.balance, t_2.val, t_2.tableoid, t_2.ctid, t_2.gp_segment_id + -> Seq Scan on public.merge_dist_pa_p4 t_3 + Output: t_3.tid, t_3.balance, t_3.val, t_3.tableoid, t_3.ctid, t_3.gp_segment_id + -> Hash + Output: s.sid, s.delta, s.ctid + -> Partition Selector (selector id: $1) + Output: s.sid, s.delta, s.ctid + -> Seq Scan on public.merge_dist_pa_s s + Output: s.sid, s.delta, s.ctid + Optimizer: Postgres query optimizer +(26 rows) + +-- Update distribution key on partitioned table with reordered/dropped columns +MERGE INTO merge_dist_pa_t t USING merge_dist_pa_s s ON t.tid = s.sid +WHEN MATCHED THEN UPDATE SET tid = tid - 1; +SELECT gp_segment_id, tableoid::regclass, * FROM merge_dist_pa_t ORDER BY tid; + gp_segment_id | tableoid | tid | balance | val +---------------+------------------+-----+---------+----- + 1 | merge_dist_pa_p1 | 0 | 100 | p1 + 1 | merge_dist_pa_p1 | 1 | 200 | p2 + 0 | merge_dist_pa_p2 | 3 | 400 | p4 +(3 rows) + +-- Verify segment placement matches direct INSERT +CREATE TABLE merge_dist_pa_verify (tid integer, balance float, val text) DISTRIBUTED BY (tid); +INSERT INTO merge_dist_pa_verify VALUES (0, 100, 'p1'), (1, 200, 'p2'), (3, 400, 'p4'); +SELECT gp_segment_id, * FROM merge_dist_pa_verify ORDER BY tid; + gp_segment_id | tid | balance | val +---------------+-----+---------+----- + 1 | 0 | 100 | p1 + 1 | 1 | 200 | p2 + 0 | 3 | 400 | p4 +(3 rows) + +DROP TABLE merge_dist_pa_verify; +-- Mixed: dist key update + insert on partitioned table +TRUNCATE merge_dist_pa_t; +INSERT INTO merge_dist_pa_t VALUES (1, 100, 'p1'), (2, 200, 'p2'); +MERGE INTO merge_dist_pa_t t USING merge_dist_pa_s s ON t.tid = s.sid +WHEN MATCHED THEN UPDATE SET tid = tid - 1 +WHEN NOT MATCHED THEN INSERT VALUES (s.sid, s.delta, 'new'); +SELECT * FROM merge_dist_pa_t ORDER BY tid; +NOTICE: One or more columns in the following table(s) do not have statistics: merge_dist_pa_t +HINT: For non-partitioned tables, run analyze <table_name>(<column_list>). For partitioned tables, run analyze rootpartition <table_name>(<column_list>). See log for columns missing statistics. + tid | balance | val +-----+---------+----- + 0 | 100 | p1 + 1 | 200 | p2 + 4 | 30 | new +(3 rows) + +DROP TABLE merge_dist_pa_t CASCADE; +DROP TABLE merge_dist_pa_s; +-- MERGE with dist key update + INSERT trigger (not UPDATE trigger) +-- INSERT triggers should fire; UPDATE triggers would block split merge +CREATE TABLE merge_trig_dist (id int, val text) DISTRIBUTED BY (id); +CREATE TABLE merge_trig_dist_s (id int, val text) DISTRIBUTED BY (id); +CREATE OR REPLACE FUNCTION merge_trig_dist_fn() RETURNS trigger AS $$ +BEGIN + RAISE NOTICE '% trigger on %, id=%, val=%', TG_OP, TG_TABLE_NAME, NEW.id, NEW.val; + RETURN NEW; +END; +$$ LANGUAGE plpgsql; +-- Only INSERT trigger, no UPDATE trigger — split merge is allowed +CREATE TRIGGER merge_trig_bi BEFORE INSERT ON merge_trig_dist + FOR EACH ROW EXECUTE FUNCTION merge_trig_dist_fn(); +INSERT INTO merge_trig_dist VALUES (1, 'old'); +NOTICE: INSERT trigger on merge_trig_dist, id=1, val=old +INSERT INTO merge_trig_dist_s VALUES (1, 'new'), (2, 'two'); +-- dist key update + insert, INSERT trigger should fire for new rows +MERGE INTO merge_trig_dist t USING merge_trig_dist_s s ON t.id = s.id +WHEN MATCHED THEN UPDATE SET id = t.id + 100, val = s.val +WHEN NOT MATCHED THEN INSERT VALUES (s.id, s.val); +NOTICE: INSERT trigger on merge_trig_dist, id=2, val=two +SELECT * FROM merge_trig_dist ORDER BY id; + id | val +-----+----- + 2 | two + 101 | new +(2 rows) + +-- Now add UPDATE trigger — split merge should be blocked +CREATE TRIGGER merge_trig_bu BEFORE UPDATE ON merge_trig_dist + FOR EACH ROW EXECUTE FUNCTION merge_trig_dist_fn(); +MERGE INTO merge_trig_dist t USING merge_trig_dist_s s ON t.id = s.id +WHEN MATCHED THEN UPDATE SET id = t.id + 100, val = s.val +WHEN NOT MATCHED THEN INSERT VALUES (s.id, s.val); +ERROR: UPDATE on distributed key column not allowed on relation with update triggers +DROP TABLE merge_trig_dist, merge_trig_dist_s; +DROP FUNCTION merge_trig_dist_fn; REVOKE ALL ON SCHEMA public FROM regress_merge_privs; DROP USER regress_merge_privs; DROP USER regress_merge_no_privs; diff --git a/src/test/regress/expected/privileges.out b/src/test/regress/expected/privileges.out index a625d39ea88..aff2b89051c 100644 --- a/src/test/regress/expected/privileges.out +++ b/src/test/regress/expected/privileges.out @@ -1025,7 +1025,7 @@ WHEN MATCHED THEN UPDATE SET b = s.b, a = t.a + 1 WHEN NOT MATCHED THEN INSERT VALUES (a, b); -ERROR: cannot update column in merge with distributed column +ERROR: permission denied for table mtarget -- fail (no SELECT on t.b) MERGE INTO mtarget t USING msource s ON t.a = s.a WHEN MATCHED AND t.b IS NOT NULL THEN diff --git a/src/test/regress/expected/triggers.out b/src/test/regress/expected/triggers.out index d4e702ed7cd..9c786db7528 100644 --- a/src/test/regress/expected/triggers.out +++ b/src/test/regress/expected/triggers.out @@ -2221,7 +2221,7 @@ ERROR: UPDATE on distributed key column not allowed on relation with update tri -- update action in merge should behave the same merge into parted_trig using (select 1) as ss on true when matched and a = 2 then update set a = 1; -ERROR: cannot update column in merge with distributed column +ERROR: UPDATE on distributed key column not allowed on relation with update triggers drop table parted_trig; -- Verify propagation of trigger arguments to partitions create table parted_trig (a int) partition by list (a); diff --git a/src/test/regress/sql/merge.sql b/src/test/regress/sql/merge.sql index 4e99ef36bfd..a22a0805ec7 100644 --- a/src/test/regress/sql/merge.sql +++ b/src/test/regress/sql/merge.sql @@ -1555,6 +1555,158 @@ DROP TABLE test1; DROP TABLE target, target2; DROP TABLE source, source2; DROP FUNCTION merge_trigfunc(); + +-- Test MERGE with distribution key updates (split-update) +-- These tests verify that MERGE ... WHEN MATCHED THEN UPDATE SET dist_col = ... +-- works correctly by using the SplitMerge mechanism (DELETE old + INSERT new). + +-- Basic: update distribution key with constant +CREATE TABLE merge_dist_t (id int, val text) DISTRIBUTED BY (id); +CREATE TABLE merge_dist_s (id int, val text) DISTRIBUTED BY (id); +INSERT INTO merge_dist_t VALUES (1, 'old'); +INSERT INTO merge_dist_s VALUES (1, 'new'); +-- Check segment before merge +SELECT gp_segment_id, * FROM merge_dist_t ORDER BY id; +MERGE INTO merge_dist_t t USING merge_dist_s s ON t.id = s.id +WHEN MATCHED THEN UPDATE SET id = 101, val = s.val; +-- Check segment after merge: should match direct INSERT of 101 +SELECT gp_segment_id, * FROM merge_dist_t ORDER BY id; +INSERT INTO merge_dist_t VALUES (101, 'direct'); +SELECT gp_segment_id, * FROM merge_dist_t WHERE id = 101 ORDER BY val; +DELETE FROM merge_dist_t WHERE val = 'direct'; + +-- EXPLAIN VERBOSE: SplitMerge plan for simple table dist key update +EXPLAIN (VERBOSE, COSTS OFF) +MERGE INTO merge_dist_t t USING merge_dist_s s ON t.id = s.id +WHEN MATCHED THEN UPDATE SET id = t.id + 100, val = s.val +WHEN NOT MATCHED THEN INSERT VALUES (s.id, s.val); + +-- Update distribution key with expression referencing old row +TRUNCATE merge_dist_t, merge_dist_s; +INSERT INTO merge_dist_t VALUES (1, 'a'), (2, 'b'); +INSERT INTO merge_dist_s VALUES (1, 'x'), (2, 'y'); +MERGE INTO merge_dist_t t USING merge_dist_s s ON t.id = s.id +WHEN MATCHED THEN UPDATE SET id = t.id + 100, val = s.val; +SELECT * FROM merge_dist_t ORDER BY id; + +-- Mixed: MATCHED update dist key + NOT MATCHED insert +TRUNCATE merge_dist_t, merge_dist_s; +INSERT INTO merge_dist_t VALUES (1, 'old'); +INSERT INTO merge_dist_s VALUES (1, 'new'), (2, 'two'); +MERGE INTO merge_dist_t t USING merge_dist_s s ON t.id = s.id +WHEN MATCHED THEN UPDATE SET id = t.id + 100, val = s.val +WHEN NOT MATCHED THEN INSERT VALUES (s.id, s.val); +SELECT * FROM merge_dist_t ORDER BY id; + +-- Multi-column distribution key: only update one dist column +CREATE TABLE merge_dist_mc (a int, b int, val text) DISTRIBUTED BY (a, b); +CREATE TABLE merge_dist_mc_s (a int, b int, val text) DISTRIBUTED BY (a); +INSERT INTO merge_dist_mc VALUES (1, 1, 'old'); +INSERT INTO merge_dist_mc_s VALUES (1, 1, 'new'); +MERGE INTO merge_dist_mc t USING merge_dist_mc_s s ON t.a = s.a AND t.b = s.b +WHEN MATCHED THEN UPDATE SET a = t.a + 100, val = s.val; +SELECT * FROM merge_dist_mc ORDER BY a; + +-- Conditional WHEN: first clause updates dist key, second does not +TRUNCATE merge_dist_t, merge_dist_s; +INSERT INTO merge_dist_t VALUES (1, 'one'), (2, 'two'); +INSERT INTO merge_dist_s VALUES (1, 'new1'), (2, 'new2'); +MERGE INTO merge_dist_t t USING merge_dist_s s ON t.id = s.id +WHEN MATCHED AND t.id = 1 THEN UPDATE SET id = t.id + 100, val = s.val +WHEN MATCHED THEN UPDATE SET val = s.val; +SELECT * FROM merge_dist_t ORDER BY id; + +-- Normal MERGE still works alongside split-update tests +TRUNCATE merge_dist_t, merge_dist_s; +INSERT INTO merge_dist_t VALUES (1, 'old'); +INSERT INTO merge_dist_s VALUES (1, 'updated'), (2, 'inserted'); +MERGE INTO merge_dist_t t USING merge_dist_s s ON t.id = s.id +WHEN MATCHED THEN UPDATE SET val = s.val +WHEN NOT MATCHED THEN INSERT VALUES (s.id, s.val); +SELECT * FROM merge_dist_t ORDER BY id; + +DROP TABLE merge_dist_t, merge_dist_s; +DROP TABLE merge_dist_mc, merge_dist_mc_s; + +-- Partitioned table: MERGE with distribution key update +CREATE TABLE merge_dist_pa_s (sid integer, delta float) DISTRIBUTED BY (sid); +INSERT INTO merge_dist_pa_s VALUES (1, 5), (2, 10), (4, 30); + +CREATE TABLE merge_dist_pa_t (tid integer, balance float, val text) + PARTITION BY LIST (tid) DISTRIBUTED BY (tid); +CREATE TABLE merge_dist_pa_p1 (tid integer, balance float, val text) DISTRIBUTED BY (tid); +CREATE TABLE merge_dist_pa_p2 (balance float, tid integer, val text) DISTRIBUTED BY (tid); +CREATE TABLE merge_dist_pa_p4 (extraid text, tid integer, balance float, val text) DISTRIBUTED BY (tid); +ALTER TABLE merge_dist_pa_p4 DROP COLUMN extraid; + +ALTER TABLE merge_dist_pa_t ATTACH PARTITION merge_dist_pa_p1 FOR VALUES IN (0, 1); +ALTER TABLE merge_dist_pa_t ATTACH PARTITION merge_dist_pa_p2 FOR VALUES IN (2, 3); +ALTER TABLE merge_dist_pa_t ATTACH PARTITION merge_dist_pa_p4 FOR VALUES IN (4, 5); + +INSERT INTO merge_dist_pa_t VALUES (1, 100, 'p1'), (2, 200, 'p2'), (4, 400, 'p4'); + +-- EXPLAIN VERBOSE: SplitMerge plan for partitioned table dist key update +EXPLAIN (VERBOSE, COSTS OFF) +MERGE INTO merge_dist_pa_t t USING merge_dist_pa_s s ON t.tid = s.sid +WHEN MATCHED THEN UPDATE SET tid = tid - 1 +WHEN NOT MATCHED THEN INSERT VALUES (s.sid, s.delta, 'new'); + +-- Update distribution key on partitioned table with reordered/dropped columns +MERGE INTO merge_dist_pa_t t USING merge_dist_pa_s s ON t.tid = s.sid +WHEN MATCHED THEN UPDATE SET tid = tid - 1; +SELECT gp_segment_id, tableoid::regclass, * FROM merge_dist_pa_t ORDER BY tid; +-- Verify segment placement matches direct INSERT +CREATE TABLE merge_dist_pa_verify (tid integer, balance float, val text) DISTRIBUTED BY (tid); +INSERT INTO merge_dist_pa_verify VALUES (0, 100, 'p1'), (1, 200, 'p2'), (3, 400, 'p4'); +SELECT gp_segment_id, * FROM merge_dist_pa_verify ORDER BY tid; +DROP TABLE merge_dist_pa_verify; + +-- Mixed: dist key update + insert on partitioned table +TRUNCATE merge_dist_pa_t; +INSERT INTO merge_dist_pa_t VALUES (1, 100, 'p1'), (2, 200, 'p2'); +MERGE INTO merge_dist_pa_t t USING merge_dist_pa_s s ON t.tid = s.sid +WHEN MATCHED THEN UPDATE SET tid = tid - 1 +WHEN NOT MATCHED THEN INSERT VALUES (s.sid, s.delta, 'new'); +SELECT * FROM merge_dist_pa_t ORDER BY tid; + +DROP TABLE merge_dist_pa_t CASCADE; +DROP TABLE merge_dist_pa_s; + +-- MERGE with dist key update + INSERT trigger (not UPDATE trigger) +-- INSERT triggers should fire; UPDATE triggers would block split merge +CREATE TABLE merge_trig_dist (id int, val text) DISTRIBUTED BY (id); +CREATE TABLE merge_trig_dist_s (id int, val text) DISTRIBUTED BY (id); +CREATE OR REPLACE FUNCTION merge_trig_dist_fn() RETURNS trigger AS $$ +BEGIN + RAISE NOTICE '% trigger on %, id=%, val=%', TG_OP, TG_TABLE_NAME, NEW.id, NEW.val; + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +-- Only INSERT trigger, no UPDATE trigger — split merge is allowed +CREATE TRIGGER merge_trig_bi BEFORE INSERT ON merge_trig_dist + FOR EACH ROW EXECUTE FUNCTION merge_trig_dist_fn(); + +INSERT INTO merge_trig_dist VALUES (1, 'old'); +INSERT INTO merge_trig_dist_s VALUES (1, 'new'), (2, 'two'); + +-- dist key update + insert, INSERT trigger should fire for new rows +MERGE INTO merge_trig_dist t USING merge_trig_dist_s s ON t.id = s.id +WHEN MATCHED THEN UPDATE SET id = t.id + 100, val = s.val +WHEN NOT MATCHED THEN INSERT VALUES (s.id, s.val); +SELECT * FROM merge_trig_dist ORDER BY id; + +-- Now add UPDATE trigger — split merge should be blocked +CREATE TRIGGER merge_trig_bu BEFORE UPDATE ON merge_trig_dist + FOR EACH ROW EXECUTE FUNCTION merge_trig_dist_fn(); + +MERGE INTO merge_trig_dist t USING merge_trig_dist_s s ON t.id = s.id +WHEN MATCHED THEN UPDATE SET id = t.id + 100, val = s.val +WHEN NOT MATCHED THEN INSERT VALUES (s.id, s.val); + +DROP TABLE merge_trig_dist, merge_trig_dist_s; +DROP FUNCTION merge_trig_dist_fn; + REVOKE ALL ON SCHEMA public FROM regress_merge_privs; DROP USER regress_merge_privs; DROP USER regress_merge_no_privs; --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
