external
/
citus
mirror of https://github.com/citusdata/citus.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

1) Restrict MERGE command INSERT to the source's distribution column 

Fixes #6672

2) Move all MERGE related routines to a new file merge_planner.c
merge_locks
Teja Mupparti 2023-01-31 18:23:44 -08:00
parent 529577209d
commit f7d838add0
7 changed files with 752 additions and 371 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

519
src/backend/distributed/planner/merge_planner.c Normal file
View File

@ -0,0 +1,519 @@
/*-------------------------------------------------------------------------
*
* merge_planner.c
*
* This file contains functions to help plan MERGE queries.
*
* Copyright (c) Citus Data, Inc.
*
*-------------------------------------------------------------------------
*/
#include <stddef.h>
#include "postgres.h"
#include "parser/parsetree.h"
#include "distributed/pg_version_constants.h"
#include "distributed/merge_planner.h"
#include "distributed/multi_logical_optimizer.h"
#include "distributed/multi_router_planner.h"
#include "distributed/listutils.h"
static bool QueryHasMergeCommand(Query *queryTree);
static DeferredErrorMessage * ErrorIfMergeHasUnsupportedTables(Query *parse,
List *rangeTableList,
PlannerRestrictionContext *
restrictionContext);
static DeferredErrorMessage * ErrorIfDistTablesNotColocated(Query *parse,
List *distTablesList,
PlannerRestrictionContext *
plannerRestrictionContext);
static bool IsPartitionColumnInMergeSource(Expr *columnExpression, Query *query, bool
skipOuterVars);
#if PG_VERSION_NUM >= PG_VERSION_15
static DeferredErrorMessage * InsertPartitionColumnMatchesSource(Query *query, RangeTblEntry *resultRte);
#endif
/*
* MergeQuerySupported does check for a MERGE command in the query, if it finds
* one, it will verify the below criteria
* - Supported tables and combinations in ErrorIfMergeHasUnsupportedTables
* - Distributed tables requirements in ErrorIfDistTablesNotColocated
* - Checks target-lists and functions-in-quals in TargetlistAndFunctionsSupported
*/
DeferredErrorMessage *
MergeQuerySupported(Query *originalQuery,
PlannerRestrictionContext *plannerRestrictionContext)
{
/* For non-MERGE commands it's a no-op */
if (!QueryHasMergeCommand(originalQuery))
{
return NULL;
}
List *rangeTableList = ExtractRangeTableEntryList(originalQuery);
RangeTblEntry *resultRte = ExtractResultRelationRTE(originalQuery);
/*
* Fast path queries cannot have merge command, and we prevent the remaining here.
* In Citus we have limited support for MERGE, it's allowed only if all
* the tables(target, source or any CTE) tables are are local i.e. a
* combination of Citus local and Non-Citus tables (regular Postgres tables)
* or distributed tables with some restrictions, please see header of routine
* ErrorIfDistTablesNotColocated for details.
*/
DeferredErrorMessage *deferredError =
ErrorIfMergeHasUnsupportedTables(originalQuery,
rangeTableList,
plannerRestrictionContext);
if (deferredError)
{
return deferredError;
}
Oid resultRelationId = resultRte->relid;
deferredError =
TargetlistAndFunctionsSupported(resultRelationId,
originalQuery->jointree,
originalQuery->jointree->quals,
originalQuery->targetList,
originalQuery->commandType,
originalQuery->returningList);
if (deferredError)
{
return deferredError;
}
#if PG_VERSION_NUM >= PG_VERSION_15
/*
* MERGE is a special case where we have multiple modify statements
* within itself. Check each INSERT/UPDATE/DELETE individually.
*/
MergeAction *action = NULL;
foreach_ptr(action, originalQuery->mergeActionList)
{
Assert(originalQuery->returningList == NULL);
deferredError =
TargetlistAndFunctionsSupported(resultRelationId,
originalQuery->jointree,
action->qual,
action->targetList,
action->commandType,
originalQuery->returningList);
if (deferredError)
{
return deferredError;
}
}
deferredError =
InsertPartitionColumnMatchesSource(originalQuery, resultRte);
if (deferredError)
{
return deferredError;
}
#endif
return NULL;
}
/*
* IsMergeAllowedOnRelation takes a relation entry and checks if MERGE command is
* permitted on special relations, such as materialized view, returns true only if
* it's a "source" relation.
*/
bool
IsMergeAllowedOnRelation(Query *parse, RangeTblEntry *rte)
{
if (!IsMergeQuery(parse))
{
return false;
}
RangeTblEntry *targetRte = rt_fetch(parse->resultRelation, parse->rtable);
/* Is it a target relation? */
if (targetRte->relid == rte->relid)
{
return false;
}
return true;
}
/*
* ErrorIfDistTablesNotColocated Checks to see if
*
* - There are a minimum of two distributed tables (source and a target).
* - All the distributed tables are indeed colocated.
* - MERGE relations are joined on the distribution column
* MERGE .. USING .. ON target.dist_key = source.dist_key
*
* If any of the conditions are not met, it raises an exception.
*/
static DeferredErrorMessage *
ErrorIfDistTablesNotColocated(Query *parse, List *distTablesList,
PlannerRestrictionContext *plannerRestrictionContext)
{
/* All MERGE tables must be distributed */
if (list_length(distTablesList) < 2)
{
return DeferredError(ERRCODE_FEATURE_NOT_SUPPORTED,
"For MERGE command, both the source and target "
"must be distributed", NULL, NULL);
}
/* All distributed tables must be colocated */
if (!AllRelationsInListColocated(distTablesList, RANGETABLE_ENTRY))
{
return DeferredError(ERRCODE_FEATURE_NOT_SUPPORTED,
"For MERGE command, all the distributed tables "
"must be colocated", NULL, NULL);
}
/* Are source and target tables joined on distribution column? */
if (!RestrictionEquivalenceForPartitionKeys(plannerRestrictionContext))
{
return DeferredError(ERRCODE_FEATURE_NOT_SUPPORTED,
"MERGE command is only supported when distributed "
"tables are joined on their distribution column",
NULL, NULL);
}
return NULL;
}
/*
* ErrorIfMergeHasUnsupportedTables checks if all the tables(target, source or any CTE
* present) in the MERGE command are local i.e. a combination of Citus local and Non-Citus
* tables (regular Postgres tables), or distributed tables with some restrictions, please
* see header of routine ErrorIfDistTablesNotColocated for details, raises an exception
* for all other combinations.
*/
static DeferredErrorMessage *
ErrorIfMergeHasUnsupportedTables(Query *parse, List *rangeTableList,
PlannerRestrictionContext *restrictionContext)
{
List *distTablesList = NIL;
bool foundLocalTables = false;
RangeTblEntry *rangeTableEntry = NULL;
foreach_ptr(rangeTableEntry, rangeTableList)
{
Oid relationId = rangeTableEntry->relid;
switch (rangeTableEntry->rtekind)
{
case RTE_RELATION:
{
/* Check the relation type */
break;
}
case RTE_SUBQUERY:
case RTE_FUNCTION:
case RTE_TABLEFUNC:
case RTE_VALUES:
case RTE_JOIN:
case RTE_CTE:
{
/* Skip them as base table(s) will be checked */
continue;
}
/*
* RTE_NAMEDTUPLESTORE is typically used in ephmeral named relations,
* such as, trigger data; until we find a genuine use case, raise an
* exception.
* RTE_RESULT is a node added by the planner and we shouldn't
* encounter it in the parse tree.
*/
case RTE_NAMEDTUPLESTORE:
case RTE_RESULT:
{
return DeferredError(ERRCODE_FEATURE_NOT_SUPPORTED,
"MERGE command is not supported with "
"Tuplestores and results",
NULL, NULL);
}
default:
{
return DeferredError(ERRCODE_FEATURE_NOT_SUPPORTED,
"MERGE command: Unrecognized range table entry.",
NULL, NULL);
}
}
/* RTE Relation can be of various types, check them now */
/* skip the regular views as they are replaced with subqueries */
if (rangeTableEntry->relkind == RELKIND_VIEW)
{
continue;
}
if (rangeTableEntry->relkind == RELKIND_MATVIEW ||
rangeTableEntry->relkind == RELKIND_FOREIGN_TABLE)
{
/* Materialized view or Foreign table as target is not allowed */
if (IsMergeAllowedOnRelation(parse, rangeTableEntry))
{
/* Non target relation is ok */
continue;
}
else
{
/* Usually we don't reach this exception as the Postgres parser catches it */
StringInfo errorMessage = makeStringInfo();
appendStringInfo(errorMessage,
"MERGE command is not allowed on "
"relation type(relkind:%c)", rangeTableEntry->relkind);
return DeferredError(ERRCODE_FEATURE_NOT_SUPPORTED, errorMessage->data,
NULL, NULL);
}
}
if (rangeTableEntry->relkind != RELKIND_RELATION &&
rangeTableEntry->relkind != RELKIND_PARTITIONED_TABLE)
{
StringInfo errorMessage = makeStringInfo();
appendStringInfo(errorMessage, "Unexpected table type(relkind:%c) "
"in MERGE command", rangeTableEntry->relkind);
return DeferredError(ERRCODE_FEATURE_NOT_SUPPORTED, errorMessage->data,
NULL, NULL);
}
Assert(rangeTableEntry->relid != 0);
/* Reference tables are not supported yet */
if (IsCitusTableType(relationId, REFERENCE_TABLE))
{
return DeferredError(ERRCODE_FEATURE_NOT_SUPPORTED,
"MERGE command is not supported on reference "
"tables yet", NULL, NULL);
}
/* Append/Range tables are not supported */
if (IsCitusTableType(relationId, APPEND_DISTRIBUTED) ||
IsCitusTableType(relationId, RANGE_DISTRIBUTED))
{
return DeferredError(ERRCODE_FEATURE_NOT_SUPPORTED,
"For MERGE command, all the distributed tables "
"must be colocated, for append/range distribution, "
"colocation is not supported", NULL,
"Consider using hash distribution instead");
}
/*
* For now, save all distributed tables, later (below) we will
* check for supported combination(s).
*/
if (IsCitusTableType(relationId, DISTRIBUTED_TABLE))
{
distTablesList = lappend(distTablesList, rangeTableEntry);
continue;
}
/* Regular Postgres tables and Citus local tables are allowed */
if (!IsCitusTable(relationId) ||
IsCitusTableType(relationId, CITUS_LOCAL_TABLE))
{
foundLocalTables = true;
continue;
}
/* Any other Citus table type missing ? */
}
/* Ensure all tables are indeed local */
if (foundLocalTables && list_length(distTablesList) == 0)
{
/* All the tables are local, supported */
return NULL;
}
else if (foundLocalTables && list_length(distTablesList) > 0)
{
return DeferredError(ERRCODE_FEATURE_NOT_SUPPORTED,
"MERGE command is not supported with "
"combination of distributed/local tables yet",
NULL, NULL);
}
/* Ensure all distributed tables are indeed co-located */
return ErrorIfDistTablesNotColocated(parse, distTablesList, restrictionContext);
}
/*
* QueryHasMergeCommand walks over the query tree and returns false if there
* is no Merge command (e.g., CMD_MERGE), true otherwise.
*/
static bool
QueryHasMergeCommand(Query *queryTree)
{
/* function is void for pre-15 versions of Postgres */
#if PG_VERSION_NUM < PG_VERSION_15
return false;
#else
/*
* Postgres currently doesn't support Merge queries inside subqueries and
* ctes, but lets be defensive and do query tree walk anyway.
*
* We do not call this path for fast-path queries to avoid this additional
* overhead.
*/
if (!ContainsMergeCommandWalker((Node *) queryTree))
{
/* No MERGE found */
return false;
}
return true;
#endif
}
/*
* IsPartitionColumnInMerge returns true if the given column is a partition column.
* The function uses FindReferencedTableColumn to find the original relation
* id and column that the column expression refers to. It then checks whether
* that column is a partition column of the relation.
*
* Also, the function returns always false for reference tables given that
* reference tables do not have partition column.
*
* If skipOuterVars is true, then it doesn't process the outervars.
*/
bool
IsPartitionColumnInMergeSource(Expr *columnExpression, Query *query, bool skipOuterVars)
{
bool isPartitionColumn = false;
Var *column = NULL;
RangeTblEntry *relationRTE = NULL;
/* ParentQueryList is same as the original query for MERGE */
FindReferencedTableColumn(columnExpression, list_make1(query), query, &column,
&relationRTE,
skipOuterVars);
Oid relationId = relationRTE ? relationRTE->relid : InvalidOid;
if (relationId != InvalidOid && column != NULL)
{
Var *partitionColumn = DistPartitionKey(relationId);
/* not all distributed tables have partition column */
if (partitionColumn != NULL && column->varattno == partitionColumn->varattno)
{
isPartitionColumn = true;
}
}
return isPartitionColumn;
}
#if PG_VERSION_NUM >= PG_VERSION_15
/*
* InsertPartitionColumnMatchesSource check to see if MERGE is inserting a
* value into the target which is not from the source table, if so, it
* raises an exception.
* Note: Inserting random values other than the joined column values will
* result in unexpected behaviour of rows ending up in incorrect shards.
*/
static DeferredErrorMessage *
InsertPartitionColumnMatchesSource(Query *query, RangeTblEntry *resultRte)
{
if (!IsCitusTableType(resultRte->relid, DISTRIBUTED_TABLE))
{
return NULL;
}
bool foundDistributionColumn = false;
MergeAction *action = NULL;
foreach_ptr(action, query->mergeActionList)
{
/* Skip MATCHED clauses */
if (action->matched)
{
continue;
}
/* NOT MATCHED can have either INSERT or DO NOTHING */
if (action->commandType == CMD_NOTHING)
{
return NULL;
}
if (action->targetList == NIL)
{
/* INSERT DEFAULT VALUES is not allowed */
return DeferredError(ERRCODE_FEATURE_NOT_SUPPORTED,
"cannot perform MERGE INSERT with DEFAULTS",
NULL, NULL);
}
Assert(action->commandType == CMD_INSERT);
Var *targetKey = PartitionColumn(resultRte->relid, 1);
TargetEntry *targetEntry = NULL;
foreach_ptr(targetEntry, action->targetList)
{
if (targetEntry->resjunk)
{
continue;
}
AttrNumber originalAttrNo = targetEntry->resno;
/* skip processing of target table non-partition columns */
if (originalAttrNo != targetKey->varattno)
{
continue;
}
foundDistributionColumn = true;
if (targetEntry->expr->type == T_Var)
{
if (IsPartitionColumnInMergeSource(targetEntry->expr, query, true))
{
return NULL;
}
else
{
return DeferredError(ERRCODE_FEATURE_NOT_SUPPORTED,
"MERGE INSERT must use the source table "
"distribution column value",
NULL, NULL);
}
}
else
{
return DeferredError(ERRCODE_FEATURE_NOT_SUPPORTED,
"MERGE INSERT must refer a source column "
"for distribution column ",
NULL, NULL);
}
}
if (!foundDistributionColumn)
{
return DeferredError(ERRCODE_FEATURE_NOT_SUPPORTED,
"MERGE INSERT must have distribution column as value",
NULL, NULL);
}
}
return NULL;
}
#endif

