From 4028b93c4b1d2fcd2a245b894b6d50b369ddd556 Mon Sep 17 00:00:00 2001 From: Greg Nancarrow Date: Tue, 19 Jan 2021 15:12:53 +1100 Subject: [PATCH v12 1/4] Enable parallel SELECT for "INSERT INTO ... SELECT ...", where it is safe to do so. Parallel SELECT can't be utilized for INSERT in the following cases: - INSERT statement uses the ON CONFLICT DO UPDATE clause - Target table is a foreign or temporary table - Target table has a parallel-unsafe trigger, index expression, column default expression or check constraint - Target table is a partitioned table with a parallel-unsafe partition key expression or support function The planner is updated to perform additional parallel-safety checks for the cases listed above, for determining whether it is safe to run INSERT in parallel-mode with an underlying parallel SELECT. The planner is further updated to consider using parallel SELECT for "INSERT INTO ... SELECT ...", provided nothing unsafe is found from the additional parallel-safety checks, or from the existing parallel-safety checks for SELECT that it currently performs. Prior to entering parallel-mode for execution of INSERT with parallel SELECT, a TransactionId is acquired and assigned to the current transaction state which is then serialized in the parallel DSM for the parallel workers to use. Discussion: https://postgr.es/m/CAJcOf-cXnB5cnMKqWEp2E2z7Mvcd04iLVmV=qpFJrR3AcrTS3g@mail.gmail.com --- src/backend/access/transam/xact.c | 22 ++ src/backend/executor/execMain.c | 3 + src/backend/optimizer/plan/planner.c | 21 +- src/backend/optimizer/util/clauses.c | 452 +++++++++++++++++++++++++++ src/include/access/xact.h | 15 + src/include/optimizer/clauses.h | 1 + 6 files changed, 504 insertions(+), 10 deletions(-) diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c index a2068e3fd4..7514106dc8 100644 --- a/src/backend/access/transam/xact.c +++ b/src/backend/access/transam/xact.c @@ -41,6 +41,7 @@ #include "libpq/be-fsstubs.h" #include "libpq/pqsignal.h" #include "miscadmin.h" +#include "optimizer/optimizer.h" #include "pg_trace.h" #include "pgstat.h" #include "replication/logical.h" @@ -1014,6 +1015,27 @@ IsInParallelMode(void) return CurrentTransactionState->parallelModeLevel != 0; } +/* + * PrepareParallelMode + * + * Prepare for entering parallel mode, based on command-type. + */ +void +PrepareParallelMode(CmdType commandType) +{ + if (IsModifySupportedInParallelMode(commandType)) + { + Assert(!IsInParallelMode()); + + /* + * Prepare for entering parallel mode by assigning a + * FullTransactionId, to be included in the transaction state that is + * serialized in the parallel DSM. + */ + (void) GetCurrentTransactionId(); + } +} + /* * CommandCounterIncrement */ diff --git a/src/backend/executor/execMain.c b/src/backend/executor/execMain.c index f4dd47acc7..c72e3e102e 100644 --- a/src/backend/executor/execMain.c +++ b/src/backend/executor/execMain.c @@ -1526,7 +1526,10 @@ ExecutePlan(EState *estate, estate->es_use_parallel_mode = use_parallel_mode; if (use_parallel_mode) + { + PrepareParallelMode(estate->es_plannedstmt->commandType); EnterParallelMode(); + } /* * Loop until we've processed the proper number of tuples from the plan. diff --git a/src/backend/optimizer/plan/planner.c b/src/backend/optimizer/plan/planner.c index 4e6497ff32..83ca943736 100644 --- a/src/backend/optimizer/plan/planner.c +++ b/src/backend/optimizer/plan/planner.c @@ -316,16 +316,16 @@ standard_planner(Query *parse, const char *query_string, int cursorOptions, /* * Assess whether it's feasible to use parallel mode for this query. We * can't do this in a standalone backend, or if the command will try to - * modify any data, or if this is a cursor operation, or if GUCs are set - * to values that don't permit parallelism, or if parallel-unsafe - * functions are present in the query tree. + * modify any data using a CTE, or if this is a cursor operation, or if + * GUCs are set to values that don't permit parallelism, or if + * parallel-unsafe functions are present in the query tree. * - * (Note that we do allow CREATE TABLE AS, SELECT INTO, and CREATE - * MATERIALIZED VIEW to use parallel plans, but as of now, only the leader - * backend writes into a completely new table. In the future, we can - * extend it to allow workers to write into the table. However, to allow - * parallel updates and deletes, we have to solve other problems, - * especially around combo CIDs.) + * (Note that we do allow CREATE TABLE AS, INSERT INTO...SELECT, SELECT + * INTO, and CREATE MATERIALIZED VIEW to use parallel plans. However, as + * of now, only the leader backend writes into a completely new table. In + * the future, we can extend it to allow workers to write into the table. + * However, to allow parallel updates and deletes, we have to solve other + * problems, especially around combo CIDs.) * * For now, we don't try to use parallel mode if we're running inside a * parallel worker. We might eventually be able to relax this @@ -334,7 +334,8 @@ standard_planner(Query *parse, const char *query_string, int cursorOptions, */ if ((cursorOptions & CURSOR_OPT_PARALLEL_OK) != 0 && IsUnderPostmaster && - parse->commandType == CMD_SELECT && + (parse->commandType == CMD_SELECT || + IsModifySupportedInParallelMode(parse->commandType)) && !parse->hasModifyingCTE && max_parallel_workers_per_gather > 0 && !IsParallelWorker()) diff --git a/src/backend/optimizer/util/clauses.c b/src/backend/optimizer/util/clauses.c index 51d26a0691..b13d57cb2a 100644 --- a/src/backend/optimizer/util/clauses.c +++ b/src/backend/optimizer/util/clauses.c @@ -19,13 +19,19 @@ #include "postgres.h" +#include "access/genam.h" #include "access/htup_details.h" +#include "access/table.h" +#include "access/xact.h" +#include "catalog/index.h" #include "catalog/pg_aggregate.h" #include "catalog/pg_class.h" +#include "catalog/pg_constraint.h" #include "catalog/pg_language.h" #include "catalog/pg_operator.h" #include "catalog/pg_proc.h" #include "catalog/pg_type.h" +#include "commands/trigger.h" #include "executor/executor.h" #include "executor/functions.h" #include "funcapi.h" @@ -43,6 +49,8 @@ #include "parser/parse_agg.h" #include "parser/parse_coerce.h" #include "parser/parse_func.h" +#include "parser/parsetree.h" +#include "partitioning/partdesc.h" #include "rewrite/rewriteManip.h" #include "tcop/tcopprot.h" #include "utils/acl.h" @@ -51,6 +59,8 @@ #include "utils/fmgroids.h" #include "utils/lsyscache.h" #include "utils/memutils.h" +#include "utils/partcache.h" +#include "utils/rel.h" #include "utils/syscache.h" #include "utils/typcache.h" @@ -148,6 +158,15 @@ static Query *substitute_actual_srf_parameters(Query *expr, static Node *substitute_actual_srf_parameters_mutator(Node *node, substitute_actual_srf_parameters_context *context); +static bool trigger_max_parallel_hazard_for_modify(TriggerDesc *trigdesc, + max_parallel_hazard_context *context); +static bool index_expr_max_parallel_hazard_for_modify(Relation rel, + max_parallel_hazard_context *context); +static bool domain_max_parallel_hazard_for_modify(Oid typid, max_parallel_hazard_context *context); +static bool rel_max_parallel_hazard_for_modify(Oid relid, + CmdType command_type, + max_parallel_hazard_context *context, + LOCKMODE lockmode); /***************************************************************************** * Aggregate-function clause manipulation @@ -553,6 +572,18 @@ max_parallel_hazard(Query *parse) context.max_interesting = PROPARALLEL_UNSAFE; context.safe_param_ids = NIL; (void) max_parallel_hazard_walker((Node *) parse, &context); + + /* + * Additional parallel-mode safety checks are required in order to + * allow an underlying parallel query to be used for a + * table-modification command that is supported in parallel-mode. + */ + + if (context.max_hazard != PROPARALLEL_UNSAFE && + IsModifySupportedInParallelMode(parse->commandType)) + { + context.max_hazard = max_parallel_hazard_for_modify(parse, context.max_hazard); + } return context.max_hazard; } @@ -786,6 +817,427 @@ max_parallel_hazard_walker(Node *node, max_parallel_hazard_context *context) context); } +/* + * trigger_max_parallel_hazard_for_modify + * + * Finds the maximum parallel-mode hazard level for the specified trigger data. + */ +static bool +trigger_max_parallel_hazard_for_modify(TriggerDesc *trigdesc, + max_parallel_hazard_context *context) +{ + int i; + + for (i = 0; i < trigdesc->numtriggers; i++) + { + int trigtype; + Trigger *trigger = &trigdesc->triggers[i]; + + if (max_parallel_hazard_test(func_parallel(trigger->tgfoid), context)) + return true; + + /* + * If the trigger type is RI_TRIGGER_FK, this indicates a FK exists in + * the relation, and this would result in creation of new CommandIds + * on insert/update/delete and this isn't supported in a parallel + * worker (but is safe in the parallel leader). + */ + trigtype = RI_FKey_trigger_type(trigger->tgfoid); + if (trigtype == RI_TRIGGER_FK) + { + if (max_parallel_hazard_test(PROPARALLEL_RESTRICTED, context)) + return true; + } + } + + return false; +} + +/* + * index_expr_max_parallel_hazard_for_modify + * + * Finds the maximum parallel-mode hazard level for any existing index + * expressions of a specified relation. + */ +static bool +index_expr_max_parallel_hazard_for_modify(Relation rel, + max_parallel_hazard_context *context) +{ + List *index_oid_list; + ListCell *lc; + LOCKMODE lockmode = AccessShareLock; + + index_oid_list = RelationGetIndexList(rel); + foreach(lc, index_oid_list) + { + Oid index_oid = lfirst_oid(lc); + Relation index_rel; + IndexInfo *index_info; + + index_rel = index_open(index_oid, lockmode); + + index_info = BuildIndexInfo(index_rel); + + if (index_info->ii_Expressions != NIL) + { + int i; + ListCell *index_expr_item = list_head(index_info->ii_Expressions); + + for (i = 0; i < index_info->ii_NumIndexAttrs; i++) + { + int keycol = index_info->ii_IndexAttrNumbers[i]; + + if (keycol == 0) + { + /* Found an index expression */ + + Node *index_expr; + + Assert(index_expr_item != NULL); + if (index_expr_item == NULL) /* shouldn't happen */ + { + elog(WARNING, "too few entries in indexprs list"); + index_close(index_rel, lockmode); + context->max_hazard = PROPARALLEL_UNSAFE; + return true; + } + + index_expr = (Node *) lfirst(index_expr_item); + index_expr = (Node *) expression_planner((Expr *) index_expr); + + if (max_parallel_hazard_walker(index_expr, context)) + { + index_close(index_rel, lockmode); + return true; + } + + index_expr_item = lnext(index_info->ii_Expressions, index_expr_item); + } + } + } + index_close(index_rel, lockmode); + } + + return false; +} + +/* + * domain_max_parallel_hazard_for_modify + * + * Finds the maximum parallel-mode hazard level for the specified DOMAIN type. + * Only any CHECK expressions are examined for parallel safety. + * DEFAULT values of DOMAIN-type columns in the target-list are already + * being checked for parallel-safety in the max_parallel_hazard() scan of the + * query tree in standard_planner(). + * + */ +static bool +domain_max_parallel_hazard_for_modify(Oid typid, max_parallel_hazard_context *context) +{ + Relation con_rel; + ScanKeyData key[1]; + SysScanDesc scan; + HeapTuple tup; + bool found_max_hazard = false; + + LOCKMODE lockmode = AccessShareLock; + + con_rel = table_open(ConstraintRelationId, lockmode); + + ScanKeyInit(&key[0], + Anum_pg_constraint_contypid, BTEqualStrategyNumber, + F_OIDEQ, ObjectIdGetDatum(typid)); + scan = systable_beginscan(con_rel, ConstraintTypidIndexId, true, + NULL, 1, key); + + while (HeapTupleIsValid((tup = systable_getnext(scan)))) + { + Form_pg_constraint con = (Form_pg_constraint) GETSTRUCT(tup); + + if (con->contype == CONSTRAINT_CHECK) + { + char *conbin; + Datum val; + bool isnull; + Expr *check_expr; + + val = SysCacheGetAttr(CONSTROID, tup, + Anum_pg_constraint_conbin, &isnull); + Assert(!isnull); + if (isnull) + { + /* + * This shouldn't ever happen, but if it does, log a WARNING + * and return UNSAFE, rather than erroring out. + */ + elog(WARNING, "null conbin for constraint %u", con->oid); + context->max_hazard = PROPARALLEL_UNSAFE; + found_max_hazard = true; + break; + } + conbin = TextDatumGetCString(val); + check_expr = stringToNode(conbin); + if (max_parallel_hazard_walker((Node *) check_expr, context)) + { + found_max_hazard = true; + break; + } + } + } + + systable_endscan(scan); + table_close(con_rel, lockmode); + return found_max_hazard; +} + +/* + * rel_max_parallel_hazard_for_modify + * + * Determines the maximum parallel-mode hazard level for modification + * of a specified relation. + */ +static bool +rel_max_parallel_hazard_for_modify(Oid relid, + CmdType command_type, + max_parallel_hazard_context *context, + LOCKMODE lockmode) +{ + Relation rel; + TupleDesc tupdesc; + int attnum; + + /* Currently only CMD_INSERT is supported */ + Assert(command_type == CMD_INSERT); + + rel = table_open(relid, lockmode); + + /* + * We can't support table modification in parallel-mode if it's a foreign + * table/partition (no FDW API for supporting parallel access) or a + * temporary table. + */ + if (rel->rd_rel->relkind == RELKIND_FOREIGN_TABLE || + RelationUsesLocalBuffers(rel)) + { + table_close(rel, lockmode); + context->max_hazard = PROPARALLEL_UNSAFE; + return true; + } + + /* + * If a partitioned table, check that each partition is safe for + * modification in parallel-mode. + */ + if (rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE) + { + int i; + PartitionDesc pdesc; + PartitionKey pkey; + ListCell *partexprs_item; + int partnatts; + List *partexprs; + + pkey = RelationGetPartitionKey(rel); + + partnatts = get_partition_natts(pkey); + partexprs = get_partition_exprs(pkey); + + partexprs_item = list_head(partexprs); + for (i = 0; i < partnatts; i++) + { + /* Check parallel-safety of partition key support functions */ + if (OidIsValid(pkey->partsupfunc[i].fn_oid)) + { + if (max_parallel_hazard_test(func_parallel(pkey->partsupfunc[i].fn_oid), context)) + { + table_close(rel, lockmode); + return true; + } + } + + /* Check parallel-safety of any expressions in the partition key */ + if (get_partition_col_attnum(pkey, i) == 0) + { + Node *check_expr = (Node *) lfirst(partexprs_item); + + if (max_parallel_hazard_walker(check_expr, context)) + { + table_close(rel, lockmode); + return true; + } + + partexprs_item = lnext(partexprs, partexprs_item); + } + } + + /* Recursively check each partition ... */ + pdesc = RelationGetPartitionDesc(rel); + for (i = 0; i < pdesc->nparts; i++) + { + if (rel_max_parallel_hazard_for_modify(pdesc->oids[i], + command_type, + context, + AccessShareLock)) + { + table_close(rel, lockmode); + return true; + } + } + } + + /* + * If there are any index expressions, check that they are parallel-mode + * safe. + */ + if (index_expr_max_parallel_hazard_for_modify(rel, context)) + { + table_close(rel, lockmode); + return true; + } + + /* + * If any triggers exist, check that they are parallel safe. + */ + if (rel->trigdesc != NULL) + { + if (trigger_max_parallel_hazard_for_modify(rel->trigdesc, context)) + { + table_close(rel, lockmode); + return true; + } + } + + /* + * Column default expressions and check constraints are only applicable to + * INSERT and UPDATE, but since only INSERT is currently supported, only + * command_type==CMD_INSERT is checked here. + */ + if (command_type == CMD_INSERT) + { + /* + * Column default expressions for columns in the target-list are + * already being checked for parallel-safety in the + * max_parallel_hazard() scan of the query tree in standard_planner(). + */ + + tupdesc = RelationGetDescr(rel); + for (attnum = 0; attnum < tupdesc->natts; attnum++) + { + Form_pg_attribute att = TupleDescAttr(tupdesc, attnum); + + /* We don't need info for dropped or generated attributes */ + if (att->attisdropped || att->attgenerated) + continue; + + /* + * If the column is of a DOMAIN type, determine whether that + * domain has any CHECK expressions that are not parallel-mode + * safe. + */ + if (get_typtype(att->atttypid) == TYPTYPE_DOMAIN) + { + if (domain_max_parallel_hazard_for_modify(att->atttypid, context)) + { + table_close(rel, lockmode); + return true; + } + } + } + + /* + * Check if there are any CHECK constraints which are not + * parallel-safe. + */ + if (tupdesc->constr != NULL && tupdesc->constr->num_check > 0) + { + int i; + + ConstrCheck *check = tupdesc->constr->check; + + for (i = 0; i < tupdesc->constr->num_check; i++) + { + Expr *check_expr = stringToNode(check->ccbin); + + if (max_parallel_hazard_walker((Node *) check_expr, context)) + { + table_close(rel, lockmode); + return true; + } + } + } + } + + table_close(rel, lockmode); + return false; +} + +/* + * max_parallel_hazard_for_modify + * + * Determines the worst parallel-mode hazard level for the specified + * table-modification statement, based on the statement attributes and + * target table. An initial max parallel hazard level may optionally be + * supplied. The search returns the earliest in the following list: + * PROPARALLEL_UNSAFE, PROPARALLEL_RESTRICTED, PROPARALLEL_SAFE + */ +char +max_parallel_hazard_for_modify(Query *parse, char initial_max_parallel_hazard) +{ + RangeTblEntry *rte; + ListCell *lc; + bool hasSubQuery; + max_parallel_hazard_context context; + + + /* + * UPDATE is not currently supported in parallel-mode, so prohibit + * INSERT...ON CONFLICT...DO UPDATE... + * In order to support update, even if only in the leader, some + * further work would need to be done. A mechanism would be needed + * for sharing combo-cids between leader and workers during + * parallel-mode, since for example, the leader might generate a + * combo-cid and it needs to be propagated to the workers. + */ + if (parse->onConflict != NULL && parse->onConflict->action == ONCONFLICT_UPDATE) + return PROPARALLEL_UNSAFE; + + /* + * If there is no underlying query, a parallel table-modification + * operation is not possible (nor desirable). + */ + hasSubQuery = false; + foreach(lc, parse->rtable) + { + rte = lfirst_node(RangeTblEntry, lc); + if (rte->rtekind == RTE_SUBQUERY) + { + hasSubQuery = true; + break; + } + } + if (!hasSubQuery) + return PROPARALLEL_UNSAFE; + + /* + * Setup the context used in finding the max parallel-mode hazard. + */ + Assert(initial_max_parallel_hazard == 0 || + initial_max_parallel_hazard == PROPARALLEL_SAFE || + initial_max_parallel_hazard == PROPARALLEL_RESTRICTED); + context.max_hazard = initial_max_parallel_hazard == 0 ? + PROPARALLEL_SAFE : initial_max_parallel_hazard; + context.max_interesting = PROPARALLEL_UNSAFE; + context.safe_param_ids = NIL; + + rte = rt_fetch(parse->resultRelation, parse->rtable); + + /* + * The target table is already locked by the caller (this is done in the + * parse/analyze phase). + */ + (void) rel_max_parallel_hazard_for_modify(rte->relid, parse->commandType, &context, NoLock); + return context.max_hazard; +} /***************************************************************************** * Check clauses for nonstrict functions diff --git a/src/include/access/xact.h b/src/include/access/xact.h index f49a57b35e..5571e8c0a2 100644 --- a/src/include/access/xact.h +++ b/src/include/access/xact.h @@ -466,5 +466,20 @@ extern void ParsePrepareRecord(uint8 info, xl_xact_prepare *xlrec, xl_xact_parse extern void EnterParallelMode(void); extern void ExitParallelMode(void); extern bool IsInParallelMode(void); +extern void PrepareParallelMode(CmdType commandType); + +/* + * IsModifySupportedInParallelMode + * + * Indicates whether execution of the specified table-modification command + * (INSERT/UPDATE/DELETE) in parallel-mode is supported, subject to certain + * parallel-safety conditions. + */ +static inline bool +IsModifySupportedInParallelMode(CmdType commandType) +{ + /* Currently only INSERT is supported */ + return (commandType == CMD_INSERT); +} #endif /* XACT_H */ diff --git a/src/include/optimizer/clauses.h b/src/include/optimizer/clauses.h index da3fc4df10..cc3cfc16b2 100644 --- a/src/include/optimizer/clauses.h +++ b/src/include/optimizer/clauses.h @@ -52,5 +52,6 @@ extern void CommuteOpExpr(OpExpr *clause); extern Query *inline_set_returning_function(PlannerInfo *root, RangeTblEntry *rte); +extern char max_parallel_hazard_for_modify(Query *parse, char initial_max_parallel_hazard); #endif /* CLAUSES_H */ -- 2.27.0