* Copyright (c) Huawei Technologies Co., Ltd. 2022. All rights reserved.
*
* 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.
* ---------------------------------------------------------------------------------------
*
* eventcmds.cpp
* Routines for CREATE/ALTER/DROP EVENT commands.
* IDENTIFICATION
* src/gausskernel/optimizer/commands/eventcmds.cpp
*
* ---------------------------------------------------------------------------------------
*/
#include "postgres.h"
#include "knl/knl_variable.h"
#include "access/genam.h"
#include "access/heapam.h"
#include "access/tableam.h"
#include "access/sysattr.h"
#include "catalog/dependency.h"
#include "catalog/indexing.h"
#include "catalog/objectaccess.h"
#include "catalog/pg_object.h"
#include "catalog/pg_proc.h"
#include "catalog/pg_type.h"
#include "catalog/gs_db_privilege.h"
#include "commands/defrem.h"
#include "commands/proclang.h"
#include "executor/executor.h"
#include "knl/knl_thread.h"
#include "miscadmin.h"
#include "optimizer/var.h"
#include "parser/parse_collate.h"
#include "parser/parse_func.h"
#include "storage/tcap.h"
#include "utils/acl.h"
#include "utils/builtins.h"
#include "utils/fmgroids.h"
#include "utils/guc.h"
#include "utils/inval.h"
#include "utils/lsyscache.h"
#include "utils/rel.h"
#include "utils/rel_gs.h"
#include "utils/syscache.h"
#include "utils/snapmgr.h"
#include "utils/plpgsql.h"
#include "storage/lmgr.h"
#include "tcop/utility.h"
#include "catalog/pg_job.h"
#include "tcop/dest.h"
#include "commands/sqladvisor.h"
#include "access/printtup.h"
#include "access/attnum.h"
#include "nodes/makefuncs.h"
#include "postgres_ext.h"
#include "catalog/gs_job_attribute.h"
#include "utils/dbe_scheduler.h"
#include "catalog/pg_job_proc.h"
#include "catalog/pg_authid.h"
#include "commands/user.h"
#include "access/htup.h"
#include "commands/dbcommands.h"
#define SHOW_EVENT_SIZE 10
#define INTERVAL_QUALITY_LENGTH 16
typedef struct field_str_map_st {
int interval;
const char *field_str;
} field_str_map;
const field_str_map g_field_str_map[] = {
{INTERVAL_MASK(YEAR), " year"},
{INTERVAL_MASK(MONTH), " month"},
{INTERVAL_MASK(DAY), " day"},
{INTERVAL_MASK(HOUR), " hour"},
{INTERVAL_MASK(MINUTE), " minute"},
{INTERVAL_MASK(SECOND), " second"},
{INTERVAL_MASK(YEAR) | INTERVAL_MASK(MONTH), " year to month"},
{INTERVAL_MASK(DAY) | INTERVAL_MASK(HOUR), " day to hour"},
{INTERVAL_MASK(DAY) | INTERVAL_MASK(HOUR) | INTERVAL_MASK(MINUTE), " day to minute"},
{INTERVAL_MASK(DAY) | INTERVAL_MASK(HOUR) | INTERVAL_MASK(MINUTE) | INTERVAL_MASK(SECOND), " day to second"},
{INTERVAL_MASK(HOUR) | INTERVAL_MASK(MINUTE), " hour to minute"},
{INTERVAL_MASK(HOUR) | INTERVAL_MASK(MINUTE) | INTERVAL_MASK(SECOND), " hour to second"},
{INTERVAL_MASK(MINUTE) | INTERVAL_MASK(SECOND), " minute to second"},
{INTERVAL_FULL_RANGE, ""},
};
enum class job_type {
ARG_JOB_DEFAULT,
ARG_JOB_ACTION,
ARG_START_DATE,
ARG_END_DATE,
ARG_REPEAT_INTERVAL,
ARG_JOB_AUTO_DROP,
ARG_JOB_ENABLED,
ARG_JOB_COMMENTS,
ARG_JOB_RENAME,
ARG_JOB_DEFINER,
ARG_JOB_DEFINER_DEFAULT
};
typedef struct job_script_map_st {
job_type act_name;
const char *act_type;
} job_script;
const job_script g_job_script[] = {{job_type::ARG_JOB_ACTION, "program_action"},
{job_type::ARG_START_DATE, "start_date"},
{job_type::ARG_REPEAT_INTERVAL, "repeat_interval"},
{job_type::ARG_JOB_AUTO_DROP, "auto_drop"},
{job_type::ARG_END_DATE, "end_date"},
{job_type::ARG_JOB_COMMENTS, "comments"},
{job_type::ARG_JOB_ENABLED, "enabled"},
{job_type::ARG_JOB_RENAME, "rename"},
{job_type::ARG_JOB_DEFINER, "owner"},
{job_type::ARG_JOB_DEFINER_DEFAULT, "owner_default"}};
bool CheckEventExists(Datum ev_name, bool miss_ok)
{
bool is_exists;
CatCList *list = SearchSysCacheList1(JOBATTRIBUTENAME, ev_name);
is_exists = (list->n_members > 0) ? true : false;
ReleaseSysCacheList(list);
if (!is_exists) {
return false;
}
if (miss_ok) {
ereport(NOTICE, (errcode(ERRCODE_DUPLICATE_OBJECT),
errmsg("event \"%s\" already exists, skipping", TextDatumGetCString(ev_name))));
return true;
} else {
ereport(ERROR, (errcode(ERRCODE_DUPLICATE_OBJECT),
errmsg("event \"%s\" already exists", TextDatumGetCString(ev_name))));
}
return false;
}
bool CheckEventNotExists(Datum ev_name, bool miss_ok)
{
bool is_exists;
CatCList *list = SearchSysCacheList1(JOBATTRIBUTENAME, ev_name);
is_exists = (list->n_members > 0) ? true : false;
ReleaseSysCacheList(list);
if (is_exists) {
return false;
}
if (miss_ok) {
ereport(NOTICE, (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
errmsg("event \"%s\" is not exists, skipping", TextDatumGetCString(ev_name))));
return true;
} else {
ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
errmsg("event \"%s\" is not exists", TextDatumGetCString(ev_name))));
}
return false;
}
const char *IntervalTypmodParse(A_Const *expr)
{
long interval_num = expr->val.val.ival;
size_t size = (sizeof(g_field_str_map) / sizeof(g_field_str_map[0]));
for (size_t i = 0; i < size; ++i) {
if (g_field_str_map[i].interval == interval_num) {
return g_field_str_map[i].field_str;
}
}
return NULL;
}
Datum ParseIntevalExpr(Node *intervalNode)
{
StringInfoData buf;
initStringInfo(&buf);
appendStringInfo(&buf, "interval");
TypeCast *tc = (TypeCast *)intervalNode;
A_Const *ac = (A_Const *)tc->arg;
A_Const *tm = (A_Const *)lfirst(list_head(tc->typname->typmods));
const char *tm_str = NULL;
tm_str = IntervalTypmodParse(tm);
if (IsA(&ac->val, Integer)) {
char *quantity_str = (char *)palloc(INTERVAL_QUALITY_LENGTH);
pg_itoa((int)ac->val.val.ival, quantity_str);
appendStringInfo(&buf, " \'%s\' ", quantity_str);
} else if (IsA(&ac->val, String) || IsA(&ac->val, Float)) {
appendStringInfo(&buf, " \'%s\' ", (char *)ac->val.val.str);
}
appendStringInfo(&buf, "%s", tm_str);
return CStringGetTextDatum(buf.data);
}
char* parseIntervalExprString(Node *intervalNode)
{
char *res;
StringInfoData buf;
initStringInfo(&buf);
TypeCast *tc = (TypeCast *)intervalNode;
A_Const *ac = (A_Const *)tc->arg;
A_Const *tm = (A_Const *)lfirst(list_head(tc->typname->typmods));
const char *tm_str = NULL;
tm_str = IntervalTypmodParse(tm);
if (IsA(&ac->val, Integer)) {
char *quantity_str = (char *)palloc(INTERVAL_QUALITY_LENGTH);
pg_itoa((int)ac->val.val.ival, quantity_str);
appendStringInfo(&buf, " \'%s\' ", quantity_str);
} else if (IsA(&ac->val, String) || IsA(&ac->val, Float)) {
appendStringInfo(&buf, " \'%s\' ", (char *)ac->val.val.str);
}
appendStringInfo(&buf, "%s", tm_str);
return pstrdup(buf.data);
}
Datum ExecTimeExpr(Node *node)
{
Oid result_type = exprType(node);
if (result_type != TIMESTAMPOID && result_type != TIMESTAMPTZOID) {
ereport(ERROR,
(errcode(ERRCODE_INVALID_DATETIME_FORMAT), errmsg("the format of the time expression is incorrect.")));
}
EState *estate = NULL;
ExprContext *econtext = NULL;
Expr *time_expr = NULL;
ExprState *exprstate = NULL;
time_expr = (Expr *)node;
bool is_null = false;
Datum result;
estate = CreateExecutorState();
econtext = GetPerTupleExprContext(estate);
exprstate = ExecPrepareExpr(time_expr, estate);
if (!PointerIsValid(exprstate)) {
ereport(ERROR, (errcode(ERRCODE_DUPLICATE_OBJECT),
errmsg("failed when making time expression state for constCompare.")));
}
result = ExecEvalExpr(exprstate, econtext, &is_null, NULL);
if (result_type == TIMESTAMPTZOID) {
result = DirectFunctionCall1(timestamptz_timestamp, DatumGetTimestampTz(result));
}
FreeExecutorState(estate);
return result;
}
char* parseTimeExprString(Node* timeEexpr)
{
Datum timeres;
timeres = ExecTimeExpr(timeEexpr);
return DatumGetCString(DirectFunctionCall1(timestamp_out, timeres));
}
void GetTimeExecResult(CreateEventStmt *stmt, Datum &start_time, Datum &interval_time, Datum &end_time)
{
Node *interval_time_expr = stmt->interval_time;
interval_time = (Datum)0;
if (interval_time_expr) {
interval_time = ParseIntevalExpr(interval_time_expr);
}
Node *end_time_expr = stmt->end_time_expr;
end_time = get_scheduler_max_timestamptz();
if (end_time_expr) {
end_time = ExecTimeExpr(end_time_expr);
}
Node *start_time_expr = stmt->start_time_expr;
start_time = DirectFunctionCall1(timestamptz_timestamp, DatumGetTimestampTz(GetCurrentTimestamp()));
if (start_time_expr) {
start_time = ExecTimeExpr(start_time_expr);
}
}
Datum TranslateArg(char *act_name, Node *act_node)
{
Datum result = (Datum)0;
job_type ev_act_type = job_type::ARG_JOB_DEFAULT;
size_t size = (sizeof(g_job_script) / sizeof(g_job_script[0]));
for (size_t i = 0; i < size; ++i) {
if (!strcmp(g_job_script[i].act_type, act_name)) {
ev_act_type = g_job_script[i].act_name;
break;
}
}
switch (ev_act_type) {
case job_type::ARG_JOB_ACTION:
case job_type::ARG_JOB_COMMENTS: {
Value *event_arg_node = (Value *)act_node;
result = CStringGetTextDatum(event_arg_node->val.str);
break;
}
case job_type::ARG_JOB_RENAME: {
Value *new_name_val = (Value *)act_node;
result = CStringGetTextDatum(new_name_val->val.str);
break;
}
case job_type::ARG_JOB_DEFINER: {
Value *ev_definer_node = (Value *)act_node;
HOLD_INTERRUPTS();
Oid user_oid = GetSysCacheOid1(AUTHNAME, CStringGetDatum(ev_definer_node->val.str));
RESUME_INTERRUPTS();
CHECK_FOR_INTERRUPTS();
if (!OidIsValid(user_oid))
ereport(ERROR, (errcode(ERRCODE_UNDEFINED_OBJECT),
errmsg("role \"%s\" does not exist.", ev_definer_node->val.str)));
char *name = (char *)palloc(MAX_JOB_NAME_LEN * sizeof(char));
pg_ltoa((int)user_oid, name);
result = CStringGetTextDatum(name);
break;
}
case job_type::ARG_JOB_DEFINER_DEFAULT: {
char *username = get_role_name_str();
result = CStringGetTextDatum(username);
break;
}
case job_type::ARG_START_DATE:
case job_type::ARG_END_DATE: {
result = ExecTimeExpr(act_node);
break;
}
case job_type::ARG_REPEAT_INTERVAL: {
result = ParseIntevalExpr(act_node);
break;
}
case job_type::ARG_JOB_ENABLED: {
Value *ev_status_node = (Value *)act_node;
EventStatus ev_status = (EventStatus)(ev_status_node->val.ival);
if (ev_status == EVENT_DISABLE_ON_SLAVE || ev_status == EVENT_DISABLE) {
result = BoolGetDatum(0);
} else if (ev_status == EVENT_ENABLE) {
result = BoolGetDatum(1);
}
break;
}
case job_type::ARG_JOB_AUTO_DROP: {
Value *ev_drop_node = (Value *)act_node;
result = ev_drop_node->val.ival ? CStringGetTextDatum("true") : CStringGetTextDatum("false");
break;
}
default:
return (Datum)0;
}
return result;
}
void CheckDefinerPriviledge(char *user_name)
{
Oid user_oid = GetSysCacheOid1(AUTHNAME, CStringGetDatum(user_name));
Oid init_user_id = 10;
if (user_oid != GetUserId()) {
if (g_instance.attr.attr_security.enablePrivilegesSeparate) {
ereport(ERROR, (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
errmsg("definer_name cannot be specified when PrivilegesSeparate is enabled.")));
} else if (user_oid == init_user_id) {
ereport(ERROR, (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
errmsg("definer_name cannot be specified as the initial user.")));
} else if (is_role_independent(user_oid) || isMonitoradmin(user_oid) || isOperatoradmin(user_oid)) {
ereport(ERROR, (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
errmsg("definer_name cannot be specified as a private user, operator admin, or monitoradmin.")));
}
}
}
Datum SetDefinerName(char *def_name, Datum program_name, char** definer_oid)
{
Datum curuser = get_priv_user(program_name, CharGetDatum(JOB_INTYPE_PLAIN));
if (def_name) {
CheckDefinerPriviledge(def_name);
if (!superuser()) {
ereport(ERROR, (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
errmsg("The current user does not have sufficient permissions to specify the definer.")));
} else {
HeapTuple oldtuple = SearchSysCache1(AUTHNAME, CStringGetDatum(def_name));
Value* definer_name = makeString(def_name);
(*definer_oid) = TextDatumGetCString(TranslateArg("owner", (Node*)definer_name));
if (!HeapTupleIsValid(oldtuple)) {
ereport(ERROR, (errcode(ERRCODE_UNDEFINED_OBJECT), errmsg("role \"%s\" does not exist.", def_name)));
} else {
ReleaseSysCache(oldtuple);
return CStringGetDatum(def_name);
}
}
}
if (SECUREC_LIKELY(*definer_oid == NULL)) {
(*definer_oid) = get_role_name_str();
}
return curuser;
}
void PrepareFuncArg(CreateEventStmt *stmt, Datum ev_name, Datum schemaName, FunctionCallInfoData *ev_arg)
{
char* job_definer_oid = NULL;
Datum definer = SetDefinerName(stmt->def_name, ev_name, &job_definer_oid);
Datum job_action = CStringGetTextDatum(stmt->event_query_str);
char *job_type_string = PLSQL_JOB_TYPE;
text *job_type_text = cstring_to_text_with_len(job_type_string, strlen(job_type_string));
Datum job_type = CStringGetTextDatum(text_to_cstring(job_type_text));
char *c_schedule_name = get_inline_schedule_name(ev_name);
Datum schedule_name = CStringGetTextDatum(c_schedule_name);
char *c_program_name =
CreateEventInlineProgram(ev_name, job_type, job_action, CStringGetTextDatum(job_definer_oid));
Datum program_name = CStringGetTextDatum(c_program_name);
Datum start_time;
Datum interval_time;
Datum end_time;
GetTimeExecResult(stmt, start_time, interval_time, end_time);
ev_arg->arg[ARG_0] = ev_name;
ev_arg->arg[ARG_1] = program_name;
ev_arg->arg[ARG_2] = schedule_name;
ev_arg->arg[ARG_3] = CStringGetTextDatum("DEFAULT_JOB_CLASS");
ev_arg->arg[ARG_4] = (stmt->event_status == EVENT_ENABLE) ? BoolGetDatum(1) : BoolGetDatum(0);
ev_arg->arg[ARG_5] = BoolGetDatum(stmt->complete_preserve);
ev_arg->arg[ARG_6] = (stmt->event_comment_str == NULL) ? (Datum)0 : CStringGetTextDatum(stmt->event_comment_str);
ev_arg->arg[ARG_7] = CStringGetTextDatum("REGULAR");
ev_arg->arg[ARG_8] = (Datum)0;
ev_arg->arg[ARG_9] = (Datum)0;
ev_arg->arg[ARG_10] = CharGetDatum(JOB_INTYPE_PLAIN);
ev_arg->arg[ARG_11] = TimeStampToText(start_time);
ev_arg->arg[ARG_12] = interval_time;
ev_arg->arg[ARG_13] = TimeStampToText(end_time);
ev_arg->arg[ARG_14] = job_action;
ev_arg->arg[ARG_15] = job_type;
ev_arg->arg[ARG_16] = definer;
ev_arg->arg[ARG_17] = schemaName;
ev_arg->arg[ARG_18] = CStringGetTextDatum(job_definer_oid);
}
void CheckEventPrivilege(char* schema_name, char* event_name, AclMode mode, bool is_create_or_alter)
{
Oid user_oid = GetUserId();
Oid schema_oid = get_namespace_oid(schema_name, true);
AclResult acl_result = pg_namespace_aclcheck(schema_oid, user_oid, mode);
if(acl_result != ACLCHECK_OK) {
aclcheck_error(acl_result, ACL_KIND_NAMESPACE, schema_name);
}
if(superuser_arg(user_oid) || systemDBA_arg(user_oid)) {
return;
}
if(is_create_or_alter) {
return;
}
HeapTuple tup = NULL;
tup = SearchSysCache2(JOBATTRIBUTENAME, CStringGetTextDatum(event_name), CStringGetTextDatum("owner"));
if(!HeapTupleIsValid(tup)) {
ereport(ERROR, (errcode(ERRCODE_UNDEFINED_OBJECT),
errmsg("event \"%s\" does not exist", event_name)));
}
bool isnull = false;
Datum owner_datum = SysCacheGetAttr(JOBATTRIBUTENAME, tup, Anum_gs_job_attribute_attribute_value, &isnull);
char* owner_id_str = TextDatumGetCString(owner_datum);
Oid owner_id = pg_atoi(owner_id_str, sizeof(int32), '\0');
if(owner_id != user_oid) {
ereport(ERROR, (errcode(ERRCODE_UNDEFINED_OBJECT),
errmsg("only event's owner have the permission to operate object \"%s\"", event_name)));
}
ReleaseSysCache(tup);
}
ObjectAddress CreateEventCommand(CreateEventStmt *stmt)
{
ObjectAddress myself;
char *event_name_str = stmt->event_name->relname;
char *schema_name_str = (stmt->event_name->schemaname) ? stmt->event_name->schemaname : get_real_search_schema();
if (!stmt->event_name->schemaname) {
stmt->event_name->schemaname = pstrdup(schema_name_str);
}
CheckEventPrivilege(schema_name_str, event_name_str, ACL_CREATE, true);
myself.classId = PgJobRelationId;
myself.objectId = 0;
myself.objectSubId = 0;
Datum schema_name = DirectFunctionCall1(namein, CStringGetDatum(schema_name_str));
Datum ev_name = CStringGetTextDatum(event_name_str);
FunctionCallInfoData ev_arg;
const short nrgs_job = ARG_19;
if (CheckEventExists(ev_name, stmt->if_not_exists)) {
return myself;
}
InitFunctionCallInfoData(ev_arg, NULL, nrgs_job, InvalidOid, NULL, NULL);
errno_t rc = memset_s(ev_arg.arg, nrgs_job * sizeof(Datum), 0, nrgs_job * sizeof(Datum));
securec_check(rc, "\0", "\0");
rc = memset_s(ev_arg.argnull, nrgs_job * sizeof(bool), 0, nrgs_job * sizeof(bool));
securec_check(rc, "\0", "\0");
PrepareFuncArg(stmt, ev_name, schema_name, &ev_arg);
create_job_raw(&ev_arg);
return myself;
}
Datum GetInlineJobName(Datum ev_name)
{
errno_t rc;
char *c_program_name = (char *)palloc(sizeof(char) * MAX_JOB_NAME_LEN);
rc = strcpy_s(c_program_name, MAX_JOB_NAME_LEN, INLINE_JOB_PROGRAM_PREFIX);
securec_check(rc, "\0", "\0");
rc = strcat_s(c_program_name, MAX_JOB_NAME_LEN, TextDatumGetCString(ev_name));
securec_check(rc, "\0", "\0");
return CStringGetTextDatum(c_program_name);
}
void UpdateMultiAttribute(Datum ev_name, Datum attValue, Datum new_name, Datum att_name)
{
Datum values[Natts_gs_job_attribute] = {0};
bool nulls[Natts_gs_job_attribute] = {0};
bool replaces[Natts_gs_job_attribute] = {0};
replaces[Anum_gs_job_attribute_attribute_value - 1] = true;
values[Anum_gs_job_attribute_attribute_value - 1] = attValue;
nulls[Anum_gs_job_attribute_attribute_value - 1] = false;
replaces[Anum_gs_job_attribute_job_name - 1] = true;
values[Anum_gs_job_attribute_job_name - 1] = new_name;
nulls[Anum_gs_job_attribute_job_name - 1] = false;
Relation gs_job_attribute_rel = heap_open(GsJobAttributeRelationId, RowExclusiveLock);
HeapTuple oldtuple = SearchSysCache2(JOBATTRIBUTENAME, ev_name, att_name);
if (oldtuple == NULL) {
ereport(ERROR, (errcode(ERRCODE_UNDEFINED_OBJECT), errmsg("Fail to update attribute."),
errdetail("Attribute entry %s not found.", TextDatumGetCString(att_name)),
errcause("attribute is not exist"), erraction("Please check object_name")));
}
HeapTuple newtuple = heap_modify_tuple(oldtuple, RelationGetDescr(gs_job_attribute_rel), values, nulls, replaces);
simple_heap_update(gs_job_attribute_rel, &newtuple->t_self, newtuple);
CatalogUpdateIndexes(gs_job_attribute_rel, newtuple);
heap_close(gs_job_attribute_rel, NoLock);
ReleaseSysCache(oldtuple);
heap_freetuple_ext(newtuple);
}
void UpdateAttributeMultiColum(Datum ev_name, char *attribute_name[], int attr_pos, DefElem *para_def,
Datum new_name_result)
{
attribute_name[attr_pos] = para_def->defname;
Datum arg_result = TranslateArg(para_def->defname, para_def->arg);
if (!strcmp(para_def->defname, "owner_default")) {
UpdateMultiAttribute(ev_name, arg_result, new_name_result, CStringGetTextDatum("owner"));
} else {
UpdateMultiAttribute(ev_name, arg_result, new_name_result, CStringGetTextDatum(para_def->defname));
}
}
void UpdateSingleAttribute(Datum ev_name, DefElem *para_def)
{
Datum arg_result = TranslateArg(para_def->defname, para_def->arg);
Datum att_name = strcmp(para_def->defname, "owner_default") ? CStringGetTextDatum(para_def->defname)
: CStringGetTextDatum("owner");
update_attribute(ev_name, att_name, arg_result);
}
void UpdateMultiRows(AlterEventStmt *stmt, int attr_pos, char *attribute_name[], Datum ev_name)
{
Datum values[Natts_gs_job_attribute] = {0};
bool nulls[Natts_gs_job_attribute] = {0};
bool replaces[Natts_gs_job_attribute] = {0};
Datum arg_result = TranslateArg(stmt->new_name->defname, (Node *)stmt->new_name->arg);
replaces[Anum_gs_job_attribute_job_name - 1] = true;
values[Anum_gs_job_attribute_job_name - 1] = arg_result;
nulls[Anum_gs_job_attribute_job_name - 1] = false;
Relation gs_job_attribute_rel = heap_open(GsJobAttributeRelationId, RowExclusiveLock);
CatCList *list = SearchSysCacheList1(JOBATTRIBUTENAME, ev_name);
bool is_null = false;
Datum enattr;
char *enattname;
bool is_repeat = false;
for (int i = 0; i < list->n_members; i++) {
HeapTuple enum_tup = t_thrd.lsc_cxt.FetchTupleFromCatCList(list, i);
enattr = heap_getattr(enum_tup, Anum_gs_job_attribute_attribute_name, RelationGetDescr(gs_job_attribute_rel),
&is_null);
enattname = TextDatumGetCString(enattr);
if (strcmp(enattname, "program_name") == 0) {
replaces[Anum_gs_job_attribute_attribute_value - 1] = true;
values[Anum_gs_job_attribute_attribute_value - 1] = GetInlineJobName(arg_result);
nulls[Anum_gs_job_attribute_attribute_value - 1] = false;
}
if (strcmp(enattname, "schedule_name") == 0) {
replaces[Anum_gs_job_attribute_attribute_value - 1] = true;
values[Anum_gs_job_attribute_attribute_value - 1] =
CStringGetTextDatum(get_inline_schedule_name(arg_result));
nulls[Anum_gs_job_attribute_attribute_value - 1] = false;
}
for (int j = 0; j < attr_pos; ++j) {
if (!strcmp(attribute_name[j], "owner_default")) {
attribute_name[j] = "owner";
};
if (strcmp(enattname, attribute_name[j]) == 0) {
is_repeat = true;
break;
}
}
if (is_repeat) {
is_repeat = false;
continue;
}
HeapTuple old_tup = enum_tup;
enum_tup = heap_copytuple(old_tup);
HeapTuple newtuple =
heap_modify_tuple(enum_tup, RelationGetDescr(gs_job_attribute_rel), values, nulls, replaces);
if (strcmp(enattname, "program_name") == 0) {
replaces[Anum_gs_job_attribute_attribute_value - 1] = false;
values[Anum_gs_job_attribute_attribute_value - 1] = (Datum)0;
nulls[Anum_gs_job_attribute_attribute_value - 1] = true;
}
if (strcmp(enattname, "schedule_name") == 0) {
replaces[Anum_gs_job_attribute_attribute_value - 1] = false;
values[Anum_gs_job_attribute_attribute_value - 1] = (Datum)0;
nulls[Anum_gs_job_attribute_attribute_value - 1] = true;
}
simple_heap_update(gs_job_attribute_rel, &newtuple->t_self, newtuple);
CatalogUpdateIndexes(gs_job_attribute_rel, newtuple);
}
ReleaseSysCacheList(list);
list = SearchSysCacheList1(JOBATTRIBUTENAME, GetInlineJobName(ev_name));
values[Anum_gs_job_attribute_job_name - 1] = GetInlineJobName(arg_result);
for (int i = 0; i < list->n_members; i++) {
HeapTuple enum_tup = t_thrd.lsc_cxt.FetchTupleFromCatCList(list, i);
enattr = heap_getattr(enum_tup, Anum_gs_job_attribute_attribute_name, RelationGetDescr(gs_job_attribute_rel),
&is_null);
enattname = TextDatumGetCString(enattr);
for (int j = 0; j < attr_pos; ++j) {
if (!strcmp(attribute_name[j], "owner_default")) {
attribute_name[j] = "owner";
};
if (strcmp(enattname, attribute_name[j]) == 0 && strcmp(enattname, "comments") != 0) {
is_repeat = true;
break;
}
}
if (is_repeat) {
is_repeat = false;
continue;
}
HeapTuple old_tup = enum_tup;
enum_tup = heap_copytuple(old_tup);
HeapTuple newtuple =
heap_modify_tuple(enum_tup, RelationGetDescr(gs_job_attribute_rel), values, nulls, replaces);
simple_heap_update(gs_job_attribute_rel, &newtuple->t_self, newtuple);
CatalogUpdateIndexes(gs_job_attribute_rel, newtuple);
}
ReleaseSysCacheList(list);
heap_close(gs_job_attribute_rel, RowExclusiveLock);
}
void UpdateAttributeParam(AlterEventStmt *stmt, Datum ev_name)
{
bool need_rename = (stmt->new_name == NULL) ? false : true;
Datum new_name_result = (Datum)0;
if (need_rename) {
new_name_result = TranslateArg(stmt->new_name->defname, stmt->new_name->arg);
}
const int attribute_num = ARG_17;
int attr_pos = 0;
char *attribute_name[attribute_num] = {0};
Datum inline_name = GetInlineJobName(ev_name);
if (stmt->event_comment_str) {
if (need_rename) {
UpdateAttributeMultiColum(ev_name, attribute_name, attr_pos++, stmt->event_comment_str, new_name_result);
} else {
UpdateSingleAttribute(ev_name, stmt->event_comment_str);
}
}
if (stmt->complete_preserve) {
if (need_rename) {
UpdateAttributeMultiColum(ev_name, attribute_name, attr_pos++, stmt->complete_preserve, new_name_result);
} else {
UpdateSingleAttribute(ev_name, stmt->complete_preserve);
}
}
if (stmt->def_name) {
if (need_rename) {
UpdateAttributeMultiColum(ev_name, attribute_name, attr_pos++, stmt->def_name, new_name_result);
UpdateAttributeMultiColum(inline_name, attribute_name, attr_pos++, stmt->def_name,
GetInlineJobName(new_name_result));
} else {
UpdateSingleAttribute(ev_name, stmt->def_name);
UpdateSingleAttribute(inline_name, stmt->def_name);
}
} else {
DefElem* defaute_definer_def = makeDefElem("owner_default", NULL);
if (need_rename) {
UpdateAttributeMultiColum(ev_name, attribute_name, attr_pos++, defaute_definer_def, new_name_result);
UpdateAttributeMultiColum(inline_name, attribute_name, attr_pos++, defaute_definer_def,
GetInlineJobName(new_name_result));
} else {
UpdateSingleAttribute(ev_name, defaute_definer_def);
UpdateSingleAttribute(inline_name, defaute_definer_def);
}
}
if (stmt->event_query_str) {
if (need_rename) {
UpdateAttributeMultiColum(inline_name, attribute_name, attr_pos++, stmt->event_query_str,
GetInlineJobName(new_name_result));
} else {
UpdateSingleAttribute(inline_name, stmt->event_query_str);
}
}
if (need_rename) {
UpdateMultiRows(stmt, attr_pos, attribute_name, ev_name);
}
}
void UpdateMultiProc(Datum attValue, Datum nameValue, Datum jobName)
{
Datum values[Natts_pg_job_proc] = {0};
bool nulls[Natts_pg_job_proc] = {0};
bool replaces[Natts_pg_job_proc] = {0};
Relation rel = heap_open(PgJobProcRelationId, RowExclusiveLock);
replaces[Anum_pg_job_proc_what - 1] = true;
values[Anum_pg_job_proc_what - 1] = attValue;
nulls[Anum_pg_job_proc_what - 1] = false;
replaces[Anum_pg_job_proc_job_name - 1] = true;
values[Anum_pg_job_proc_job_name - 1] = nameValue;
nulls[Anum_pg_job_proc_job_name - 1] = false;
HeapTuple oldtuple = search_from_pg_job_proc(rel, jobName);
HeapTuple newtuple = heap_modify_tuple(oldtuple, RelationGetDescr(rel), values, nulls, replaces);
simple_heap_update(rel, &newtuple->t_self, newtuple);
CatalogUpdateIndexes(rel, newtuple);
heap_close(rel, RowExclusiveLock);
heap_freetuple_ext(oldtuple);
heap_freetuple_ext(newtuple);
}
void UpdatePgJobProcName(Datum job_name, DefElem *new_name)
{
Datum values[Natts_pg_job_proc] = {0};
bool nulls[Natts_pg_job_proc] = {0};
bool replaces[Natts_pg_job_proc] = {0};
Relation rel = heap_open(PgJobProcRelationId, RowExclusiveLock);
Datum arg_result = TranslateArg(new_name->defname, new_name->arg);
replaces[Anum_pg_job_proc_job_name - 1] = true;
values[Anum_pg_job_proc_job_name - 1] = arg_result;
nulls[Anum_pg_job_proc_job_name - 1] = false;
HeapTuple oldtuple = search_from_pg_job_proc(rel, job_name);
HeapTuple newtuple = heap_modify_tuple(oldtuple, RelationGetDescr(rel), values, nulls, replaces);
simple_heap_update(rel, &newtuple->t_self, newtuple);
CatalogUpdateIndexes(rel, newtuple);
heap_close(rel, RowExclusiveLock);
heap_freetuple_ext(oldtuple);
heap_freetuple_ext(newtuple);
}
void UpdatePgJobProcParam(AlterEventStmt *stmt, Datum ev_name)
{
if (stmt->event_query_str) {
Datum arg_result = TranslateArg(stmt->event_query_str->defname, stmt->event_query_str->arg);
if (stmt->new_name) {
Datum ev_new_name = TranslateArg(stmt->new_name->defname, stmt->new_name->arg);
UpdateMultiProc(arg_result, ev_new_name, ev_name);
} else {
dbe_update_pg_job_proc(arg_result, ev_name);
}
} else {
if (stmt->new_name) {
UpdatePgJobProcName(ev_name, stmt->new_name);
}
}
}
void UpdateMultiJob(Datum job_name, Datum *values, bool *nulls, bool *replaces)
{
Relation pg_job_rel = heap_open(PgJobRelationId, RowExclusiveLock);
HeapTuple oldtuple = search_from_pg_job(pg_job_rel, job_name);
if (!HeapTupleIsValid(oldtuple)) {
heap_close(pg_job_rel, NoLock);
ereport(ERROR, (errcode(ERRCODE_UNDEFINED_DATABASE),
errmsg("event \"%s\" does not exist", TextDatumGetCString(job_name))));
}
HeapTuple newtuple = heap_modify_tuple(oldtuple, RelationGetDescr(pg_job_rel), values, nulls, replaces);
simple_heap_update(pg_job_rel, &newtuple->t_self, newtuple);
CatalogUpdateIndexes(pg_job_rel, newtuple);
heap_close(pg_job_rel, NoLock);
heap_freetuple_ext(newtuple);
heap_freetuple_ext(oldtuple);
}
void UpdatePgJobParam(AlterEventStmt *stmt, Datum ev_name)
{
Datum arg_result;
Datum values[Natts_pg_job] = {0};
bool nulls[Natts_pg_job] = {0};
bool replaces[Natts_pg_job] = {0};
Datum definer = DirectFunctionCall1(namein, CStringGetDatum(GetUserNameFromId(GetUserId())));
if (stmt->def_name) {
Value *definerVal = (Value *)stmt->def_name->arg;
definer = DirectFunctionCall1(namein, CStringGetDatum(definerVal->val.str));
}
values[Anum_pg_job_log_user - 1] = definer;
nulls[Anum_pg_job_log_user - 1] = false;
replaces[Anum_pg_job_log_user - 1] = true;
values[Anum_pg_job_priv_user - 1] = definer;
nulls[Anum_pg_job_priv_user - 1] = false;
replaces[Anum_pg_job_priv_user - 1] = true;
if (stmt->end_time_expr) {
arg_result = TranslateArg(stmt->end_time_expr->defname, stmt->end_time_expr->arg);
values[Anum_pg_job_end_date - 1] = arg_result;
nulls[Anum_pg_job_end_date - 1] = false;
replaces[Anum_pg_job_end_date - 1] = true;
}
if (stmt->start_time_expr) {
arg_result = TranslateArg(stmt->start_time_expr->defname, stmt->start_time_expr->arg);
values[Anum_pg_job_start_date - 1] = arg_result;
nulls[Anum_pg_job_start_date - 1] = false;
replaces[Anum_pg_job_start_date - 1] = true;
values[Anum_pg_job_next_run_date - 1] = arg_result;
nulls[Anum_pg_job_next_run_date - 1] = false;
replaces[Anum_pg_job_next_run_date - 1] = true;
}
if (stmt->interval_time) {
arg_result = (stmt->interval_time->arg == NULL)
? CStringGetTextDatum("null")
: TranslateArg(stmt->interval_time->defname, stmt->interval_time->arg);
values[Anum_pg_job_interval - 1] = arg_result;
nulls[Anum_pg_job_interval - 1] = false;
replaces[Anum_pg_job_interval - 1] = true;
}
if (stmt->event_status) {
arg_result = TranslateArg(stmt->event_status->defname, stmt->event_status->arg);
values[Anum_pg_job_enable - 1] = arg_result;
nulls[Anum_pg_job_enable - 1] = false;
replaces[Anum_pg_job_enable - 1] = true;
}
if (stmt->new_name) {
arg_result = TranslateArg(stmt->new_name->defname, stmt->new_name->arg);
values[Anum_pg_job_job_name - 1] = arg_result;
nulls[Anum_pg_job_job_name - 1] = false;
replaces[Anum_pg_job_job_name - 1] = true;
}
UpdateMultiJob(ev_name, values, nulls, replaces);
}
ObjectAddress AlterEventCommand(AlterEventStmt *stmt)
{
ObjectAddress myself;
char *event_name_str = stmt->event_name->relname;
char *schema_name_str = (stmt->event_name->schemaname) ? stmt->event_name->schemaname : get_real_search_schema();
if (!stmt->event_name->schemaname) {
stmt->event_name->schemaname = pstrdup(schema_name_str);
}
Datum ev_name = CStringGetTextDatum(event_name_str);
CheckEventPrivilege(schema_name_str, event_name_str, ACL_USAGE, false);
check_object_is_visible(ev_name, false);
if (stmt->def_name) {
Value *ev_definer_node = (Value *)stmt->def_name->arg;
CheckDefinerPriviledge(ev_definer_node->val.str);
if (!superuser()) {
ereport(ERROR, (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
errmsg("The current user does not have sufficient permissions to specify the definer.")));
}
}
UpdateAttributeParam(stmt, ev_name);
UpdatePgJobProcParam(stmt, ev_name);
UpdatePgJobParam(stmt, ev_name);
myself.classId = PgJobRelationId;
myself.objectId = 0;
myself.objectSubId = 0;
return myself;
}
void DropEventCommand(DropEventStmt *stmt)
{
char *event_name_str = stmt->event_name->relname;
char *schema_name_str = (stmt->event_name->schemaname) ? stmt->event_name->schemaname : get_real_search_schema();
if (!stmt->event_name->schemaname) {
stmt->event_name->schemaname = pstrdup(schema_name_str);
}
Datum ev_name = CStringGetTextDatum(event_name_str);
if (CheckEventNotExists(ev_name, stmt->missing_ok)) {
return;
}
CheckEventPrivilege(schema_name_str, event_name_str, ACL_USAGE, false);
FunctionCallInfoData ev_arg;
const short nrgs_job = ARG_3;
InitFunctionCallInfoData(ev_arg, NULL, nrgs_job, InvalidOid, NULL, NULL);
errno_t rc = memset_s(ev_arg.arg, nrgs_job * sizeof(Datum), 0, nrgs_job * sizeof(Datum));
securec_check(rc, "\0", "\0");
rc = memset_s(ev_arg.argnull, nrgs_job * sizeof(bool), 0, nrgs_job * sizeof(bool));
securec_check(rc, "\0", "\0");
ev_arg.arg[ARG_0] = ev_name;
ev_arg.arg[ARG_1] = BoolGetDatum(0);
ev_arg.arg[ARG_2] = BoolGetDatum(0);
drop_single_job_internal(&ev_arg);
}
StmtResult *SearchEventInfo(ShowEventStmt *stmt)
{
StringInfoData buf;
initStringInfo(&buf);
appendStringInfo(&buf, "SELECT ");
char* intervalColName;
if (u_sess->attr.attr_sql.dolphin) {
intervalColName = "`interval`";
} else {
intervalColName = "interval";
}
appendStringInfo(&buf,
"job_name,"
"nspname,"
"log_user,"
"priv_user,"
"job_status,"
"start_date,"
"%s,"
"end_date,"
"enable,"
"failure_msg ",
intervalColName);
appendStringInfo(&buf, "FROM PG_JOB ");
appendStringInfo(&buf, "WHERE dbname=\'%s\' AND ", get_database_name(u_sess->proc_cxt.MyDatabaseId));
if (stmt->from_clause) {
A_Const *fc = (A_Const *)stmt->from_clause;
appendStringInfo(&buf, "nspname=\'%s\' ", (char *)fc->val.val.str);
} else {
char *schema_name = get_real_search_schema();
appendStringInfo(&buf, "nspname=\'%s\' ", schema_name);
}
appendStringInfo(&buf, " AND job_name IS NOT NULL ");
if (stmt->where_clause) {
appendStringInfo(&buf, " AND ");
appendStringInfo(&buf, " %s ", stmt->where_clause);
}
return execute_stmt(buf.data, true);
}
void ShowEventCommand(ShowEventStmt *stmt, DestReceiver *dest)
{
TupOutputState *tstate = NULL;
TupleDesc tupdesc;
Datum values[SHOW_EVENT_SIZE] = {0};
bool isnull[SHOW_EVENT_SIZE] = {false};
tupdesc = CreateTemplateTupleDesc(SHOW_EVENT_SIZE, false, TableAmHeap);
TupleDescInitEntry(tupdesc, (AttrNumber)ARG_1, "job_name", TEXTOID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)ARG_2, "schema_name", TEXTOID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)ARG_3, "log_user", TEXTOID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)ARG_4, "priv_user", TEXTOID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)ARG_5, "job_status", TEXTOID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)ARG_6, "start_date", TEXTOID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)ARG_7, "interval", TEXTOID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)ARG_8, "end_date", TEXTOID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)ARG_9, "enable", TEXTOID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)ARG_10, "failure_msg", TEXTOID, -1, 0);
tstate = begin_tup_output_tupdesc(dest, tupdesc);
StmtResult *res = SearchEventInfo(stmt);
List *eventList = res->tuples;
ListCell *eventCell = NULL;
foreach (eventCell, eventList) {
List *eventTuple = (List *)eventCell->data.ptr_value;
ListCell *eventDesc = NULL;
char *ival;
int eventPos = 0;
foreach (eventDesc, eventTuple) {
ival = (char *)eventDesc->data.ptr_value;
values[eventPos++] = PointerGetDatum(cstring_to_text(ival));
}
do_tup_output(tstate, values, SHOW_EVENT_SIZE, isnull, SHOW_EVENT_SIZE);
for (int i = 0; i < SHOW_EVENT_SIZE; ++i) {
pfree(DatumGetPointer(values[i]));
}
}
end_tup_output(tstate);
}
TupleDesc GetEventResultDesc()
{
TupleDesc tupdesc;
tupdesc = CreateTemplateTupleDesc(SHOW_EVENT_SIZE, false, TableAmHeap);
TupleDescInitEntry(tupdesc, (AttrNumber)ARG_1, "job_name", TEXTOID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)ARG_2, "schema_name", TEXTOID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)ARG_3, "log_user", TEXTOID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)ARG_4, "priv_user", TEXTOID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)ARG_5, "job_status", TEXTOID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)ARG_6, "start_date", TEXTOID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)ARG_7, "interval", TEXTOID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)ARG_8, "end_date", TEXTOID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)ARG_9, "enable", TEXTOID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)ARG_10, "failure_msg", TEXTOID, -1, 0);
return tupdesc;
}