350
src/backend/distributed/planner/multi_router_planner.c
View File

@ -33,6 +33,7 @@
#include "distributed/intermediate_result_pruning.h"
#include "distributed/metadata_utility.h"
#include "distributed/coordinator_protocol.h"
#include "distributed/merge_planner.h"
#include "distributed/metadata_cache.h"
#include "distributed/multi_executor.h"
#include "distributed/multi_join_order.h"
@ -179,12 +180,8 @@ static void ReorderTaskPlacementsByTaskAssignmentPolicy(Job *job,
static bool ModifiesLocalTableWithRemoteCitusLocalTable(List *rangeTableList);
static DeferredErrorMessage * DeferErrorIfUnsupportedLocalTableJoin(List *rangeTableList);
static bool IsLocallyAccessibleCitusLocalTable(Oid relationId);
static DeferredErrorMessage * TargetlistAndFunctionsSupported(Oid resultRelationId,
FromExpr *joinTree,
Node *quals,
List *targetList,
CmdType commandType,
List *returningList);
/*
* CreateRouterPlan attempts to create a router executor plan for the given
* SELECT statement. ->planningError is set if planning fails.
@ -521,7 +518,7 @@ IsTidColumn(Node *node)
* updating distribution column, etc.
* Note: This subset of checks are repeated for each MERGE modify action.
*/
static DeferredErrorMessage *
DeferredErrorMessage *
TargetlistAndFunctionsSupported(Oid resultRelationId, FromExpr *joinTree, Node *quals,
List *targetList,
CmdType commandType, List *returningList)
@ -904,85 +901,6 @@ NodeIsFieldStore(Node *node)
}
/*
* MergeQuerySupported does check for a MERGE command in the query, if it finds
* one, it will verify the below criteria
* - Supported tables and combinations in ErrorIfMergeHasUnsupportedTables
* - Distributed tables requirements in ErrorIfDistTablesNotColocated
* - Checks target-lists and functions-in-quals in TargetlistAndFunctionsSupported
*/
static DeferredErrorMessage *
MergeQuerySupported(Query *originalQuery,
PlannerRestrictionContext *plannerRestrictionContext)
{
/* For non-MERGE commands it's a no-op */
if (!QueryHasMergeCommand(originalQuery))
{
return NULL;
}
List *rangeTableList = ExtractRangeTableEntryList(originalQuery);
RangeTblEntry *resultRte = ExtractResultRelationRTE(originalQuery);
/*
* Fast path queries cannot have merge command, and we prevent the remaining here.
* In Citus we have limited support for MERGE, it's allowed only if all
* the tables(target, source or any CTE) tables are are local i.e. a
* combination of Citus local and Non-Citus tables (regular Postgres tables)
* or distributed tables with some restrictions, please see header of routine
* ErrorIfDistTablesNotColocated for details.
*/
DeferredErrorMessage *deferredError =
ErrorIfMergeHasUnsupportedTables(originalQuery,
rangeTableList,
plannerRestrictionContext);
if (deferredError)
{
return deferredError;
}
Oid resultRelationId = resultRte->relid;
deferredError =
TargetlistAndFunctionsSupported(resultRelationId,
originalQuery->jointree,
originalQuery->jointree->quals,
originalQuery->targetList,
originalQuery->commandType,
originalQuery->returningList);
if (deferredError)
{
return deferredError;
}
#if PG_VERSION_NUM >= PG_VERSION_15
/*
* MERGE is a special case where we have multiple modify statements
* within itself. Check each INSERT/UPDATE/DELETE individually.
*/
MergeAction *action = NULL;
foreach_ptr(action, originalQuery->mergeActionList)
{
Assert(originalQuery->returningList == NULL);
deferredError =
TargetlistAndFunctionsSupported(resultRelationId,
originalQuery->jointree,
action->qual,
action->targetList,
action->commandType,
originalQuery->returningList);
if (deferredError)
{
return deferredError;
}
}
#endif
return NULL;
}
/*
* ModifyQuerySupported returns NULL if the query only contains supported
* features, otherwise it returns an error description.
@ -4057,263 +3975,3 @@ CompareInsertValuesByShardId(const void *leftElement, const void *rightElement)
}
}
}
/*
* IsMergeAllowedOnRelation takes a relation entry and checks if MERGE command is
* permitted on special relations, such as materialized view, returns true only if
* it's a "source" relation.
*/
bool
IsMergeAllowedOnRelation(Query *parse, RangeTblEntry *rte)
{
if (!IsMergeQuery(parse))
{
return false;
}
RangeTblEntry *targetRte = rt_fetch(parse->resultRelation, parse->rtable);
/* Is it a target relation? */
if (targetRte->relid == rte->relid)
{
return false;
}
return true;
}
/*
* ErrorIfDistTablesNotColocated Checks to see if
*
* - There are a minimum of two distributed tables (source and a target).
* - All the distributed tables are indeed colocated.
* - MERGE relations are joined on the distribution column
* MERGE .. USING .. ON target.dist_key = source.dist_key
*
* If any of the conditions are not met, it raises an exception.
*/
static DeferredErrorMessage *
ErrorIfDistTablesNotColocated(Query *parse, List *distTablesList,
PlannerRestrictionContext *plannerRestrictionContext)
{
/* All MERGE tables must be distributed */
if (list_length(distTablesList) < 2)
{
return DeferredError(ERRCODE_FEATURE_NOT_SUPPORTED,
"For MERGE command, both the source and target "
"must be distributed", NULL, NULL);
}
/* All distributed tables must be colocated */
if (!AllRelationsInListColocated(distTablesList, RANGETABLE_ENTRY))
{
return DeferredError(ERRCODE_FEATURE_NOT_SUPPORTED,
"For MERGE command, all the distributed tables "
"must be colocated", NULL, NULL);
}
/* Are source and target tables joined on distribution column? */
if (!RestrictionEquivalenceForPartitionKeys(plannerRestrictionContext))
{
return DeferredError(ERRCODE_FEATURE_NOT_SUPPORTED,
"MERGE command is only supported when distributed "
"tables are joined on their distribution column",
NULL, NULL);
}
return NULL;
}
/*
* ErrorIfMergeHasUnsupportedTables checks if all the tables(target, source or any CTE
* present) in the MERGE command are local i.e. a combination of Citus local and Non-Citus
* tables (regular Postgres tables), or distributed tables with some restrictions, please
* see header of routine ErrorIfDistTablesNotColocated for details, raises an exception
* for all other combinations.
*/
static DeferredErrorMessage *
ErrorIfMergeHasUnsupportedTables(Query *parse, List *rangeTableList,
PlannerRestrictionContext *restrictionContext)
{
List *distTablesList = NIL;
bool foundLocalTables = false;
RangeTblEntry *rangeTableEntry = NULL;
foreach_ptr(rangeTableEntry, rangeTableList)
{
Oid relationId = rangeTableEntry->relid;
switch (rangeTableEntry->rtekind)
{
case RTE_RELATION:
{
/* Check the relation type */
break;
}
case RTE_SUBQUERY:
case RTE_FUNCTION:
case RTE_TABLEFUNC:
case RTE_VALUES:
case RTE_JOIN:
case RTE_CTE:
{
/* Skip them as base table(s) will be checked */
continue;
}
/*
* RTE_NAMEDTUPLESTORE is typically used in ephmeral named relations,
* such as, trigger data; until we find a genuine use case, raise an
* exception.
* RTE_RESULT is a node added by the planner and we shouldn't
* encounter it in the parse tree.
*/
case RTE_NAMEDTUPLESTORE:
case RTE_RESULT:
{
return DeferredError(ERRCODE_FEATURE_NOT_SUPPORTED,
"MERGE command is not supported with "
"Tuplestores and results",
NULL, NULL);
}
default:
{
return DeferredError(ERRCODE_FEATURE_NOT_SUPPORTED,
"MERGE command: Unrecognized range table entry.",
NULL, NULL);
}
}
/* RTE Relation can be of various types, check them now */
/* skip the regular views as they are replaced with subqueries */
if (rangeTableEntry->relkind == RELKIND_VIEW)
{
continue;
}
if (rangeTableEntry->relkind == RELKIND_MATVIEW ||
rangeTableEntry->relkind == RELKIND_FOREIGN_TABLE)
{
/* Materialized view or Foreign table as target is not allowed */
if (IsMergeAllowedOnRelation(parse, rangeTableEntry))
{
/* Non target relation is ok */
continue;
}
else
{
/* Usually we don't reach this exception as the Postgres parser catches it */
StringInfo errorMessage = makeStringInfo();
appendStringInfo(errorMessage,
"MERGE command is not allowed on "
"relation type(relkind:%c)", rangeTableEntry->relkind);
return DeferredError(ERRCODE_FEATURE_NOT_SUPPORTED, errorMessage->data,
NULL, NULL);
}
}
if (rangeTableEntry->relkind != RELKIND_RELATION &&
rangeTableEntry->relkind != RELKIND_PARTITIONED_TABLE)
{
StringInfo errorMessage = makeStringInfo();
appendStringInfo(errorMessage, "Unexpected table type(relkind:%c) "
"in MERGE command", rangeTableEntry->relkind);
return DeferredError(ERRCODE_FEATURE_NOT_SUPPORTED, errorMessage->data,
NULL, NULL);
}
Assert(rangeTableEntry->relid != 0);
/* Reference tables are not supported yet */
if (IsCitusTableType(relationId, REFERENCE_TABLE))
{
return DeferredError(ERRCODE_FEATURE_NOT_SUPPORTED,
"MERGE command is not supported on reference "
"tables yet", NULL, NULL);
}
/* Append/Range tables are not supported */
if (IsCitusTableType(relationId, APPEND_DISTRIBUTED) ||
IsCitusTableType(relationId, RANGE_DISTRIBUTED))
{
return DeferredError(ERRCODE_FEATURE_NOT_SUPPORTED,
"For MERGE command, all the distributed tables "
"must be colocated, for append/range distribution, "
"colocation is not supported", NULL,
"Consider using hash distribution instead");
}
/*
* For now, save all distributed tables, later (below) we will
* check for supported combination(s).
*/
if (IsCitusTableType(relationId, DISTRIBUTED_TABLE))
{
distTablesList = lappend(distTablesList, rangeTableEntry);
continue;
}
/* Regular Postgres tables and Citus local tables are allowed */
if (!IsCitusTable(relationId) ||
IsCitusTableType(relationId, CITUS_LOCAL_TABLE))
{
foundLocalTables = true;
continue;
}
/* Any other Citus table type missing ? */
}
/* Ensure all tables are indeed local */
if (foundLocalTables && list_length(distTablesList) == 0)
{
/* All the tables are local, supported */
return NULL;
}
else if (foundLocalTables && list_length(distTablesList) > 0)
{
return DeferredError(ERRCODE_FEATURE_NOT_SUPPORTED,
"MERGE command is not supported with "
"combination of distributed/local tables yet",
NULL, NULL);
}
/* Ensure all distributed tables are indeed co-located */
return ErrorIfDistTablesNotColocated(parse, distTablesList, restrictionContext);
}
/*
* QueryHasMergeCommand walks over the query tree and returns false if there
* is no Merge command (e.g., CMD_MERGE), true otherwise.
*/
static bool
QueryHasMergeCommand(Query *queryTree)
{
/* function is void for pre-15 versions of Postgres */
#if PG_VERSION_NUM < PG_VERSION_15
return false;
#else
/*
* Postgres currently doesn't support Merge queries inside subqueries and
* ctes, but lets be defensive and do query tree walk anyway.
*
* We do not call this path for fast-path queries to avoid this additional
* overhead.
*/
if (!ContainsMergeCommandWalker((Node *) queryTree))
{
/* No MERGE found */
return false;
}
return true;
#endif
}

