mingyu17
/
openGauss-server
forked from huawei/openGauss-server
0
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

!943 GPC适回合2.0.0分支 

Merge pull request !943 from 赵文浩/2.0.0
This commit is contained in:
opengauss-bot 2021-05-06 17:45:16 +08:00 committed by Gitee
parent e7bdddbfb7 20c720785c
commit a6bca8da91
18 changed files with 1440 additions and 877 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
src/bin/gs_guc/cluster_guc.conf
View File

@ -186,6 +186,7 @@ enable_compress_hll|bool|0,0|NULL|NULL|
enable_fast_numeric|bool|0,0|NULL|Enable numeric optimize.|
enable_force_vector_engine|bool|0,0|NULL|NULL|
enable_global_plancache|bool|0,0|NULL|NULL|
gpc_clean_timeout|int|300,86400|NULL|NULL|
enable_twophase_commit|bool|0,0|NULL|NULL|
enable_hashagg|bool|0,0|NULL|NULL|
enable_hashjoin|bool|0,0|NULL|NULL|

37
src/common/backend/utils/cache/plancache.cpp vendored
View File

@ -139,6 +139,25 @@ void InitPlanCache(void)
CacheRegisterSyscacheCallback(AMOPOPID, PlanCacheSysCallback, (Datum)0);
}
/* ddl no need to global it */
static bool IsSupportGPCStmt(const Node* node)
{
bool isSupportGPC = false;
switch (nodeTag(node)) {
case T_InsertStmt:
case T_DeleteStmt:
case T_UpdateStmt:
case T_MergeStmt:
case T_SelectStmt:
isSupportGPC = false;
break;
default:
isSupportGPC = true;
break;
}
return isSupportGPC;
}
/*
* CreateCachedPlan: initially create a plan cache entry.
*
@ -174,15 +193,8 @@ CachedPlanSource* CreateCachedPlan(Node* raw_parse_tree, const char* query_strin
MemoryContext oldcxt;
Assert(query_string != NULL); /* required as of 8.4 */
bool enable_pbe_gpc = false;
if (stmt_name != NULL && stmt_name[0] != '\0') {
#ifdef ENABLE_MULTIPLE_NODES
/* TransactionStmt do not support shared plan */
enable_pbe_gpc = (ENABLE_CN_GPC && raw_parse_tree && !IsA(raw_parse_tree, TransactionStmt)) || ENABLE_DN_GPC;
#else
enable_pbe_gpc = ENABLE_GPC && raw_parse_tree && !IsA(raw_parse_tree, TransactionStmt);
#endif
}
bool isSupportGPC = raw_parse_tree && IsSupportGPCStmt(raw_parse_tree);
bool enable_pbe_gpc = ENABLE_GPC && stmt_name != NULL && stmt_name[0] != '\0' && !isSupportGPC;
if(!enable_pbe_gpc && !(ENABLE_CN_GPC && enable_spi_gpc)) {
/*
@ -2660,12 +2672,13 @@ void DropCachedPlanInternal(CachedPlanSource* plansource)
}
plansource->lightProxyObj = NULL;
} else {
if (plansource->opFusionObj != NULL) {
if (!plansource->gpc.status.InShareTable() && plansource->opFusionObj != NULL) {
OpFusion *opfusion = (OpFusion *)plansource->opFusionObj;
if (opfusion->m_portalName == NULL || OpFusion::locateFusion(opfusion->m_portalName) == NULL) {
if (opfusion->m_local.m_portalName == NULL ||
OpFusion::locateFusion(opfusion->m_local.m_portalName) == NULL) {
OpFusion::tearDown(opfusion);
} else {
opfusion->m_psrc = NULL;
opfusion->m_global->m_psrc = NULL;
}
plansource->opFusionObj = NULL;
}

15
src/common/backend/utils/misc/guc.cpp
View File

@ -6893,6 +6893,21 @@ static void InitConfigureNamesInt()
NULL,
NULL},
#endif
/* The I/O upper limit of batch flush dirty page every second */
{{"gpc_clean_timeout",
PGC_SIGHUP,
CLIENT_CONN,
gettext_noop("Set the maximum allowed duration of any unused global plancache."),
NULL,
GUC_UNIT_S},
&u_sess->attr.attr_common.gpc_clean_timeout,
30 * 60, /* 30min */
5 * 60, /* 5min */
24 * 60 * 60, /* 24h */
NULL,
NULL,
NULL},
/* End-of-list marker */
{{NULL, (GucContext)0, (config_group)0, NULL, NULL}, NULL, 0, 0, 0, NULL, NULL, NULL}};

4
src/common/backend/utils/resowner/resowner.cpp
View File

