database/openGauss-server
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add reform info view && drc info view && dms cmd view

Browse Source
This commit is contained in:
jihanyu
2023-11-23 10:03:01 +08:00
parent d0097a9da7
commit ed34160649
11 changed files with 364 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
src/common/backend/catalog/builtin_funcs.ini
View File

@ -12967,3 +12967,11 @@ AddFuncGroup(
"query_page_distribution_info", 1,
AddBuiltinFunc(_0(2866), _1("query_page_distribution_info"), _2(3), _3(true), _4(true), _5(query_page_distribution_info), _6(2249), _7(PG_CATALOG_NAMESPACE), _8(BOOTSTRAP_SUPERUSERID), _9(INTERNALlanguageId), _10(1), _11(64), _12(0), _13(0), _14(false), _15(false), _16(false), _17(false), _18('v'), _19(0), _20(3, TEXTOID, INT4OID, INT4OID), _21(11, TEXTOID, INT4OID, INT4OID, INT4OID, BOOLOID, BOOLOID, BOOLOID, TEXTOID, OIDOID, OIDOID, BOOLOID), _22(11, 'i', 'i', 'i', 'o', 'o', 'o', 'o', 'o', 'o', 'o', 'o'), _23(11, "relname", "fork", "blockno", "instance_id", "is_master", "is_owner", "is_copy", "lock_mode", "mem_lsn", "disk_lsn", "is_dirty"), _24(NULL), _25("query_page_distribution_info"), _26(NULL), _27(NULL), _28(NULL), _29(0), _30(false), _31(NULL), _32(false), _33("statistics: query page distribution information "), _34('f'), _35(NULL), _36(0), _37(false), _38(NULL), _39(NULL), _40(0))
),
AddFuncGroup(
"query_node_reform_info_from_dms", 1,
AddBuiltinFunc(_0(2869), _1("query_node_reform_info_from_dms"), _2(1), _3(true), _4(true), _5(query_node_reform_info_from_dms), _6(2249), _7(PG_CATALOG_NAMESPACE), _8(BOOTSTRAP_SUPERUSERID), _9(INTERNALlanguageId), _10(1), _11(27), _12(0), _13(0), _14(false), _15(false), _16(false), _17(false), _18('v'), _19(0), _20(1, INT4OID), _21(3, INT4OID, TEXTOID, TEXTOID), _22(3, 'i', 'o', 'o'), _23(3, "TYPE", "NAME", "DESCRIPTION"), _24(NULL), _25("query_node_reform_info_from_dms"), _26(NULL), _27(NULL), _28(NULL), _29(0), _30(false), _31(NULL), _32(false), _33("query node reform info from dms"), _34('f'), _35(NULL), _36(0), _37(false), _38(NULL), _39(NULL), _40(0))
),
AddFuncGroup(
"query_all_drc_info", 1,
AddBuiltinFunc(_0(2870), _1("query_all_drc_info"), _2(1), _3(true), _4(true), _5(query_all_drc_info), _6(2249), _7(PG_CATALOG_NAMESPACE), _8(BOOTSTRAP_SUPERUSERID), _9(INTERNALlanguageId), _10(1), _11(27), _12(0), _13(0), _14(false), _15(false), _16(false), _17(false), _18('v'), _19(0), _20(1, INT4OID), _21(19, INT4OID, TEXTOID, INT4OID, INT8OID, INT4OID, INT4OID, INT4OID, INT4OID, INT4OID, INT4OID, INT2OID, INT8OID, INT8OID, INT2OID, INT4OID, INT4OID, INT4OID, INT4OID, INT4OID), _22(19, 'i', 'o', 'o', 'o','o','o','o','o','o','o', 'o', 'o', 'o','o','o','o','o','o','o'), _23(19, "TYPE", "RESOURCE_ID", "MASTER_ID", "COPY_INSTS", "CLAIMED_OWNER", "LOCK_MODE", "LAST_EDP", "TYPE", "IN_RECOVERY", "COPY_PROMOTE", "PART_ID", "EDP_MAP", "LSN", "LEN", "RECOVERY_SKIP", "RECYCLING", "CONVERTING_INST_ID", "CONVERTING_CURR_MODE", "CONVERTING_REQ_MODE"), _24(NULL), _25("query_all_drc_info"), _26(NULL), _27(NULL), _28(NULL), _29(0), _30(false), _31(NULL), _32(false), _33("query all drc info"), _34('f'), _35(NULL), _36(0), _37(false), _38(NULL), _39(NULL), _40(0))
),