2
src/include/distributed/distributed_planner.h
View File

@ -254,8 +254,6 @@ extern struct DistributedPlan * CreateDistributedPlan(uint64 planId,
bool hasUnresolvedParams,
PlannerRestrictionContext *
plannerRestrictionContext);
extern bool IsMergeAllowedOnRelation(Query *parse, RangeTblEntry *rte);
extern bool ConjunctionContainsColumnFilter(Node *node,
Var *column,
Node **distributionKeyValue);

25
src/include/distributed/merge_planner.h Normal file
View File

@ -0,0 +1,25 @@
/*-------------------------------------------------------------------------
*
* merge_planner.h
*
* Declarations for public functions and types related to router planning.
*
* Copyright (c) Citus Data, Inc.
*
*-------------------------------------------------------------------------
*/
#ifndef MERGE_PLANNER_H
#define MERGE_PLANNER_H
#include "c.h"
#include "nodes/parsenodes.h"
#include "distributed/distributed_planner.h"
#include "distributed/errormessage.h"
extern bool IsMergeAllowedOnRelation(Query *parse, RangeTblEntry *rte);
extern DeferredErrorMessage * MergeQuerySupported(Query *originalQuery,
PlannerRestrictionContext *
plannerRestrictionContext);
#endif /* MERGE_PLANNER_H */