@ -431,6 +431,8 @@ static void ResourceOwnerFreeOwner(ResourceOwner owner)
pfree(owner->partmaprefs);
if (owner->fakepartrefs)
pfree(owner->fakepartrefs);
if (owner->globalMemContexts)
pfree(owner->globalMemContexts);
pfree(owner);
}
@ -1700,7 +1702,7 @@ void ResourceOwnerEnlargeGMemContext(ResourceOwner owner)
if (owner->globalMemContexts == NULL) {
newmax = 2;
owner->globalMemContexts = (MemoryContext*)MemoryContextAlloc(
THREAD_GET_MEM_CXT_GROUP(MEMORY_CONTEXT_CBB), newmax * sizeof(MemoryContext));
THREAD_GET_MEM_CXT_GROUP(MEMORY_CONTEXT_EXECUTOR), newmax * sizeof(MemoryContext));
} else {
newmax = owner->maxGlobalMemContexts * 2;
owner->globalMemContexts = (MemoryContext*)repalloc(owner->globalMemContexts, newmax * sizeof(MemoryContext));

23
src/gausskernel/optimizer/commands/prepare.cpp
View File

@ -233,12 +233,19 @@ void ExecuteQuery(ExecuteStmt* stmt, IntoClause* intoClause, const char* querySt
OpFusion::clearForCplan((OpFusion*)psrc->opFusionObj, psrc);
if (psrc->opFusionObj != NULL) {
((OpFusion*)psrc->opFusionObj)->setPreparedDestReceiver(dest);
((OpFusion*)psrc->opFusionObj)->useOuterParameter(paramLI);
((OpFusion*)psrc->opFusionObj)->setCurrentOpFusionObj((OpFusion*)psrc->opFusionObj);
OpFusion *opFusionObj = (OpFusion *)(psrc->opFusionObj);
if (opFusionObj->IsGlobal()) {
opFusionObj = (OpFusion *)OpFusion::FusionFactory(opFusionObj->m_global->m_type,
u_sess->cache_mem_cxt, psrc, NULL, params);
Assert(opFusionObj != NULL);
}
opFusionObj->setPreparedDestReceiver(dest);
opFusionObj->useOuterParameter(paramLI);
opFusionObj->setCurrentOpFusionObj((OpFusion*)psrc->opFusionObj);
CachedPlanSource* cps = ((OpFusion*)psrc->opFusionObj)->m_psrc;
if (cps != NULL && cps->gplan) {
CachedPlanSource* cps = opFusionObj->m_global->m_psrc;
bool needBucketId = cps != NULL && cps->gplan;
if (needBucketId) {
setCachedPlanBucketId(cps->gplan, paramLI);
}
@ -310,10 +317,12 @@ void ExecuteQuery(ExecuteStmt* stmt, IntoClause* intoClause, const char* querySt
eflags = 0;
count = FETCH_ALL;
}
bool checkSQLBypass = IS_PGXC_DATANODE && (psrc->cplan == NULL) && (psrc->is_checked_opfusion == false);
bool checkSQLBypass = IS_PGXC_DATANODE && !psrc->gpc.status.InShareTable() &&
(psrc->cplan == NULL) && (psrc->is_checked_opfusion == false);
if (checkSQLBypass) {
psrc->opFusionObj =
OpFusion::FusionFactory(OpFusion::getFusionType(cplan, paramLI, NULL), psrc->context, psrc, NULL, paramLI);
OpFusion::FusionFactory(OpFusion::getFusionType(cplan, paramLI, NULL),
u_sess->cache_mem_cxt, psrc, NULL, paramLI);
psrc->is_checked_opfusion = true;
if (psrc->opFusionObj != NULL) {
((OpFusion*)psrc->opFusionObj)->setPreparedDestReceiver(dest);

11
src/gausskernel/process/globalplancache/globalplancache.cpp
View File

@ -33,6 +33,7 @@
#include "executor/lightProxy.h"
#include "executor/spi_priv.h"
#include "optimizer/nodegroups.h"
#include "opfusion/opfusion.h"
#include "pgxc/groupmgr.h"
#include "pgxc/pgxcnode.h"
#include "utils/dynahash.h"
@ -248,6 +249,10 @@ bool GlobalPlanCache::TryStore(CachedPlanSource *plansource, PreparedStatement
INSTR_TIME_SET_CURRENT(entry->val.last_use_time);
/* off the link */
plansource->next_saved = NULL;
plansource->is_checked_opfusion = true;
if (plansource->opFusionObj != NULL) {
OpFusion::SaveInGPC((OpFusion*)(plansource->opFusionObj));
}
/* initialize the ref count .*/
#ifdef ENABLE_MULTIPLE_NODES
/* dn only count reference on cur_stmt_psrc, no prepare statement.
@ -386,6 +391,9 @@ void GlobalPlanCache::DropInvalid()
curr->magic = 0;
MemoryContextUnSeal(curr->context);
MemoryContextUnSeal(curr->query_context);
if (curr->opFusionObj) {
OpFusion::DropGlobalOpfusion((OpFusion*)(curr->opFusionObj));
}
MemoryContextDelete(curr->context);
cell = next;
@ -949,6 +957,9 @@ void GlobalPlanCache::CleanUpByTime()
cur_plansource->magic = 0;
MemoryContextUnSeal(cur_plansource->context);
MemoryContextUnSeal(cur_plansource->query_context);
if (cur_plansource->opFusionObj) {
OpFusion::DropGlobalOpfusion((OpFusion*)(cur_plansource->opFusionObj));
}
MemoryContextDelete(cur_plansource->context);
m_array[bucket_id].count--;
}

5
src/gausskernel/process/globalplancache/globalplancache_util.cpp
View File

@ -307,9 +307,8 @@ List* CopyLocalStmt(const List* stmt_list, const MemoryContext parent_cxt, Memor
*plan_context = AllocSetContextCreate(parent_cxt,
"CopyedStmt",
ALLOCSET_DEFAULT_MINSIZE,
16 * 1024,
ALLOCSET_DEFAULT_MAXSIZE,
STACK_CONTEXT);
ALLOCSET_DEFAULT_INITSIZE,
ALLOCSET_DEFAULT_MAXSIZE);
/*
* Copy plan into the new context.
*/

4
src/gausskernel/process/globalplancache/globalplancache_view.cpp
View File

@ -29,6 +29,7 @@
#include "access/xact.h"
#include "catalog/pgxc_node.h"
#include "commands/prepare.h"
#include "opfusion/opfusion.h"
#include "optimizer/nodegroups.h"
#include "pgxc/groupmgr.h"
#include "pgxc/pgxcnode.h"
@ -211,6 +212,9 @@ Datum GlobalPlanCache::PlanClean()
hash_search(m_array[bucket_id].hash_tbl, (void *) &(entry->key), HASH_REMOVE, &found);
MemoryContextUnSeal(cur->context);
MemoryContextUnSeal(cur->query_context);
if (cur->opFusionObj) {
OpFusion::DropGlobalOpfusion((OpFusion*)(cur->opFusionObj));
}
MemoryContextDelete(cur->context);
m_array[bucket_id].count--;
}

157
src/gausskernel/process/tcop/postgres.cpp
View File

@ -3323,7 +3323,7 @@ static void exec_parse_message(const char* query_string, /* string to execute */
query = parse_analyze_varparams(raw_parse_tree, query_string, &paramTypes, &numParams);
#else
query = parse_analyze_varparams(raw_parse_tree, query_string, &paramTypes, &numParams, paramTypeNames);
#endif
#endif
#ifdef ENABLE_MOT
/* check cross engine queries */
StorageEngineType storageEngineType = SE_TYPE_UNSPECIFIED;
@ -3341,7 +3341,11 @@ static void exec_parse_message(const char* query_string, /* string to execute */
// MOT LLVM
TryMotJitCodegenQuery(query_string, psrc, query);
}
/* gpc does not support MOT engine */
if (ENABLE_CN_GPC && psrc->gpc.status.IsSharePlan() &&
(psrc->storageEngineType == SE_TYPE_MOT || psrc->storageEngineType == SE_TYPE_MIXED)) {
psrc->gpc.status.SetKind(GPC_UNSHARED);
}
if (!IsTransactionExitStmt(raw_parse_tree) && CheckMotIndexedColumnUpdate(query)) {
ereport(ERROR, (errcode(ERRCODE_FDW_UPDATE_INDEXED_FIELD_NOT_SUPPORTED), errmodule(MOD_MOT),
errmsg("Update of indexed column is not supported for memory table")));
@ -4109,22 +4113,14 @@ static void exec_bind_message(StringInfo input_message)
#ifdef ENABLE_MOT
/* set transaction storage engine and check for cross transaction violation */
SetCurrentTransactionStorageEngine(psrc->storageEngineType);
if (!IsTransactionExitStmt(psrc->raw_parse_tree) && IsMixedEngineUsed()) {
if (!IsTransactionExitStmt(psrc->raw_parse_tree) && IsMixedEngineUsed())
ereport(ERROR, (errcode(ERRCODE_FDW_CROSS_STORAGE_ENGINE_TRANSACTION_NOT_SUPPORTED), errmodule(MOD_MOT),
errmsg("Cross storage engine transaction is not supported")));
}
errmsg("Cross storage engine transaction is not supported")));
/* block MOT engine queries in sub-transactions */
if (!IsTransactionExitStmt(psrc->raw_parse_tree) && IsMOTEngineUsedInParentTransaction() && IsMOTEngineUsed()) {
if (!IsTransactionExitStmt(psrc->raw_parse_tree) && IsMOTEngineUsedInParentTransaction() && IsMOTEngineUsed())
ereport(ERROR, (errcode(ERRCODE_FDW_OPERATION_NOT_SUPPORTED), errmodule(MOD_MOT),
errmsg("SubTransaction is not supported for memory table")));
}
if (IsTransactionPrepareStmt(psrc->raw_parse_tree) && (IsMOTEngineUsed() || IsMixedEngineUsed())) {
/* Explicit prepare transaction is not supported for memory table */
ereport(ERROR, (errcode(ERRCODE_FDW_OPERATION_NOT_SUPPORTED), errmodule(MOD_MOT),
errmsg("Explicit prepare transaction is not supported for memory table")));
}
errmsg("SubTransaction is not supported for memory table")));
/*
* MOT JIT Execution:
@ -4144,21 +4140,26 @@ static void exec_bind_message(StringInfo input_message)
OpFusion::clearForCplan((OpFusion*)psrc->opFusionObj, psrc);
if (psrc->opFusionObj != NULL) {
Assert(psrc->cplan == NULL);
(void)RevalidateCachedQuery(psrc);
if (psrc->opFusionObj != NULL) {
Assert(psrc->cplan == NULL);
((OpFusion*)psrc->opFusionObj)->clean();
((OpFusion*)psrc->opFusionObj)->updatePreAllocParamter(input_message);
((OpFusion*)psrc->opFusionObj)->setCurrentOpFusionObj((OpFusion*)psrc->opFusionObj);
if (portal_name[0] != '\0')
((OpFusion *)psrc->opFusionObj)->storeFusion(portal_name);
CachedPlanSource* cps = ((OpFusion*)psrc->opFusionObj)->m_psrc;
if (cps != NULL && cps->gplan) {
setCachedPlanBucketId(cps->gplan, ((OpFusion*)psrc->opFusionObj)->m_params);
OpFusion *opFusionObj = (OpFusion *)(psrc->opFusionObj);
if (opFusionObj != NULL) {
if (opFusionObj->IsGlobal()) {
opFusionObj = (OpFusion *)OpFusion::FusionFactory(opFusionObj->m_global->m_type,
u_sess->cache_mem_cxt, psrc, NULL, params);
Assert(opFusionObj != NULL);
}
opFusionObj->clean();
opFusionObj->updatePreAllocParamter(input_message);
opFusionObj->setCurrentOpFusionObj(opFusionObj);
if (portal_name[0] != '\0')
opFusionObj->storeFusion(portal_name);
CachedPlanSource* cps = opFusionObj->m_global->m_psrc;
if (cps != NULL && cps->gplan) {
setCachedPlanBucketId(cps->gplan, opFusionObj->m_local.m_params);
}
if (t_thrd.postgres_cxt.whereToSendOutput == DestRemote)
pq_putemptymessage('2');
gstrace_exit(GS_TRC_ID_exec_bind_message);
@ -4543,13 +4544,10 @@ static void exec_bind_message(StringInfo input_message)
*/
PortalDefineQuery(portal, saved_stmt_name, query_string, psrc->commandTag, cplan->stmt_list, cplan);
if (ENABLE_GPC && psrc->gplan) {
portal->stmts = CopyLocalStmt(cplan->stmt_list, u_sess->top_portal_cxt, &portal->copyCxt);
}
if (IS_PGXC_DATANODE && psrc->cplan == NULL && psrc->is_checked_opfusion == false) {
if (IS_PGXC_DATANODE && psrc->cplan == NULL && !psrc->gpc.status.InShareTable() &&
psrc->is_checked_opfusion == false) {
psrc->opFusionObj = OpFusion::FusionFactory(OpFusion::getFusionType(cplan, params, NULL),
u_sess->cache_mem_cxt, psrc, NULL, params);
u_sess->cache_mem_cxt, psrc, NULL, params);
psrc->is_checked_opfusion = true;
if (psrc->opFusionObj != NULL) {
((OpFusion*)psrc->opFusionObj)->clean();
@ -4569,6 +4567,10 @@ static void exec_bind_message(StringInfo input_message)
}
}
if (ENABLE_GPC && psrc->gplan) {
portal->stmts = CopyLocalStmt(cplan->stmt_list, u_sess->top_portal_cxt, &portal->copyCxt);
}
/* Done with the snapshot used for parameter I/O and parsing/planning */
if (snapshot_set)
PopActiveSnapshot();
@ -8404,8 +8406,12 @@ int PostgresMain(int argc, char* argv[], const char* dbname, const char* usernam
if (IS_PGXC_DATANODE && closeTarget[0] != '\0') {
OpFusion* curr = OpFusion::locateFusion(closeTarget);
if (curr != NULL) {
curr->clean();
OpFusion::removeFusionFromHtab(closeTarget);
if (curr->IsGlobal()) {
OpFusion::tearDown(curr);
} else {
curr->clean();
OpFusion::removeFusionFromHtab(closeTarget);
}
}
}
@ -9773,25 +9779,45 @@ void execute_simple_query(const char* query_string)
* If query is not SELECT/INSERT/UPDATE/DELETE, return completionTag, else NULL.
*/
static void exec_one_in_batch(CachedPlanSource* psrc, ParamListInfo params, int numRFormats, int16* rformats,
bool send_DP_msg, CommandDest dest, char* completionTag, const char* stmt_name, List* gpcCopyStmts)
bool send_DP_msg, CommandDest dest, char* completionTag,
const char* stmt_name, List** gpcCopyStmts, MemoryContext* tmpCxt, PreparedStatement* pstmt)
{
CachedPlan* cplan = NULL;
Portal portal;
DestReceiver* receiver = NULL;
bool completed = false;
if (ENABLE_GPC && psrc->stmt_name && psrc->stmt_name[0] != '\0' &&
g_instance.plan_cache->CheckRecreateCachePlan(psrc)) {
g_instance.plan_cache->RecreateCachePlan(psrc, stmt_name, pstmt, NULL, NULL);
#ifdef ENABLE_MULTIPLE_NODES
psrc = IS_PGXC_DATANODE ? u_sess->pcache_cxt.cur_stmt_psrc : pstmt->plansource;
#else
psrc = pstmt->plansource;
#endif
t_thrd.postgres_cxt.debug_query_string = psrc->query_string;
}
int generation = psrc->generation;
OpFusion::clearForCplan((OpFusion*)psrc->opFusionObj, psrc);
if (psrc->opFusionObj != NULL) {
(void)RevalidateCachedQuery(psrc);
OpFusion *opFusionObj = (OpFusion *)(psrc->opFusionObj);
if (psrc->opFusionObj != NULL) {
Assert(psrc->cplan == NULL);
((OpFusion*)psrc->opFusionObj)->bindClearPosition();
((OpFusion*)psrc->opFusionObj)->useOuterParameter(params);
((OpFusion*)psrc->opFusionObj)->setCurrentOpFusionObj((OpFusion*)psrc->opFusionObj);
if (opFusionObj->IsGlobal()) {
opFusionObj = (OpFusion *)OpFusion::FusionFactory(opFusionObj->m_global->m_type,
u_sess->cache_mem_cxt, psrc, NULL, params);
Assert(opFusionObj != NULL);
}
opFusionObj->bindClearPosition();
opFusionObj->useOuterParameter(params);
opFusionObj->setCurrentOpFusionObj(opFusionObj);
CachedPlanSource* cps = ((OpFusion*)psrc->opFusionObj)->m_psrc;
CachedPlanSource* cps = opFusionObj->m_global->m_psrc;
if (cps != NULL && cps->gplan) {
setCachedPlanBucketId(cps->gplan, params);
}
@ -9856,8 +9882,20 @@ static void exec_one_in_batch(CachedPlanSource* psrc, ParamListInfo params, int
cplan->stmt_list,
cplan);
if (ENABLE_GPC && gpcCopyStmts != NULL) {
portal->stmts = gpcCopyStmts;
if (ENABLE_GPC) {
/* generated new gplan, copy it incase someone change it */
if (generation != psrc->generation && psrc->gplan) {
if (*tmpCxt != NULL)
MemoryContextDelete(*tmpCxt);
*gpcCopyStmts = CopyLocalStmt(psrc->gplan->stmt_list, u_sess->temp_mem_cxt, tmpCxt);
} else if (*gpcCopyStmts == NULL && psrc->gpc.status.InShareTable()) {
/* copy for shared plan */
if (*tmpCxt != NULL)
MemoryContextDelete(*tmpCxt);
*gpcCopyStmts = CopyLocalStmt(psrc->gplan->stmt_list, u_sess->temp_mem_cxt, tmpCxt);
}
if (*gpcCopyStmts != NULL)
portal->stmts = *gpcCopyStmts;
}
#ifdef ENABLE_MOT
@ -9873,13 +9911,14 @@ static void exec_one_in_batch(CachedPlanSource* psrc, ParamListInfo params, int
}
#endif
bool checkSQLBypass = IS_PGXC_DATANODE && (psrc->cplan == NULL) && (psrc->is_checked_opfusion == false);
bool checkSQLBypass = IS_PGXC_DATANODE && !psrc->gpc.status.InShareTable() &&
(psrc->cplan == NULL) && (psrc->is_checked_opfusion == false);
if (checkSQLBypass) {
psrc->opFusionObj =
OpFusion::FusionFactory(OpFusion::getFusionType(cplan, params, NULL), psrc->context, psrc, NULL, params);
psrc->opFusionObj = OpFusion::FusionFactory(OpFusion::getFusionType(cplan, params, NULL),
u_sess->cache_mem_cxt, psrc, NULL, params);
psrc->is_checked_opfusion = true;
if (psrc->opFusionObj != NULL) {
((OpFusion*)psrc->opFusionObj)->bindClearPosition();
((OpFusion*)psrc->opFusionObj)->bindClearPosition();
((OpFusion*)psrc->opFusionObj)->useOuterParameter(params);
((OpFusion*)psrc->opFusionObj)->setCurrentOpFusionObj((OpFusion*)psrc->opFusionObj);
((OpFusion*)psrc->opFusionObj)->CopyFormats(rformats, numRFormats);
@ -10342,6 +10381,7 @@ static void exec_batch_bind_execute(StringInfo input_message)
*portal_name ? portal_name : "<unnamed>",
*stmt_name ? stmt_name : "<unnamed>",
batch_count)));
PreparedStatement *pstmt = NULL;
/* Find prepared statement */
if (stmt_name[0] != '\0') {
@ -10351,7 +10391,6 @@ static void exec_batch_bind_execute(StringInfo input_message)
ereport(ERROR, (errcode(ERRCODE_UNDEFINED_PSTATEMENT),
errmsg("dn gpc's prepared statement %s does not exist", stmt_name)));
} else {
PreparedStatement *pstmt = NULL;
pstmt = FetchPreparedStatement(stmt_name, true, true);
psrc = pstmt->plansource;
}
@ -10859,21 +10898,26 @@ static void exec_batch_bind_execute(StringInfo input_message)
} else {
char* completionTag = (char*)palloc0(COMPLETION_TAG_BUFSIZE * sizeof(char));
List* copyedStmts = NULL;
MemoryContext tmpCxt = NULL;
if (u_sess->attr.attr_resource.use_workload_manager && g_instance.wlm_cxt->gscgroup_init_done &&
!IsAbortedTransactionBlockState()) {
u_sess->wlm_cxt->cgroup_last_stmt = u_sess->wlm_cxt->cgroup_stmt;
u_sess->wlm_cxt->cgroup_stmt = WLMIsSpecialCommand(psrc->raw_parse_tree, NULL);
}
if (ENABLE_GPC && psrc->gplan && psrc->gpc.status.IsSharePlan()) {
MemoryContext tmpCxt = NULL;
copyedStmts = CopyLocalStmt(psrc->gplan->stmt_list, u_sess->temp_mem_cxt, &tmpCxt);
}
if (use_original_logic) {
for (int i = 0; i < batch_count; i++) {
exec_one_in_batch(psrc, params_set[i], numRFormats, rformats,
(i == 0) ? send_DP_msg : false, dest, completionTag, stmt_name, copyedStmts);
(i == 0) ? send_DP_msg : false, dest, completionTag,
stmt_name, &copyedStmts, &tmpCxt, pstmt);
if (ENABLE_GPC && stmt_name[0] != '\0') {
#ifdef ENABLE_MULTIPLE_NODES
psrc = IS_PGXC_DATANODE ? u_sess->pcache_cxt.cur_stmt_psrc : pstmt->plansource;
#else
psrc = pstmt->plansource;
#endif
}
}
/* only send the last commandTag */
@ -10883,7 +10927,15 @@ static void exec_batch_bind_execute(StringInfo input_message)
for (int i = 0; i < batch_count; i++) {
exec_one_in_batch(psrc, params_set[i], numRFormats, rformats,
(i == 0) ? send_DP_msg : false, dest, completionTag, stmt_name, copyedStmts);
(i == 0) ? send_DP_msg : false, dest, completionTag,
stmt_name, &copyedStmts, &tmpCxt, pstmt);
if (ENABLE_GPC && stmt_name[0] != '\0') {
#ifdef ENABLE_MULTIPLE_NODES
psrc = IS_PGXC_DATANODE ? u_sess->pcache_cxt.cur_stmt_psrc : pstmt->plansource;
#else
psrc = pstmt->plansource;
#endif
}
/* Get process_count (X) from completionTag */
if (completionTag[0] == 'I') {
@ -10897,7 +10949,6 @@ static void exec_batch_bind_execute(StringInfo input_message)
Assert(completionTag[0] == 'U' || completionTag[0] == 'D' || completionTag[0] == 'S');
tmp_count = pg_atoi(&completionTag[7], sizeof(int32), '\0');
}
process_count += tmp_count;
}
}

18
src/gausskernel/process/threadpool/threadpool_scheduler.cpp
View File

@ -46,17 +46,31 @@
#define MAX_HANG_TIME 100
#define REDUCE_THREAD_TIME 100
#define SHUTDOWN_THREAD_TIME 1000
#define GPC_CLEAN_TIME 600
#define GPC_CLEAN_TIME 300
static void SchedulerSIGKILLHandler(SIGNAL_ARGS)
{
proc_exit(0);
}
static void SchedulerSIGHUPHandler(SIGNAL_ARGS)
{
t_thrd.threadpool_cxt.scheduler->m_getSIGHUP = true;
}
static void reloadConfigFileIfNecessary()
{
if (t_thrd.threadpool_cxt.scheduler->m_getSIGHUP) {
t_thrd.threadpool_cxt.scheduler->m_getSIGHUP = false;
ProcessConfigFile(PGC_SIGHUP);
}
}
void TpoolSchedulerMain(ThreadPoolScheduler *scheduler)
{
int gpc_count = 0;
(void)gspqsignal(SIGHUP, SchedulerSIGHUPHandler);
(void)gspqsignal(SIGKILL, SchedulerSIGKILLHandler);
gs_signal_setmask(&t_thrd.libpq_cxt.UnBlockSig, NULL);
(void)gs_signal_unblock_sigusr2();
@ -70,6 +84,7 @@ void TpoolSchedulerMain(ThreadPoolScheduler *scheduler)
while (true) {
pg_usleep(SCHEDULER_TIME_UNIT);
reloadConfigFileIfNecessary();
scheduler->DynamicAdjustThreadPool();
scheduler->GPCScheduleCleaner(&gpc_count);
g_threadPoolControler->GetSessionCtrl()->CheckSessionTimeout();
@ -85,6 +100,7 @@ ThreadPoolScheduler::ThreadPoolScheduler(int groupNum, ThreadPoolGroup** groups)
m_freeTestCount = (uint *)palloc0(sizeof(uint) * groupNum);
m_freeStreamCount = (uint *)palloc0(sizeof(uint) * groupNum);
m_gpcContext = NULL;
m_getSIGHUP = false;
}
ThreadPoolScheduler::~ThreadPoolScheduler()

1711
src/gausskernel/runtime/executor/opfusion.cpp
View File

File diff suppressed because it is too large Load Diff

13
src/gausskernel/runtime/executor/opfusion_util.cpp
View File

@ -1129,3 +1129,16 @@ void tpslot_free_heaptuple(TupleTableSlot* reslot)
reslot->tts_tuple = NULL;
}
}
/* judge plan node is partiterator */
Node* JudgePlanIsPartIterator(Plan* plan)
{
Node* node = NULL;
if (IsA(plan, PartIterator)) {
node = (Node*)plan->lefttree;
} else {
node = (Node*)plan;
}
return node;
}

9
src/gausskernel/storage/access/transam/xact.cpp
View File

@ -6284,6 +6284,11 @@ static void CommitSubTransaction(bool STP_commit)
ShowTransactionState("CommitSubTransaction");
/* clean hash table for sub transaction in opfusion */
if (IS_PGXC_DATANODE) {
OpFusion::ClearInSubUnexpectSituation(s->curTransactionOwner);
}
if (s->state != TRANS_INPROGRESS) {
ereport(WARNING, (errmsg("CommitSubTransaction while in %s state", TransStateAsString(s->state))));
}
@ -6415,6 +6420,10 @@ void AbortSubTransaction(bool STP_rollback)
{
TransactionState s = CurrentTransactionState;
t_thrd.xact_cxt.bInAbortTransaction = true;
/* clean hash table for sub transaction in opfusion */
if (IS_PGXC_DATANODE) {
OpFusion::ClearInSubUnexpectSituation(s->curTransactionOwner);
}
/*
* @dfs

1
src/include/knl/knl_guc/knl_session_attr_common.h
View File

@ -203,6 +203,7 @@ typedef struct knl_session_attr_common {
char* router_att;
bool enable_router;
int gpc_clean_timeout;
} knl_session_attr_common;
#endif /* SRC_INCLUDE_KNL_KNL_SESSION_ATTR_COMMON_H_ */

304
src/include/opfusion/opfusion.h
View File

@ -45,6 +45,16 @@ typedef struct pnFusionObj {
#define HASH_TBL_LEN 64
/*
* The variables in OpFusion is always in two parts: global's variables and local's variables.
* Global variable means it can be shared in each session.
* Local variable means it will be change in local session, so it cannot be shared.
*
* Global variables be saved into struct OpFusionGlobalVariable, and we access the global variables
* from pointer m_global, and m_global's mem context is under global cachedplansource.
* Local variables be saved into struct OpFusionLocaleVariable, and we access the local variables
* from object m_local, and m_local's context is under session context.
*/
class OpFusion : public BaseObject {
public:
OpFusion(MemoryContext context, CachedPlanSource* psrc, List* plantree_list);
@ -60,6 +70,14 @@ public:
static bool process(int op, StringInfo msg, char* completionTag, bool isTopLevel, bool* isQueryCompleted);
static void SaveInGPC(OpFusion* obj);
static void DropGlobalOpfusion(OpFusion* obj);
void InitGlobals(MemoryContext context, CachedPlanSource* psrc, List* plantree_list);
void InitLocals(MemoryContext context);
void CopyFormats(int16* formats, int numRFormats);
void updatePreAllocParamter(StringInfo msg);
@ -80,6 +98,7 @@ public:
return;
}
void copyGlobalOpfusionVar(OpFusion);
void setPreparedDestReceiver(DestReceiver* preparedDest);
Datum CalFuncNodeVal(Oid functionId, List* args, bool* is_null, Datum* values, bool* isNulls);
@ -98,7 +117,7 @@ public:
void executeInit();
void executeEnd(const char* portal_name, bool* completionTag);
bool executeEnd(const char* portal_name, bool* completionTag);
void auditRecord();
@ -112,6 +131,8 @@ public:
static void ClearInUnexpectSituation();
static void ClearInSubUnexpectSituation(ResourceOwner owner);
void storeFusion(const char *portalname);
static OpFusion *locateFusion(const char *portalname);
@ -120,66 +141,104 @@ public:
static void refreshCurFusion(StringInfo msg);
inline bool IsGlobal()
{
pg_memory_barrier();
return (m_global && m_global->m_is_global);
}
public:
struct ParamLoc {
int paramId;
int scanKeyIndx;
};
struct ConstLoc {
Datum constValue;
bool constIsNull;
int constLoc;
};
/*
* these variables can be shared, mem context on global plancache
*/
struct OpFusionGlobalVariable {
CachedPlanSource* m_psrc; /* to get m_cacheplan in PBE */
CachedPlan* m_cacheplan;
PlannedStmt* m_planstmt; /* m_cacheplan->stmt_list in PBE, plantree in non-PBE */
MemoryContext m_context;
bool m_is_pbe_query;
ParamLoc* m_paramLoc; /* location of m_params, include paramId and the location in indexqual */
Oid m_reloid; /* relation oid of range table */
int m_paramNum;
CachedPlanSource* m_psrc; /* to get m_cacheplan in PBE */
TableAmType m_table_type;
int16* m_attrno; /* target attribute number, length is m_tupDesc->natts */
bool m_is_bucket_rel;
CachedPlan* m_cacheplan;
FusionType m_type;
PlannedStmt* m_planstmt; /* m_cacheplan->stmt_list in PBE, plantree in non-PBE */
int m_natts;
bool m_isFirst; /* be true if is the fisrt execute in PBE */
volatile bool m_is_global;
MemoryContext m_context;
TupleDesc m_tupDesc; /* tuple descriptor */
MemoryContext m_tmpContext; /* use for tmp memory allocation. */
};
ParamListInfo m_params;
OpFusionGlobalVariable *m_global;
/*
* other variables need change each BE, mem context on session cache context
*/
struct OpFusionLocaleVariable {
MemoryContext m_localContext; /* use for local variables */
ParamListInfo m_outParams; /* use outer side parameter. */
MemoryContext m_tmpContext; /* use for tmp memory allocation. */
bool m_isFirst; /* be true if is the fisrt execute in PBE */
ParamListInfo m_outParams; /* use outer side parameter. */
int m_paramNum;
ParamListInfo m_params;
ParamLoc* m_paramLoc; /* location of m_params, include paramId and the location in indexqual */
TupleTableSlot* m_reslot; /* result slot */
Datum* m_values;
bool* m_isnull;
Datum* m_tmpvals; /* for mapping m_values */
bool* m_tmpisnull; /* for mapping m_isnull */
DestReceiver* m_receiver;
bool m_isInsideRec;
int16* m_rformats;
bool m_isCompleted;
long m_position;
const char *m_portalName;
Snapshot m_snapshot;
Oid m_reloid; /* relation oid of range table */
class ScanFusion* m_scan;
TupleDesc m_tupDesc; /* tuple descriptor */
TupleTableSlot* m_reslot; /* result slot */
int16* m_attrno; /* target attribute number, length is m_tupDesc->natts */
Datum* m_values;
bool* m_isnull;
Datum* m_tmpvals; /* for mapping m_values */
bool* m_tmpisnull; /* for mapping m_isnull */
DestReceiver* m_receiver;
bool m_isInsideRec;
bool m_is_pbe_query;
int16* m_rformats;
bool m_isCompleted;
long m_position;
const char *m_portalName;
Snapshot m_snapshot;
class ScanFusion* m_scan;
ResourceOwner m_resOwner;
};
OpFusionLocaleVariable m_local;
private:
#ifdef ENABLE_MOT
static FusionType GetMotFusionType(PlannedStmt* plannedStmt);
@ -196,12 +255,18 @@ public:
void close();
private:
int64 m_limitCount;
void InitLocals(ParamListInfo params);
int64 m_limitOffset;
void InitGlobals();
private:
struct SelectFusionGlobalVariable {
int64 m_limitCount;
int64 m_limitOffset;
};
SelectFusionGlobalVariable* m_c_global;
};
class InsertFusion : public OpFusion {
public:
InsertFusion(MemoryContext context, CachedPlanSource* psrc, List* plantree_list, ParamListInfo params);
@ -210,23 +275,33 @@ public:
bool execute(long max_rows, char* completionTag);
void InitLocals(ParamListInfo params);
void InitGlobals();
private:
void refreshParameterIfNecessary();
EState* m_estate;
struct InsertFusionGlobalVariable {
/* for func/op expr calculation */
FuncExprInfo* m_targetFuncNodes;
int m_targetFuncNum;
int m_targetParamNum;
/* for func/op expr calculation */
FuncExprInfo* m_targetFuncNodes;
int m_targetConstNum;
int m_targetFuncNum;
ConstLoc* m_targetConstLoc;
};
InsertFusionGlobalVariable* m_c_global;
int m_targetParamNum;
struct InsertFusionLocaleVariable {
EState* m_estate;
Datum* m_curVarValue;
bool* m_curVarIsnull;
};
Datum* m_curVarValue;
bool* m_curVarIsnull;
bool m_is_bucket_rel;
InsertFusionLocaleVariable m_c_local;
};
class UpdateFusion : public OpFusion {
@ -237,44 +312,47 @@ public:
bool execute(long max_rows, char* completionTag);
void InitLocals(ParamListInfo params);
void InitGlobals();
private:
HeapTuple heapModifyTuple(HeapTuple tuple);
void refreshTargetParameterIfNecessary();
EState* m_estate;
/* targetlist */
int m_targetNum;
int m_targetParamNum;
Datum* m_targetValues;
bool* m_targetIsnull;
Datum* m_curVarValue;
struct VarLoc {
int varNo;
int scanKeyIndx;
};
struct UpdateFusionGlobalVariable {
/* targetlist */
int m_targetConstNum;
int m_targetParamNum;
VarLoc* m_targetVarLoc;
int m_varNum;
ConstLoc* m_targetConstLoc;
ParamLoc* m_targetParamLoc;
/* for func/op expr calculation */
FuncExprInfo* m_targetFuncNodes;
int m_targetFuncNum;
};
UpdateFusionGlobalVariable* m_c_global;
VarLoc* m_targetVarLoc;
struct UpdateFusionLocaleVariable {
EState* m_estate;
Datum* m_curVarValue;
bool* m_curVarIsnull;
};
int m_varNum;
bool* m_curVarIsnull;
int* m_targetConstLoc;
ParamLoc* m_targetParamLoc;
/* for func/op expr calculation */
FuncExprInfo* m_targetFuncNodes;
int m_targetFuncNum;
bool m_is_bucket_rel;
UpdateFusionLocaleVariable m_c_local;
};
class DeleteFusion : public OpFusion {
@ -285,10 +363,15 @@ public:
bool execute(long max_rows, char* completionTag);
private:
EState* m_estate;
void InitLocals(ParamListInfo params);
bool m_is_bucket_rel;
void InitGlobals();
private:
struct DeleteFusionLocaleVariable {
EState* m_estate;
};
DeleteFusionLocaleVariable m_c_local;
};
#ifdef ENABLE_MOT
@ -299,6 +382,10 @@ public:
~MotJitSelectFusion() {};
bool execute(long max_rows, char *completionTag);
void InitLocals(ParamListInfo params);
void InitGlobals();
};
class MotJitModifyFusion : public OpFusion {
@ -309,9 +396,15 @@ public:
bool execute(long max_rows, char *completionTag);
void InitLocals(ParamListInfo params);
void InitGlobals();
private:
EState* m_estate;
CmdType m_cmdType;
struct MotJitModifyFusionLocaleVariable {
CmdType m_cmdType;
EState* m_estate;
};
MotJitModifyFusionLocaleVariable m_c_local;
};
#endif
@ -325,14 +418,20 @@ public:
void close();
void InitLocals(ParamListInfo params);
void InitGlobals();
private:
int64 m_limitCount;
struct SelectForUpdateFusionGlobalVariable {
int64 m_limitCount;
int64 m_limitOffset;
};
SelectForUpdateFusionGlobalVariable* m_c_global;
EState* m_estate;
int64 m_limitOffset;
bool m_is_bucket_rel;
struct SelectForUpdateFusionLocaleVariable {
EState* m_estate;
};
SelectForUpdateFusionLocaleVariable m_c_local;
};
class AggFusion : public OpFusion {
@ -344,6 +443,10 @@ public:
bool execute(long max_rows, char* completionTag);
void InitLocals(ParamListInfo params);
void InitGlobals();
protected:
typedef void (AggFusion::*aggSumFun)(Datum *transVal, bool transIsNull, Datum *inVal, bool inIsNull);
@ -368,7 +471,10 @@ protected:
dest->buf = NULL; /* digits array is not palloc'd */
}
aggSumFun m_aggSumFunc;
struct AggFusionGlobalVariable {
aggSumFun m_aggSumFunc;
};
AggFusionGlobalVariable* m_c_global;
};
class SortFusion: public OpFusion {
@ -381,8 +487,16 @@ public:
bool execute(long max_rows, char *completionTag);
void InitLocals(ParamListInfo params);
void InitGlobals();
protected:
TupleDesc m_scanDesc;
struct SortFusionLocaleVariable {
TupleDesc m_scanDesc;
};
SortFusionLocaleVariable m_c_local;
};
#endif /* SRC_INCLUDE_OPFUSION_OPFUSION_H_ */

1
src/include/opfusion/opfusion_util.h
View File

@ -38,6 +38,7 @@ const int FUSION_DESCRIB = 1;
extern int namestrcmp(Name name, const char* str);
extern void report_qps_type(CmdType commandType);
void InitOpfusionFunctionId();
Node* JudgePlanIsPartIterator(Plan* plan);
enum FusionType {
NONE_FUSION,

1
src/include/threadpool/threadpool_scheduler.h
View File

@ -46,6 +46,7 @@ public:
return m_has_shutdown;
}
MemoryContext m_gpcContext;
bool m_getSIGHUP;
private:
void AdjustWorkerPool(int idx);
void AdjustStreamPool(int idx);

2
src/include/utils/globalplancore.h
View File

@ -36,7 +36,7 @@
#define GPC_HTAB_SIZE (128)
#define GLOBALPLANCACHEKEY_MAGIC (953717831)
#define CAS_SLEEP_DURATION (2)
#define GPC_CLEAN_WAIT_TIME (3600)
#define GPC_CLEAN_WAIT_TIME (u_sess->attr.attr_common.gpc_clean_timeout)
#define ENABLE_GPC (g_instance.attr.attr_common.enable_global_plancache == true && \
g_instance.attr.attr_common.enable_thread_pool == true)