From 7cad3cf052856ec9f5e087f1edec1c24b920dc74 Mon Sep 17 00:00:00 2001 From: houzj Date: Mon, 31 May 2021 09:32:54 +0800 Subject: [PATCH v14 2/4] parallel-SELECT-for-INSERT Enable parallel select for insert. Prepare for entering parallel mode by assigning a TransactionId. --- src/backend/access/transam/xact.c | 26 +++++++++ src/backend/executor/execMain.c | 3 + src/backend/optimizer/plan/planner.c | 21 +++---- src/backend/optimizer/util/clauses.c | 87 +++++++++++++++++++++++++++- src/include/access/xact.h | 15 +++++ src/include/optimizer/clauses.h | 2 + 6 files changed, 143 insertions(+), 11 deletions(-) diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c index 441445927e..2d68e4633a 100644 --- a/src/backend/access/transam/xact.c +++ b/src/backend/access/transam/xact.c @@ -1014,6 +1014,32 @@ IsInParallelMode(void) return CurrentTransactionState->parallelModeLevel != 0; } +/* + * PrepareParallelModePlanExec + * + * Prepare for entering parallel mode plan execution, based on command-type. + */ +void +PrepareParallelModePlanExec(CmdType commandType) +{ + if (IsModifySupportedInParallelMode(commandType)) + { + Assert(!IsInParallelMode()); + + /* + * Prepare for entering parallel mode by assigning a TransactionId. + * Failure to do this now would result in heap_insert() subsequently + * attempting to assign a TransactionId whilst in parallel-mode, which + * is not allowed. + * + * This approach has a disadvantage in that if the underlying SELECT + * does not return any rows, then the TransactionId is not used, + * however that shouldn't happen in practice in many cases. + */ + (void) GetCurrentTransactionId(); + } +} + /* * CommandCounterIncrement */ diff --git a/src/backend/executor/execMain.c b/src/backend/executor/execMain.c index b3ce4bae53..ea685f0846 100644 --- a/src/backend/executor/execMain.c +++ b/src/backend/executor/execMain.c @@ -1535,7 +1535,10 @@ ExecutePlan(EState *estate, estate->es_use_parallel_mode = use_parallel_mode; if (use_parallel_mode) + { + PrepareParallelModePlanExec(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 1868c4eff4..7736813230 100644 --- a/src/backend/optimizer/plan/planner.c +++ b/src/backend/optimizer/plan/planner.c @@ -314,16 +314,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 (except for Insert), 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 @@ -332,7 +332,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 || + is_parallel_allowed_for_modify(parse)) && !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 7187f17da5..ac0f243bf1 100644 --- a/src/backend/optimizer/util/clauses.c +++ b/src/backend/optimizer/util/clauses.c @@ -20,6 +20,8 @@ #include "postgres.h" #include "access/htup_details.h" +#include "access/table.h" +#include "access/xact.h" #include "catalog/pg_aggregate.h" #include "catalog/pg_class.h" #include "catalog/pg_language.h" @@ -43,6 +45,7 @@ #include "parser/parse_agg.h" #include "parser/parse_coerce.h" #include "parser/parse_func.h" +#include "parser/parsetree.h" #include "rewrite/rewriteHandler.h" #include "rewrite/rewriteManip.h" #include "tcop/tcopprot.h" @@ -51,6 +54,7 @@ #include "utils/fmgroids.h" #include "utils/lsyscache.h" #include "utils/memutils.h" +#include "utils/rel.h" #include "utils/syscache.h" #include "utils/typcache.h" @@ -151,6 +155,7 @@ static Query *substitute_actual_srf_parameters(Query *expr, int nargs, List *args); static Node *substitute_actual_srf_parameters_mutator(Node *node, substitute_actual_srf_parameters_context *context); +static bool max_parallel_hazard_test(char proparallel, max_parallel_hazard_context *context); /***************************************************************************** @@ -618,12 +623,34 @@ contain_volatile_functions_not_nextval_walker(Node *node, void *context) char max_parallel_hazard(Query *parse) { + bool max_hazard_found; max_parallel_hazard_context context; context.max_hazard = PROPARALLEL_SAFE; context.max_interesting = PROPARALLEL_UNSAFE; context.safe_param_ids = NIL; - (void) max_parallel_hazard_walker((Node *) parse, &context); + + max_hazard_found = max_parallel_hazard_walker((Node *) parse, &context); + + if (!max_hazard_found && + IsModifySupportedInParallelMode(parse->commandType)) + { + RangeTblEntry *rte; + Relation target_rel; + + rte = rt_fetch(parse->resultRelation, parse->rtable); + + /* + * The target table is already locked by the caller (this is done in the + * parse/analyze phase), and remains locked until end-of-transaction. + */ + target_rel = table_open(rte->relid, NoLock); + + (void) max_parallel_hazard_test(target_rel->rd_rel->relparalleldml, + &context); + table_close(target_rel, NoLock); + } + return context.max_hazard; } @@ -857,6 +884,64 @@ max_parallel_hazard_walker(Node *node, max_parallel_hazard_context *context) context); } +/* + * is_parallel_allowed_for_modify + * + * Check at a high-level if parallel mode is able to be used for the specified + * table-modification statement. Currently, we support only Inserts. + * + * It's not possible in the following cases: + * + * 1) INSERT...ON CONFLICT...DO UPDATE + * 2) INSERT without SELECT + * + * (Note: we don't do in-depth parallel-safety checks here, we do only the + * cheaper tests that can quickly exclude obvious cases for which + * parallelism isn't supported, to avoid having to do further parallel-safety + * checks for these) + */ +bool +is_parallel_allowed_for_modify(Query *parse) +{ + bool hasSubQuery; + RangeTblEntry *rte; + ListCell *lc; + + if (!IsModifySupportedInParallelMode(parse->commandType)) + return false; + + /* + * 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->commandType == CMD_INSERT && + parse->onConflict != NULL && + parse->onConflict->action == ONCONFLICT_UPDATE) + return false; + + /* + * If there is no underlying SELECT, a parallel insert operation is not + * desirable. + */ + hasSubQuery = false; + foreach(lc, parse->rtable) + { + rte = lfirst_node(RangeTblEntry, lc); + if (rte->rtekind == RTE_SUBQUERY) + { + hasSubQuery = true; + break; + } + } + + return hasSubQuery; +} /***************************************************************************** * Check clauses for nonstrict functions diff --git a/src/include/access/xact.h b/src/include/access/xact.h index 134f6862da..fd3f86bf7c 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 PrepareParallelModePlanExec(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 0673887a85..32b56565e5 100644 --- a/src/include/optimizer/clauses.h +++ b/src/include/optimizer/clauses.h @@ -53,4 +53,6 @@ extern void CommuteOpExpr(OpExpr *clause); extern Query *inline_set_returning_function(PlannerInfo *root, RangeTblEntry *rte); +extern bool is_parallel_allowed_for_modify(Query *parse); + #endif /* CLAUSES_H */ -- 2.27.0