7
src/include/distributed/multi_router_planner.h
View File

@ -100,6 +100,13 @@ extern PlannedStmt * FastPathPlanner(Query *originalQuery, Query *parse, ParamLi
extern bool FastPathRouterQuery(Query *query, Node **distributionKeyValue);
extern bool JoinConditionIsOnFalse(List *relOptInfo);
extern Oid ResultRelationOidForQuery(Query *query);
extern DeferredErrorMessage * TargetlistAndFunctionsSupported(Oid resultRelationId,
FromExpr *joinTree,
Node *quals,
List *targetList,
CmdType commandType,
List *returningList);
#endif /* MULTI_ROUTER_PLANNER_H */

147
src/test/regress/expected/merge.out
View File

@ -17,8 +17,9 @@ CREATE SCHEMA merge_schema;
SET search_path TO merge_schema;
SET citus.shard_count TO 4;
SET citus.next_shard_id TO 4000000;
SET citus.explain_all_tasks to true;
SET citus.explain_all_tasks TO true;
SET citus.shard_replication_factor TO 1;
SET citus.max_adaptive_executor_pool_size TO 1;
SELECT 1 FROM master_add_node('localhost', :master_port, groupid => 0);
NOTICE: localhost:xxxxx is the coordinator and already contains metadata, skipping syncing the metadata
?column?
@ -1959,7 +1960,7 @@ ON pg_target.id = sub.id AND pg_target.id = $1
WHEN MATCHED THEN
UPDATE SET val = 'Updated by prepare using ' || sub.val
WHEN NOT MATCHED THEN
DO NOTHING;
INSERT VALUES (sub.id, sub.val);
PREPARE citus_prep(int) AS
MERGE INTO citus_target
USING (SELECT * FROM citus_source) sub
@ -1967,15 +1968,20 @@ ON citus_target.id = sub.id AND citus_target.id = $1
WHEN MATCHED THEN
UPDATE SET val = 'Updated by prepare using ' || sub.val
WHEN NOT MATCHED THEN
DO NOTHING;
INSERT VALUES (sub.id, sub.val);
BEGIN;
SET citus.log_remote_commands to true;
SELECT * FROM pg_target WHERE id = 500; -- before merge
id | val
---------------------------------------------------------------------
500 | target
(1 row)
SELECT count(*) FROM pg_target; -- before merge
count
---------------------------------------------------------------------
251
(1 row)
EXECUTE pg_prep(500);
SELECT * FROM pg_target WHERE id = 500; -- non-cached
id | val
@ -1994,18 +2000,33 @@ SELECT * FROM pg_target WHERE id = 500; -- cached
500 | Updated by prepare using source
(1 row)
SELECT count(*) FROM pg_target; -- cached
count
---------------------------------------------------------------------
3245
(1 row)
SELECT * FROM citus_target WHERE id = 500; -- before merge
NOTICE: issuing BEGIN TRANSACTION ISOLATION LEVEL READ COMMITTED;SELECT assign_distributed_transaction_id(xx, xx, 'xxxxxxx');
DETAIL: on server postgres@localhost:xxxxx connectionId: xxxxxxx
NOTICE: issuing SELECT id, val FROM merge_schema.citus_target_xxxxxxx citus_target WHERE (id OPERATOR(pg_catalog.=) 500)
DETAIL: on server postgres@localhost:xxxxx connectionId: xxxxxxx
id | val
---------------------------------------------------------------------
500 | target
(1 row)
SELECT count(*) FROM citus_target; -- before merge
count
---------------------------------------------------------------------
251
(1 row)
SET citus.log_remote_commands to true;
EXECUTE citus_prep(500);
NOTICE: issuing MERGE INTO merge_schema.citus_target_xxxxxxx citus_target USING (SELECT citus_source.id, citus_source.val FROM merge_schema.citus_source_xxxxxxx citus_source) sub ON ((citus_target.id OPERATOR(pg_catalog.=) sub.id) AND (citus_target.id OPERATOR(pg_catalog.=) $1)) WHEN MATCHED THEN UPDATE SET val = (('Updated by prepare using '::text OPERATOR(pg_catalog.||) (sub.val COLLATE "default")) COLLATE "default") WHEN NOT MATCHED THEN DO NOTHING
NOTICE: issuing MERGE INTO merge_schema.citus_target_xxxxxxx citus_target USING (SELECT citus_source.id, citus_source.val FROM merge_schema.citus_source_xxxxxxx citus_source) sub ON ((citus_target.id OPERATOR(pg_catalog.=) sub.id) AND (citus_target.id OPERATOR(pg_catalog.=) $1)) WHEN MATCHED THEN UPDATE SET val = (('Updated by prepare using '::text OPERATOR(pg_catalog.||) (sub.val COLLATE "default")) COLLATE "default") WHEN NOT MATCHED THEN INSERT (id, val) VALUES (sub.id, sub.val)
DETAIL: on server postgres@localhost:xxxxx connectionId: xxxxxxx
NOTICE: issuing MERGE INTO merge_schema.citus_target_xxxxxxx citus_target USING (SELECT citus_source.id, citus_source.val FROM merge_schema.citus_source_xxxxxxx citus_source) sub ON ((citus_target.id OPERATOR(pg_catalog.=) sub.id) AND (citus_target.id OPERATOR(pg_catalog.=) $1)) WHEN MATCHED THEN UPDATE SET val = (('Updated by prepare using '::text OPERATOR(pg_catalog.||) (sub.val COLLATE "default")) COLLATE "default") WHEN NOT MATCHED THEN INSERT (id, val) VALUES (sub.id, sub.val)
DETAIL: on server postgres@localhost:xxxxx connectionId: xxxxxxx
NOTICE: issuing MERGE INTO merge_schema.citus_target_xxxxxxx citus_target USING (SELECT citus_source.id, citus_source.val FROM merge_schema.citus_source_xxxxxxx citus_source) sub ON ((citus_target.id OPERATOR(pg_catalog.=) sub.id) AND (citus_target.id OPERATOR(pg_catalog.=) $1)) WHEN MATCHED THEN UPDATE SET val = (('Updated by prepare using '::text OPERATOR(pg_catalog.||) (sub.val COLLATE "default")) COLLATE "default") WHEN NOT MATCHED THEN INSERT (id, val) VALUES (sub.id, sub.val)
DETAIL: on server postgres@localhost:xxxxx connectionId: xxxxxxx
NOTICE: issuing MERGE INTO merge_schema.citus_target_xxxxxxx citus_target USING (SELECT citus_source.id, citus_source.val FROM merge_schema.citus_source_xxxxxxx citus_source) sub ON ((citus_target.id OPERATOR(pg_catalog.=) sub.id) AND (citus_target.id OPERATOR(pg_catalog.=) $1)) WHEN MATCHED THEN UPDATE SET val = (('Updated by prepare using '::text OPERATOR(pg_catalog.||) (sub.val COLLATE "default")) COLLATE "default") WHEN NOT MATCHED THEN INSERT (id, val) VALUES (sub.id, sub.val)
DETAIL: on server postgres@localhost:xxxxx connectionId: xxxxxxx
SELECT * FROM citus_target WHERE id = 500; -- non-cached
NOTICE: issuing SELECT id, val FROM merge_schema.citus_target_xxxxxxx citus_target WHERE (id OPERATOR(pg_catalog.=) 500)
@ -2016,29 +2037,63 @@ DETAIL: on server postgres@localhost:xxxxx connectionId: xxxxxxx
(1 row)
EXECUTE citus_prep(500);
NOTICE: issuing MERGE INTO merge_schema.citus_target_xxxxxxx citus_target USING (SELECT citus_source.id, citus_source.val FROM merge_schema.citus_source_xxxxxxx citus_source) sub ON ((citus_target.id OPERATOR(pg_catalog.=) sub.id) AND (citus_target.id OPERATOR(pg_catalog.=) $1)) WHEN MATCHED THEN UPDATE SET val = (('Updated by prepare using '::text OPERATOR(pg_catalog.||) (sub.val COLLATE "default")) COLLATE "default") WHEN NOT MATCHED THEN DO NOTHING
NOTICE: issuing MERGE INTO merge_schema.citus_target_xxxxxxx citus_target USING (SELECT citus_source.id, citus_source.val FROM merge_schema.citus_source_xxxxxxx citus_source) sub ON ((citus_target.id OPERATOR(pg_catalog.=) sub.id) AND (citus_target.id OPERATOR(pg_catalog.=) $1)) WHEN MATCHED THEN UPDATE SET val = (('Updated by prepare using '::text OPERATOR(pg_catalog.||) (sub.val COLLATE "default")) COLLATE "default") WHEN NOT MATCHED THEN INSERT (id, val) VALUES (sub.id, sub.val)
DETAIL: on server postgres@localhost:xxxxx connectionId: xxxxxxx
NOTICE: issuing MERGE INTO merge_schema.citus_target_xxxxxxx citus_target USING (SELECT citus_source.id, citus_source.val FROM merge_schema.citus_source_xxxxxxx citus_source) sub ON ((citus_target.id OPERATOR(pg_catalog.=) sub.id) AND (citus_target.id OPERATOR(pg_catalog.=) $1)) WHEN MATCHED THEN UPDATE SET val = (('Updated by prepare using '::text OPERATOR(pg_catalog.||) (sub.val COLLATE "default")) COLLATE "default") WHEN NOT MATCHED THEN INSERT (id, val) VALUES (sub.id, sub.val)
DETAIL: on server postgres@localhost:xxxxx connectionId: xxxxxxx
NOTICE: issuing MERGE INTO merge_schema.citus_target_xxxxxxx citus_target USING (SELECT citus_source.id, citus_source.val FROM merge_schema.citus_source_xxxxxxx citus_source) sub ON ((citus_target.id OPERATOR(pg_catalog.=) sub.id) AND (citus_target.id OPERATOR(pg_catalog.=) $1)) WHEN MATCHED THEN UPDATE SET val = (('Updated by prepare using '::text OPERATOR(pg_catalog.||) (sub.val COLLATE "default")) COLLATE "default") WHEN NOT MATCHED THEN INSERT (id, val) VALUES (sub.id, sub.val)
DETAIL: on server postgres@localhost:xxxxx connectionId: xxxxxxx
NOTICE: issuing MERGE INTO merge_schema.citus_target_xxxxxxx citus_target USING (SELECT citus_source.id, citus_source.val FROM merge_schema.citus_source_xxxxxxx citus_source) sub ON ((citus_target.id OPERATOR(pg_catalog.=) sub.id) AND (citus_target.id OPERATOR(pg_catalog.=) $1)) WHEN MATCHED THEN UPDATE SET val = (('Updated by prepare using '::text OPERATOR(pg_catalog.||) (sub.val COLLATE "default")) COLLATE "default") WHEN NOT MATCHED THEN INSERT (id, val) VALUES (sub.id, sub.val)
DETAIL: on server postgres@localhost:xxxxx connectionId: xxxxxxx
EXECUTE citus_prep(500);
NOTICE: issuing MERGE INTO merge_schema.citus_target_xxxxxxx citus_target USING (SELECT citus_source.id, citus_source.val FROM merge_schema.citus_source_xxxxxxx citus_source) sub ON ((citus_target.id OPERATOR(pg_catalog.=) sub.id) AND (citus_target.id OPERATOR(pg_catalog.=) $1)) WHEN MATCHED THEN UPDATE SET val = (('Updated by prepare using '::text OPERATOR(pg_catalog.||) (sub.val COLLATE "default")) COLLATE "default") WHEN NOT MATCHED THEN DO NOTHING
NOTICE: issuing MERGE INTO merge_schema.citus_target_xxxxxxx citus_target USING (SELECT citus_source.id, citus_source.val FROM merge_schema.citus_source_xxxxxxx citus_source) sub ON ((citus_target.id OPERATOR(pg_catalog.=) sub.id) AND (citus_target.id OPERATOR(pg_catalog.=) $1)) WHEN MATCHED THEN UPDATE SET val = (('Updated by prepare using '::text OPERATOR(pg_catalog.||) (sub.val COLLATE "default")) COLLATE "default") WHEN NOT MATCHED THEN INSERT (id, val) VALUES (sub.id, sub.val)
DETAIL: on server postgres@localhost:xxxxx connectionId: xxxxxxx
NOTICE: issuing MERGE INTO merge_schema.citus_target_xxxxxxx citus_target USING (SELECT citus_source.id, citus_source.val FROM merge_schema.citus_source_xxxxxxx citus_source) sub ON ((citus_target.id OPERATOR(pg_catalog.=) sub.id) AND (citus_target.id OPERATOR(pg_catalog.=) $1)) WHEN MATCHED THEN UPDATE SET val = (('Updated by prepare using '::text OPERATOR(pg_catalog.||) (sub.val COLLATE "default")) COLLATE "default") WHEN NOT MATCHED THEN INSERT (id, val) VALUES (sub.id, sub.val)
DETAIL: on server postgres@localhost:xxxxx connectionId: xxxxxxx
NOTICE: issuing MERGE INTO merge_schema.citus_target_xxxxxxx citus_target USING (SELECT citus_source.id, citus_source.val FROM merge_schema.citus_source_xxxxxxx citus_source) sub ON ((citus_target.id OPERATOR(pg_catalog.=) sub.id) AND (citus_target.id OPERATOR(pg_catalog.=) $1)) WHEN MATCHED THEN UPDATE SET val = (('Updated by prepare using '::text OPERATOR(pg_catalog.||) (sub.val COLLATE "default")) COLLATE "default") WHEN NOT MATCHED THEN INSERT (id, val) VALUES (sub.id, sub.val)
DETAIL: on server postgres@localhost:xxxxx connectionId: xxxxxxx
NOTICE: issuing MERGE INTO merge_schema.citus_target_xxxxxxx citus_target USING (SELECT citus_source.id, citus_source.val FROM merge_schema.citus_source_xxxxxxx citus_source) sub ON ((citus_target.id OPERATOR(pg_catalog.=) sub.id) AND (citus_target.id OPERATOR(pg_catalog.=) $1)) WHEN MATCHED THEN UPDATE SET val = (('Updated by prepare using '::text OPERATOR(pg_catalog.||) (sub.val COLLATE "default")) COLLATE "default") WHEN NOT MATCHED THEN INSERT (id, val) VALUES (sub.id, sub.val)
DETAIL: on server postgres@localhost:xxxxx connectionId: xxxxxxx
EXECUTE citus_prep(500);
NOTICE: issuing MERGE INTO merge_schema.citus_target_xxxxxxx citus_target USING (SELECT citus_source.id, citus_source.val FROM merge_schema.citus_source_xxxxxxx citus_source) sub ON ((citus_target.id OPERATOR(pg_catalog.=) sub.id) AND (citus_target.id OPERATOR(pg_catalog.=) $1)) WHEN MATCHED THEN UPDATE SET val = (('Updated by prepare using '::text OPERATOR(pg_catalog.||) (sub.val COLLATE "default")) COLLATE "default") WHEN NOT MATCHED THEN DO NOTHING
NOTICE: issuing MERGE INTO merge_schema.citus_target_xxxxxxx citus_target USING (SELECT citus_source.id, citus_source.val FROM merge_schema.citus_source_xxxxxxx citus_source) sub ON ((citus_target.id OPERATOR(pg_catalog.=) sub.id) AND (citus_target.id OPERATOR(pg_catalog.=) $1)) WHEN MATCHED THEN UPDATE SET val = (('Updated by prepare using '::text OPERATOR(pg_catalog.||) (sub.val COLLATE "default")) COLLATE "default") WHEN NOT MATCHED THEN INSERT (id, val) VALUES (sub.id, sub.val)
DETAIL: on server postgres@localhost:xxxxx connectionId: xxxxxxx
NOTICE: issuing MERGE INTO merge_schema.citus_target_xxxxxxx citus_target USING (SELECT citus_source.id, citus_source.val FROM merge_schema.citus_source_xxxxxxx citus_source) sub ON ((citus_target.id OPERATOR(pg_catalog.=) sub.id) AND (citus_target.id OPERATOR(pg_catalog.=) $1)) WHEN MATCHED THEN UPDATE SET val = (('Updated by prepare using '::text OPERATOR(pg_catalog.||) (sub.val COLLATE "default")) COLLATE "default") WHEN NOT MATCHED THEN INSERT (id, val) VALUES (sub.id, sub.val)
DETAIL: on server postgres@localhost:xxxxx connectionId: xxxxxxx
NOTICE: issuing MERGE INTO merge_schema.citus_target_xxxxxxx citus_target USING (SELECT citus_source.id, citus_source.val FROM merge_schema.citus_source_xxxxxxx citus_source) sub ON ((citus_target.id OPERATOR(pg_catalog.=) sub.id) AND (citus_target.id OPERATOR(pg_catalog.=) $1)) WHEN MATCHED THEN UPDATE SET val = (('Updated by prepare using '::text OPERATOR(pg_catalog.||) (sub.val COLLATE "default")) COLLATE "default") WHEN NOT MATCHED THEN INSERT (id, val) VALUES (sub.id, sub.val)
DETAIL: on server postgres@localhost:xxxxx connectionId: xxxxxxx
NOTICE: issuing MERGE INTO merge_schema.citus_target_xxxxxxx citus_target USING (SELECT citus_source.id, citus_source.val FROM merge_schema.citus_source_xxxxxxx citus_source) sub ON ((citus_target.id OPERATOR(pg_catalog.=) sub.id) AND (citus_target.id OPERATOR(pg_catalog.=) $1)) WHEN MATCHED THEN UPDATE SET val = (('Updated by prepare using '::text OPERATOR(pg_catalog.||) (sub.val COLLATE "default")) COLLATE "default") WHEN NOT MATCHED THEN INSERT (id, val) VALUES (sub.id, sub.val)
DETAIL: on server postgres@localhost:xxxxx connectionId: xxxxxxx
EXECUTE citus_prep(500);
NOTICE: issuing MERGE INTO merge_schema.citus_target_xxxxxxx citus_target USING (SELECT citus_source.id, citus_source.val FROM merge_schema.citus_source_xxxxxxx citus_source) sub ON ((citus_target.id OPERATOR(pg_catalog.=) sub.id) AND (citus_target.id OPERATOR(pg_catalog.=) $1)) WHEN MATCHED THEN UPDATE SET val = (('Updated by prepare using '::text OPERATOR(pg_catalog.||) (sub.val COLLATE "default")) COLLATE "default") WHEN NOT MATCHED THEN DO NOTHING
NOTICE: issuing MERGE INTO merge_schema.citus_target_xxxxxxx citus_target USING (SELECT citus_source.id, citus_source.val FROM merge_schema.citus_source_xxxxxxx citus_source) sub ON ((citus_target.id OPERATOR(pg_catalog.=) sub.id) AND (citus_target.id OPERATOR(pg_catalog.=) $1)) WHEN MATCHED THEN UPDATE SET val = (('Updated by prepare using '::text OPERATOR(pg_catalog.||) (sub.val COLLATE "default")) COLLATE "default") WHEN NOT MATCHED THEN INSERT (id, val) VALUES (sub.id, sub.val)
DETAIL: on server postgres@localhost:xxxxx connectionId: xxxxxxx
NOTICE: issuing MERGE INTO merge_schema.citus_target_xxxxxxx citus_target USING (SELECT citus_source.id, citus_source.val FROM merge_schema.citus_source_xxxxxxx citus_source) sub ON ((citus_target.id OPERATOR(pg_catalog.=) sub.id) AND (citus_target.id OPERATOR(pg_catalog.=) $1)) WHEN MATCHED THEN UPDATE SET val = (('Updated by prepare using '::text OPERATOR(pg_catalog.||) (sub.val COLLATE "default")) COLLATE "default") WHEN NOT MATCHED THEN INSERT (id, val) VALUES (sub.id, sub.val)
DETAIL: on server postgres@localhost:xxxxx connectionId: xxxxxxx
NOTICE: issuing MERGE INTO merge_schema.citus_target_xxxxxxx citus_target USING (SELECT citus_source.id, citus_source.val FROM merge_schema.citus_source_xxxxxxx citus_source) sub ON ((citus_target.id OPERATOR(pg_catalog.=) sub.id) AND (citus_target.id OPERATOR(pg_catalog.=) $1)) WHEN MATCHED THEN UPDATE SET val = (('Updated by prepare using '::text OPERATOR(pg_catalog.||) (sub.val COLLATE "default")) COLLATE "default") WHEN NOT MATCHED THEN INSERT (id, val) VALUES (sub.id, sub.val)
DETAIL: on server postgres@localhost:xxxxx connectionId: xxxxxxx
NOTICE: issuing MERGE INTO merge_schema.citus_target_xxxxxxx citus_target USING (SELECT citus_source.id, citus_source.val FROM merge_schema.citus_source_xxxxxxx citus_source) sub ON ((citus_target.id OPERATOR(pg_catalog.=) sub.id) AND (citus_target.id OPERATOR(pg_catalog.=) $1)) WHEN MATCHED THEN UPDATE SET val = (('Updated by prepare using '::text OPERATOR(pg_catalog.||) (sub.val COLLATE "default")) COLLATE "default") WHEN NOT MATCHED THEN INSERT (id, val) VALUES (sub.id, sub.val)
DETAIL: on server postgres@localhost:xxxxx connectionId: xxxxxxx
EXECUTE citus_prep(500);
NOTICE: issuing MERGE INTO merge_schema.citus_target_xxxxxxx citus_target USING (SELECT citus_source.id, citus_source.val FROM merge_schema.citus_source_xxxxxxx citus_source) sub ON ((citus_target.id OPERATOR(pg_catalog.=) sub.id) AND (citus_target.id OPERATOR(pg_catalog.=) $1)) WHEN MATCHED THEN UPDATE SET val = (('Updated by prepare using '::text OPERATOR(pg_catalog.||) (sub.val COLLATE "default")) COLLATE "default") WHEN NOT MATCHED THEN DO NOTHING
NOTICE: issuing MERGE INTO merge_schema.citus_target_xxxxxxx citus_target USING (SELECT citus_source.id, citus_source.val FROM merge_schema.citus_source_xxxxxxx citus_source) sub ON ((citus_target.id OPERATOR(pg_catalog.=) sub.id) AND (citus_target.id OPERATOR(pg_catalog.=) $1)) WHEN MATCHED THEN UPDATE SET val = (('Updated by prepare using '::text OPERATOR(pg_catalog.||) (sub.val COLLATE "default")) COLLATE "default") WHEN NOT MATCHED THEN INSERT (id, val) VALUES (sub.id, sub.val)
DETAIL: on server postgres@localhost:xxxxx connectionId: xxxxxxx
NOTICE: issuing MERGE INTO merge_schema.citus_target_xxxxxxx citus_target USING (SELECT citus_source.id, citus_source.val FROM merge_schema.citus_source_xxxxxxx citus_source) sub ON ((citus_target.id OPERATOR(pg_catalog.=) sub.id) AND (citus_target.id OPERATOR(pg_catalog.=) $1)) WHEN MATCHED THEN UPDATE SET val = (('Updated by prepare using '::text OPERATOR(pg_catalog.||) (sub.val COLLATE "default")) COLLATE "default") WHEN NOT MATCHED THEN INSERT (id, val) VALUES (sub.id, sub.val)
DETAIL: on server postgres@localhost:xxxxx connectionId: xxxxxxx
NOTICE: issuing MERGE INTO merge_schema.citus_target_xxxxxxx citus_target USING (SELECT citus_source.id, citus_source.val FROM merge_schema.citus_source_xxxxxxx citus_source) sub ON ((citus_target.id OPERATOR(pg_catalog.=) sub.id) AND (citus_target.id OPERATOR(pg_catalog.=) $1)) WHEN MATCHED THEN UPDATE SET val = (('Updated by prepare using '::text OPERATOR(pg_catalog.||) (sub.val COLLATE "default")) COLLATE "default") WHEN NOT MATCHED THEN INSERT (id, val) VALUES (sub.id, sub.val)
DETAIL: on server postgres@localhost:xxxxx connectionId: xxxxxxx
NOTICE: issuing MERGE INTO merge_schema.citus_target_xxxxxxx citus_target USING (SELECT citus_source.id, citus_source.val FROM merge_schema.citus_source_xxxxxxx citus_source) sub ON ((citus_target.id OPERATOR(pg_catalog.=) sub.id) AND (citus_target.id OPERATOR(pg_catalog.=) $1)) WHEN MATCHED THEN UPDATE SET val = (('Updated by prepare using '::text OPERATOR(pg_catalog.||) (sub.val COLLATE "default")) COLLATE "default") WHEN NOT MATCHED THEN INSERT (id, val) VALUES (sub.id, sub.val)
DETAIL: on server postgres@localhost:xxxxx connectionId: xxxxxxx
SET citus.log_remote_commands to false;
SELECT * FROM citus_target WHERE id = 500; -- cached
NOTICE: issuing SELECT id, val FROM merge_schema.citus_target_xxxxxxx citus_target WHERE (id OPERATOR(pg_catalog.=) 500)
DETAIL: on server postgres@localhost:xxxxx connectionId: xxxxxxx
id | val
---------------------------------------------------------------------
500 | Updated by prepare using source
(1 row)
SET citus.log_remote_commands to false;
SELECT count(*) FROM citus_target; -- cached
count
---------------------------------------------------------------------
3245
(1 row)
SELECT compare_tables();
compare_tables
---------------------------------------------------------------------
@ -2168,6 +2223,64 @@ ROLLBACK;
--
-- Error and Unsupported scenarios
--
-- Grouping sets not supported
MERGE INTO citus_target t
USING (SELECT count(*), id FROM citus_source GROUP BY GROUPING SETS (id, val)) subq
ON subq.id = t.id
WHEN MATCHED AND t.id > 350 THEN
UPDATE SET val = t.val || 'Updated'
WHEN NOT MATCHED THEN
INSERT VALUES (subq.id, 99)
WHEN MATCHED AND t.id < 350 THEN
DELETE;
ERROR: cannot push down this subquery
DETAIL: could not run distributed query with GROUPING SETS, CUBE, or ROLLUP
WITH subq AS
(
SELECT count(*), id FROM citus_source GROUP BY GROUPING SETS (id, val)
)
MERGE INTO citus_target t
USING subq
ON subq.id = t.id
WHEN MATCHED AND t.id > 350 THEN
UPDATE SET val = t.val || 'Updated'
WHEN NOT MATCHED THEN
INSERT VALUES (subq.id, 99)
WHEN MATCHED AND t.id < 350 THEN
DELETE;
ERROR: cannot push down this subquery
DETAIL: could not run distributed query with GROUPING SETS, CUBE, or ROLLUP
-- try inserting unmatched distribution column value
MERGE INTO citus_target t
USING citus_source s
ON t.id = s.id
WHEN NOT MATCHED THEN
INSERT DEFAULT VALUES;
ERROR: cannot perform MERGE INSERT with DEFAULTS
MERGE INTO citus_target t
USING citus_source s
ON t.id = s.id
WHEN NOT MATCHED THEN
INSERT VALUES(10000);
ERROR: MERGE INSERT must refer a source column for distribution column
MERGE INTO citus_target t
USING citus_source s
ON t.id = s.id
WHEN NOT MATCHED THEN
INSERT (id) VALUES(1000);
ERROR: MERGE INSERT must refer a source column for distribution column
MERGE INTO t1 t
USING s1 s
ON t.id = s.id
WHEN NOT MATCHED THEN
INSERT (id) VALUES(s.val);
ERROR: MERGE INSERT must use the source table distribution column value
MERGE INTO t1 t
USING s1 s
ON t.id = s.id
WHEN NOT MATCHED THEN
INSERT (val) VALUES(s.val);
ERROR: MERGE INSERT must have distribution column as value
-- try updating the distribution key column
BEGIN;
MERGE INTO target_cj t

73
src/test/regress/sql/merge.sql
View File

@ -18,8 +18,9 @@ CREATE SCHEMA merge_schema;
SET search_path TO merge_schema;
SET citus.shard_count TO 4;
SET citus.next_shard_id TO 4000000;
SET citus.explain_all_tasks to true;
SET citus.explain_all_tasks TO true;
SET citus.shard_replication_factor TO 1;
SET citus.max_adaptive_executor_pool_size TO 1;
SELECT 1 FROM master_add_node('localhost', :master_port, groupid => 0);
CREATE TABLE source
@ -1287,7 +1288,7 @@ ON pg_target.id = sub.id AND pg_target.id = $1
WHEN MATCHED THEN
UPDATE SET val = 'Updated by prepare using ' || sub.val
WHEN NOT MATCHED THEN
DO NOTHING;
INSERT VALUES (sub.id, sub.val);
PREPARE citus_prep(int) AS
MERGE INTO citus_target
@ -1296,12 +1297,12 @@ ON citus_target.id = sub.id AND citus_target.id = $1
WHEN MATCHED THEN
UPDATE SET val = 'Updated by prepare using ' || sub.val
WHEN NOT MATCHED THEN
DO NOTHING;
INSERT VALUES (sub.id, sub.val);
BEGIN;
SET citus.log_remote_commands to true;
SELECT * FROM pg_target WHERE id = 500; -- before merge
SELECT count(*) FROM pg_target; -- before merge
EXECUTE pg_prep(500);
SELECT * FROM pg_target WHERE id = 500; -- non-cached
EXECUTE pg_prep(500);
@ -1310,8 +1311,11 @@ EXECUTE pg_prep(500);
EXECUTE pg_prep(500);
EXECUTE pg_prep(500);
SELECT * FROM pg_target WHERE id = 500; -- cached
SELECT count(*) FROM pg_target; -- cached
SELECT * FROM citus_target WHERE id = 500; -- before merge
SELECT count(*) FROM citus_target; -- before merge
SET citus.log_remote_commands to true;
EXECUTE citus_prep(500);
SELECT * FROM citus_target WHERE id = 500; -- non-cached
EXECUTE citus_prep(500);
@ -1319,9 +1323,10 @@ EXECUTE citus_prep(500);
EXECUTE citus_prep(500);
EXECUTE citus_prep(500);
EXECUTE citus_prep(500);
SELECT * FROM citus_target WHERE id = 500; -- cached
SET citus.log_remote_commands to false;
SELECT * FROM citus_target WHERE id = 500; -- cached
SELECT count(*) FROM citus_target; -- cached
SELECT compare_tables();
ROLLBACK;
@ -1421,6 +1426,62 @@ ROLLBACK;
-- Error and Unsupported scenarios
--
-- Grouping sets not supported
MERGE INTO citus_target t
USING (SELECT count(*), id FROM citus_source GROUP BY GROUPING SETS (id, val)) subq
ON subq.id = t.id
WHEN MATCHED AND t.id > 350 THEN
UPDATE SET val = t.val || 'Updated'
WHEN NOT MATCHED THEN
INSERT VALUES (subq.id, 99)
WHEN MATCHED AND t.id < 350 THEN
DELETE;
WITH subq AS
(
SELECT count(*), id FROM citus_source GROUP BY GROUPING SETS (id, val)
)
MERGE INTO citus_target t
USING subq
ON subq.id = t.id
WHEN MATCHED AND t.id > 350 THEN
UPDATE SET val = t.val || 'Updated'
WHEN NOT MATCHED THEN
INSERT VALUES (subq.id, 99)
WHEN MATCHED AND t.id < 350 THEN
DELETE;
-- try inserting unmatched distribution column value
MERGE INTO citus_target t
USING citus_source s
ON t.id = s.id
WHEN NOT MATCHED THEN
INSERT DEFAULT VALUES;
MERGE INTO citus_target t
USING citus_source s
ON t.id = s.id
WHEN NOT MATCHED THEN
INSERT VALUES(10000);
MERGE INTO citus_target t
USING citus_source s
ON t.id = s.id
WHEN NOT MATCHED THEN
INSERT (id) VALUES(1000);
MERGE INTO t1 t
USING s1 s
ON t.id = s.id
WHEN NOT MATCHED THEN
INSERT (id) VALUES(s.val);
MERGE INTO t1 t
USING s1 s
ON t.id = s.id
WHEN NOT MATCHED THEN
INSERT (val) VALUES(s.val);
-- try updating the distribution key column
BEGIN;
MERGE INTO target_cj t