179
src/common/backend/utils/adt/pgstatfuncs.cpp
View File

@ -82,6 +82,8 @@
#include "ddes/dms/ss_dms_bufmgr.h"
#include "storage/file/fio_device.h"
#include "ddes/dms/ss_dms_recovery.h"
#include "utils/json.h"
#include "utils/jsonapi.h"
#define UINT32_ACCESS_ONCE(var) ((uint32)(*((volatile uint32*)&(var))))
#define NUM_PG_LOCKTAG_ID 12
@ -15144,7 +15146,7 @@ int compute_copy_insts_count(uint64 bitmap)
/* this struct is used to control the iteration during query_page_distribution_info */
typedef struct st_dms_iterate {
stat_drc_info_t *drc_info;
dv_drc_buf_info *drc_info;
uint8 iterate_idx;
} dms_iterate_t;
@ -15170,7 +15172,7 @@ Datum query_page_distribution_info_internal(text* relname, ForkNumber fork, Bloc
funcctx = SRF_FIRSTCALL_INIT();
MemoryContext oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);
stat_drc_info_t *drc_info = (stat_drc_info_t*)palloc0(sizeof(stat_drc_info_t));
dv_drc_buf_info *drc_info = (dv_drc_buf_info*)palloc0(sizeof(dv_drc_buf_info));
InitDmsBufContext(&drc_info->dms_ctx, tag);
drc_info->claimed_owner = CM_INVALID_ID8;
drc_info->buf_info[0] = buf_info;
@ -15262,4 +15264,175 @@ Datum query_page_distribution_info(PG_FUNCTION_ARGS)
ForkNumber fork = PG_GETARG_INT64(1);
BlockNumber blockno = PG_GETARG_INT64(2);
return query_page_distribution_info_internal(relname, fork, blockno, fcinfo);
}
}
#define REFORM_INFO_ROW_NUM 27
TupleDesc create_query_node_reform_info_from_dms_tupdesc()
{
int column = 2;
TupleDesc tupdesc = CreateTemplateTupleDesc(column, false);
TupleDescInitEntry(tupdesc, (AttrNumber)1, "NAME", TEXTOID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)2, "DESCRIPTION", TEXTOID, -1, 0);
BlessTupleDesc(tupdesc);
return tupdesc;
}
typedef struct RecordsetState {
JsonLexContext *lex;
Datum row_info[REFORM_INFO_ROW_NUM][2];
int row_id;
int cur_row_id;
char *saved_scalar;
} RecordsetState;
static void recordset_object_field_end(void *state, char *fname, bool isnull)
{
RecordsetState *_state = (RecordsetState *)state;
if (_state->row_id >= REFORM_INFO_ROW_NUM) {
ereport(ERROR, (errmsg("the row number returned from dms exceeds max row number")));
}
_state->row_info[_state->row_id][0] = CStringGetTextDatum(fname);
_state->row_info[_state->row_id][1] = CStringGetTextDatum(_state->saved_scalar);
_state->row_id++;
}
static void recordset_scalar(void *state, char *token, JsonTokenType tokentype)
{
RecordsetState *_state = (RecordsetState *)state;
if (_state->saved_scalar != NULL) {
pfree(_state->saved_scalar);
}
_state->saved_scalar = token;
}
Datum query_node_reform_info_from_dms(PG_FUNCTION_ARGS)
{
dms_info_id_e reform_info_id =
PG_GETARG_INT64(0) == 0 ? dms_info_id_e::DMS_INFO_REFORM_LAST : dms_info_id_e::DMS_INFO_REFORM_CURRENT;
if (!ENABLE_DMS) {
ereport(ERROR, (errmsg("[SS] cannot query query_node_reform_info without shared storage deployment!")));
}
FuncCallContext *funcctx = NULL;
if (SRF_IS_FIRSTCALL()) {
funcctx = SRF_FIRSTCALL_INIT();
MemoryContext oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);
char *json = (char *)palloc0(4096 * sizeof(char)); // 4k is enough
json[0] = '\0';
if (!dms_info(json, 4096, reform_info_id) == GS_SUCCESS) {
ereport(ERROR, (errmsg("[SS] get reform infomation from dms fail!")));
}
if (json[0] == '\0') {
ereport(WARNING, (errmsg("[SS] dms not init!")));
SRF_RETURN_DONE(funcctx);
}
RecordsetState *state = (RecordsetState *)palloc0(sizeof(RecordsetState));
state->row_id = 0;
JsonLexContext *lex = makeJsonLexContext(cstring_to_text(json), true);
pfree(json);
JsonSemAction *sem = (JsonSemAction *)palloc0(sizeof(JsonSemAction));
sem->semstate = (void *)state;
sem->scalar = recordset_scalar;
sem->object_field_end = recordset_object_field_end;
state->lex = lex;
pg_parse_json(lex, sem);
state->cur_row_id = 0;
funcctx->user_fctx = (void *)state;
funcctx->tuple_desc = create_query_node_reform_info_from_dms_tupdesc();
MemoryContextSwitchTo(oldcontext);
}
funcctx = SRF_PERCALL_SETUP();
RecordsetState *state = (RecordsetState *)funcctx->user_fctx;
if (state->cur_row_id < state->row_id) {
bool nulls[2] = {false};
HeapTuple tuple = heap_form_tuple(funcctx->tuple_desc, state->row_info[state->cur_row_id], nulls);
state->cur_row_id++;
SRF_RETURN_NEXT(funcctx, HeapTupleGetDatum(tuple));
}
pfree_ext(state);
SRF_RETURN_DONE(funcctx);
}
TupleDesc create_query_all_drc_info_tupdesc()
{
int column = 18;
TupleDesc tupdesc = CreateTemplateTupleDesc(column, false);
TupleDescInitEntry(tupdesc, (AttrNumber)1, "RESOURCE_ID", TEXTOID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)2, "MASTER_ID", INT4OID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)3, "COPY_INSTS", INT8OID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)4, "CLAIMED_OWNER", INT4OID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)5, "LOCK_MODE", INT4OID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)6, "LAST_EDP", INT4OID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)7, "TYPE", INT4OID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)8, "IN_RECOVERY", INT4OID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)9, "COPY_PROMOTE", INT4OID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)10, "PART_ID", INT2OID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)11, "EDP_MAP", INT8OID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)12, "LSN", INT8OID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)13, "LEN", INT2OID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)14, "RECOVERY_SKIP", INT4OID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)15, "RECYCLING", INT4OID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)16, "CONVERTING_INST_ID", INT4OID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)17, "CONVERTING_CURR_MODE", INT4OID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber)18, "CONVERTING_REQ_MODE", INT4OID, -1, 0);
BlessTupleDesc(tupdesc);
return tupdesc;
}
void fill_drc_info_to_values(dv_drc_buf_info *drc_info, Datum *values)
{
values[0] = CStringGetTextDatum(drc_info->data);
values[1] = UInt32GetDatum((uint32)drc_info->master_id);
values[2] = UInt64GetDatum(drc_info->copy_insts);
values[3] = UInt32GetDatum((uint32)drc_info->claimed_owner);
values[4] = UInt32GetDatum((uint32)drc_info->lock_mode);
values[5] = UInt32GetDatum((uint32)drc_info->last_edp);
values[6] = UInt32GetDatum((uint32)drc_info->type);
values[7] = UInt32GetDatum((uint32)drc_info->in_recovery);
values[8] = UInt32GetDatum((uint32)drc_info->copy_promote);
values[9] = UInt16GetDatum(drc_info->part_id);
values[10] = UInt64GetDatum(drc_info->edp_map);
values[11] = UInt64GetDatum(drc_info->lsn);
values[12] = UInt16GetDatum(drc_info->len);
values[13] = UInt32GetDatum((uint32)drc_info->recovery_skip);
values[14] = UInt32GetDatum((uint32)drc_info->recycling);
values[15] = UInt32GetDatum((uint32)drc_info->converting_req_info_inst_id);
values[16] = UInt32GetDatum((uint32)drc_info->converting_req_info_curr_mod);
values[17] = UInt32GetDatum((uint32)drc_info->converting_req_info_req_mod);
}
Datum query_all_drc_info(PG_FUNCTION_ARGS)
{
int type = PG_GETARG_INT64(0) == 0 ? en_drc_res_type::DRC_RES_PAGE_TYPE : en_drc_res_type::DRC_RES_LOCK_TYPE;
if (!ENABLE_DMS) {
ereport(ERROR, (errmsg("[SS] cannot query query_node_reform_info without shared storage deployment!")));
}
FuncCallContext *funcctx = NULL;
if (SRF_IS_FIRSTCALL()) {
funcctx = SRF_FIRSTCALL_INIT();
MemoryContext oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);
funcctx->tuple_desc = create_query_all_drc_info_tupdesc();
MemoryContextSwitchTo(oldcontext);
}
funcctx = SRF_PERCALL_SETUP();
dv_drc_buf_info drc_info = {0};
unsigned long long rowid = funcctx->call_cntr;
dms_get_buf_res(&rowid, &drc_info, type);
Datum values[18];
bool nulls[18] = {false};
if (drc_info.is_valid) {
fill_drc_info_to_values(&drc_info, values);
HeapTuple tuple = heap_form_tuple(funcctx->tuple_desc, values, nulls);
SRF_RETURN_NEXT(funcctx, HeapTupleGetDatum(tuple));
}
SRF_RETURN_DONE(funcctx);
}

