forked from huawei/openGauss-server
!252 并行查询增加NUMA绑核
Merge pull request !252 from TotaJ/feature/parallel_performance
This commit is contained in:
Gitee
commit
7662afaa4c
|
|
@ -1,61 +1,61 @@
|
|||
#
|
||||
# Copyright (c) 2020 Huawei Technologies Co.,Ltd.
|
||||
#
|
||||
# openGauss is licensed under Mulan PSL v2.
|
||||
# You can use this software according to the terms and conditions of the Mulan PSL v2.
|
||||
# You may obtain a copy of Mulan PSL v2 at:
|
||||
#
|
||||
# http://license.coscl.org.cn/MulanPSL2
|
||||
#
|
||||
# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
|
||||
# EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
|
||||
# MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
|
||||
# See the Mulan PSL v2 for more details.
|
||||
# ---------------------------------------------------------------------------------------
|
||||
#
|
||||
# Makefile
|
||||
# Makefile for the mysql_fdw
|
||||
#
|
||||
# IDENTIFICATION
|
||||
# contrib/mysql_fdw/Makefile
|
||||
#
|
||||
# ---------------------------------------------------------------------------------------
|
||||
|
||||
all:mysql_fdw_target
|
||||
install:install-data
|
||||
|
||||
top_builddir ?= ../../
|
||||
MYSQL_FDW_DIR=$(top_builddir)/third_party/dependency/mysql_fdw
|
||||
MYSQL_FDW_PACKAGE=mysql_fdw-REL-2_5_3
|
||||
MYSQL_FDW_PATCH=huawei_mysql_fdw-2.5.3_patch
|
||||
MYSQL_FDW_MEGRED_SOURCES_DIR=$(MYSQL_FDW_DIR)/code
|
||||
|
||||
.PHONY: mysql_fdw_target
|
||||
mysql_fdw_target:
|
||||
@$(call create_mysql_fdw_sources)
|
||||
@make -C $(MYSQL_FDW_MEGRED_SOURCES_DIR)/$(MYSQL_FDW_PACKAGE)
|
||||
|
||||
.PHONY: install-data
|
||||
install-data: mysql_fdw_target
|
||||
@make -C $(MYSQL_FDW_MEGRED_SOURCES_DIR)/$(MYSQL_FDW_PACKAGE) install
|
||||
|
||||
uninstall distclean clean:
|
||||
@rm -rf $(MYSQL_FDW_MEGRED_SOURCES_DIR)
|
||||
|
||||
MYSQL_FDW_RELEVANT_SOURCES = connection.c deparse.c mysql_fdw.c mysql_query.c option.c
|
||||
|
||||
define create_mysql_fdw_sources
|
||||
rm -rf $(MYSQL_FDW_MEGRED_SOURCES_DIR); \
|
||||
mkdir $(MYSQL_FDW_MEGRED_SOURCES_DIR); \
|
||||
tar xfzv $(MYSQL_FDW_DIR)/$(MYSQL_FDW_PACKAGE).tar.gz -C $(MYSQL_FDW_MEGRED_SOURCES_DIR) &> /dev/null; \
|
||||
for ((i=1;i<=99;i++)); \
|
||||
do \
|
||||
file_name="$(MYSQL_FDW_DIR)/$$i-mysql_fdw-2.5.3_patch.patch"; \
|
||||
if [ ! -f "$$file_name" ]; then \
|
||||
exit 0; \
|
||||
fi; \
|
||||
patch -p0 -d $(MYSQL_FDW_MEGRED_SOURCES_DIR)/$(MYSQL_FDW_PACKAGE) < $$file_name &> /dev/null; \
|
||||
done
|
||||
rename ".c" ".cpp" $(MYSQL_FDW_MEGRED_SOURCES_DIR)/$(MYSQL_FDW_PACKAGE)/*.c; \
|
||||
patch -p0 -d $(MYSQL_FDW_MEGRED_SOURCES_DIR)/$(MYSQL_FDW_PACKAGE) < $(MYSQL_FDW_DIR)/$(MYSQL_FDW_PATCH).patch &> /dev/null;
|
||||
endef
|
||||
#
|
||||
# Copyright (c) 2020 Huawei Technologies Co.,Ltd.
|
||||
#
|
||||
# openGauss is licensed under Mulan PSL v2.
|
||||
# You can use this software according to the terms and conditions of the Mulan PSL v2.
|
||||
# You may obtain a copy of Mulan PSL v2 at:
|
||||
#
|
||||
# http://license.coscl.org.cn/MulanPSL2
|
||||
#
|
||||
# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
|
||||
# EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
|
||||
# MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
|
||||
# See the Mulan PSL v2 for more details.
|
||||
# ---------------------------------------------------------------------------------------
|
||||
#
|
||||
# Makefile
|
||||
# Makefile for the mysql_fdw
|
||||
#
|
||||
# IDENTIFICATION
|
||||
# contrib/mysql_fdw/Makefile
|
||||
#
|
||||
# ---------------------------------------------------------------------------------------
|
||||
|
||||
all:mysql_fdw_target
|
||||
install:install-data
|
||||
|
||||
top_builddir ?= ../../
|
||||
MYSQL_FDW_DIR=$(top_builddir)/third_party/dependency/mysql_fdw
|
||||
MYSQL_FDW_PACKAGE=mysql_fdw-REL-2_5_3
|
||||
MYSQL_FDW_PATCH=openGauss_mysql_fdw-2.5.3_patch
|
||||
MYSQL_FDW_MEGRED_SOURCES_DIR=$(MYSQL_FDW_DIR)/code
|
||||
|
||||
.PHONY: mysql_fdw_target
|
||||
mysql_fdw_target:
|
||||
@$(call create_mysql_fdw_sources)
|
||||
@make -C $(MYSQL_FDW_MEGRED_SOURCES_DIR)/$(MYSQL_FDW_PACKAGE)
|
||||
|
||||
.PHONY: install-data
|
||||
install-data: mysql_fdw_target
|
||||
@make -C $(MYSQL_FDW_MEGRED_SOURCES_DIR)/$(MYSQL_FDW_PACKAGE) install
|
||||
|
||||
uninstall distclean clean:
|
||||
@rm -rf $(MYSQL_FDW_MEGRED_SOURCES_DIR)
|
||||
|
||||
MYSQL_FDW_RELEVANT_SOURCES = connection.c deparse.c mysql_fdw.c mysql_query.c option.c
|
||||
|
||||
define create_mysql_fdw_sources
|
||||
rm -rf $(MYSQL_FDW_MEGRED_SOURCES_DIR); \
|
||||
mkdir $(MYSQL_FDW_MEGRED_SOURCES_DIR); \
|
||||
tar xfzv $(MYSQL_FDW_DIR)/$(MYSQL_FDW_PACKAGE).tar.gz -C $(MYSQL_FDW_MEGRED_SOURCES_DIR) &> /dev/null; \
|
||||
for ((i=1;i<=99;i++)); \
|
||||
do \
|
||||
file_name="$(MYSQL_FDW_DIR)/$$i-mysql_fdw-2.5.3_patch.patch"; \
|
||||
if [ ! -f "$$file_name" ]; then \
|
||||
exit 0; \
|
||||
fi; \
|
||||
patch -p0 -d $(MYSQL_FDW_MEGRED_SOURCES_DIR)/$(MYSQL_FDW_PACKAGE) < $$file_name &> /dev/null; \
|
||||
done
|
||||
rename ".c" ".cpp" $(MYSQL_FDW_MEGRED_SOURCES_DIR)/$(MYSQL_FDW_PACKAGE)/*.c; \
|
||||
patch -p0 -d $(MYSQL_FDW_MEGRED_SOURCES_DIR)/$(MYSQL_FDW_PACKAGE) < $(MYSQL_FDW_DIR)/$(MYSQL_FDW_PATCH).patch &> /dev/null;
|
||||
endef
|
||||
|
|
|
|||
|
|
@ -1,59 +1,59 @@
|
|||
#
|
||||
# Copyright (c) 2020 Huawei Technologies Co.,Ltd.
|
||||
#
|
||||
# openGauss is licensed under Mulan PSL v2.
|
||||
# You can use this software according to the terms and conditions of the Mulan PSL v2.
|
||||
# You may obtain a copy of Mulan PSL v2 at:
|
||||
#
|
||||
# http://license.coscl.org.cn/MulanPSL2
|
||||
#
|
||||
# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
|
||||
# EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
|
||||
# MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
|
||||
# See the Mulan PSL v2 for more details.
|
||||
# ---------------------------------------------------------------------------------------
|
||||
#
|
||||
# Makefile
|
||||
# Makefile for the oracle_fdw
|
||||
#
|
||||
# IDENTIFICATION
|
||||
# contrib/oracle_fdw/Makefile
|
||||
#
|
||||
# ---------------------------------------------------------------------------------------
|
||||
|
||||
all:oracle_fdw_target
|
||||
install:install-data
|
||||
|
||||
top_builddir ?= ../../
|
||||
ORACLE_FDW_DIR=$(top_builddir)/third_party/dependency/oracle_fdw
|
||||
ORACLE_FDW_PACKAGE=oracle_fdw-ORACLE_FDW_2_2_0
|
||||
ORACLE_FDW_PATCH=huawei_oracle_fdw-2.2.0_patch
|
||||
ORACLE_FDW_MEGRED_SOURCES_DIR=$(ORACLE_FDW_DIR)/code
|
||||
|
||||
.PHONY: oracle_fdw_target
|
||||
oracle_fdw_target:
|
||||
@$(call create_oracle_fdw_sources)
|
||||
@make -C $(ORACLE_FDW_MEGRED_SOURCES_DIR)/$(ORACLE_FDW_PACKAGE) NO_PGXS=1
|
||||
|
||||
.PHONY: install-data
|
||||
install-data: oracle_fdw_target
|
||||
@make -C $(ORACLE_FDW_MEGRED_SOURCES_DIR)/$(ORACLE_FDW_PACKAGE) NO_PGXS=1 install
|
||||
|
||||
uninstall distclean clean:
|
||||
@rm -rf $(ORACLE_FDW_MEGRED_SOURCES_DIR)
|
||||
|
||||
define create_oracle_fdw_sources
|
||||
rm -rf $(ORACLE_FDW_MEGRED_SOURCES_DIR); \
|
||||
mkdir $(ORACLE_FDW_MEGRED_SOURCES_DIR); \
|
||||
tar xfzv $(ORACLE_FDW_DIR)/$(ORACLE_FDW_PACKAGE).tar.gz -C $(ORACLE_FDW_MEGRED_SOURCES_DIR) &> /dev/null; \
|
||||
for ((i=1;i<=99;i++)); \
|
||||
do \
|
||||
file_name="$(ORACLE_FDW_DIR)/$$i-oracle_fdw-2.2.0_patch.patch"; \
|
||||
if [ ! -f "$$file_name" ]; then \
|
||||
exit 0; \
|
||||
fi; \
|
||||
patch -p0 -d $(ORACLE_FDW_MEGRED_SOURCES_DIR)/$(ORACLE_FDW_PACKAGE) < $$file_name &> /dev/null; \
|
||||
done
|
||||
rename ".c" ".cpp" $(ORACLE_FDW_MEGRED_SOURCES_DIR)/$(ORACLE_FDW_PACKAGE)/*.c; \
|
||||
patch -p0 -d $(ORACLE_FDW_MEGRED_SOURCES_DIR)/$(ORACLE_FDW_PACKAGE) < $(ORACLE_FDW_DIR)/$(ORACLE_FDW_PATCH).patch &> /dev/null;
|
||||
endef
|
||||
#
|
||||
# Copyright (c) 2020 Huawei Technologies Co.,Ltd.
|
||||
#
|
||||
# openGauss is licensed under Mulan PSL v2.
|
||||
# You can use this software according to the terms and conditions of the Mulan PSL v2.
|
||||
# You may obtain a copy of Mulan PSL v2 at:
|
||||
#
|
||||
# http://license.coscl.org.cn/MulanPSL2
|
||||
#
|
||||
# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
|
||||
# EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
|
||||
# MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
|
||||
# See the Mulan PSL v2 for more details.
|
||||
# ---------------------------------------------------------------------------------------
|
||||
#
|
||||
# Makefile
|
||||
# Makefile for the oracle_fdw
|
||||
#
|
||||
# IDENTIFICATION
|
||||
# contrib/oracle_fdw/Makefile
|
||||
#
|
||||
# ---------------------------------------------------------------------------------------
|
||||
|
||||
all:oracle_fdw_target
|
||||
install:install-data
|
||||
|
||||
top_builddir ?= ../../
|
||||
ORACLE_FDW_DIR=$(top_builddir)/third_party/dependency/oracle_fdw
|
||||
ORACLE_FDW_PACKAGE=oracle_fdw-ORACLE_FDW_2_2_0
|
||||
ORACLE_FDW_PATCH=openGauss_oracle_fdw-2.2.0_patch
|
||||
ORACLE_FDW_MEGRED_SOURCES_DIR=$(ORACLE_FDW_DIR)/code
|
||||
|
||||
.PHONY: oracle_fdw_target
|
||||
oracle_fdw_target:
|
||||
@$(call create_oracle_fdw_sources)
|
||||
@make -C $(ORACLE_FDW_MEGRED_SOURCES_DIR)/$(ORACLE_FDW_PACKAGE) NO_PGXS=1
|
||||
|
||||
.PHONY: install-data
|
||||
install-data: oracle_fdw_target
|
||||
@make -C $(ORACLE_FDW_MEGRED_SOURCES_DIR)/$(ORACLE_FDW_PACKAGE) NO_PGXS=1 install
|
||||
|
||||
uninstall distclean clean:
|
||||
@rm -rf $(ORACLE_FDW_MEGRED_SOURCES_DIR)
|
||||
|
||||
define create_oracle_fdw_sources
|
||||
rm -rf $(ORACLE_FDW_MEGRED_SOURCES_DIR); \
|
||||
mkdir $(ORACLE_FDW_MEGRED_SOURCES_DIR); \
|
||||
tar xfzv $(ORACLE_FDW_DIR)/$(ORACLE_FDW_PACKAGE).tar.gz -C $(ORACLE_FDW_MEGRED_SOURCES_DIR) &> /dev/null; \
|
||||
for ((i=1;i<=99;i++)); \
|
||||
do \
|
||||
file_name="$(ORACLE_FDW_DIR)/$$i-oracle_fdw-2.2.0_patch.patch"; \
|
||||
if [ ! -f "$$file_name" ]; then \
|
||||
exit 0; \
|
||||
fi; \
|
||||
patch -p0 -d $(ORACLE_FDW_MEGRED_SOURCES_DIR)/$(ORACLE_FDW_PACKAGE) < $$file_name &> /dev/null; \
|
||||
done
|
||||
rename ".c" ".cpp" $(ORACLE_FDW_MEGRED_SOURCES_DIR)/$(ORACLE_FDW_PACKAGE)/*.c; \
|
||||
patch -p0 -d $(ORACLE_FDW_MEGRED_SOURCES_DIR)/$(ORACLE_FDW_PACKAGE) < $(ORACLE_FDW_DIR)/$(ORACLE_FDW_PATCH).patch &> /dev/null;
|
||||
endef
|
||||
|
|
|
|||
|
|
@ -528,7 +528,6 @@ max_inner_tool_connections|int|1,8388607|NULL|NULL|
|
|||
max_keep_log_seg|int|0,2147483647|NULL|NULL|
|
||||
max_background_workers|int|0,262143|NULL|NULL|
|
||||
min_parallel_table_scan_size|int|0,715827882|kB|NULL|
|
||||
max_parallel_workers|int|0,1024|NULL|NULL|
|
||||
max_parallel_workers_per_gather|int|0,1024|NULL|NULL|
|
||||
parallel_tuple_cost|real|0,1.79769e+308|NULL|NULL|
|
||||
parallel_setup_cost|real|0,1.79769e+308|NULL|NULL|
|
||||
|
|
|
|||
|
|
@ -9229,22 +9229,6 @@ static void init_configure_names_int()
|
|||
NULL,
|
||||
NULL
|
||||
},
|
||||
{
|
||||
{
|
||||
"max_parallel_workers",
|
||||
PGC_USERSET,
|
||||
RESOURCES_ASYNCHRONOUS,
|
||||
gettext_noop("Sets the maximum number of parallel workers that can be active at one time."),
|
||||
NULL
|
||||
},
|
||||
&g_instance.attr.attr_common.max_parallel_workers,
|
||||
8,
|
||||
0,
|
||||
MAX_PARALLEL_WORKER_LIMIT,
|
||||
NULL,
|
||||
NULL,
|
||||
NULL
|
||||
},
|
||||
{
|
||||
{
|
||||
"max_parallel_workers_per_gather",
|
||||
|
|
@ -9253,7 +9237,7 @@ static void init_configure_names_int()
|
|||
gettext_noop("Sets the maximum number of parallel processes per executor node."),
|
||||
NULL
|
||||
},
|
||||
&g_instance.attr.attr_common.max_parallel_workers_per_gather,
|
||||
&u_sess->attr.attr_sql.max_parallel_workers_per_gather,
|
||||
2,
|
||||
0,
|
||||
MAX_PARALLEL_WORKER_LIMIT,
|
||||
|
|
|
|||
|
|
@ -898,17 +898,19 @@ bool HeapTupleSatisfiesMVCC(HeapTuple htup, Snapshot snapshot, Buffer buffer)
|
|||
TransactionIdStatus hintstatus;
|
||||
Page page = BufferGetPage(buffer);
|
||||
|
||||
ereport(DEBUG1,
|
||||
(errmsg("HeapTupleSatisfiesMVCC self(%u,%u) ctid(%u,%u) cur_xid " XID_FMT " xmin " XID_FMT
|
||||
" xmax " XID_FMT " csn " CSN_FMT,
|
||||
ItemPointerGetBlockNumber(&htup->t_self),
|
||||
ItemPointerGetOffsetNumber(&htup->t_self),
|
||||
ItemPointerGetBlockNumber(&tuple->t_ctid),
|
||||
ItemPointerGetOffsetNumber(&tuple->t_ctid),
|
||||
GetCurrentTransactionIdIfAny(),
|
||||
HeapTupleHeaderGetXmin(page, tuple),
|
||||
HeapTupleHeaderGetXmax(page, tuple),
|
||||
snapshot->snapshotcsn)));
|
||||
if (SHOW_DEBUG_MESSAGE()) {
|
||||
ereport(DEBUG1,
|
||||
(errmsg("HeapTupleSatisfiesMVCC self(%u,%u) ctid(%u,%u) cur_xid " XID_FMT " xmin " XID_FMT
|
||||
" xmax " XID_FMT " csn " CSN_FMT,
|
||||
ItemPointerGetBlockNumber(&htup->t_self),
|
||||
ItemPointerGetOffsetNumber(&htup->t_self),
|
||||
ItemPointerGetBlockNumber(&tuple->t_ctid),
|
||||
ItemPointerGetOffsetNumber(&tuple->t_ctid),
|
||||
GetCurrentTransactionIdIfAny(),
|
||||
HeapTupleHeaderGetXmin(page, tuple),
|
||||
HeapTupleHeaderGetXmax(page, tuple),
|
||||
snapshot->snapshotcsn)));
|
||||
}
|
||||
|
||||
/*
|
||||
* Just valid for read-only transaction when u_sess->attr.attr_common.XactReadOnly is true.
|
||||
|
|
|
|||
|
|
@ -899,7 +899,7 @@ static void set_plain_rel_pathlist(PlannerInfo* root, RelOptInfo* rel, RangeTblE
|
|||
* sophisticated, but we need something here for now.
|
||||
*/
|
||||
while (rel->pages > parallel_threshold * 3 &&
|
||||
parallel_degree < g_instance.attr.attr_common.max_parallel_workers_per_gather) {
|
||||
parallel_degree < u_sess->attr.attr_sql.max_parallel_workers_per_gather) {
|
||||
parallel_degree++;
|
||||
parallel_threshold *= 3;
|
||||
if (parallel_threshold >= PG_INT32_MAX / 3)
|
||||
|
|
|
|||
|
|
@ -462,7 +462,7 @@ PlannedStmt* standard_planner(Query* parse, int cursorOptions, ParamListInfo bou
|
|||
*/
|
||||
glob->parallelModeOK = (cursorOptions & CURSOR_OPT_PARALLEL_OK) != 0 && IsUnderPostmaster &&
|
||||
parse->commandType == CMD_SELECT && !parse->hasModifyingCTE && parse->utilityStmt == NULL &&
|
||||
g_instance.attr.attr_common.max_parallel_workers_per_gather > 0 && !IsParallelWorker() &&
|
||||
u_sess->attr.attr_sql.max_parallel_workers_per_gather > 0 && !IsParallelWorker() &&
|
||||
!IsolationIsSerializable() && !has_parallel_hazard((Node *)parse, true);
|
||||
|
||||
/*
|
||||
|
|
|
|||
|
|
@ -683,7 +683,7 @@ void StartBackgroundWorker(void* bgWorkerSlotShmAddr)
|
|||
BackgroundWorker *worker = t_thrd.bgworker_cxt.my_bgworker_entry;
|
||||
bgworker_main_type entrypt;
|
||||
|
||||
t_thrd.proc_cxt.MyProgName = "BackgroundWorker";
|
||||
knl_thread_set_name("BgWorker");
|
||||
/*
|
||||
* Create memory context and buffer used for RowDescription messages. As
|
||||
* SendRowDescriptionMessage(), via exec_describe_statement_message(), is
|
||||
|
|
|
|||
File diff suppressed because it is too large
Load Diff
|
|
@ -1,434 +1,436 @@
|
|||
/* -------------------------------------------------------------------------
|
||||
*
|
||||
* nodeGather.c
|
||||
* Support routines for scanning a plan via multiple workers.
|
||||
*
|
||||
* Portions Copyright (c) 1996-2015, PostgreSQL Global Development Group
|
||||
* Portions Copyright (c) 1994, Regents of the University of California
|
||||
*
|
||||
* A Gather executor launches parallel workers to run multiple copies of a
|
||||
* plan. It can also run the plan itself, if the workers are not available
|
||||
* or have not started up yet. It then merges all of the results it produces
|
||||
* and the results from the workers into a single output stream. Therefore,
|
||||
* it will normally be used with a plan where running multiple copies of the
|
||||
* same plan does not produce duplicate output, such as parallel-aware
|
||||
* SeqScan.
|
||||
*
|
||||
* Alternatively, a Gather node can be configured to use just one worker
|
||||
* and the single-copy flag can be set. In this case, the Gather node will
|
||||
* run the plan in one worker and will not execute the plan itself. In
|
||||
* this case, it simply returns whatever tuples were returned by the worker.
|
||||
* If a worker cannot be obtained, then it will run the plan itself and
|
||||
* return the results. Therefore, a plan used with a single-copy Gather
|
||||
* node need not be parallel-aware.
|
||||
*
|
||||
* IDENTIFICATION
|
||||
* src/backend/executor/nodeGather.c
|
||||
*
|
||||
* -------------------------------------------------------------------------
|
||||
*/
|
||||
|
||||
#include "postgres.h"
|
||||
|
||||
#include "access/relscan.h"
|
||||
#include "access/xact.h"
|
||||
#include "executor/execdebug.h"
|
||||
#include "executor/execParallel.h"
|
||||
#include "executor/nodeGather.h"
|
||||
#include "executor/nodeSubplan.h"
|
||||
#include "executor/tqueue.h"
|
||||
#include "miscadmin.h"
|
||||
#include "utils/memutils.h"
|
||||
#include "utils/rel.h"
|
||||
|
||||
|
||||
static TupleTableSlot *gather_getnext(GatherState *gatherstate);
|
||||
static HeapTuple gather_readnext(GatherState *gatherstate);
|
||||
static void ExecShutdownGatherWorkers(GatherState *node);
|
||||
|
||||
|
||||
/* ----------------------------------------------------------------
|
||||
* ExecInitGather
|
||||
* ----------------------------------------------------------------
|
||||
*/
|
||||
GatherState *ExecInitGather(Gather *node, EState *estate, int eflags)
|
||||
{
|
||||
bool hasoid = false;
|
||||
|
||||
/* Gather node doesn't have innerPlan node. */
|
||||
Assert(innerPlan(node) == NULL);
|
||||
|
||||
/*
|
||||
* create state structure
|
||||
*/
|
||||
GatherState *gatherstate = makeNode(GatherState);
|
||||
gatherstate->ps.plan = (Plan *)node;
|
||||
gatherstate->ps.state = estate;
|
||||
gatherstate->need_to_scan_locally = !node->single_copy &&
|
||||
u_sess->attr.attr_sql.parallel_leader_participation;
|
||||
|
||||
/*
|
||||
* Miscellaneous initialization
|
||||
*
|
||||
* create expression context for node
|
||||
*/
|
||||
ExecAssignExprContext(estate, &gatherstate->ps);
|
||||
|
||||
/*
|
||||
* initialize child expressions
|
||||
*/
|
||||
gatherstate->ps.targetlist = (List *)ExecInitExpr((Expr *)node->plan.targetlist, (PlanState *)gatherstate);
|
||||
gatherstate->ps.qual = (List *)ExecInitExpr((Expr *)node->plan.qual, (PlanState *)gatherstate);
|
||||
|
||||
/*
|
||||
* tuple table initialization
|
||||
*/
|
||||
gatherstate->funnel_slot = ExecInitExtraTupleSlot(estate);
|
||||
ExecInitResultTupleSlot(estate, &gatherstate->ps);
|
||||
|
||||
/*
|
||||
* now initialize outer plan
|
||||
*/
|
||||
Plan *outerNode = outerPlan(node);
|
||||
outerPlanState(gatherstate) = ExecInitNode(outerNode, estate, eflags);
|
||||
|
||||
gatherstate->ps.ps_TupFromTlist = false;
|
||||
|
||||
/*
|
||||
* Initialize result tuple type and projection info.
|
||||
*/
|
||||
ExecAssignResultTypeFromTL(&gatherstate->ps);
|
||||
ExecAssignProjectionInfo(&gatherstate->ps, NULL);
|
||||
|
||||
/*
|
||||
* Initialize funnel slot to same tuple descriptor as outer plan.
|
||||
*/
|
||||
if (!ExecContextForcesOids(&gatherstate->ps, &hasoid))
|
||||
hasoid = false;
|
||||
TupleDesc tupDesc = ExecTypeFromTL(outerNode->targetlist, hasoid);
|
||||
ExecSetSlotDescriptor(gatherstate->funnel_slot, tupDesc);
|
||||
|
||||
return gatherstate;
|
||||
}
|
||||
|
||||
/* ----------------------------------------------------------------
|
||||
* ExecGather(node)
|
||||
*
|
||||
* Scans the relation via multiple workers and returns
|
||||
* the next qualifying tuple.
|
||||
* ----------------------------------------------------------------
|
||||
*/
|
||||
TupleTableSlot *ExecGather(GatherState *node)
|
||||
{
|
||||
TupleTableSlot *fslot = node->funnel_slot;
|
||||
int i;
|
||||
TupleTableSlot *slot = NULL;
|
||||
TupleTableSlot *resultSlot = NULL;
|
||||
ExprDoneCond isDone;
|
||||
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
|
||||
/*
|
||||
* Initialize the parallel context and workers on first execution. We do
|
||||
* this on first execution rather than during node initialization, as it
|
||||
* needs to allocate large dynamic segement, so it is better to do if it
|
||||
* is really needed.
|
||||
*/
|
||||
if (!node->initialized) {
|
||||
EState *estate = node->ps.state;
|
||||
Gather *gather = (Gather *)node->ps.plan;
|
||||
t_thrd.subrole = BACKGROUND_LEADER;
|
||||
|
||||
/*
|
||||
* Sometimes we might have to run without parallelism; but if
|
||||
* parallel mode is active then we can try to fire up some workers.
|
||||
*/
|
||||
if (gather->num_workers > 0 && IsInParallelMode()) {
|
||||
bool got_any_worker = false;
|
||||
|
||||
/* Initialize the workers required to execute Gather node. */
|
||||
if (!node->pei)
|
||||
node->pei = ExecInitParallelPlan(node->ps.lefttree, estate, gather->num_workers);
|
||||
|
||||
/*
|
||||
* Register backend workers. We might not get as many as we
|
||||
* requested, or indeed any at all.
|
||||
*/
|
||||
ParallelContext *pcxt = node->pei->pcxt;
|
||||
LaunchParallelWorkers(pcxt);
|
||||
|
||||
/* Set up tuple queue readers to read the results. */
|
||||
if (pcxt->nworkers > 0) {
|
||||
node->nreaders = 0;
|
||||
node->reader = (TupleQueueReader **)palloc(pcxt->nworkers * sizeof(TupleQueueReader *));
|
||||
|
||||
for (i = 0; i < pcxt->nworkers; ++i) {
|
||||
if (pcxt->worker[i].bgwhandle == NULL)
|
||||
continue;
|
||||
|
||||
shm_mq_set_handle(node->pei->tqueue[i], pcxt->worker[i].bgwhandle);
|
||||
node->reader[node->nreaders++] =
|
||||
CreateTupleQueueReader(node->pei->tqueue[i], fslot->tts_tupleDescriptor);
|
||||
got_any_worker = true;
|
||||
}
|
||||
}
|
||||
|
||||
/* No workers? Then never mind. */
|
||||
if (!got_any_worker)
|
||||
ExecShutdownGatherWorkers(node);
|
||||
}
|
||||
|
||||
/* Run plan locally if no workers or not single-copy. */
|
||||
node->need_to_scan_locally = (node->reader == NULL) ||
|
||||
(!gather->single_copy && u_sess->attr.attr_sql.parallel_leader_participation);
|
||||
node->initialized = true;
|
||||
}
|
||||
|
||||
/*
|
||||
* Check to see if we're still projecting out tuples from a previous scan
|
||||
* tuple (because there is a function-returning-set in the projection
|
||||
* expressions). If so, try to project another one.
|
||||
*/
|
||||
if (node->ps.ps_TupFromTlist) {
|
||||
resultSlot = ExecProject(node->ps.ps_ProjInfo, &isDone);
|
||||
if (isDone == ExprMultipleResult)
|
||||
return resultSlot;
|
||||
/* Done with that source tuple... */
|
||||
node->ps.ps_TupFromTlist = false;
|
||||
}
|
||||
|
||||
/*
|
||||
* Reset per-tuple memory context to free any expression evaluation
|
||||
* storage allocated in the previous tuple cycle. Note we can't do this
|
||||
* until we're done projecting. This will also clear any previous tuple
|
||||
* returned by a TupleQueueReader; to make sure we don't leave a dangling
|
||||
* pointer around, clear the working slot first.
|
||||
*/
|
||||
(void)ExecClearTuple(node->funnel_slot);
|
||||
ExprContext *econtext = node->ps.ps_ExprContext;
|
||||
ResetExprContext(econtext);
|
||||
|
||||
/* Get and return the next tuple, projecting if necessary. */
|
||||
for (;;) {
|
||||
/*
|
||||
* Get next tuple, either from one of our workers, or by running the
|
||||
* plan ourselves.
|
||||
*/
|
||||
slot = gather_getnext(node);
|
||||
if (TupIsNull(slot))
|
||||
return NULL;
|
||||
|
||||
/*
|
||||
* form the result tuple using ExecProject(), and return it --- unless
|
||||
* the projection produces an empty set, in which case we must loop
|
||||
* back around for another tuple
|
||||
*/
|
||||
econtext->ecxt_outertuple = slot;
|
||||
resultSlot = ExecProject(node->ps.ps_ProjInfo, &isDone);
|
||||
|
||||
if (isDone != ExprEndResult) {
|
||||
node->ps.ps_TupFromTlist = (isDone == ExprMultipleResult);
|
||||
return resultSlot;
|
||||
}
|
||||
}
|
||||
|
||||
return slot;
|
||||
}
|
||||
|
||||
/* ----------------------------------------------------------------
|
||||
* ExecEndGather
|
||||
*
|
||||
* frees any storage allocated through C routines.
|
||||
* ----------------------------------------------------------------
|
||||
*/
|
||||
void ExecEndGather(GatherState *node)
|
||||
{
|
||||
ExecShutdownGather(node);
|
||||
ExecFreeExprContext(&node->ps);
|
||||
(void)ExecClearTuple(node->ps.ps_ResultTupleSlot);
|
||||
ExecEndNode(outerPlanState(node));
|
||||
}
|
||||
|
||||
/*
|
||||
* Read the next tuple. We might fetch a tuple from one of the tuple queues
|
||||
* using gather_readnext, or if no tuple queue contains a tuple and the
|
||||
* single_copy flag is not set, we might generate one locally instead.
|
||||
*/
|
||||
static TupleTableSlot *gather_getnext(GatherState *gatherstate)
|
||||
{
|
||||
PlanState *outerPlan = outerPlanState(gatherstate);
|
||||
TupleTableSlot *fslot = gatherstate->funnel_slot;
|
||||
|
||||
while (gatherstate->reader != NULL || gatherstate->need_to_scan_locally) {
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
|
||||
if (gatherstate->reader != NULL) {
|
||||
HeapTuple tup = gather_readnext(gatherstate);
|
||||
if (HeapTupleIsValid(tup)) {
|
||||
(void)ExecStoreTuple(tup, /* tuple to store */
|
||||
fslot, /* slot in which to store the tuple */
|
||||
InvalidBuffer, /* buffer associated with this tuple */
|
||||
true); /* pfree this pointer if not from heap */
|
||||
return fslot;
|
||||
}
|
||||
}
|
||||
|
||||
if (gatherstate->need_to_scan_locally) {
|
||||
TupleTableSlot *outerTupleSlot = ExecProcNode(outerPlan);
|
||||
|
||||
if (!TupIsNull(outerTupleSlot))
|
||||
return outerTupleSlot;
|
||||
|
||||
gatherstate->need_to_scan_locally = false;
|
||||
}
|
||||
}
|
||||
|
||||
return ExecClearTuple(fslot);
|
||||
}
|
||||
|
||||
/*
|
||||
* Attempt to read a tuple from one of our parallel workers.
|
||||
*/
|
||||
static HeapTuple gather_readnext(GatherState *gatherstate)
|
||||
{
|
||||
int nvisited = 0;
|
||||
|
||||
for (;;) {
|
||||
bool readerdone = false;
|
||||
|
||||
/* Check for async events, particularly messages from workers. */
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
|
||||
/* Attempt to read a tuple, but don't block if none is available. */
|
||||
TupleQueueReader *reader = gatherstate->reader[gatherstate->nextreader];
|
||||
HeapTuple tup = TupleQueueReaderNext(reader, true, &readerdone);
|
||||
|
||||
/*
|
||||
* If this reader is done, remove it. If all readers are done,
|
||||
* clean up remaining worker state.
|
||||
*/
|
||||
if (readerdone) {
|
||||
Assert(!tup);
|
||||
DestroyTupleQueueReader(reader);
|
||||
--gatherstate->nreaders;
|
||||
if (gatherstate->nreaders == 0) {
|
||||
ExecShutdownGatherWorkers(gatherstate);
|
||||
return NULL;
|
||||
}
|
||||
Size remainSize = sizeof(TupleQueueReader *) * (gatherstate->nreaders - gatherstate->nextreader);
|
||||
if (remainSize != 0) {
|
||||
int rc = memmove_s(&gatherstate->reader[gatherstate->nextreader], remainSize,
|
||||
&gatherstate->reader[gatherstate->nextreader + 1], remainSize);
|
||||
securec_check(rc, "", "");
|
||||
}
|
||||
if (gatherstate->nextreader >= gatherstate->nreaders) {
|
||||
gatherstate->nextreader = 0;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
/* If we got a tuple, return it. */
|
||||
if (tup)
|
||||
return tup;
|
||||
|
||||
/*
|
||||
* Advance nextreader pointer in round-robin fashion. Note that we
|
||||
* only reach this code if we weren't able to get a tuple from the
|
||||
* current worker. We used to advance the nextreader pointer after
|
||||
* every tuple, but it turns out to be much more efficient to keep
|
||||
* reading from the same queue until that would require blocking.
|
||||
*/
|
||||
gatherstate->nextreader++;
|
||||
if (gatherstate->nextreader >= gatherstate->nreaders)
|
||||
gatherstate->nextreader = 0;
|
||||
|
||||
/* Have we visited every (surviving) TupleQueueReader? */
|
||||
nvisited++;
|
||||
if (nvisited >= gatherstate->nreaders) {
|
||||
/*
|
||||
* If (still) running plan locally, return NULL so caller can
|
||||
* generate another tuple from the local copy of the plan.
|
||||
*/
|
||||
if (gatherstate->need_to_scan_locally)
|
||||
return NULL;
|
||||
|
||||
/* Nothing to do except wait for developments. */
|
||||
(void)WaitLatch(&t_thrd.proc->procLatch, WL_LATCH_SET, 0);
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
ResetLatch(&t_thrd.proc->procLatch);
|
||||
nvisited = 0;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/* ----------------------------------------------------------------
|
||||
* ExecShutdownGatherWorkers
|
||||
*
|
||||
* Destroy the parallel workers. Collect all the stats after
|
||||
* workers are stopped, else some work done by workers won't be
|
||||
* accounted.
|
||||
* ----------------------------------------------------------------
|
||||
*/
|
||||
static void ExecShutdownGatherWorkers(GatherState *node)
|
||||
{
|
||||
/* Shut down tuple queue readers before shutting down workers. */
|
||||
if (node->reader != NULL) {
|
||||
for (int i = 0; i < node->nreaders; ++i)
|
||||
DestroyTupleQueueReader(node->reader[i]);
|
||||
|
||||
pfree(node->reader);
|
||||
node->reader = NULL;
|
||||
}
|
||||
|
||||
/* Now shut down the workers. */
|
||||
if (node->pei != NULL)
|
||||
ExecParallelFinish(node->pei);
|
||||
}
|
||||
|
||||
/* ----------------------------------------------------------------
|
||||
* ExecShutdownGather
|
||||
*
|
||||
* Destroy the setup for parallel workers including parallel context.
|
||||
* Collect all the stats after workers are stopped, else some work
|
||||
* done by workers won't be accounted.
|
||||
* ----------------------------------------------------------------
|
||||
*/
|
||||
void ExecShutdownGather(GatherState *node)
|
||||
{
|
||||
ExecShutdownGatherWorkers(node);
|
||||
|
||||
/* Now destroy the parallel context. */
|
||||
if (node->pei != NULL) {
|
||||
ExecParallelCleanup(node->pei);
|
||||
node->pei = NULL;
|
||||
}
|
||||
}
|
||||
|
||||
/* ----------------------------------------------------------------
|
||||
* Join Support
|
||||
* ----------------------------------------------------------------
|
||||
*/
|
||||
/* ----------------------------------------------------------------
|
||||
* ExecReScanGather
|
||||
*
|
||||
* Re-initialize the workers and rescans a relation via them.
|
||||
* ----------------------------------------------------------------
|
||||
*/
|
||||
void ExecReScanGather(GatherState *node)
|
||||
{
|
||||
/*
|
||||
* Re-initialize the parallel workers to perform rescan of relation.
|
||||
* We want to gracefully shutdown all the workers so that they
|
||||
* should be able to propagate any error or other information to master
|
||||
* backend before dying. Parallel context will be reused for rescan.
|
||||
*/
|
||||
ExecShutdownGatherWorkers(node);
|
||||
|
||||
node->initialized = false;
|
||||
|
||||
if (node->pei)
|
||||
ExecParallelReinitialize(node->pei);
|
||||
|
||||
ExecReScan(node->ps.lefttree);
|
||||
}
|
||||
|
||||
/* -------------------------------------------------------------------------
|
||||
*
|
||||
* nodeGather.c
|
||||
* Support routines for scanning a plan via multiple workers.
|
||||
*
|
||||
* Portions Copyright (c) 1996-2015, PostgreSQL Global Development Group
|
||||
* Portions Copyright (c) 1994, Regents of the University of California
|
||||
*
|
||||
* A Gather executor launches parallel workers to run multiple copies of a
|
||||
* plan. It can also run the plan itself, if the workers are not available
|
||||
* or have not started up yet. It then merges all of the results it produces
|
||||
* and the results from the workers into a single output stream. Therefore,
|
||||
* it will normally be used with a plan where running multiple copies of the
|
||||
* same plan does not produce duplicate output, such as parallel-aware
|
||||
* SeqScan.
|
||||
*
|
||||
* Alternatively, a Gather node can be configured to use just one worker
|
||||
* and the single-copy flag can be set. In this case, the Gather node will
|
||||
* run the plan in one worker and will not execute the plan itself. In
|
||||
* this case, it simply returns whatever tuples were returned by the worker.
|
||||
* If a worker cannot be obtained, then it will run the plan itself and
|
||||
* return the results. Therefore, a plan used with a single-copy Gather
|
||||
* node need not be parallel-aware.
|
||||
*
|
||||
* IDENTIFICATION
|
||||
* src/backend/executor/nodeGather.c
|
||||
*
|
||||
* -------------------------------------------------------------------------
|
||||
*/
|
||||
|
||||
#include "postgres.h"
|
||||
|
||||
#include "access/relscan.h"
|
||||
#include "access/xact.h"
|
||||
#include "executor/execdebug.h"
|
||||
#include "executor/execParallel.h"
|
||||
#include "executor/nodeGather.h"
|
||||
#include "executor/nodeSubplan.h"
|
||||
#include "executor/tqueue.h"
|
||||
#include "miscadmin.h"
|
||||
#include "utils/memutils.h"
|
||||
#include "utils/rel.h"
|
||||
|
||||
|
||||
static TupleTableSlot *gather_getnext(GatherState *gatherstate);
|
||||
static HeapTuple gather_readnext(GatherState *gatherstate);
|
||||
static void ExecShutdownGatherWorkers(GatherState *node);
|
||||
|
||||
|
||||
/* ----------------------------------------------------------------
|
||||
* ExecInitGather
|
||||
* ----------------------------------------------------------------
|
||||
*/
|
||||
GatherState *ExecInitGather(Gather *node, EState *estate, int eflags)
|
||||
{
|
||||
bool hasoid = false;
|
||||
|
||||
/* Gather node doesn't have innerPlan node. */
|
||||
Assert(innerPlan(node) == NULL);
|
||||
|
||||
/*
|
||||
* create state structure
|
||||
*/
|
||||
GatherState *gatherstate = makeNode(GatherState);
|
||||
gatherstate->ps.plan = (Plan *)node;
|
||||
gatherstate->ps.state = estate;
|
||||
gatherstate->need_to_scan_locally = !node->single_copy &&
|
||||
u_sess->attr.attr_sql.parallel_leader_participation;
|
||||
|
||||
/*
|
||||
* Miscellaneous initialization
|
||||
*
|
||||
* create expression context for node
|
||||
*/
|
||||
ExecAssignExprContext(estate, &gatherstate->ps);
|
||||
|
||||
/*
|
||||
* initialize child expressions
|
||||
*/
|
||||
gatherstate->ps.targetlist = (List *)ExecInitExpr((Expr *)node->plan.targetlist, (PlanState *)gatherstate);
|
||||
gatherstate->ps.qual = (List *)ExecInitExpr((Expr *)node->plan.qual, (PlanState *)gatherstate);
|
||||
|
||||
/*
|
||||
* tuple table initialization
|
||||
*/
|
||||
gatherstate->funnel_slot = ExecInitExtraTupleSlot(estate);
|
||||
ExecInitResultTupleSlot(estate, &gatherstate->ps);
|
||||
|
||||
/*
|
||||
* now initialize outer plan
|
||||
*/
|
||||
Plan *outerNode = outerPlan(node);
|
||||
outerPlanState(gatherstate) = ExecInitNode(outerNode, estate, eflags);
|
||||
|
||||
gatherstate->ps.ps_TupFromTlist = false;
|
||||
|
||||
/*
|
||||
* Initialize result tuple type and projection info.
|
||||
*/
|
||||
ExecAssignResultTypeFromTL(&gatherstate->ps);
|
||||
ExecAssignProjectionInfo(&gatherstate->ps, NULL);
|
||||
|
||||
/*
|
||||
* Initialize funnel slot to same tuple descriptor as outer plan.
|
||||
*/
|
||||
if (!ExecContextForcesOids(&gatherstate->ps, &hasoid))
|
||||
hasoid = false;
|
||||
TupleDesc tupDesc = ExecTypeFromTL(outerNode->targetlist, hasoid);
|
||||
ExecSetSlotDescriptor(gatherstate->funnel_slot, tupDesc);
|
||||
|
||||
return gatherstate;
|
||||
}
|
||||
|
||||
/* ----------------------------------------------------------------
|
||||
* ExecGather(node)
|
||||
*
|
||||
* Scans the relation via multiple workers and returns
|
||||
* the next qualifying tuple.
|
||||
* ----------------------------------------------------------------
|
||||
*/
|
||||
TupleTableSlot *ExecGather(GatherState *node)
|
||||
{
|
||||
TupleTableSlot *fslot = node->funnel_slot;
|
||||
int i;
|
||||
TupleTableSlot *slot = NULL;
|
||||
TupleTableSlot *resultSlot = NULL;
|
||||
ExprDoneCond isDone;
|
||||
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
|
||||
/*
|
||||
* Initialize the parallel context and workers on first execution. We do
|
||||
* this on first execution rather than during node initialization, as it
|
||||
* needs to allocate large dynamic segement, so it is better to do if it
|
||||
* is really needed.
|
||||
*/
|
||||
if (!node->initialized) {
|
||||
EState *estate = node->ps.state;
|
||||
Gather *gather = (Gather *)node->ps.plan;
|
||||
|
||||
/*
|
||||
* Sometimes we might have to run without parallelism; but if
|
||||
* parallel mode is active then we can try to fire up some workers.
|
||||
*/
|
||||
if (gather->num_workers > 0 && IsInParallelMode()) {
|
||||
bool got_any_worker = false;
|
||||
|
||||
/* Initialize the workers required to execute Gather node. */
|
||||
if (!node->pei)
|
||||
node->pei = ExecInitParallelPlan(node->ps.lefttree, estate, gather->num_workers);
|
||||
|
||||
/*
|
||||
* Register backend workers. We might not get as many as we
|
||||
* requested, or indeed any at all.
|
||||
*/
|
||||
ParallelContext *pcxt = node->pei->pcxt;
|
||||
LaunchParallelWorkers(pcxt);
|
||||
|
||||
/* Set up tuple queue readers to read the results. */
|
||||
if (pcxt->nworkers > 0) {
|
||||
node->nreaders = 0;
|
||||
node->reader = (TupleQueueReader **)palloc(pcxt->nworkers * sizeof(TupleQueueReader *));
|
||||
|
||||
for (i = 0; i < pcxt->nworkers; ++i) {
|
||||
if (pcxt->worker[i].bgwhandle == NULL)
|
||||
continue;
|
||||
|
||||
shm_mq_set_handle(node->pei->tqueue[i], pcxt->worker[i].bgwhandle);
|
||||
node->reader[node->nreaders++] =
|
||||
CreateTupleQueueReader(node->pei->tqueue[i], fslot->tts_tupleDescriptor);
|
||||
got_any_worker = true;
|
||||
}
|
||||
}
|
||||
|
||||
/* No workers? Then never mind. */
|
||||
if (!got_any_worker) {
|
||||
ExecShutdownGatherWorkers(node);
|
||||
} else {
|
||||
t_thrd.subrole = BACKGROUND_LEADER;
|
||||
}
|
||||
}
|
||||
|
||||
/* Run plan locally if no workers or not single-copy. */
|
||||
node->need_to_scan_locally = (node->reader == NULL) ||
|
||||
(!gather->single_copy && u_sess->attr.attr_sql.parallel_leader_participation);
|
||||
node->initialized = true;
|
||||
}
|
||||
|
||||
/*
|
||||
* Check to see if we're still projecting out tuples from a previous scan
|
||||
* tuple (because there is a function-returning-set in the projection
|
||||
* expressions). If so, try to project another one.
|
||||
*/
|
||||
if (node->ps.ps_TupFromTlist) {
|
||||
resultSlot = ExecProject(node->ps.ps_ProjInfo, &isDone);
|
||||
if (isDone == ExprMultipleResult)
|
||||
return resultSlot;
|
||||
/* Done with that source tuple... */
|
||||
node->ps.ps_TupFromTlist = false;
|
||||
}
|
||||
|
||||
/*
|
||||
* Reset per-tuple memory context to free any expression evaluation
|
||||
* storage allocated in the previous tuple cycle. Note we can't do this
|
||||
* until we're done projecting. This will also clear any previous tuple
|
||||
* returned by a TupleQueueReader; to make sure we don't leave a dangling
|
||||
* pointer around, clear the working slot first.
|
||||
*/
|
||||
(void)ExecClearTuple(node->funnel_slot);
|
||||
ExprContext *econtext = node->ps.ps_ExprContext;
|
||||
ResetExprContext(econtext);
|
||||
|
||||
/* Get and return the next tuple, projecting if necessary. */
|
||||
for (;;) {
|
||||
/*
|
||||
* Get next tuple, either from one of our workers, or by running the
|
||||
* plan ourselves.
|
||||
*/
|
||||
slot = gather_getnext(node);
|
||||
if (TupIsNull(slot))
|
||||
return NULL;
|
||||
|
||||
/*
|
||||
* form the result tuple using ExecProject(), and return it --- unless
|
||||
* the projection produces an empty set, in which case we must loop
|
||||
* back around for another tuple
|
||||
*/
|
||||
econtext->ecxt_outertuple = slot;
|
||||
resultSlot = ExecProject(node->ps.ps_ProjInfo, &isDone);
|
||||
|
||||
if (isDone != ExprEndResult) {
|
||||
node->ps.ps_TupFromTlist = (isDone == ExprMultipleResult);
|
||||
return resultSlot;
|
||||
}
|
||||
}
|
||||
|
||||
return slot;
|
||||
}
|
||||
|
||||
/* ----------------------------------------------------------------
|
||||
* ExecEndGather
|
||||
*
|
||||
* frees any storage allocated through C routines.
|
||||
* ----------------------------------------------------------------
|
||||
*/
|
||||
void ExecEndGather(GatherState *node)
|
||||
{
|
||||
ExecShutdownGather(node);
|
||||
ExecFreeExprContext(&node->ps);
|
||||
(void)ExecClearTuple(node->ps.ps_ResultTupleSlot);
|
||||
ExecEndNode(outerPlanState(node));
|
||||
}
|
||||
|
||||
/*
|
||||
* Read the next tuple. We might fetch a tuple from one of the tuple queues
|
||||
* using gather_readnext, or if no tuple queue contains a tuple and the
|
||||
* single_copy flag is not set, we might generate one locally instead.
|
||||
*/
|
||||
static TupleTableSlot *gather_getnext(GatherState *gatherstate)
|
||||
{
|
||||
PlanState *outerPlan = outerPlanState(gatherstate);
|
||||
TupleTableSlot *fslot = gatherstate->funnel_slot;
|
||||
|
||||
while (gatherstate->reader != NULL || gatherstate->need_to_scan_locally) {
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
|
||||
if (gatherstate->reader != NULL) {
|
||||
HeapTuple tup = gather_readnext(gatherstate);
|
||||
if (HeapTupleIsValid(tup)) {
|
||||
(void)ExecStoreTuple(tup, /* tuple to store */
|
||||
fslot, /* slot in which to store the tuple */
|
||||
InvalidBuffer, /* buffer associated with this tuple */
|
||||
true); /* pfree this pointer if not from heap */
|
||||
return fslot;
|
||||
}
|
||||
}
|
||||
|
||||
if (gatherstate->need_to_scan_locally) {
|
||||
TupleTableSlot *outerTupleSlot = ExecProcNode(outerPlan);
|
||||
|
||||
if (!TupIsNull(outerTupleSlot))
|
||||
return outerTupleSlot;
|
||||
|
||||
gatherstate->need_to_scan_locally = false;
|
||||
}
|
||||
}
|
||||
|
||||
return ExecClearTuple(fslot);
|
||||
}
|
||||
|
||||
/*
|
||||
* Attempt to read a tuple from one of our parallel workers.
|
||||
*/
|
||||
static HeapTuple gather_readnext(GatherState *gatherstate)
|
||||
{
|
||||
int nvisited = 0;
|
||||
|
||||
for (;;) {
|
||||
bool readerdone = false;
|
||||
|
||||
/* Check for async events, particularly messages from workers. */
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
|
||||
/* Attempt to read a tuple, but don't block if none is available. */
|
||||
TupleQueueReader *reader = gatherstate->reader[gatherstate->nextreader];
|
||||
HeapTuple tup = TupleQueueReaderNext(reader, true, &readerdone);
|
||||
|
||||
/*
|
||||
* If this reader is done, remove it. If all readers are done,
|
||||
* clean up remaining worker state.
|
||||
*/
|
||||
if (readerdone) {
|
||||
Assert(!tup);
|
||||
DestroyTupleQueueReader(reader);
|
||||
--gatherstate->nreaders;
|
||||
if (gatherstate->nreaders == 0) {
|
||||
ExecShutdownGatherWorkers(gatherstate);
|
||||
return NULL;
|
||||
}
|
||||
Size remainSize = sizeof(TupleQueueReader *) * (gatherstate->nreaders - gatherstate->nextreader);
|
||||
if (remainSize != 0) {
|
||||
int rc = memmove_s(&gatherstate->reader[gatherstate->nextreader], remainSize,
|
||||
&gatherstate->reader[gatherstate->nextreader + 1], remainSize);
|
||||
securec_check(rc, "", "");
|
||||
}
|
||||
if (gatherstate->nextreader >= gatherstate->nreaders) {
|
||||
gatherstate->nextreader = 0;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
/* If we got a tuple, return it. */
|
||||
if (tup)
|
||||
return tup;
|
||||
|
||||
/*
|
||||
* Advance nextreader pointer in round-robin fashion. Note that we
|
||||
* only reach this code if we weren't able to get a tuple from the
|
||||
* current worker. We used to advance the nextreader pointer after
|
||||
* every tuple, but it turns out to be much more efficient to keep
|
||||
* reading from the same queue until that would require blocking.
|
||||
*/
|
||||
gatherstate->nextreader++;
|
||||
if (gatherstate->nextreader >= gatherstate->nreaders)
|
||||
gatherstate->nextreader = 0;
|
||||
|
||||
/* Have we visited every (surviving) TupleQueueReader? */
|
||||
nvisited++;
|
||||
if (nvisited >= gatherstate->nreaders) {
|
||||
/*
|
||||
* If (still) running plan locally, return NULL so caller can
|
||||
* generate another tuple from the local copy of the plan.
|
||||
*/
|
||||
if (gatherstate->need_to_scan_locally)
|
||||
return NULL;
|
||||
|
||||
/* Nothing to do except wait for developments. */
|
||||
(void)WaitLatch(&t_thrd.proc->procLatch, WL_LATCH_SET, 0);
|
||||
CHECK_FOR_INTERRUPTS();
|
||||
ResetLatch(&t_thrd.proc->procLatch);
|
||||
nvisited = 0;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/* ----------------------------------------------------------------
|
||||
* ExecShutdownGatherWorkers
|
||||
*
|
||||
* Destroy the parallel workers. Collect all the stats after
|
||||
* workers are stopped, else some work done by workers won't be
|
||||
* accounted.
|
||||
* ----------------------------------------------------------------
|
||||
*/
|
||||
static void ExecShutdownGatherWorkers(GatherState *node)
|
||||
{
|
||||
/* Shut down tuple queue readers before shutting down workers. */
|
||||
if (node->reader != NULL) {
|
||||
for (int i = 0; i < node->nreaders; ++i)
|
||||
DestroyTupleQueueReader(node->reader[i]);
|
||||
|
||||
pfree(node->reader);
|
||||
node->reader = NULL;
|
||||
}
|
||||
|
||||
/* Now shut down the workers. */
|
||||
if (node->pei != NULL)
|
||||
ExecParallelFinish(node->pei);
|
||||
}
|
||||
|
||||
/* ----------------------------------------------------------------
|
||||
* ExecShutdownGather
|
||||
*
|
||||
* Destroy the setup for parallel workers including parallel context.
|
||||
* Collect all the stats after workers are stopped, else some work
|
||||
* done by workers won't be accounted.
|
||||
* ----------------------------------------------------------------
|
||||
*/
|
||||
void ExecShutdownGather(GatherState *node)
|
||||
{
|
||||
ExecShutdownGatherWorkers(node);
|
||||
|
||||
/* Now destroy the parallel context. */
|
||||
if (node->pei != NULL) {
|
||||
ExecParallelCleanup(node->pei);
|
||||
node->pei = NULL;
|
||||
}
|
||||
}
|
||||
|
||||
/* ----------------------------------------------------------------
|
||||
* Join Support
|
||||
* ----------------------------------------------------------------
|
||||
*/
|
||||
/* ----------------------------------------------------------------
|
||||
* ExecReScanGather
|
||||
*
|
||||
* Re-initialize the workers and rescans a relation via them.
|
||||
* ----------------------------------------------------------------
|
||||
*/
|
||||
void ExecReScanGather(GatherState *node)
|
||||
{
|
||||
/*
|
||||
* Re-initialize the parallel workers to perform rescan of relation.
|
||||
* We want to gracefully shutdown all the workers so that they
|
||||
* should be able to propagate any error or other information to master
|
||||
* backend before dying. Parallel context will be reused for rescan.
|
||||
*/
|
||||
ExecShutdownGatherWorkers(node);
|
||||
|
||||
node->initialized = false;
|
||||
|
||||
if (node->pei)
|
||||
ExecParallelReinitialize(node->pei);
|
||||
|
||||
ExecReScan(node->ps.lefttree);
|
||||
}
|
||||
|
||||
|
|
|
|||
File diff suppressed because it is too large
Load Diff
|
|
@ -398,18 +398,16 @@ IndexTuple index_truncate_tuple(TupleDesc tupleDescriptor, IndexTuple olditup, i
|
|||
TupleDesc itupdesc = CreateTupleDescCopyConstr(tupleDescriptor);
|
||||
Datum values[INDEX_MAX_KEYS];
|
||||
bool isnull[INDEX_MAX_KEYS];
|
||||
IndexTuple newitup;
|
||||
int indnatts = tupleDescriptor->natts;
|
||||
|
||||
Assert(indnatts <= INDEX_MAX_KEYS);
|
||||
Assert(tupleDescriptor->natts <= INDEX_MAX_KEYS);
|
||||
Assert(new_indnatts > 0);
|
||||
Assert(new_indnatts < indnatts);
|
||||
Assert(new_indnatts < tupleDescriptor->natts);
|
||||
|
||||
index_deform_tuple(olditup, tupleDescriptor, values, isnull);
|
||||
|
||||
/* form new tuple that will contain only key attributes */
|
||||
itupdesc->natts = new_indnatts;
|
||||
newitup = index_form_tuple(itupdesc, values, isnull);
|
||||
IndexTuple newitup = index_form_tuple(itupdesc, values, isnull);
|
||||
newitup->t_tid = olditup->t_tid;
|
||||
|
||||
FreeTupleDesc(itupdesc);
|
||||
|
|
|
|||
|
|
@ -366,9 +366,11 @@ void heapgetpage(HeapScanDesc scan, BlockNumber page)
|
|||
scan->rs_vistuples[ntup++] = line_off;
|
||||
}
|
||||
|
||||
ereport(DEBUG1,
|
||||
(errmsg(
|
||||
"heapgetpage xid %lu ctid(%u,%d) valid %d", GetCurrentTransactionIdIfAny(), page, line_off, valid)));
|
||||
if (SHOW_DEBUG_MESSAGE()) {
|
||||
ereport(DEBUG1,
|
||||
(errmsg(
|
||||
"heapgetpage xid %lu ctid(%u,%d) valid %d", GetCurrentTransactionIdIfAny(), page, line_off, valid)));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
File diff suppressed because it is too large
Load Diff
|
|
@ -1,63 +1,79 @@
|
|||
/* -------------------------------------------------------------------------
|
||||
*
|
||||
* dsm.c
|
||||
* manage dynamic shared memory segments
|
||||
*
|
||||
* This file provides a set of services to make programming with dynamic
|
||||
* shared memory segments more convenient. Unlike the low-level
|
||||
* facilities provided by dsm_impl.h and dsm_impl.c, mappings and segments
|
||||
* created using this module will be cleaned up automatically. Mappings
|
||||
* will be removed when the resource owner under which they were created
|
||||
* is cleaned up, unless dsm_pin_mapping() is used, in which case they
|
||||
* have session lifespan. Segments will be removed when there are no
|
||||
* remaining mappings, or at postmaster shutdown in any case. After a
|
||||
* hard postmaster crash, remaining segments will be removed, if they
|
||||
* still exist, at the next postmaster startup.
|
||||
*
|
||||
* Portions Copyright (c) 2020 Huawei Technologies Co.,Ltd
|
||||
* Portions Copyright (c) 1996-2019, PostgreSQL Global Development Group
|
||||
* Portions Copyright (c) 1994, Regents of the University of California
|
||||
*
|
||||
*
|
||||
* IDENTIFICATION
|
||||
* src/gausskernel/storage/ipc/dsm.c
|
||||
*
|
||||
* -------------------------------------------------------------------------
|
||||
*/
|
||||
|
||||
#include "postgres.h"
|
||||
#include "storage/dsm.h"
|
||||
#include "knl/knl_session.h"
|
||||
#include "utils/memutils.h"
|
||||
#include "postmaster/bgworker_internals.h"
|
||||
|
||||
void dsm_detach(void **seg)
|
||||
{
|
||||
Assert(*seg != NULL);
|
||||
knl_u_parallel_context *ctx = (knl_u_parallel_context *)*seg;
|
||||
MemoryContextDelete(ctx->memCtx);
|
||||
ctx->memCtx = NULL;
|
||||
ctx->pwCtx = NULL;
|
||||
ctx->used = false;
|
||||
}
|
||||
|
||||
void *dsm_create(void)
|
||||
{
|
||||
for (int i = 0; i < DSM_MAX_ITEM_PER_QUERY; i++) {
|
||||
if (u_sess->parallel_ctx[i].used == false) {
|
||||
u_sess->parallel_ctx[i].memCtx = AllocSetContextCreate(u_sess->top_mem_cxt, "parallel query",
|
||||
ALLOCSET_DEFAULT_MINSIZE, ALLOCSET_DEFAULT_INITSIZE, ALLOCSET_DEFAULT_MAXSIZE, SHARED_CONTEXT);
|
||||
|
||||
MemoryContext oldContext = MemoryContextSwitchTo(u_sess->parallel_ctx[i].memCtx);
|
||||
u_sess->parallel_ctx[i].pwCtx = (ParallelInfoContext *)palloc0(sizeof(ParallelInfoContext));
|
||||
(void)MemoryContextSwitchTo(oldContext);
|
||||
|
||||
u_sess->parallel_ctx[i].used = true;
|
||||
return &(u_sess->parallel_ctx[i]);
|
||||
}
|
||||
}
|
||||
|
||||
ereport(ERROR, (errcode(ERRCODE_INSUFFICIENT_RESOURCES), errmsg("too many dynamic shared memory segments")));
|
||||
return NULL;
|
||||
}
|
||||
|
||||
/* -------------------------------------------------------------------------
|
||||
*
|
||||
* dsm.c
|
||||
* manage dynamic shared memory segments
|
||||
*
|
||||
* This file provides a set of services to make programming with dynamic
|
||||
* shared memory segments more convenient. Unlike the low-level
|
||||
* facilities provided by dsm_impl.h and dsm_impl.c, mappings and segments
|
||||
* created using this module will be cleaned up automatically. Mappings
|
||||
* will be removed when the resource owner under which they were created
|
||||
* is cleaned up, unless dsm_pin_mapping() is used, in which case they
|
||||
* have session lifespan. Segments will be removed when there are no
|
||||
* remaining mappings, or at postmaster shutdown in any case. After a
|
||||
* hard postmaster crash, remaining segments will be removed, if they
|
||||
* still exist, at the next postmaster startup.
|
||||
*
|
||||
* Portions Copyright (c) 2020 Huawei Technologies Co.,Ltd
|
||||
* Portions Copyright (c) 1996-2019, PostgreSQL Global Development Group
|
||||
* Portions Copyright (c) 1994, Regents of the University of California
|
||||
*
|
||||
*
|
||||
* IDENTIFICATION
|
||||
* src/gausskernel/storage/ipc/dsm.c
|
||||
*
|
||||
* -------------------------------------------------------------------------
|
||||
*/
|
||||
|
||||
#include "postgres.h"
|
||||
#include "storage/dsm.h"
|
||||
#include "knl/knl_session.h"
|
||||
#include "utils/memutils.h"
|
||||
#include "postmaster/bgworker_internals.h"
|
||||
|
||||
#ifdef __USE_NUMA
|
||||
static void RestoreCpuAffinity(cpu_set_t *cpuset)
|
||||
{
|
||||
/* Resotre CPU affinity after parallel query is done. */
|
||||
if (cpuset != NULL) {
|
||||
int rc = pthread_setaffinity_np(t_thrd.proc->pid, sizeof(cpu_set_t), cpuset);
|
||||
if (rc != 0) {
|
||||
ereport(WARNING, (errmsg("pthread_setaffinity_np failed:%d", rc)));
|
||||
}
|
||||
}
|
||||
}
|
||||
#endif
|
||||
|
||||
void dsm_detach(void **seg)
|
||||
{
|
||||
Assert(*seg != NULL);
|
||||
knl_u_parallel_context *ctx = (knl_u_parallel_context *)*seg;
|
||||
#ifdef __USE_NUMA
|
||||
RestoreCpuAffinity(ctx->pwCtx->cpuset);
|
||||
#endif
|
||||
MemoryContextDelete(ctx->memCtx);
|
||||
ctx->memCtx = NULL;
|
||||
ctx->pwCtx = NULL;
|
||||
ctx->used = false;
|
||||
}
|
||||
|
||||
void *dsm_create(void)
|
||||
{
|
||||
for (int i = 0; i < DSM_MAX_ITEM_PER_QUERY; i++) {
|
||||
if (u_sess->parallel_ctx[i].used == false) {
|
||||
u_sess->parallel_ctx[i].memCtx = AllocSetContextCreate(u_sess->top_mem_cxt, "parallel query",
|
||||
ALLOCSET_DEFAULT_MINSIZE, ALLOCSET_DEFAULT_INITSIZE, ALLOCSET_DEFAULT_MAXSIZE, SHARED_CONTEXT);
|
||||
|
||||
MemoryContext oldContext = MemoryContextSwitchTo(u_sess->parallel_ctx[i].memCtx);
|
||||
u_sess->parallel_ctx[i].pwCtx = (ParallelInfoContext *)palloc0(sizeof(ParallelInfoContext));
|
||||
(void)MemoryContextSwitchTo(oldContext);
|
||||
|
||||
u_sess->parallel_ctx[i].used = true;
|
||||
return &(u_sess->parallel_ctx[i]);
|
||||
}
|
||||
}
|
||||
|
||||
ereport(ERROR, (errcode(ERRCODE_INSUFFICIENT_RESOURCES), errmsg("too many dynamic shared memory segments")));
|
||||
return NULL;
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -608,8 +608,8 @@ shm_mq_result shm_mq_receive(shm_mq_handle *mqh, Size *nbytesp, void **datap, bo
|
|||
lengthbytes = sizeof(Size) - mqh->mqh_partial_bytes;
|
||||
else
|
||||
lengthbytes = rb;
|
||||
errno_t rc = memcpy_s(&mqh->mqh_buffer[mqh->mqh_partial_bytes], lengthbytes,
|
||||
rawdata, lengthbytes);
|
||||
errno_t rc = memcpy_s(&mqh->mqh_buffer[mqh->mqh_partial_bytes],
|
||||
mqh->mqh_buflen - mqh->mqh_partial_bytes, rawdata, lengthbytes);
|
||||
securec_check(rc, "\0", "\0");
|
||||
mqh->mqh_partial_bytes += lengthbytes;
|
||||
mqh->mqh_consume_pending += MAXALIGN(lengthbytes);
|
||||
|
|
@ -671,7 +671,8 @@ shm_mq_result shm_mq_receive(shm_mq_handle *mqh, Size *nbytesp, void **datap, bo
|
|||
/* Copy as much as we can. */
|
||||
Assert(mqh->mqh_partial_bytes + rb <= nbytes);
|
||||
if (rb != 0) {
|
||||
errno_t rc = memcpy_s(&mqh->mqh_buffer[mqh->mqh_partial_bytes], rb, rawdata, rb);
|
||||
errno_t rc = memcpy_s(&mqh->mqh_buffer[mqh->mqh_partial_bytes],
|
||||
mqh->mqh_buflen - mqh->mqh_partial_bytes, rawdata, rb);
|
||||
securec_check(rc, "\0", "\0");
|
||||
mqh->mqh_partial_bytes += rb;
|
||||
|
||||
|
|
@ -897,8 +898,8 @@ static shm_mq_result shm_mq_send_bytes(shm_mq_handle *mqh, Size nbytes, const vo
|
|||
* subsequent write to mq_ring, we need a full barrier here.)
|
||||
*/
|
||||
pg_memory_barrier();
|
||||
errno_t rc = memcpy_s(&mq->mq_ring[mq->mq_ring_offset + offset], sendnow,
|
||||
(char*)data + sent, sendnow);
|
||||
errno_t rc = memcpy_s(&mq->mq_ring[mq->mq_ring_offset + offset],
|
||||
ringsize - offset, (char*)data + sent, sendnow);
|
||||
securec_check(rc, "\0", "\0");
|
||||
sent += sendnow;
|
||||
|
||||
|
|
|
|||
|
|
@ -76,10 +76,6 @@ typedef struct knl_instance_attr_common {
|
|||
bool enable_alarm;
|
||||
char* Alarm_component;
|
||||
char* MOTConfigFileName;
|
||||
|
||||
int max_worker_processes;
|
||||
int max_parallel_workers;
|
||||
int max_parallel_workers_per_gather;
|
||||
} knl_instance_attr_common;
|
||||
|
||||
#endif /* SRC_INCLUDE_KNL_KNL_INSTANCE_ATTR_COMMON_H_ */
|
||||
|
|
|
|||
|
|
@ -205,6 +205,7 @@ typedef struct knl_session_attr_sql {
|
|||
int opfusion_debug_mode;
|
||||
int single_shard_stmt;
|
||||
int force_parallel_mode;
|
||||
int max_parallel_workers_per_gather;
|
||||
} knl_session_attr_sql;
|
||||
|
||||
#endif /* SRC_INCLUDE_KNL_KNL_SESSION_ATTR_SQL */
|
||||
|
|
|
|||
|
|
@ -2089,7 +2089,10 @@ typedef struct ParallelInfoContext {
|
|||
char *tupleQueue;
|
||||
struct SharedExecutorInstrumentation *instrumentation;
|
||||
char *namespace_search_path;
|
||||
|
||||
#ifdef __USE_NUMA
|
||||
int numaNode;
|
||||
cpu_set_t *cpuset;
|
||||
#endif
|
||||
/* Mutex protects remaining fields. */
|
||||
slock_t mutex;
|
||||
/* Maximum XactLastRecEnd of any worker. */
|
||||
|
|
|
|||
|
|
@ -0,0 +1,167 @@
|
|||
create table parallel_t1(a int);
|
||||
insert into parallel_t1 values(generate_series(1,100000));
|
||||
--normal plan for seq scan
|
||||
explain (costs off) select count(*) from parallel_t1;
|
||||
QUERY PLAN
|
||||
-------------------------------
|
||||
Aggregate
|
||||
-> Seq Scan on parallel_t1
|
||||
(2 rows)
|
||||
|
||||
explain (costs off) select count(*) from parallel_t1 where a = 5000;
|
||||
QUERY PLAN
|
||||
-------------------------------
|
||||
Aggregate
|
||||
-> Seq Scan on parallel_t1
|
||||
Filter: (a = 5000)
|
||||
(3 rows)
|
||||
|
||||
explain (costs off) select count(*) from parallel_t1 where a > 5000;
|
||||
QUERY PLAN
|
||||
-------------------------------
|
||||
Aggregate
|
||||
-> Seq Scan on parallel_t1
|
||||
Filter: (a > 5000)
|
||||
(3 rows)
|
||||
|
||||
explain (costs off) select count(*) from parallel_t1 where a < 5000;
|
||||
QUERY PLAN
|
||||
-------------------------------
|
||||
Aggregate
|
||||
-> Seq Scan on parallel_t1
|
||||
Filter: (a < 5000)
|
||||
(3 rows)
|
||||
|
||||
explain (costs off) select count(*) from parallel_t1 where a <> 5000;
|
||||
QUERY PLAN
|
||||
-------------------------------
|
||||
Aggregate
|
||||
-> Seq Scan on parallel_t1
|
||||
Filter: (a <> 5000)
|
||||
(3 rows)
|
||||
|
||||
select count(*) from parallel_t1;
|
||||
count
|
||||
--------
|
||||
100000
|
||||
(1 row)
|
||||
|
||||
select count(*) from parallel_t1 where a = 5000;
|
||||
count
|
||||
-------
|
||||
1
|
||||
(1 row)
|
||||
|
||||
select count(*) from parallel_t1 where a > 5000;
|
||||
count
|
||||
-------
|
||||
95000
|
||||
(1 row)
|
||||
|
||||
select count(*) from parallel_t1 where a < 5000;
|
||||
count
|
||||
-------
|
||||
4999
|
||||
(1 row)
|
||||
|
||||
select count(*) from parallel_t1 where a <> 5000;
|
||||
count
|
||||
-------
|
||||
99999
|
||||
(1 row)
|
||||
|
||||
--set parallel parameter
|
||||
set force_parallel_mode=on;
|
||||
set parallel_setup_cost=0;
|
||||
set parallel_tuple_cost=0.000005;
|
||||
set max_parallel_workers_per_gather=2;
|
||||
set min_parallel_table_scan_size=0;
|
||||
set parallel_leader_participation=on;
|
||||
--parallel plan for seq scan
|
||||
explain (costs off) select count(*) from parallel_t1;
|
||||
QUERY PLAN
|
||||
----------------------------------------------
|
||||
Aggregate
|
||||
-> Gather
|
||||
Number of Workers: 2
|
||||
-> Parallel Seq Scan on parallel_t1
|
||||
(4 rows)
|
||||
|
||||
explain (costs off) select count(*) from parallel_t1 where a = 5000;
|
||||
QUERY PLAN
|
||||
----------------------------------------------
|
||||
Aggregate
|
||||
-> Gather
|
||||
Number of Workers: 2
|
||||
-> Parallel Seq Scan on parallel_t1
|
||||
Filter: (a = 5000)
|
||||
(5 rows)
|
||||
|
||||
explain (costs off) select count(*) from parallel_t1 where a > 5000;
|
||||
QUERY PLAN
|
||||
----------------------------------------------
|
||||
Aggregate
|
||||
-> Gather
|
||||
Number of Workers: 2
|
||||
-> Parallel Seq Scan on parallel_t1
|
||||
Filter: (a > 5000)
|
||||
(5 rows)
|
||||
|
||||
explain (costs off) select count(*) from parallel_t1 where a < 5000;
|
||||
QUERY PLAN
|
||||
----------------------------------------------
|
||||
Aggregate
|
||||
-> Gather
|
||||
Number of Workers: 2
|
||||
-> Parallel Seq Scan on parallel_t1
|
||||
Filter: (a < 5000)
|
||||
(5 rows)
|
||||
|
||||
explain (costs off) select count(*) from parallel_t1 where a <> 5000;
|
||||
QUERY PLAN
|
||||
----------------------------------------------
|
||||
Aggregate
|
||||
-> Gather
|
||||
Number of Workers: 2
|
||||
-> Parallel Seq Scan on parallel_t1
|
||||
Filter: (a <> 5000)
|
||||
(5 rows)
|
||||
|
||||
select count(*) from parallel_t1;
|
||||
count
|
||||
--------
|
||||
100000
|
||||
(1 row)
|
||||
|
||||
select count(*) from parallel_t1 where a = 5000;
|
||||
count
|
||||
-------
|
||||
1
|
||||
(1 row)
|
||||
|
||||
select count(*) from parallel_t1 where a > 5000;
|
||||
count
|
||||
-------
|
||||
95000
|
||||
(1 row)
|
||||
|
||||
select count(*) from parallel_t1 where a < 5000;
|
||||
count
|
||||
-------
|
||||
4999
|
||||
(1 row)
|
||||
|
||||
select count(*) from parallel_t1 where a <> 5000;
|
||||
count
|
||||
-------
|
||||
99999
|
||||
(1 row)
|
||||
|
||||
--clean up
|
||||
drop table parallel_t1;
|
||||
reset force_parallel_mode;
|
||||
reset parallel_setup_cost;
|
||||
reset parallel_tuple_cost;
|
||||
reset max_parallel_workers_per_gather;
|
||||
reset min_parallel_table_scan_size;
|
||||
reset parallel_leader_participation;
|
||||
|
|
@ -595,5 +595,8 @@ test: create_procedure create_function pg_compatibility postgres_fdw
|
|||
# autonomous transaction Test
|
||||
test: autonomous_transaction
|
||||
|
||||
# parallel query
|
||||
test: parallel_query
|
||||
|
||||
# gs_basebackup
|
||||
test: gs_basebackup
|
||||
|
|
|
|||
|
|
@ -25,3 +25,6 @@ test: upsert_grammer_test_01 upsert_unlog_test upsert_tmp_test
|
|||
test: upsert_grammer_test_02 upsert_restriction upsert_composite
|
||||
test: upsert_trigger_test upsert_explain
|
||||
test: upsert_clean
|
||||
|
||||
# test parallel query
|
||||
test: parallel_query
|
||||
|
|
|
|||
|
|
@ -0,0 +1,42 @@
|
|||
create table parallel_t1(a int);
|
||||
insert into parallel_t1 values(generate_series(1,100000));
|
||||
--normal plan for seq scan
|
||||
explain (costs off) select count(*) from parallel_t1;
|
||||
explain (costs off) select count(*) from parallel_t1 where a = 5000;
|
||||
explain (costs off) select count(*) from parallel_t1 where a > 5000;
|
||||
explain (costs off) select count(*) from parallel_t1 where a < 5000;
|
||||
explain (costs off) select count(*) from parallel_t1 where a <> 5000;
|
||||
select count(*) from parallel_t1;
|
||||
select count(*) from parallel_t1 where a = 5000;
|
||||
select count(*) from parallel_t1 where a > 5000;
|
||||
select count(*) from parallel_t1 where a < 5000;
|
||||
select count(*) from parallel_t1 where a <> 5000;
|
||||
|
||||
--set parallel parameter
|
||||
set force_parallel_mode=on;
|
||||
set parallel_setup_cost=0;
|
||||
set parallel_tuple_cost=0.000005;
|
||||
set max_parallel_workers_per_gather=2;
|
||||
set min_parallel_table_scan_size=0;
|
||||
set parallel_leader_participation=on;
|
||||
|
||||
--parallel plan for seq scan
|
||||
explain (costs off) select count(*) from parallel_t1;
|
||||
explain (costs off) select count(*) from parallel_t1 where a = 5000;
|
||||
explain (costs off) select count(*) from parallel_t1 where a > 5000;
|
||||
explain (costs off) select count(*) from parallel_t1 where a < 5000;
|
||||
explain (costs off) select count(*) from parallel_t1 where a <> 5000;
|
||||
select count(*) from parallel_t1;
|
||||
select count(*) from parallel_t1 where a = 5000;
|
||||
select count(*) from parallel_t1 where a > 5000;
|
||||
select count(*) from parallel_t1 where a < 5000;
|
||||
select count(*) from parallel_t1 where a <> 5000;
|
||||
|
||||
--clean up
|
||||
drop table parallel_t1;
|
||||
reset force_parallel_mode;
|
||||
reset parallel_setup_cost;
|
||||
reset parallel_tuple_cost;
|
||||
reset max_parallel_workers_per_gather;
|
||||
reset min_parallel_table_scan_size;
|
||||
reset parallel_leader_participation;
|
||||
Loading…
Reference in New Issue