65
src/gausskernel/cbb/instruments/event/instr_waitevent.cpp
View File

@ -397,22 +397,52 @@ static void set_io_event_tuple_value(WaitInfo* gsInstrWaitInfo, Datum* values, i
values[++i] = TimestampTzGetDatum(gsInstrWaitInfo->event_info.io_info[eventId].last_updated);
}
static void set_dms_event_tuple_value(WaitInfo* gsInstrWaitInfo, Datum* values, int i, uint32 eventId)
static bool set_dms_event_tuple_value(WaitInfo* gsInstrWaitInfo, Datum* values, int i, uint32 eventId)
{
values[++i] = CStringGetTextDatum("DMS_EVENT");
values[++i] = CStringGetTextDatum(pgstat_get_wait_dms(WaitEventDMS(eventId + PG_WAIT_DMS)));
if (!g_instance.dms_cxt.dmsInited) {
ereport(WARNING, (errmsg("[SS] dms not init!")));
return false;
}
unsigned long long cnt = 0;
unsigned long long time = 0;
if (g_instance.dms_cxt.dmsInited) {
dms_get_event(dms_wait_event_t(eventId), &cnt, &time);
}
dms_get_event(dms_wait_event_t(eventId), &cnt, &time);
values[++i] = Int64GetDatum(cnt);
values[++i] = Int64GetDatum(gsInstrWaitInfo->event_info.dms_info[eventId].failed_counter);
values[++i] = Int64GetDatum(INT64_MIN);
values[++i] = Int64GetDatum(time);
values[++i] = Int64GetDatum(cnt == 0 ? 0 : time / cnt);
values[++i] = Int64GetDatum(gsInstrWaitInfo->event_info.dms_info[eventId].max_duration);
values[++i] = Int64GetDatum(gsInstrWaitInfo->event_info.dms_info[eventId].min_duration);
values[++i] = Int64GetDatum(INT64_MIN);
values[++i] = Int64GetDatum(INT64_MIN);
values[++i] = TimestampTzGetDatum(gsInstrWaitInfo->event_info.dms_info[eventId].last_updated);
return true;
}
static bool set_dms_cmd_tuple_value(WaitInfo *gsInstrWaitInfo, Datum *values, int i, uint32 eventId)
{
values[++i] = CStringGetTextDatum("DMS_CMD");
if (!g_instance.dms_cxt.dmsInited) {
ereport(WARNING, (errmsg("[SS] dms not init!")));
return false;
}
wait_cmd_stat_result_t cmd_stat_result;
dms_get_cmd_stat(eventId, &cmd_stat_result);
if (!cmd_stat_result.is_valid) {
return false;
}
values[++i] = CStringGetTextDatum(cmd_stat_result.name);
values[++i] = Int64GetDatum(cmd_stat_result.wait_count);
values[++i] = Int64GetDatum(INT64_MIN);
values[++i] = Int64GetDatum(cmd_stat_result.wait_time);
values[++i] =
Int64GetDatum(cmd_stat_result.wait_count == 0 ? 0 : cmd_stat_result.wait_time / cmd_stat_result.wait_count);
values[++i] = Int64GetDatum(INT64_MIN);
values[++i] = Int64GetDatum(INT64_MIN);
values[++i] = TimestampTzGetDatum(gsInstrWaitInfo->event_info.dms_info[eventId].last_updated);
return true;
}
static void set_lock_event_tuple_value(WaitInfo* gsInstrWaitInfo, Datum* values, int i, uint32 eventId)
@ -447,7 +477,7 @@ static void set_lwlock_event_tuple_value(WaitInfo* gsInstrWaitInfo, Datum* value
values[++i] = TimestampTzGetDatum(gsInstrWaitInfo->event_info.lwlock_info[eventId].last_updated);
}
static void set_tuple_value(
static bool set_tuple_value(
WaitInfo* gsInstrWaitInfo, Datum* values, bool* nulls, int i, uint32 eventId, uint32 call_cn)
{
values[++i] = CStringGetTextDatum(g_instance.attr.attr_common.PGXCNodeName);
@ -465,8 +495,12 @@ static void set_tuple_value(
set_lwlock_event_tuple_value(gsInstrWaitInfo, values, i, eventId, nulls);
} else if (call_cn < LOCK_EVENT_NUM + IO_EVENT_NUM + STATE_WAIT_NUM + LWLOCK_EVENT_NUM + DMS_EVENT_NUM) {
eventId = call_cn - LOCK_EVENT_NUM - IO_EVENT_NUM - STATE_WAIT_NUM - LWLOCK_EVENT_NUM;
set_dms_event_tuple_value(gsInstrWaitInfo, values, i, eventId);
return set_dms_event_tuple_value(gsInstrWaitInfo, values, i, eventId);
} else {
eventId = call_cn - LOCK_EVENT_NUM - IO_EVENT_NUM - STATE_WAIT_NUM - LWLOCK_EVENT_NUM - DMS_EVENT_NUM;
return set_dms_cmd_tuple_value(gsInstrWaitInfo, values, i, eventId);
}
return true;
}
Datum get_instr_wait_event(PG_FUNCTION_ARGS)
@ -495,7 +529,6 @@ Datum get_instr_wait_event(PG_FUNCTION_ARGS)
}
funcctx->user_fctx = read_current_instr_wait_info();
funcctx->max_calls = STATE_WAIT_NUM + IO_EVENT_NUM + LOCK_EVENT_NUM + LWLOCK_EVENT_NUM + DMS_EVENT_NUM;
MemoryContextSwitchTo(oldcontext);
@ -504,7 +537,7 @@ Datum get_instr_wait_event(PG_FUNCTION_ARGS)
}
funcctx = SRF_PERCALL_SETUP();
if (funcctx->user_fctx != NULL && funcctx->call_cntr < funcctx->max_calls) {
if (funcctx->user_fctx != NULL) {
Datum values[INSTR_WAITEVENT_ATTRUM];
bool nulls[INSTR_WAITEVENT_ATTRUM] = {false};
HeapTuple tuple = NULL;
@ -518,11 +551,11 @@ Datum get_instr_wait_event(PG_FUNCTION_ARGS)
rc = memset_s(nulls, sizeof(nulls), 0, sizeof(nulls));
securec_check(rc, "\0", "\0");
set_tuple_value(gsInstrWaitInfo, values, nulls, i, eventId, funcctx->call_cntr);
tuple = heap_form_tuple(funcctx->tuple_desc, values, nulls);
result = HeapTupleGetDatum(tuple);
SRF_RETURN_NEXT(funcctx, result);
if (set_tuple_value(gsInstrWaitInfo, values, nulls, i, eventId, funcctx->call_cntr)) {
tuple = heap_form_tuple(funcctx->tuple_desc, values, nulls);
result = HeapTupleGetDatum(tuple);
SRF_RETURN_NEXT(funcctx, result);
}
}
pfree_ext(funcctx->user_fctx);

24
src/gausskernel/ddes/adapter/ss_dms.cpp
View File

@ -130,6 +130,9 @@ int ss_dms_func_init()
SS_RETURN_IFERR(DMS_LOAD_SYMBOL_FUNC(dms_get_mes_max_watting_rooms));
SS_RETURN_IFERR(DMS_LOAD_SYMBOL_FUNC(dms_send_opengauss_oldest_xmin));
SS_RETURN_IFERR(DMS_LOAD_SYMBOL_FUNC(dms_get_drc_info));
SS_RETURN_IFERR(DMS_LOAD_SYMBOL_FUNC(dms_info));
SS_RETURN_IFERR(DMS_LOAD_SYMBOL_FUNC(dms_get_buf_res));
SS_RETURN_IFERR(DMS_LOAD_SYMBOL_FUNC(dms_get_cmd_stat));
g_ss_dms_func.inited = true;
return DMS_SUCCESS;
@ -361,7 +364,26 @@ int dms_send_opengauss_oldest_xmin(dms_context_t *dms_ctx, unsigned long long ol
return g_ss_dms_func.dms_send_opengauss_oldest_xmin(dms_ctx, oldest_xmin, dest_id);
}
int get_drc_info(int* is_found, stat_drc_info_t* drc_info)
int get_drc_info(int *is_found, dv_drc_buf_info *drc_info)
{
return g_ss_dms_func.dms_get_drc_info(is_found, drc_info);
}
int dms_info(char *buf, unsigned int len, dms_info_id_e id)
{
if (g_ss_dms_func.inited) {
return g_ss_dms_func.dms_info(buf, len, id);
} else {
return DMS_SUCCESS;
}
}
void dms_get_buf_res(unsigned long long *row_id, dv_drc_buf_info *drc_info, int type)
{
g_ss_dms_func.dms_get_buf_res(row_id, drc_info, type);
}
void dms_get_cmd_stat(int index, wait_cmd_stat_result_t *cmd_stat_result)
{
g_ss_dms_func.dms_get_cmd_stat(index, cmd_stat_result);
}

2
src/gausskernel/ddes/ddes_commit_id
View File

@ -1,3 +1,3 @@
dms_commit_id=d2024b8747272b35d59b46b3ca38497ef6dc255e
dms_commit_id=5f086645dc59e43258d1179cb37f82cc95200e5d
dss_commit_id=a304c8a21cc67282215d7653d5aa596810688149
cbb_commit_id=100323132dff039c9225216e64770050ad8a6d2b

5
src/include/catalog/upgrade_sql/rollback_catalog_maindb/rollback-post_catalog_maindb_92_920.sql
View File

@ -7,3 +7,8 @@ CREATE VIEW pg_catalog.pg_publication_tables AS
FROM pg_publication P, pg_class C
JOIN pg_namespace N ON (N.oid = C.relnamespace)
WHERE C.oid IN (SELECT relid FROM pg_catalog.pg_get_publication_tables(P.pubname));
-- drop reform info functions
DROP FUNCTION IF EXISTS pg_catalog.query_node_reform_info_from_dms() CASCADE;
-- drop drc info functions
DROP FUNCTION IF EXISTS pg_catalog.query_all_drc_info() CASCADE;

5
src/include/catalog/upgrade_sql/rollback_catalog_otherdb/rollback-post_catalog_otherdb_92_920.sql
View File

@ -7,3 +7,8 @@ CREATE VIEW pg_catalog.pg_publication_tables AS
FROM pg_publication P, pg_class C
JOIN pg_namespace N ON (N.oid = C.relnamespace)
WHERE C.oid IN (SELECT relid FROM pg_catalog.pg_get_publication_tables(P.pubname));
-- drop reform info functions
DROP FUNCTION IF EXISTS pg_catalog.query_node_reform_info_from_dms() CASCADE;
-- drop drc info functions
DROP FUNCTION IF EXISTS pg_catalog.query_all_drc_info() CASCADE;

38
src/include/catalog/upgrade_sql/upgrade_catalog_maindb/upgrade-post_catalog_maindb_92_920.sql
View File

@ -11,3 +11,41 @@ CREATE VIEW pg_catalog.pg_publication_tables AS
pg_class C
JOIN pg_namespace N ON (N.oid = C.relnamespace)
WHERE C.oid = gpt.relid;
-- create query_node_reform_info_from_dms
DROP FUNCTION IF EXISTS pg_catalog.query_node_reform_info_from_dms() CASCADE;
SET LOCAL inplace_upgrade_next_system_object_oids=IUO_PROC, 2869;
CREATE FUNCTION pg_catalog.query_node_reform_info_from_dms
(
int4,
out name text,
out description text
)
RETURNS SETOF record LANGUAGE INTERNAL as 'query_node_reform_info_from_dms';
-- create query_all_drc_info
DROP FUNCTION IF EXISTS pg_catalog.query_all_drc_info() CASCADE;
SET LOCAL inplace_upgrade_next_system_object_oids=IUO_PROC, 2870;
CREATE FUNCTION pg_catalog.query_all_drc_info
(
int4,
out RESOURCE_ID text,
out MASTER_ID int4,
out COPY_INSTS int8,
out CLAIMED_OWNER int4,
out LOCK_MODE int4,
out LAST_EDP int4,
out TYPE int4,
out IN_RECOVERY char,
out COPY_PROMOTE int4,
out PART_ID int4,
out EDP_MAP int8,
out LSN int8,
out LEN int4,
out RECOVERY_SKIP int4,
out RECYCLING char,
out CONVERTING_INST_ID int4,
out CONVERTING_CURR_MODE int4,
out CONVERTING_REQ_MODE int4
)
RETURNS SETOF record LANGUAGE INTERNAL as 'query_all_drc_info';

38
src/include/catalog/upgrade_sql/upgrade_catalog_otherdb/upgrade-post_catalog_otherdb_92_920.sql
View File

@ -11,3 +11,41 @@ CREATE VIEW pg_catalog.pg_publication_tables AS
pg_class C
JOIN pg_namespace N ON (N.oid = C.relnamespace)
WHERE C.oid = gpt.relid;
-- create query_node_reform_info_from_dms
DROP FUNCTION IF EXISTS pg_catalog.query_node_reform_info_from_dms() CASCADE;
SET LOCAL inplace_upgrade_next_system_object_oids=IUO_PROC, 2869;
CREATE FUNCTION pg_catalog.query_node_reform_info_from_dms
(
int4,
out name text,
out description text
)
RETURNS SETOF record LANGUAGE INTERNAL as 'query_node_reform_info_from_dms';
-- create query_all_drc_info
DROP FUNCTION IF EXISTS pg_catalog.query_all_drc_info() CASCADE;
SET LOCAL inplace_upgrade_next_system_object_oids=IUO_PROC, 2870;
CREATE FUNCTION pg_catalog.query_all_drc_info
(
int4,
out RESOURCE_ID text,
out MASTER_ID int4,
out COPY_INSTS int8,
out CLAIMED_OWNER int4,
out LOCK_MODE int4,
out LAST_EDP int4,
out TYPE int4,
out IN_RECOVERY char,
out COPY_PROMOTE int4,
out PART_ID int4,
out EDP_MAP int8,
out LSN int8,
out LEN int4,
out RECOVERY_SKIP int4,
out RECYCLING char,
out CONVERTING_INST_ID int4,
out CONVERTING_CURR_MODE int4,
out CONVERTING_REQ_MODE int4
)
RETURNS SETOF record LANGUAGE INTERNAL as 'query_all_drc_info';

41
src/include/ddes/dms/dms_api.h
View File

@ -32,7 +32,7 @@ extern "C" {
#define DMS_LOCAL_MINOR_VER_WEIGHT 1000
#define DMS_LOCAL_MAJOR_VERSION 0
#define DMS_LOCAL_MINOR_VERSION 0
#define DMS_LOCAL_VERSION 120
#define DMS_LOCAL_VERSION 121
#define DMS_SUCCESS 0
#define DMS_ERROR (-1)
@ -676,33 +676,9 @@ typedef struct st_stat_buf_info {
/*
* used by openGauss server to get DRC information
*/
typedef struct st_stat_drc_info {
typedef struct st_dv_drc_buf_info {
stat_buf_info_t buf_info[DMS_MAX_INSTANCES]; /* save buffer related information */
dms_context_t dms_ctx;
unsigned char master_id;
unsigned long long copy_insts; /* bitmap for owners, for S mode, more than one owner may exist */
unsigned char claimed_owner; /* owner */
unsigned char lock_mode; /* current DRC lock mode */
unsigned char last_edp; /* the newest edp instance id */
unsigned char type; /* page or lock */
unsigned char in_recovery; /* in recovery or not */
unsigned char copy_promote; /* copy promote to owner, can not release, may need flush */
unsigned short part_id; /* which partition id that current page belongs to */
unsigned long long edp_map; /* indicate which instance has current page's EDP(Earlier Dirty Page) */
unsigned long long lsn; /* the newest edp LSN of current page in the cluster */
unsigned short len; /* the length of data below */
unsigned char recovery_skip; /* DRC is accessed in recovery and skip because drc has owner */
unsigned char recycling;
char data[DMS_RESID_SIZE]; /* user defined resource(page) identifier */
} stat_drc_info_t;
typedef enum en_broadcast_scope {
DMS_BROADCAST_OLDIN_LIST = 0, // default value
DMS_BROADCAST_ONLINE_LIST = 1,
DMS_BROADCAST_TYPE_COUNT,
} dms_broadcast_scope_e;
typedef struct st_dv_drc_buf_info {
char data[DMS_MAX_NAME_LEN]; /* user defined resource(page) identifier */
unsigned char master_id;
unsigned long long copy_insts; /* bitmap for owners, for S mode, more than one owner may exist */
@ -719,11 +695,17 @@ typedef struct st_dv_drc_buf_info {
unsigned char recovery_skip; /* DRC is accessed in recovery and skip because drc has owner */
unsigned char recycling;
unsigned char converting_req_info_inst_id;
unsigned char converting_req_info_curr_mode;
unsigned char converting_req_info_req_mode;
unsigned char converting_req_info_curr_mod;
unsigned char converting_req_info_req_mod;
unsigned char is_valid;
} dv_drc_buf_info;
typedef enum en_broadcast_scope {
DMS_BROADCAST_OLDIN_LIST = 0, // default value
DMS_BROADCAST_ONLINE_LIST = 1,
DMS_BROADCAST_TYPE_COUNT,
} dms_broadcast_scope_e;
typedef struct st_dms_reform_start_context {
dms_role_t role;
dms_reform_type_t reform_type;
@ -1109,6 +1091,9 @@ typedef enum en_dms_info_id {
DMS_INFO_REFORM_LAST = 1,
} dms_info_id_e;
/*
* used by openGauss server to get dms cmd information
*/
typedef struct st_wait_cmd_stat_result {
char name[DMS_MAX_NAME_LEN];
char p1[DMS_MAX_NAME_LEN];

10
src/include/ddes/dms/ss_dms.h
View File

@ -86,7 +86,10 @@ typedef struct st_ss_dms_func {
int *redo_status);
unsigned int (*dms_get_mes_max_watting_rooms)(void);
int (*dms_send_opengauss_oldest_xmin)(dms_context_t *dms_ctx, unsigned long long oldest_xmin, unsigned char dest_id);
int (*dms_get_drc_info)(int* is_found, stat_drc_info_t* drc_info);
int (*dms_get_drc_info)(int *is_found, dv_drc_buf_info *drc_info);
int (*dms_info)(char *buf, unsigned int len, dms_info_id_e id);
void (*dms_get_buf_res)(unsigned long long *row_id, dv_drc_buf_info *drc_info, int type);
void (*dms_get_cmd_stat)(int index, wait_cmd_stat_result_t *cmd_stat_result);
} ss_dms_func_t;
int ss_dms_func_init();
@ -136,7 +139,10 @@ int dms_reform_req_opengauss_ondemand_redo_buffer(dms_context_t *dms_ctx, void *
unsigned int dms_get_mes_max_watting_rooms(void);
int dms_send_opengauss_oldest_xmin(dms_context_t *dms_ctx, unsigned long long oldest_xmin, unsigned char dest_id);
int get_drc_info(int* is_found, stat_drc_info_t* drc_info);
int get_drc_info(int *is_found, dv_drc_buf_info *drc_info);
int dms_info(char *buf, unsigned int len, dms_info_id_e id);
void dms_get_buf_res(unsigned long long *row_id, dv_drc_buf_info *drc_info, int type);
void dms_get_cmd_stat(int index, wait_cmd_stat_result_t *cmd_stat_result);
#ifdef __cplusplus
}