xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge pull request #24010 from taosdata/feat/TS-4229 

feat: compact monitor and kill
This commit is contained in:
Hongze Cheng 2023-12-15 16:20:13 +08:00 committed by GitHub
parent 556eaa22ac 5e7aead4d8
commit 4ae13f411e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
45 changed files with 7076 additions and 4075 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
include/common/systable.h
View File

@ -50,6 +50,8 @@ extern "C" {
#define TSDB_INS_TABLE_STREAM_TASKS "ins_stream_tasks"
#define TSDB_INS_TABLE_USER_PRIVILEGES "ins_user_privileges"
#define TSDB_INS_TABLE_VIEWS "ins_views"
#define TSDB_INS_TABLE_COMPACTS "ins_compacts"
#define TSDB_INS_TABLE_COMPACT_DETAILS "ins_compact_details"
#define TSDB_PERFORMANCE_SCHEMA_DB "performance_schema"
#define TSDB_PERFS_TABLE_SMAS "perf_smas"

1
include/common/tglobal.h
View File

@ -196,6 +196,7 @@ extern int64_t tsWalFsyncDataSizeLimit;
// internal
extern int32_t tsTransPullupInterval;
extern int32_t tsCompactPullupInterval;
extern int32_t tsMqRebalanceInterval;
extern int32_t tsStreamCheckpointInterval;
extern float tsSinkDataRate;

56
include/common/tmsg.h
View File

@ -144,6 +144,8 @@ typedef enum _mgmt_table {
TSDB_MGMT_TABLE_STREAM_TASKS,
TSDB_MGMT_TABLE_PRIVILEGES,
TSDB_MGMT_TABLE_VIEWS,
TSDB_MGMT_TABLE_COMPACT,
TSDB_MGMT_TABLE_COMPACT_DETAIL,
TSDB_MGMT_TABLE_MAX,
} EShowType;
@ -307,6 +309,7 @@ typedef enum ENodeType {
QUERY_NODE_KILL_CONNECTION_STMT,
QUERY_NODE_KILL_QUERY_STMT,
QUERY_NODE_KILL_TRANSACTION_STMT,
QUERY_NODE_KILL_COMPACT_STMT,
QUERY_NODE_DELETE_STMT,
QUERY_NODE_INSERT_STMT,
QUERY_NODE_QUERY,
@ -353,6 +356,8 @@ typedef enum ENodeType {
QUERY_NODE_SHOW_VNODES_STMT,
QUERY_NODE_SHOW_USER_PRIVILEGES_STMT,
QUERY_NODE_SHOW_VIEWS_STMT,
QUERY_NODE_SHOW_COMPACTS_STMT,
QUERY_NODE_SHOW_COMPACT_DETAILS_STMT,
// logic plan node
QUERY_NODE_LOGIC_PLAN_SCAN = 1000,
@ -1384,6 +1389,24 @@ int32_t tSerializeSCompactDbReq(void* buf, int32_t bufLen, SCompactDbReq* pReq);
int32_t tDeserializeSCompactDbReq(void* buf, int32_t bufLen, SCompactDbReq* pReq);
void tFreeSCompactDbReq(SCompactDbReq* pReq);
typedef struct {
int32_t compactId;
int8_t bAccepted;
} SCompactDbRsp;
int32_t tSerializeSCompactDbRsp(void* buf, int32_t bufLen, SCompactDbRsp* pRsp);
int32_t tDeserializeSCompactDbRsp(void* buf, int32_t bufLen, SCompactDbRsp* pRsp);
typedef struct {
int32_t compactId;
int32_t sqlLen;
char* sql;
} SKillCompactReq;
int32_t tSerializeSKillCompactReq(void* buf, int32_t bufLen, SKillCompactReq* pReq);
int32_t tDeserializeSKillCompactReq(void* buf, int32_t bufLen, SKillCompactReq* pReq);
void tFreeSKillCompactReq(SKillCompactReq *pReq);
typedef struct {
char name[TSDB_FUNC_NAME_LEN];
int8_t igExists;
@ -1662,6 +1685,26 @@ int32_t tSerializeSCreateVnodeReq(void* buf, int32_t bufLen, SCreateVnodeReq* pR
int32_t tDeserializeSCreateVnodeReq(void* buf, int32_t bufLen, SCreateVnodeReq* pReq);
int32_t tFreeSCreateVnodeReq(SCreateVnodeReq* pReq);
typedef struct {
int32_t compactId;
int32_t vgId;
int32_t dnodeId;
} SQueryCompactProgressReq;
int32_t tSerializeSQueryCompactProgressReq(void* buf, int32_t bufLen, SQueryCompactProgressReq* pReq);
int32_t tDeserializeSQueryCompactProgressReq(void* buf, int32_t bufLen, SQueryCompactProgressReq* pReq);
typedef struct {
int32_t compactId;
int32_t vgId;
int32_t dnodeId;
int32_t numberFileset;
int32_t finished;
} SQueryCompactProgressRsp;
int32_t tSerializeSQueryCompactProgressRsp(void* buf, int32_t bufLen, SQueryCompactProgressRsp* pReq);
int32_t tDeserializeSQueryCompactProgressRsp(void* buf, int32_t bufLen, SQueryCompactProgressRsp* pReq);
typedef struct {
int32_t vgId;
int32_t dnodeId;
@ -1689,11 +1732,21 @@ typedef struct {
char db[TSDB_DB_FNAME_LEN];
int64_t compactStartTime;
STimeWindow tw;
int32_t compactId;
} SCompactVnodeReq;
int32_t tSerializeSCompactVnodeReq(void* buf, int32_t bufLen, SCompactVnodeReq* pReq);
int32_t tDeserializeSCompactVnodeReq(void* buf, int32_t bufLen, SCompactVnodeReq* pReq);
typedef struct {
int32_t compactId;
int32_t vgId;
int32_t dnodeId;
} SVKillCompactReq;
int32_t tSerializeSVKillCompactReq(void* buf, int32_t bufLen, SVKillCompactReq* pReq);
int32_t tDeserializeSVKillCompactReq(void* buf, int32_t bufLen, SVKillCompactReq* pReq);
typedef struct {
int32_t vgVersion;
int32_t buffer;
@ -1889,8 +1942,9 @@ typedef struct {
char db[TSDB_DB_FNAME_LEN];
char tb[TSDB_TABLE_NAME_LEN];
char user[TSDB_USER_LEN];
char filterTb[TSDB_TABLE_NAME_LEN];
char filterTb[TSDB_TABLE_NAME_LEN]; // for ins_columns
int64_t showId;
int64_t compactId; // for compact
} SRetrieveTableReq;
typedef struct SSysTableSchema {

6
include/common/tmsgdef.h
View File

@ -217,6 +217,8 @@
TD_DEF_MSG_TYPE(TDMT_MND_CREATE_VIEW, "create-view", SCMCreateViewReq, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_DROP_VIEW, "drop-view", SCMDropViewReq, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_VIEW_META, "view-meta", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_KILL_COMPACT, "kill-compact", SKillCompactReq, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_COMPACT_TIMER, "compact-tmr", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_MAX_MSG, "mnd-max", NULL, NULL)
TD_CLOSE_MSG_SEG(TDMT_END_MND_MSG)
@ -256,7 +258,7 @@
TD_DEF_MSG_TYPE(TDMT_VND_EXEC_RSMA, "vnode-exec-rsma", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_DELETE, "delete-data", SVDeleteReq, SVDeleteRsp)
TD_DEF_MSG_TYPE(TDMT_VND_BATCH_DEL, "batch-delete", SBatchDeleteReq, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_ALTER_CONFIG, "alter-config", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_ALTER_CONFIG, "alter-config", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_ALTER_REPLICA, "alter-replica", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_ALTER_CONFIRM, "alter-confirm", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_ALTER_HASHRANGE, "alter-hashrange", NULL, NULL)
@ -267,6 +269,8 @@ TD_DEF_MSG_TYPE(TDMT_VND_ALTER_CONFIG, "alter-config", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_CREATE_INDEX, "vnode-create-index", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_DROP_INDEX, "vnode-drop-index", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_DISABLE_WRITE, "vnode-disable-write", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_QUERY_COMPACT_PROGRESS, "vnode-query-compact-progress", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_KILL_COMPACT, "kill-compact", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_MAX_MSG, "vnd-max", NULL, NULL)
TD_CLOSE_MSG_SEG(TDMT_END_VND_MSG)

352
include/common/ttokendef.h
View File

@ -189,184 +189,180 @@
#define TK_ALIVE 170
#define TK_VIEWS 171
#define TK_VIEW 172
#define TK_NORMAL 173
#define TK_CHILD 174
#define TK_LIKE 175
#define TK_TBNAME 176
#define TK_QTAGS 177
#define TK_AS 178
#define TK_SYSTEM 179
#define TK_INDEX 180
#define TK_FUNCTION 181
#define TK_INTERVAL 182
#define TK_COUNT 183
#define TK_LAST_ROW 184
#define TK_META 185
#define TK_ONLY 186
#define TK_TOPIC 187
#define TK_CONSUMER 188
#define TK_GROUP 189
#define TK_DESC 190
#define TK_DESCRIBE 191
#define TK_RESET 192
#define TK_QUERY 193
#define TK_CACHE 194
#define TK_EXPLAIN 195
#define TK_ANALYZE 196
#define TK_VERBOSE 197
#define TK_NK_BOOL 198
#define TK_RATIO 199
#define TK_NK_FLOAT 200
#define TK_OUTPUTTYPE 201
#define TK_AGGREGATE 202
#define TK_BUFSIZE 203
#define TK_LANGUAGE 204
#define TK_REPLACE 205
#define TK_STREAM 206
#define TK_INTO 207
#define TK_PAUSE 208
#define TK_RESUME 209
#define TK_TRIGGER 210
#define TK_AT_ONCE 211
#define TK_WINDOW_CLOSE 212
#define TK_IGNORE 213
#define TK_EXPIRED 214
#define TK_FILL_HISTORY 215
#define TK_UPDATE 216
#define TK_SUBTABLE 217
#define TK_UNTREATED 218
#define TK_KILL 219
#define TK_CONNECTION 220
#define TK_TRANSACTION 221
#define TK_BALANCE 222
#define TK_VGROUP 223
#define TK_LEADER 224
#define TK_MERGE 225
#define TK_REDISTRIBUTE 226
#define TK_SPLIT 227
#define TK_DELETE 228
#define TK_INSERT 229
#define TK_NULL 230
#define TK_NK_QUESTION 231
#define TK_NK_ALIAS 232
#define TK_NK_ARROW 233
#define TK_ROWTS 234
#define TK_QSTART 235
#define TK_QEND 236
#define TK_QDURATION 237
#define TK_WSTART 238
#define TK_WEND 239
#define TK_WDURATION 240
#define TK_IROWTS 241
#define TK_ISFILLED 242
#define TK_CAST 243
#define TK_NOW 244
#define TK_TODAY 245
#define TK_TIMEZONE 246
#define TK_CLIENT_VERSION 247
#define TK_SERVER_VERSION 248
#define TK_SERVER_STATUS 249
#define TK_CURRENT_USER 250
#define TK_CASE 251
#define TK_WHEN 252
#define TK_THEN 253
#define TK_ELSE 254
#define TK_BETWEEN 255
#define TK_IS 256
#define TK_NK_LT 257
#define TK_NK_GT 258
#define TK_NK_LE 259
#define TK_NK_GE 260
#define TK_NK_NE 261
#define TK_MATCH 262
#define TK_NMATCH 263
#define TK_CONTAINS 264
#define TK_IN 265
#define TK_JOIN 266
#define TK_INNER 267
#define TK_SELECT 268
#define TK_NK_HINT 269
#define TK_DISTINCT 270
#define TK_WHERE 271
#define TK_PARTITION 272
#define TK_BY 273
#define TK_SESSION 274
#define TK_STATE_WINDOW 275
#define TK_EVENT_WINDOW 276
#define TK_SLIDING 277
#define TK_FILL 278
#define TK_VALUE 279
#define TK_VALUE_F 280
#define TK_NONE 281
#define TK_PREV 282
#define TK_NULL_F 283
#define TK_LINEAR 284
#define TK_NEXT 285
#define TK_HAVING 286
#define TK_RANGE 287
#define TK_EVERY 288
#define TK_ORDER 289
#define TK_SLIMIT 290
#define TK_SOFFSET 291
#define TK_LIMIT 292
#define TK_OFFSET 293
#define TK_ASC 294
#define TK_NULLS 295
#define TK_ABORT 296
#define TK_AFTER 297
#define TK_ATTACH 298
#define TK_BEFORE 299
#define TK_BEGIN 300
#define TK_BITAND 301
#define TK_BITNOT 302
#define TK_BITOR 303
#define TK_BLOCKS 304
#define TK_CHANGE 305
#define TK_COMMA 306
#define TK_CONCAT 307
#define TK_CONFLICT 308
#define TK_COPY 309
#define TK_DEFERRED 310
#define TK_DELIMITERS 311
#define TK_DETACH 312
#define TK_DIVIDE 313
#define TK_DOT 314
#define TK_EACH 315
#define TK_FAIL 316
#define TK_FILE 317
#define TK_FOR 318
#define TK_GLOB 319
#define TK_ID 320
#define TK_IMMEDIATE 321
#define TK_IMPORT 322
#define TK_INITIALLY 323
#define TK_INSTEAD 324
#define TK_ISNULL 325
#define TK_KEY 326
#define TK_MODULES 327
#define TK_NK_BITNOT 328
#define TK_NK_SEMI 329
#define TK_NOTNULL 330
#define TK_OF 331
#define TK_PLUS 332
#define TK_PRIVILEGE 333
#define TK_RAISE 334
#define TK_RESTRICT 335
#define TK_ROW 336
#define TK_SEMI 337
#define TK_STAR 338
#define TK_STATEMENT 339
#define TK_STRICT 340
#define TK_STRING 341
#define TK_TIMES 342
#define TK_VALUES 343
#define TK_VARIABLE 344
#define TK_WAL 345
#define TK_COMPACTS 173
#define TK_NORMAL 174
#define TK_CHILD 175
#define TK_LIKE 176
#define TK_TBNAME 177
#define TK_QTAGS 178
#define TK_AS 179
#define TK_SYSTEM 180
#define TK_INDEX 181
#define TK_FUNCTION 182
#define TK_INTERVAL 183
#define TK_COUNT 184
#define TK_LAST_ROW 185
#define TK_META 186
#define TK_ONLY 187
#define TK_TOPIC 188
#define TK_CONSUMER 189
#define TK_GROUP 190
#define TK_DESC 191
#define TK_DESCRIBE 192
#define TK_RESET 193
#define TK_QUERY 194
#define TK_CACHE 195
#define TK_EXPLAIN 196
#define TK_ANALYZE 197
#define TK_VERBOSE 198
#define TK_NK_BOOL 199
#define TK_RATIO 200
#define TK_NK_FLOAT 201
#define TK_OUTPUTTYPE 202
#define TK_AGGREGATE 203
#define TK_BUFSIZE 204
#define TK_LANGUAGE 205
#define TK_REPLACE 206
#define TK_STREAM 207
#define TK_INTO 208
#define TK_PAUSE 209
#define TK_RESUME 210
#define TK_TRIGGER 211
#define TK_AT_ONCE 212
#define TK_WINDOW_CLOSE 213
#define TK_IGNORE 214
#define TK_EXPIRED 215
#define TK_FILL_HISTORY 216
#define TK_UPDATE 217
#define TK_SUBTABLE 218
#define TK_UNTREATED 219
#define TK_KILL 220
#define TK_CONNECTION 221
#define TK_TRANSACTION 222
#define TK_BALANCE 223
#define TK_VGROUP 224
#define TK_LEADER 225
#define TK_MERGE 226
#define TK_REDISTRIBUTE 227
#define TK_SPLIT 228
#define TK_DELETE 229
#define TK_INSERT 230
#define TK_NULL 231
#define TK_NK_QUESTION 232
#define TK_NK_ALIAS 233
#define TK_NK_ARROW 234
#define TK_ROWTS 235
#define TK_QSTART 236
#define TK_QEND 237
#define TK_QDURATION 238
#define TK_WSTART 239
#define TK_WEND 240
#define TK_WDURATION 241
#define TK_IROWTS 242
#define TK_ISFILLED 243
#define TK_CAST 244
#define TK_NOW 245
#define TK_TODAY 246
#define TK_TIMEZONE 247
#define TK_CLIENT_VERSION 248
#define TK_SERVER_VERSION 249
#define TK_SERVER_STATUS 250
#define TK_CURRENT_USER 251
#define TK_CASE 252
#define TK_WHEN 253
#define TK_THEN 254
#define TK_ELSE 255
#define TK_BETWEEN 256
#define TK_IS 257
#define TK_NK_LT 258
#define TK_NK_GT 259
#define TK_NK_LE 260
#define TK_NK_GE 261
#define TK_NK_NE 262
#define TK_MATCH 263
#define TK_NMATCH 264
#define TK_CONTAINS 265
#define TK_IN 266
#define TK_JOIN 267
#define TK_INNER 268
#define TK_SELECT 269
#define TK_NK_HINT 270
#define TK_DISTINCT 271
#define TK_WHERE 272
#define TK_PARTITION 273
#define TK_BY 274
#define TK_SESSION 275
#define TK_STATE_WINDOW 276
#define TK_EVENT_WINDOW 277
#define TK_SLIDING 278
#define TK_FILL 279
#define TK_VALUE 280
#define TK_VALUE_F 281
#define TK_NONE 282
#define TK_PREV 283
#define TK_NULL_F 284
#define TK_LINEAR 285
#define TK_NEXT 286
#define TK_HAVING 287
#define TK_RANGE 288
#define TK_EVERY 289
#define TK_ORDER 290
#define TK_SLIMIT 291
#define TK_SOFFSET 292
#define TK_LIMIT 293
#define TK_OFFSET 294
#define TK_ASC 295
#define TK_NULLS 296
#define TK_ABORT 297
#define TK_AFTER 298
#define TK_ATTACH 299
#define TK_BEFORE 300
#define TK_BEGIN 301
#define TK_BITAND 302
#define TK_BITNOT 303
#define TK_BITOR 304
#define TK_BLOCKS 305
#define TK_CHANGE 306
#define TK_COMMA 307
#define TK_CONCAT 308
#define TK_CONFLICT 309
#define TK_COPY 310
#define TK_DEFERRED 311
#define TK_DELIMITERS 312
#define TK_DETACH 313
#define TK_DIVIDE 314
#define TK_DOT 315
#define TK_EACH 316
#define TK_FAIL 317
#define TK_FILE 318
#define TK_FOR 319
#define TK_GLOB 320
#define TK_ID 321
#define TK_IMMEDIATE 322
#define TK_IMPORT 323
#define TK_INITIALLY 324
#define TK_INSTEAD 325
#define TK_ISNULL 326
#define TK_KEY 327
#define TK_MODULES 328
#define TK_NK_BITNOT 329
#define TK_NK_SEMI 330
#define TK_NOTNULL 331
#define TK_OF 332
#define TK_PLUS 333
#define TK_PRIVILEGE 334
#define TK_RAISE 335
#define TK_RESTRICT 336
#define TK_ROW 337
#define TK_SEMI 338
#define TK_STAR 339
#define TK_STATEMENT 340
#define TK_STRICT 341
#define TK_STRING 342
#define TK_TIMES 343
#define TK_VALUES 344
#define TK_VARIABLE 345
#define TK_WAL 346
#define TK_NK_SPACE 600
#define TK_NK_COMMENT 601

13
include/libs/nodes/cmdnodes.h
View File

@ -46,6 +46,10 @@ extern "C" {
#define SHOW_LOCAL_VARIABLES_RESULT_FIELD2_LEN (TSDB_CONFIG_VALUE_LEN + VARSTR_HEADER_SIZE)
#define SHOW_LOCAL_VARIABLES_RESULT_FIELD3_LEN (TSDB_CONFIG_SCOPE_LEN + VARSTR_HEADER_SIZE)
#define COMPACT_DB_RESULT_COLS 3
#define COMPACT_DB_RESULT_FIELD1_LEN 32
#define COMPACT_DB_RESULT_FIELD3_LEN 128
#define SHOW_ALIVE_RESULT_COLS 1
#define BIT_FLAG_MASK(n) (1 << n)
@ -335,6 +339,15 @@ typedef struct SShowTableTagsStmt {
SNodeList* pTags;
} SShowTableTagsStmt;
typedef struct SShowCompactsStmt {
ENodeType type;
} SShowCompactsStmt;
typedef struct SShowCompactDetailsStmt {
ENodeType type;
SNode* pCompactId;
} SShowCompactDetailsStmt;
typedef enum EIndexType { INDEX_TYPE_SMA = 1, INDEX_TYPE_FULLTEXT, INDEX_TYPE_NORMAL } EIndexType;
typedef struct SIndexOptions {

115
source/client/src/clientMsgHandler.c
View File

@ -16,6 +16,7 @@
#include "catalog.h"
#include "clientInt.h"
#include "clientLog.h"
#include "cmdnodes.h"
#include "os.h"
#include "query.h"
#include "systable.h"
@ -542,6 +543,118 @@ int32_t processShowVariablesRsp(void* param, SDataBuf* pMsg, int32_t code) {
return code;
}
static int32_t buildCompactDbBlock(SCompactDbRsp* pRsp, SSDataBlock** block) {
SSDataBlock* pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock));
pBlock->info.hasVarCol = true;
pBlock->pDataBlock = taosArrayInit(COMPACT_DB_RESULT_COLS, sizeof(SColumnInfoData));
SColumnInfoData infoData = {0};
infoData.info.type = TSDB_DATA_TYPE_VARCHAR;
infoData.info.bytes = COMPACT_DB_RESULT_FIELD1_LEN;
taosArrayPush(pBlock->pDataBlock, &infoData);
infoData.info.type = TSDB_DATA_TYPE_INT;
infoData.info.bytes = tDataTypes[TSDB_DATA_TYPE_INT].bytes;
taosArrayPush(pBlock->pDataBlock, &infoData);
infoData.info.type = TSDB_DATA_TYPE_VARCHAR;
infoData.info.bytes = COMPACT_DB_RESULT_FIELD3_LEN;
taosArrayPush(pBlock->pDataBlock, &infoData);
blockDataEnsureCapacity(pBlock, 1);
SColumnInfoData* pResultCol = taosArrayGet(pBlock->pDataBlock, 0);
SColumnInfoData* pIdCol = taosArrayGet(pBlock->pDataBlock, 1);
SColumnInfoData* pReasonCol = taosArrayGet(pBlock->pDataBlock, 2);
char result[COMPACT_DB_RESULT_FIELD1_LEN] = {0};
char reason[COMPACT_DB_RESULT_FIELD3_LEN] = {0};
if (pRsp->bAccepted) {
STR_TO_VARSTR(result, "accepted");
colDataSetVal(pResultCol, 0, result, false);
colDataSetVal(pIdCol, 0, (void*)&pRsp->compactId, false);
STR_TO_VARSTR(reason, "success");
colDataSetVal(pReasonCol, 0, reason, false);
} else {
STR_TO_VARSTR(result, "rejected");
colDataSetVal(pResultCol, 0, result, false);
colDataSetNULL(pIdCol, 0);
STR_TO_VARSTR(reason, "compaction is ongoing");
colDataSetVal(pReasonCol, 0, reason, false);
}
pBlock->info.rows = 1;
*block = pBlock;
return TSDB_CODE_SUCCESS;
}
static int32_t buildRetriveTableRspForCompactDb(SCompactDbRsp* pCompactDb, SRetrieveTableRsp** pRsp) {
SSDataBlock* pBlock = NULL;
int32_t code = buildCompactDbBlock(pCompactDb, &pBlock);
if (code) {
return code;
}
size_t rspSize = sizeof(SRetrieveTableRsp) + blockGetEncodeSize(pBlock);
*pRsp = taosMemoryCalloc(1, rspSize);
if (NULL == *pRsp) {
blockDataDestroy(pBlock);
return TSDB_CODE_OUT_OF_MEMORY;
}
(*pRsp)->useconds = 0;
(*pRsp)->completed = 1;
(*pRsp)->precision = 0;
(*pRsp)->compressed = 0;
(*pRsp)->compLen = 0;
(*pRsp)->numOfRows = htobe64((int64_t)pBlock->info.rows);
(*pRsp)->numOfCols = htonl(COMPACT_DB_RESULT_COLS);
int32_t len = blockEncode(pBlock, (*pRsp)->data, COMPACT_DB_RESULT_COLS);
blockDataDestroy(pBlock);
if (len != rspSize - sizeof(SRetrieveTableRsp)) {
uError("buildRetriveTableRspForCompactDb error, len:%d != rspSize - sizeof(SRetrieveTableRsp):%" PRIu64, len,
(uint64_t)(rspSize - sizeof(SRetrieveTableRsp)));
return TSDB_CODE_TSC_INVALID_INPUT;
}
return TSDB_CODE_SUCCESS;
}
int32_t processCompactDbRsp(void* param, SDataBuf* pMsg, int32_t code) {
SRequestObj* pRequest = param;
if (code != TSDB_CODE_SUCCESS) {
setErrno(pRequest, code);
} else {
SCompactDbRsp rsp = {0};
SRetrieveTableRsp* pRes = NULL;
code = tDeserializeSCompactDbRsp(pMsg->pData, pMsg->len, &rsp);
if (TSDB_CODE_SUCCESS == code) {
code = buildRetriveTableRspForCompactDb(&rsp, &pRes);
}
if (TSDB_CODE_SUCCESS == code) {
code = setQueryResultFromRsp(&pRequest->body.resInfo, pRes, false, true);
}
if (code != 0) {
taosMemoryFree(pRes);
}
}
taosMemoryFree(pMsg->pData);
taosMemoryFree(pMsg->pEpSet);
if (pRequest->body.queryFp != NULL) {
pRequest->body.queryFp(((SSyncQueryParam *)pRequest->body.interParam)->userParam, pRequest, code);
} else {
tsem_post(&pRequest->body.rspSem);
}
return code;
}
__async_send_cb_fn_t getMsgRspHandle(int32_t msgType) {
switch (msgType) {
case TDMT_MND_CONNECT:
@ -558,6 +671,8 @@ __async_send_cb_fn_t getMsgRspHandle(int32_t msgType) {
return processAlterStbRsp;
case TDMT_MND_SHOW_VARIABLES:
return processShowVariablesRsp;
case TDMT_MND_COMPACT_DB:
return processCompactDbRsp;
default:
return genericRspCallback;
}

16
source/common/src/systable.c
View File

@ -334,6 +334,20 @@ static const SSysDbTableSchema userViewsSchema[] = {
// {.name = "column_list", .bytes = 2048 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = false},
};
static const SSysDbTableSchema userCompactsSchema[] = {
{.name = "compact_id", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = false},
{.name = "db_name", .bytes = SYSTABLE_SCH_DB_NAME_LEN, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = false},
{.name = "start_time", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP, .sysInfo = false},
};
static const SSysDbTableSchema userCompactsDetailSchema[] = {
{.name = "compact_id", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = false},
{.name = "vgroup_id", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = false},
{.name = "dnode_id", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = false},
{.name = "number_fileset", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = false},
{.name = "finished", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = false},
{.name = "start_time", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP, .sysInfo = false},
};
static const SSysTableMeta infosMeta[] = {
{TSDB_INS_TABLE_DNODES, dnodesSchema, tListLen(dnodesSchema), true},
@ -362,6 +376,8 @@ static const SSysTableMeta infosMeta[] = {
{TSDB_INS_TABLE_VNODES, vnodesSchema, tListLen(vnodesSchema), true},
{TSDB_INS_TABLE_USER_PRIVILEGES, userUserPrivilegesSchema, tListLen(userUserPrivilegesSchema), true},
{TSDB_INS_TABLE_VIEWS, userViewsSchema, tListLen(userViewsSchema), false},
{TSDB_INS_TABLE_COMPACTS, userCompactsSchema, tListLen(userCompactsSchema), false},
{TSDB_INS_TABLE_COMPACT_DETAILS, userCompactsDetailSchema, tListLen(userCompactsDetailSchema), false},
};
static const SSysDbTableSchema connectionsSchema[] = {

6
source/common/src/tglobal.c
View File

@ -248,6 +248,7 @@ int32_t tsTtlBatchDropNum = 10000; // number of tables dropped per batch
// internal
int32_t tsTransPullupInterval = 2;
int32_t tsCompactPullupInterval = 10;
int32_t tsMqRebalanceInterval = 2;
int32_t tsStreamCheckpointInterval = 60;
float tsSinkDataRate = 2.0;
@ -707,6 +708,9 @@ static int32_t taosAddServerCfg(SConfig *pCfg) {
if (cfgAddInt32(pCfg, "transPullupInterval", tsTransPullupInterval, 1, 10000, CFG_SCOPE_SERVER, CFG_DYN_ENT_SERVER) !=
0)
return -1;
if (cfgAddInt32(pCfg, "compactPullupInterval", tsCompactPullupInterval, 1, 10000, CFG_SCOPE_SERVER, CFG_DYN_ENT_SERVER) !=
0)
return -1;
if (cfgAddInt32(pCfg, "mqRebalanceInterval", tsMqRebalanceInterval, 1, 10000, CFG_SCOPE_SERVER, CFG_DYN_ENT_SERVER) !=
0)
return -1;
@ -1156,6 +1160,7 @@ static int32_t taosSetServerCfg(SConfig *pCfg) {
tmqMaxTopicNum = cfgGetItem(pCfg, "tmqMaxTopicNum")->i32;
tsTransPullupInterval = cfgGetItem(pCfg, "transPullupInterval")->i32;
tsCompactPullupInterval = cfgGetItem(pCfg, "compactPullupInterval")->i32;
tsMqRebalanceInterval = cfgGetItem(pCfg, "mqRebalanceInterval")->i32;
tsTtlUnit = cfgGetItem(pCfg, "ttlUnit")->i32;
tsTtlPushIntervalSec = cfgGetItem(pCfg, "ttlPushInterval")->i32;
@ -1470,6 +1475,7 @@ static int32_t taosCfgDynamicOptionsForServer(SConfig *pCfg, char *name) {
{"timeseriesThreshold", &tsTimeSeriesThreshold},
{"tmqMaxTopicNum", &tmqMaxTopicNum},
{"transPullupInterval", &tsTransPullupInterval},
{"compactPullupInterval", &tsCompactPullupInterval},
{"trimVDbIntervalSec", &tsTrimVDbIntervalSec},
{"ttlBatchDropNum", &tsTtlBatchDropNum},
{"ttlFlushThreshold", &tsTtlFlushThreshold},

197
source/common/src/tmsg.c
View File

@ -1827,7 +1827,7 @@ int32_t tSerializeSGetUserAuthRspImpl(SEncoder *pEncoder, SGetUserAuthRsp *pRsp)
char *tb = taosHashIterate(pRsp->readTbs, NULL);
while (tb != NULL) {
size_t keyLen = 0;
void * key = taosHashGetKey(tb, &keyLen);
void *key = taosHashGetKey(tb, &keyLen);
if (tEncodeI32(pEncoder, keyLen) < 0) return -1;
if (tEncodeCStr(pEncoder, key) < 0) return -1;
@ -1842,7 +1842,7 @@ int32_t tSerializeSGetUserAuthRspImpl(SEncoder *pEncoder, SGetUserAuthRsp *pRsp)
tb = taosHashIterate(pRsp->writeTbs, NULL);
while (tb != NULL) {
size_t keyLen = 0;
void * key = taosHashGetKey(tb, &keyLen);
void *key = taosHashGetKey(tb, &keyLen);
if (tEncodeI32(pEncoder, keyLen) < 0) return -1;
if (tEncodeCStr(pEncoder, key) < 0) return -1;
@ -1857,7 +1857,7 @@ int32_t tSerializeSGetUserAuthRspImpl(SEncoder *pEncoder, SGetUserAuthRsp *pRsp)
tb = taosHashIterate(pRsp->alterTbs, NULL);
while (tb != NULL) {
size_t keyLen = 0;
void * key = taosHashGetKey(tb, &keyLen);
void *key = taosHashGetKey(tb, &keyLen);
if (tEncodeI32(pEncoder, keyLen) < 0) return -1;
if (tEncodeCStr(pEncoder, key) < 0) return -1;
@ -1872,7 +1872,7 @@ int32_t tSerializeSGetUserAuthRspImpl(SEncoder *pEncoder, SGetUserAuthRsp *pRsp)
tb = taosHashIterate(pRsp->readViews, NULL);
while (tb != NULL) {
size_t keyLen = 0;
void * key = taosHashGetKey(tb, &keyLen);
void *key = taosHashGetKey(tb, &keyLen);
if (tEncodeI32(pEncoder, keyLen) < 0) return -1;
if (tEncodeCStr(pEncoder, key) < 0) return -1;
@ -1887,7 +1887,7 @@ int32_t tSerializeSGetUserAuthRspImpl(SEncoder *pEncoder, SGetUserAuthRsp *pRsp)
tb = taosHashIterate(pRsp->writeViews, NULL);
while (tb != NULL) {
size_t keyLen = 0;
void * key = taosHashGetKey(tb, &keyLen);
void *key = taosHashGetKey(tb, &keyLen);
if (tEncodeI32(pEncoder, keyLen) < 0) return -1;
if (tEncodeCStr(pEncoder, key) < 0) return -1;
@ -1902,7 +1902,7 @@ int32_t tSerializeSGetUserAuthRspImpl(SEncoder *pEncoder, SGetUserAuthRsp *pRsp)
tb = taosHashIterate(pRsp->alterViews, NULL);
while (tb != NULL) {
size_t keyLen = 0;
void * key = taosHashGetKey(tb, &keyLen);
void *key = taosHashGetKey(tb, &keyLen);
if (tEncodeI32(pEncoder, keyLen) < 0) return -1;
if (tEncodeCStr(pEncoder, key) < 0) return -1;
@ -1917,7 +1917,7 @@ int32_t tSerializeSGetUserAuthRspImpl(SEncoder *pEncoder, SGetUserAuthRsp *pRsp)
int32_t *useDb = taosHashIterate(pRsp->useDbs, NULL);
while (useDb != NULL) {
size_t keyLen = 0;
void * key = taosHashGetKey(useDb, &keyLen);
void *key = taosHashGetKey(useDb, &keyLen);
if (tEncodeI32(pEncoder, keyLen) < 0) return -1;
if (tEncodeCStr(pEncoder, key) < 0) return -1;
@ -3424,6 +3424,66 @@ int32_t tDeserializeSCompactDbReq(void *buf, int32_t bufLen, SCompactDbReq *pReq
void tFreeSCompactDbReq(SCompactDbReq *pReq) { FREESQL(); }
int32_t tSerializeSCompactDbRsp(void *buf, int32_t bufLen, SCompactDbRsp *pRsp) {
SEncoder encoder = {0};
tEncoderInit(&encoder, buf, bufLen);
if (tStartEncode(&encoder) < 0) return -1;
if (tEncodeI32(&encoder, pRsp->compactId) < 0) return -1;
if (tEncodeI8(&encoder, pRsp->bAccepted) < 0) return -1;
tEndEncode(&encoder);
int32_t tlen = encoder.pos;
tEncoderClear(&encoder);
return tlen;
}
int32_t tDeserializeSCompactDbRsp(void *buf, int32_t bufLen, SCompactDbRsp *pRsp) {
SDecoder decoder = {0};
tDecoderInit(&decoder, buf, bufLen);
if (tStartDecode(&decoder) < 0) return -1;
if (tDecodeI32(&decoder, &pRsp->compactId) < 0) return -1;
if (tDecodeI8(&decoder, &pRsp->bAccepted) < 0) return -1;
tEndDecode(&decoder);
tDecoderClear(&decoder);
return 0;
}
int32_t tSerializeSKillCompactReq(void *buf, int32_t bufLen, SKillCompactReq *pReq) {
SEncoder encoder = {0};
tEncoderInit(&encoder, buf, bufLen);
if (tStartEncode(&encoder) < 0) return -1;
if (tEncodeI32(&encoder, pReq->compactId) < 0) return -1;
ENCODESQL();
tEndEncode(&encoder);
int32_t tlen = encoder.pos;
tEncoderClear(&encoder);
return tlen;
}
int32_t tDeserializeSKillCompactReq(void *buf, int32_t bufLen, SKillCompactReq *pReq) {
SDecoder decoder = {0};
tDecoderInit(&decoder, buf, bufLen);
if (tStartDecode(&decoder) < 0) return -1;
if (tDecodeI32(&decoder, &pReq->compactId) < 0) return -1;
DECODESQL();
tEndDecode(&decoder);
tDecoderClear(&decoder);
return 0;
}
void tFreeSKillCompactReq(SKillCompactReq *pReq) { FREESQL(); }
int32_t tSerializeSUseDbRspImp(SEncoder *pEncoder, const SUseDbRsp *pRsp) {
if (tEncodeCStr(pEncoder, pRsp->db) < 0) return -1;
if (tEncodeI64(pEncoder, pRsp->uid) < 0) return -1;
@ -4259,6 +4319,7 @@ int32_t tSerializeSRetrieveTableReq(void *buf, int32_t bufLen, SRetrieveTableReq
if (tEncodeCStr(&encoder, pReq->tb) < 0) return -1;
if (tEncodeCStr(&encoder, pReq->filterTb) < 0) return -1;
if (tEncodeCStr(&encoder, pReq->user) < 0) return -1;
if (tEncodeI64(&encoder, pReq->compactId) < 0) return -1;
tEndEncode(&encoder);
int32_t tlen = encoder.pos;
@ -4276,6 +4337,11 @@ int32_t tDeserializeSRetrieveTableReq(void *buf, int32_t bufLen, SRetrieveTableR
if (tDecodeCStrTo(&decoder, pReq->tb) < 0) return -1;
if (tDecodeCStrTo(&decoder, pReq->filterTb) < 0) return -1;
if (tDecodeCStrTo(&decoder, pReq->user) < 0) return -1;
if (!tDecodeIsEnd(&decoder)) {
if (tDecodeI64(&decoder, &pReq->compactId) < 0) return -1;
} else {
pReq->compactId = -1;
}
tEndDecode(&decoder);
tDecoderClear(&decoder);
@ -5051,6 +5117,75 @@ int32_t tFreeSCreateVnodeReq(SCreateVnodeReq *pReq) {
return 0;
}
int32_t tSerializeSQueryCompactProgressReq(void *buf, int32_t bufLen, SQueryCompactProgressReq *pReq) {
SEncoder encoder = {0};
tEncoderInit(&encoder, buf, bufLen);
if (tStartEncode(&encoder) < 0) return -1;
if (tEncodeI32(&encoder, pReq->compactId) < 0) return -1;
if (tEncodeI32(&encoder, pReq->vgId) < 0) return -1;
if (tEncodeI32(&encoder, pReq->dnodeId) < 0) return -1;
tEndEncode(&encoder);
int32_t tlen = encoder.pos;
tEncoderClear(&encoder);
return tlen;
}
int32_t tDeserializeSQueryCompactProgressReq(void *buf, int32_t bufLen, SQueryCompactProgressReq *pReq) {
int32_t headLen = sizeof(SMsgHead);
SDecoder decoder = {0};
tDecoderInit(&decoder, ((uint8_t *)buf) + headLen, bufLen - headLen);
if (tStartDecode(&decoder) < 0) return -1;
if (tDecodeI32(&decoder, &pReq->compactId) < 0) return -1;
if (tDecodeI32(&decoder, &pReq->vgId) < 0) return -1;
if (tDecodeI32(&decoder, &pReq->dnodeId) < 0) return -1;
tEndDecode(&decoder);
tDecoderClear(&decoder);
return 0;
}
int32_t tSerializeSQueryCompactProgressRsp(void *buf, int32_t bufLen, SQueryCompactProgressRsp *pReq) {
SEncoder encoder = {0};
tEncoderInit(&encoder, buf, bufLen);
if (tStartEncode(&encoder) < 0) return -1;
if (tEncodeI32(&encoder, pReq->compactId) < 0) return -1;
if (tEncodeI32(&encoder, pReq->vgId) < 0) return -1;
if (tEncodeI32(&encoder, pReq->dnodeId) < 0) return -1;
if (tEncodeI32(&encoder, pReq->numberFileset) < 0) return -1;
if (tEncodeI32(&encoder, pReq->finished) < 0) return -1;
tEndEncode(&encoder);
int32_t tlen = encoder.pos;
tEncoderClear(&encoder);
return tlen;
}
int32_t tDeserializeSQueryCompactProgressRsp(void *buf, int32_t bufLen, SQueryCompactProgressRsp *pReq) {
SDecoder decoder = {0};
tDecoderInit(&decoder, buf, bufLen);
if (tStartDecode(&decoder) < 0) return -1;
if (tDecodeI32(&decoder, &pReq->compactId) < 0) return -1;
if (tDecodeI32(&decoder, &pReq->vgId) < 0) return -1;
if (tDecodeI32(&decoder, &pReq->dnodeId) < 0) return -1;
if (tDecodeI32(&decoder, &pReq->numberFileset) < 0) return -1;
if (tDecodeI32(&decoder, &pReq->finished) < 0) return -1;
tEndDecode(&decoder);
tDecoderClear(&decoder);
return 0;
}
int32_t tSerializeSDropVnodeReq(void *buf, int32_t bufLen, SDropVnodeReq *pReq) {
SEncoder encoder = {0};
tEncoderInit(&encoder, buf, bufLen);
@ -5139,6 +5274,8 @@ int32_t tSerializeSCompactVnodeReq(void *buf, int32_t bufLen, SCompactVnodeReq *
if (tEncodeI64(&encoder, pReq->tw.skey) < 0) return -1;
if (tEncodeI64(&encoder, pReq->tw.ekey) < 0) return -1;
if (tEncodeI32(&encoder, pReq->compactId) < 0) return -1;
tEndEncode(&encoder);
int32_t tlen = encoder.pos;
@ -5165,6 +5302,42 @@ int32_t tDeserializeSCompactVnodeReq(void *buf, int32_t bufLen, SCompactVnodeReq
if (tDecodeI64(&decoder, &pReq->tw.ekey) < 0) return -1;
}
if (!tDecodeIsEnd(&decoder)) {
if (tDecodeI32(&decoder, &pReq->compactId) < 0) return -1;
}
tEndDecode(&decoder);
tDecoderClear(&decoder);
return 0;
}
int32_t tSerializeSVKillCompactReq(void *buf, int32_t bufLen, SVKillCompactReq *pReq) {
SEncoder encoder = {0};
tEncoderInit(&encoder, buf, bufLen);
if (tStartEncode(&encoder) < 0) return -1;
if (tEncodeI32(&encoder, pReq->compactId) < 0) return -1;
if (tEncodeI32(&encoder, pReq->vgId) < 0) return -1;
if (tEncodeI32(&encoder, pReq->dnodeId) < 0) return -1;
tEndEncode(&encoder);
int32_t tlen = encoder.pos;
tEncoderClear(&encoder);
return tlen;
}
int32_t tDeserializeSVKillCompactReq(void *buf, int32_t bufLen, SVKillCompactReq *pReq) {
SDecoder decoder = {0};
tDecoderInit(&decoder, buf, bufLen);
if (tStartDecode(&decoder) < 0) return -1;
if (tDecodeI32(&decoder, &pReq->compactId) < 0) return -1;
if (tDecodeI32(&decoder, &pReq->vgId) < 0) return -1;
if (tDecodeI32(&decoder, &pReq->dnodeId) < 0) return -1;
tEndDecode(&decoder);
tDecoderClear(&decoder);
return 0;
@ -8169,7 +8342,7 @@ int32_t tEncodeMqDataRsp(SEncoder *pEncoder, const SMqDataRsp *pRsp) {
for (int32_t i = 0; i < pRsp->blockNum; i++) {
int32_t bLen = *(int32_t *)taosArrayGet(pRsp->blockDataLen, i);
void * data = taosArrayGetP(pRsp->blockData, i);
void *data = taosArrayGetP(pRsp->blockData, i);
if (tEncodeBinary(pEncoder, (const uint8_t *)data, bLen) < 0) return -1;
if (pRsp->withSchema) {
SSchemaWrapper *pSW = (SSchemaWrapper *)taosArrayGetP(pRsp->blockSchema, i);
@ -8202,7 +8375,7 @@ int32_t tDecodeMqDataRsp(SDecoder *pDecoder, SMqDataRsp *pRsp) {
}
for (int32_t i = 0; i < pRsp->blockNum; i++) {
void * data;
void *data;
uint64_t bLen;
if (tDecodeBinaryAlloc(pDecoder, &data, &bLen) < 0) return -1;
taosArrayPush(pRsp->blockData, &data);
@ -8248,7 +8421,7 @@ int32_t tEncodeSTaosxRsp(SEncoder *pEncoder, const STaosxRsp *pRsp) {
if (tEncodeI32(pEncoder, pRsp->createTableNum) < 0) return -1;
if (pRsp->createTableNum) {
for (int32_t i = 0; i < pRsp->createTableNum; i++) {
void * createTableReq = taosArrayGetP(pRsp->createTableReq, i);
void *createTableReq = taosArrayGetP(pRsp->createTableReq, i);
int32_t createTableLen = *(int32_t *)taosArrayGet(pRsp->createTableLen, i);
if (tEncodeBinary(pEncoder, createTableReq, createTableLen) < 0) return -1;
}
@ -8264,7 +8437,7 @@ int32_t tDecodeSTaosxRsp(SDecoder *pDecoder, STaosxRsp *pRsp) {
pRsp->createTableLen = taosArrayInit(pRsp->createTableNum, sizeof(int32_t));
pRsp->createTableReq = taosArrayInit(pRsp->createTableNum, sizeof(void *));
for (int32_t i = 0; i < pRsp->createTableNum; i++) {
void * pCreate = NULL;
void *pCreate = NULL;
uint64_t len;
if (tDecodeBinaryAlloc(pDecoder, &pCreate, &len) < 0) return -1;
int32_t l = (int32_t)len;
@ -8566,7 +8739,7 @@ void tDestroySubmitTbData(SSubmitTbData *pTbData, int32_t flag) {
taosArrayDestroy(pTbData->aCol);
} else {
int32_t nRow = TARRAY_SIZE(pTbData->aRowP);
SRow ** rows = (SRow **)TARRAY_DATA(pTbData->aRowP);
SRow **rows = (SRow **)TARRAY_DATA(pTbData->aRowP);
for (int32_t i = 0; i < nRow; ++i) {
tRowDestroy(rows[i]);

3
source/dnode/mgmt/mgmt_mnode/src/mmHandle.c
View File

@ -192,6 +192,8 @@ SArray *mmGetMsgHandles() {
if (dmSetMgmtHandle(pArray, TDMT_MND_CREATE_VIEW, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_MND_DROP_VIEW, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_MND_VIEW_META, mmPutMsgToReadQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_MND_KILL_COMPACT, mmPutMsgToReadQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_VND_QUERY_COMPACT_PROGRESS_RSP, mmPutMsgToReadQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_SCH_QUERY, mmPutMsgToQueryQueue, 1) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_SCH_MERGE_QUERY, mmPutMsgToQueryQueue, 1) == NULL) goto _OVER;
@ -221,6 +223,7 @@ SArray *mmGetMsgHandles() {
if (dmSetMgmtHandle(pArray, TDMT_VND_STREAM_TASK_UPDATE_RSP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_VND_STREAM_TASK_RESET_RSP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_MND_STREAM_HEARTBEAT, mmPutMsgToReadQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_VND_KILL_COMPACT_RSP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_CONFIG_RSP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_REPLICA_RSP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;

2
source/dnode/mgmt/mgmt_vnode/src/vmHandle.c
View File

@ -818,6 +818,8 @@ SArray *vmGetMsgHandles() {
if (dmSetMgmtHandle(pArray, TDMT_SCH_QUERY_HEARTBEAT, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_VND_CREATE_INDEX, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_VND_DROP_INDEX, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_VND_QUERY_COMPACT_PROGRESS, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_VND_KILL_COMPACT, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_DEPLOY, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_DROP, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;

56
source/dnode/mnode/impl/inc/mndCompact.h Normal file
View File

@ -0,0 +1,56 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef _TD_MND_COMPACT_H_
#define _TD_MND_COMPACT_H_
#include "mndInt.h"
#ifdef __cplusplus
extern "C" {
#endif
int32_t mndInitCompact(SMnode *pMnode);
void mndCleanupCompact(SMnode *pMnode);
void tFreeCompactObj(SCompactObj *pCompact);
int32_t tSerializeSCompactObj(void *buf, int32_t bufLen, const SCompactObj *pObj);
int32_t tDeserializeSCompactObj(void *buf, int32_t bufLen, SCompactObj *pObj);
SSdbRaw* mndCompactActionEncode(SCompactObj *pCompact);
SSdbRow* mndCompactActionDecode(SSdbRaw *pRaw);
int32_t mndCompactActionInsert(SSdb *pSdb, SCompactObj *pCompact);
int32_t mndCompactActionDelete(SSdb *pSdb, SCompactObj *pCompact);
int32_t mndCompactActionUpdate(SSdb *pSdb, SCompactObj *pOldCompact, SCompactObj *pNewCompact);
int32_t mndAddCompactToTran(SMnode *pMnode, STrans *pTrans, SCompactObj* pCompact, SDbObj *pDb, SCompactDbRsp *rsp);
int32_t mndRetrieveCompact(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows);
int32_t mndProcessKillCompactReq(SRpcMsg *pReq);
int32_t mndProcessQueryCompactRsp(SRpcMsg *pReq);
SCompactObj *mndAcquireCompact(SMnode *pMnode, int64_t compactId);
void mndReleaseCompact(SMnode *pMnode, SCompactObj *pCompact);
void mndCompactSendProgressReq(SMnode *pMnode, SCompactObj *pCompact);
#ifdef __cplusplus
}
#endif
#endif /*_TD_MND_COMPACT_H_*/

48
source/dnode/mnode/impl/inc/mndCompactDetail.h Normal file
View File

@ -0,0 +1,48 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef _TD_MND_COMPACT_DETAIL_H_
#define _TD_MND_COMPACT_DETAIL_H_
#include "mndInt.h"
#ifdef __cplusplus
extern "C" {
#endif
int32_t mndInitCompactDetail(SMnode *pMnode);
void mndCleanupCompactDetail(SMnode *pMnode);
void tFreeCompactDetailObj(SCompactDetailObj *pCompact);
int32_t tSerializeSCompactDetailObj(void *buf, int32_t bufLen, const SCompactDetailObj *pObj);
int32_t tDeserializeSCompactDetailObj(void *buf, int32_t bufLen, SCompactDetailObj *pObj);
SSdbRaw* mndCompactDetailActionEncode(SCompactDetailObj *pCompact);
SSdbRow* mndCompactDetailActionDecode(SSdbRaw *pRaw);
int32_t mndCompactDetailActionInsert(SSdb *pSdb, SCompactDetailObj *pCompact);
int32_t mndCompactDetailActionDelete(SSdb *pSdb, SCompactDetailObj *pCompact);
int32_t mndCompactDetailActionUpdate(SSdb *pSdb, SCompactDetailObj *pOldCompact, SCompactDetailObj *pNewCompact);
int32_t mndAddCompactDetailToTran(SMnode *pMnode, STrans *pTrans, SCompactObj* pCompact, SVgObj *pVgroup,
SVnodeGid *pVgid, int32_t index);
int32_t mndRetrieveCompactDetail(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows);
#ifdef __cplusplus
}
#endif
#endif /*_TD_MND_COMPACT_DETAIL_H_*/

19
source/dnode/mnode/impl/inc/mndDef.h
View File

@ -741,6 +741,25 @@ int32_t tEncodeSViewObj(SEncoder* pEncoder, const SViewObj* pObj);
int32_t tDecodeSViewObj(SDecoder* pDecoder, SViewObj* pObj, int32_t sver);
void tFreeSViewObj(SViewObj* pObj);
typedef struct {
int32_t compactDetailId;
int32_t compactId;
int32_t vgId;
int32_t dnodeId;
int32_t numberFileset;
int32_t finished;
int64_t startTime;
int32_t newNumberFileset;
int32_t newFinished;
} SCompactDetailObj;
typedef struct {
int32_t compactId;
char dbname[TSDB_TABLE_FNAME_LEN];
int64_t startTime;
SArray* compactDetail;
} SCompactObj;
#ifdef __cplusplus
}
#endif

741
source/dnode/mnode/impl/src/mndCompact.c Normal file
View File

@ -0,0 +1,741 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "mndCompact.h"
#include "mndTrans.h"
#include "mndShow.h"
#include "mndDb.h"
#include "mndCompactDetail.h"
#include "mndVgroup.h"
#include "tmsgcb.h"
#include "mndDnode.h"
#include "tmisce.h"
#include "audit.h"
#include "mndPrivilege.h"
#include "mndTrans.h"
#define MND_COMPACT_VER_NUMBER 1
static int32_t mndProcessCompactTimer(SRpcMsg *pReq);
int32_t mndInitCompact(SMnode *pMnode) {
mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_COMPACT, mndRetrieveCompact);
mndSetMsgHandle(pMnode, TDMT_MND_KILL_COMPACT, mndProcessKillCompactReq);
mndSetMsgHandle(pMnode, TDMT_VND_QUERY_COMPACT_PROGRESS_RSP, mndProcessQueryCompactRsp);
mndSetMsgHandle(pMnode, TDMT_MND_COMPACT_TIMER, mndProcessCompactTimer);
mndSetMsgHandle(pMnode, TDMT_VND_KILL_COMPACT_RSP, mndTransProcessRsp);
SSdbTable table = {
.sdbType = SDB_COMPACT,
.keyType = SDB_KEY_INT32,
.encodeFp = (SdbEncodeFp)mndCompactActionEncode,
.decodeFp = (SdbDecodeFp)mndCompactActionDecode,
.insertFp = (SdbInsertFp)mndCompactActionInsert,
.updateFp = (SdbUpdateFp)mndCompactActionUpdate,
.deleteFp = (SdbDeleteFp)mndCompactActionDelete,
};
return sdbSetTable(pMnode->pSdb, table);
}
void mndCleanupCompact(SMnode *pMnode) {
mDebug("mnd compact cleanup");
}
void tFreeCompactObj(SCompactObj *pCompact) {
}
int32_t tSerializeSCompactObj(void *buf, int32_t bufLen, const SCompactObj *pObj) {
SEncoder encoder = {0};
tEncoderInit(&encoder, buf, bufLen);
if (tStartEncode(&encoder) < 0) return -1;
if (tEncodeI32(&encoder, pObj->compactId) < 0) return -1;
if (tEncodeCStr(&encoder, pObj->dbname) < 0) return -1;
if (tEncodeI64(&encoder, pObj->startTime) < 0) return -1;
tEndEncode(&encoder);
int32_t tlen = encoder.pos;
tEncoderClear(&encoder);
return tlen;
}
int32_t tDeserializeSCompactObj(void *buf, int32_t bufLen, SCompactObj *pObj) {
int8_t ex = 0;
SDecoder decoder = {0};
tDecoderInit(&decoder, buf, bufLen);
if (tStartDecode(&decoder) < 0) return -1;
if (tDecodeI32(&decoder, &pObj->compactId) < 0) return -1;
if (tDecodeCStrTo(&decoder, pObj->dbname) < 0) return -1;
if (tDecodeI64(&decoder, &pObj->startTime) < 0) return -1;
tEndDecode(&decoder);
tDecoderClear(&decoder);
return 0;
}
SSdbRaw *mndCompactActionEncode(SCompactObj *pCompact) {
terrno = TSDB_CODE_SUCCESS;
void *buf = NULL;
SSdbRaw *pRaw = NULL;
int32_t tlen = tSerializeSCompactObj(NULL, 0, pCompact);
if (tlen < 0) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
goto OVER;
}
int32_t size = sizeof(int32_t) + tlen;
pRaw = sdbAllocRaw(SDB_COMPACT, MND_COMPACT_VER_NUMBER, size);
if (pRaw == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
goto OVER;
}
buf = taosMemoryMalloc(tlen);
if (buf == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
goto OVER;
}
tlen = tSerializeSCompactObj(buf, tlen, pCompact);
if (tlen < 0) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
goto OVER;
}
int32_t dataPos = 0;
SDB_SET_INT32(pRaw, dataPos, tlen, OVER);
SDB_SET_BINARY(pRaw, dataPos, buf, tlen, OVER);
SDB_SET_DATALEN(pRaw, dataPos, OVER);
OVER:
taosMemoryFreeClear(buf);
if (terrno != TSDB_CODE_SUCCESS) {
mError("compact:%" PRId32 ", failed to encode to raw:%p since %s", pCompact->compactId, pRaw, terrstr());
sdbFreeRaw(pRaw);
return NULL;
}
mTrace("compact:%" PRId32 ", encode to raw:%p, row:%p", pCompact->compactId, pRaw, pCompact);
return pRaw;
}
SSdbRow *mndCompactActionDecode(SSdbRaw *pRaw) {
SSdbRow *pRow = NULL;
SCompactObj *pCompact = NULL;
void *buf = NULL;
terrno = TSDB_CODE_SUCCESS;
int8_t sver = 0;
if (sdbGetRawSoftVer(pRaw, &sver) != 0) {
goto OVER;
}
if (sver != MND_COMPACT_VER_NUMBER) {
terrno = TSDB_CODE_SDB_INVALID_DATA_VER;
mError("compact read invalid ver, data ver: %d, curr ver: %d", sver, MND_COMPACT_VER_NUMBER);
goto OVER;
}
pRow = sdbAllocRow(sizeof(SCompactObj));
if (pRow == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
goto OVER;
}
pCompact = sdbGetRowObj(pRow);
if (pCompact == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
goto OVER;
}
int32_t tlen;
int32_t dataPos = 0;
SDB_GET_INT32(pRaw, dataPos, &tlen, OVER);
buf = taosMemoryMalloc(tlen + 1);
if (buf == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
goto OVER;
}
SDB_GET_BINARY(pRaw, dataPos, buf, tlen, OVER);
if (tDeserializeSCompactObj(buf, tlen, pCompact) < 0) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
goto OVER;
}
//taosInitRWLatch(&pView->lock);
OVER:
taosMemoryFreeClear(buf);
if (terrno != TSDB_CODE_SUCCESS) {
mError("compact:%" PRId32 ", failed to decode from raw:%p since %s", pCompact->compactId, pRaw, terrstr());
taosMemoryFreeClear(pRow);
return NULL;
}
mTrace("compact:%" PRId32 ", decode from raw:%p, row:%p", pCompact->compactId, pRaw, pCompact);
return pRow;
}
int32_t mndCompactActionInsert(SSdb *pSdb, SCompactObj *pCompact) {
mTrace("compact:%" PRId32 ", perform insert action", pCompact->compactId);
return 0;
}
int32_t mndCompactActionDelete(SSdb *pSdb, SCompactObj *pCompact) {
mTrace("compact:%" PRId32 ", perform insert action", pCompact->compactId);
tFreeCompactObj(pCompact);
return 0;
}
int32_t mndCompactActionUpdate(SSdb *pSdb, SCompactObj *pOldCompact, SCompactObj *pNewCompact) {
mTrace("compact:%" PRId32 ", perform update action, old row:%p new row:%p",
pOldCompact->compactId, pOldCompact, pNewCompact);
return 0;
}
SCompactObj *mndAcquireCompact(SMnode *pMnode, int64_t compactId) {
SSdb *pSdb = pMnode->pSdb;
SCompactObj *pCompact = sdbAcquire(pSdb, SDB_COMPACT, &compactId);
if (pCompact == NULL && terrno == TSDB_CODE_SDB_OBJ_NOT_THERE) {
terrno = TSDB_CODE_SUCCESS;
}
return pCompact;
}
void mndReleaseCompact(SMnode *pMnode, SCompactObj *pCompact) {
SSdb *pSdb = pMnode->pSdb;
sdbRelease(pSdb, pCompact);
}
//compact db
int32_t mndAddCompactToTran(SMnode *pMnode, STrans *pTrans, SCompactObj* pCompact, SDbObj *pDb, SCompactDbRsp *rsp){
pCompact->compactId = tGenIdPI32();
strcpy(pCompact->dbname, pDb->name);
pCompact->startTime = taosGetTimestampMs();
SSdbRaw *pVgRaw = mndCompactActionEncode(pCompact);
if (pVgRaw == NULL) return -1;
if (mndTransAppendRedolog(pTrans, pVgRaw) != 0) {
sdbFreeRaw(pVgRaw);
return -1;
}
(void)sdbSetRawStatus(pVgRaw, SDB_STATUS_READY);
rsp->compactId = pCompact->compactId;
return 0;
}
//retrieve compact
int32_t mndRetrieveCompact(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows){
SMnode *pMnode = pReq->info.node;
SSdb *pSdb = pMnode->pSdb;
int32_t numOfRows = 0;
SCompactObj *pCompact = NULL;
char *sep = NULL;
SDbObj *pDb = NULL;
if (strlen(pShow->db) > 0) {
sep = strchr(pShow->db, '.');
if (sep && ((0 == strcmp(sep + 1, TSDB_INFORMATION_SCHEMA_DB) || (0 == strcmp(sep + 1, TSDB_PERFORMANCE_SCHEMA_DB))))) {
sep++;
} else {
pDb = mndAcquireDb(pMnode, pShow->db);
if (pDb == NULL) return terrno;
}
}
while (numOfRows < rows) {
pShow->pIter = sdbFetch(pSdb, SDB_COMPACT, pShow->pIter, (void **)&pCompact);
if (pShow->pIter == NULL) break;
SColumnInfoData *pColInfo;
SName n;
int32_t cols = 0;
char tmpBuf[TSDB_SHOW_SQL_LEN + VARSTR_HEADER_SIZE] = {0};
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataSetVal(pColInfo, numOfRows, (const char *)&pCompact->compactId, false);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
if (pDb != NULL || !IS_SYS_DBNAME(pCompact->dbname)) {
SName name = {0};
tNameFromString(&name, pCompact->dbname, T_NAME_ACCT | T_NAME_DB);
tNameGetDbName(&name, varDataVal(tmpBuf));
} else {
strncpy(varDataVal(tmpBuf), pCompact->dbname, strlen(pCompact->dbname) + 1);
}
varDataSetLen(tmpBuf, strlen(varDataVal(tmpBuf)));
colDataSetVal(pColInfo, numOfRows, (const char *)tmpBuf, false);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataSetVal(pColInfo, numOfRows, (const char *)&pCompact->startTime, false);
numOfRows++;
sdbRelease(pSdb, pCompact);
}
pShow->numOfRows += numOfRows;
mndReleaseDb(pMnode, pDb);
return numOfRows;
}
//kill compact
static void *mndBuildKillCompactReq(SMnode *pMnode, SVgObj *pVgroup, int32_t *pContLen,
int32_t compactId, int32_t dnodeid) {
SVKillCompactReq req = {0};
req.compactId = compactId;
req.vgId = pVgroup->vgId;
req.dnodeId = dnodeid;
mInfo("vgId:%d, build compact vnode config req", pVgroup->vgId);
int32_t contLen = tSerializeSVKillCompactReq(NULL, 0, &req);
if (contLen < 0) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return NULL;
}
contLen += sizeof(SMsgHead);
void *pReq = taosMemoryMalloc(contLen);
if (pReq == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return NULL;
}
SMsgHead *pHead = pReq;
pHead->contLen = htonl(contLen);
pHead->vgId = htonl(pVgroup->vgId);
tSerializeSVKillCompactReq((char *)pReq + sizeof(SMsgHead), contLen, &req);
*pContLen = contLen;
return pReq;
}
static int32_t mndAddKillCompactAction(SMnode *pMnode, STrans *pTrans, SVgObj *pVgroup,
int32_t compactId, int32_t dnodeid) {
STransAction action = {0};
SDnodeObj *pDnode = mndAcquireDnode(pMnode, dnodeid);
if (pDnode == NULL) return -1;
action.epSet = mndGetDnodeEpset(pDnode);
mndReleaseDnode(pMnode, pDnode);
int32_t contLen = 0;
void *pReq = mndBuildKillCompactReq(pMnode, pVgroup, &contLen, compactId, dnodeid);
if (pReq == NULL) return -1;
action.pCont = pReq;
action.contLen = contLen;
action.msgType = TDMT_VND_KILL_COMPACT;
if (mndTransAppendRedoAction(pTrans, &action) != 0) {
taosMemoryFree(pReq);
return -1;
}
return 0;
}
static int32_t mndKillCompact(SMnode *pMnode, SRpcMsg *pReq, SCompactObj *pCompact) {
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_NOTHING, pReq, "kill-compact");
if (pTrans == NULL) {
mError("compact:%" PRId32 ", failed to drop since %s" , pCompact->compactId, terrstr());
return -1;
}
mInfo("trans:%d, used to kill compact:%" PRId32, pTrans->id, pCompact->compactId);
SSdbRaw *pCommitRaw = mndCompactActionEncode(pCompact);
if (pCommitRaw == NULL || mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) {
mError("trans:%d, failed to append commit log since %s", pTrans->id, terrstr());
mndTransDrop(pTrans);
return -1;
}
(void)sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY);
void *pIter = NULL;
while (1) {
SCompactDetailObj *pDetail = NULL;
pIter = sdbFetch(pMnode->pSdb, SDB_COMPACT_DETAIL, pIter, (void **)&pDetail);
if (pIter == NULL) break;
if (pDetail->compactId == pCompact->compactId) {
SVgObj *pVgroup = mndAcquireVgroup(pMnode, pDetail->vgId);
if(pVgroup == NULL){
mError("trans:%d, failed to append redo action since %s", pTrans->id, terrstr());
mndTransDrop(pTrans);
return -1;
}
if(mndAddKillCompactAction(pMnode, pTrans, pVgroup, pCompact->compactId, pDetail->dnodeId) != 0){
mError("trans:%d, failed to append redo action since %s", pTrans->id, terrstr());
mndTransDrop(pTrans);
return -1;
}
mndReleaseVgroup(pMnode, pVgroup);
/*
SSdbRaw *pCommitRaw = mndCompactDetailActionEncode(pDetail);
if (pCommitRaw == NULL || mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) {
mError("trans:%d, failed to append commit log since %s", pTrans->id, terrstr());
mndTransDrop(pTrans);
return -1;
}
(void)sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED);
*/
}
sdbRelease(pMnode->pSdb, pDetail);
}
if (mndTransPrepare(pMnode, pTrans) != 0) {
mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
mndTransDrop(pTrans);
return -1;
}
mndTransDrop(pTrans);
return 0;
}
int32_t mndProcessKillCompactReq(SRpcMsg *pReq){
SKillCompactReq killCompactReq = {0};
if (tDeserializeSKillCompactReq(pReq->pCont, pReq->contLen, &killCompactReq) != 0) {
terrno = TSDB_CODE_INVALID_MSG;
return -1;
}
mInfo("start to kill compact:%" PRId32, killCompactReq.compactId);
SMnode *pMnode = pReq->info.node;
int32_t code = -1;
SCompactObj *pCompact = mndAcquireCompact(pMnode, killCompactReq.compactId);
if(pCompact == NULL){
tFreeSKillCompactReq(&killCompactReq);
return -1;
}
if (0 != mndCheckOperPrivilege(pMnode, pReq->info.conn.user, MND_OPER_COMPACT_DB)) {
goto _OVER;
}
if (mndKillCompact(pMnode, pReq, pCompact) < 0) {
goto _OVER;
}
code = TSDB_CODE_ACTION_IN_PROGRESS;
char obj[10] = {0};
sprintf(obj, "%d", pCompact->compactId);
auditRecord(pReq, pMnode->clusterId, "killCompact", pCompact->dbname, obj, killCompactReq.sql, killCompactReq.sqlLen);
_OVER:
if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) {
mError("failed to kill compact %" PRId32 " since %s", killCompactReq.compactId, terrstr());
}
tFreeSKillCompactReq(&killCompactReq);
sdbRelease(pMnode->pSdb, pCompact);
return code;
}
//update progress
static int32_t mndUpdateCompactProgress(SMnode *pMnode, SRpcMsg *pReq, int32_t compactId, SQueryCompactProgressRsp* rsp) {
void* pIter = NULL;
while (1) {
SCompactDetailObj *pDetail = NULL;
pIter = sdbFetch(pMnode->pSdb, SDB_COMPACT_DETAIL, pIter, (void **)&pDetail);
if (pIter == NULL) break;
if (pDetail->compactId == compactId && pDetail->vgId == rsp->vgId && pDetail->dnodeId == rsp->dnodeId) {
pDetail->newNumberFileset = rsp->numberFileset;
pDetail->newFinished = rsp->finished;
sdbRelease(pMnode->pSdb, pDetail);
return 0;
}
sdbRelease(pMnode->pSdb, pDetail);
}
return -1;
}
int32_t mndProcessQueryCompactRsp(SRpcMsg *pReq){
SQueryCompactProgressRsp req = {0};
if (tDeserializeSQueryCompactProgressRsp(pReq->pCont, pReq->contLen, &req) != 0) {
terrno = TSDB_CODE_INVALID_MSG;
return -1;
}
mDebug("compact:%d, receive query response, vgId:%d, dnodeId:%d, numberFileset:%d, finished:%d",
req.compactId, req.vgId, req.dnodeId, req.numberFileset, req.finished);
SMnode *pMnode = pReq->info.node;
int32_t code = -1;
if(mndUpdateCompactProgress(pMnode, pReq, req.compactId, &req) != 0){
mError("compact:%d, failed to update progress, vgId:%d, dnodeId:%d, numberFileset:%d, finished:%d",
req.compactId, req.vgId, req.dnodeId, req.numberFileset, req.finished);
return -1;
}
return 0;
}
//timer
void mndCompactSendProgressReq(SMnode *pMnode, SCompactObj *pCompact){
void *pIter = NULL;
while (1) {
SCompactDetailObj *pDetail = NULL;
pIter = sdbFetch(pMnode->pSdb, SDB_COMPACT_DETAIL, pIter, (void **)&pDetail);
if (pIter == NULL) break;
if (pDetail->compactId == pCompact->compactId) {
SEpSet epSet = {0};
SDnodeObj *pDnode = mndAcquireDnode(pMnode, pDetail->dnodeId);
if(pDnode == NULL) break;
addEpIntoEpSet(&epSet, pDnode->fqdn, pDnode->port);
mndReleaseDnode(pMnode, pDnode);
SQueryCompactProgressReq req;
req.compactId = pDetail->compactId;
req.vgId = pDetail->vgId;
req.dnodeId = pDetail->dnodeId;
int32_t contLen = tSerializeSQueryCompactProgressReq(NULL, 0, &req);
if (contLen < 0) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
continue;
}
contLen += sizeof(SMsgHead);
SMsgHead *pHead = rpcMallocCont(contLen);
if (pHead == NULL) {
sdbCancelFetch(pMnode->pSdb, pDetail);
sdbRelease(pMnode->pSdb, pDetail);
continue;
}
pHead->contLen = htonl(contLen);
pHead->vgId = htonl(pDetail->vgId);
tSerializeSQueryCompactProgressReq((char *)pHead + sizeof(SMsgHead), contLen - sizeof(SMsgHead), &req);
SRpcMsg rpcMsg = {.msgType = TDMT_VND_QUERY_COMPACT_PROGRESS,
.contLen = contLen};
//rpcMsg.pCont = rpcMallocCont(contLen);
//if (rpcMsg.pCont == NULL) {
// return;
//}
//memcpy(rpcMsg.pCont, pHead, contLen);
rpcMsg.pCont = pHead;
char detail[1024] = {0};
int32_t len = snprintf(detail, sizeof(detail), "msgType:%s numOfEps:%d inUse:%d", TMSG_INFO(TDMT_VND_QUERY_COMPACT_PROGRESS),
epSet.numOfEps, epSet.inUse);
for (int32_t i = 0; i < epSet.numOfEps; ++i) {
len += snprintf(detail + len, sizeof(detail) - len, " ep:%d-%s:%u", i, epSet.eps[i].fqdn,
epSet.eps[i].port);
}
mDebug("compact:%d, send update progress msg to %s", pDetail->compactId, detail);
tmsgSendReq(&epSet, &rpcMsg);
}
sdbRelease(pMnode->pSdb, pDetail);
}
}
static int32_t mndSaveCompactProgress(SMnode *pMnode, int32_t compactId) {
bool needSave = false;
void* pIter = NULL;
while (1) {
SCompactDetailObj *pDetail = NULL;
pIter = sdbFetch(pMnode->pSdb, SDB_COMPACT_DETAIL, pIter, (void **)&pDetail);
if (pIter == NULL) break;
mDebug("compact:%d, check save progress, vgId:%d, dnodeId:%d, numberFileset:%d, finished:%d, "
"newNumberFileset:%d, newFinished:%d",
pDetail->compactId, pDetail->vgId, pDetail->dnodeId, pDetail->numberFileset, pDetail->finished,
pDetail->newNumberFileset, pDetail->newFinished);
if (pDetail->compactId == compactId) {
if(pDetail->numberFileset < pDetail->newNumberFileset || pDetail->finished < pDetail->newFinished)
needSave = true;
}
sdbRelease(pMnode->pSdb, pDetail);
}
if(!needSave) {
mDebug("compact:%" PRId32 ", no need to save" , compactId);
return 0;
}
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_NOTHING, NULL, "update-compact-progress");
if (pTrans == NULL) {
mError("trans:%" PRId32 ", failed to create since %s" , pTrans->id, terrstr());
return -1;
}
mInfo("trans:%d, used to update compact progress:%" PRId32, pTrans->id, compactId);
SCompactObj *pCompact = mndAcquireCompact(pMnode, compactId);
pIter = NULL;
while (1) {
SCompactDetailObj *pDetail = NULL;
pIter = sdbFetch(pMnode->pSdb, SDB_COMPACT_DETAIL, pIter, (void **)&pDetail);
if (pIter == NULL) break;
if (pDetail->compactId == compactId) {
mInfo("trans:%d, check compact progress:%d, vgId:%d, dnodeId:%d, numberFileset:%d, finished:%d, "
"newNumberFileset:%d, newFinished:%d",
pTrans->id, pDetail->compactId, pDetail->vgId, pDetail->dnodeId, pDetail->numberFileset, pDetail->finished,
pDetail->newNumberFileset, pDetail->newFinished);
pDetail->numberFileset = pDetail->newNumberFileset;
pDetail->finished = pDetail->newFinished;
SSdbRaw *pCommitRaw = mndCompactDetailActionEncode(pDetail);
if (pCommitRaw == NULL || mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) {
mError("trans:%d, failed to append commit log since %s", pTrans->id, terrstr());
mndTransDrop(pTrans);
return -1;
}
(void)sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY);
}
sdbRelease(pMnode->pSdb, pDetail);
}
bool allFinished = true;
while (1) {
SCompactDetailObj *pDetail = NULL;
pIter = sdbFetch(pMnode->pSdb, SDB_COMPACT_DETAIL, pIter, (void **)&pDetail);
if (pIter == NULL) break;
mInfo("trans:%d, check compact finished:%d, vgId:%d, dnodeId:%d, numberFileset:%d, finished:%d",
pTrans->id, pDetail->compactId, pDetail->vgId, pDetail->dnodeId, pDetail->numberFileset, pDetail->finished);
if(pDetail->numberFileset == -1 && pDetail->finished == -1){
allFinished = false;
sdbRelease(pMnode->pSdb, pDetail);
break;
}
if (pDetail->numberFileset != -1 && pDetail->finished != -1 &&
pDetail->numberFileset != pDetail->finished) {
allFinished = false;
sdbRelease(pMnode->pSdb, pDetail);
break;
}
sdbRelease(pMnode->pSdb, pDetail);
}
if(allFinished){
while (1) {
SCompactDetailObj *pDetail = NULL;
pIter = sdbFetch(pMnode->pSdb, SDB_COMPACT_DETAIL, pIter, (void **)&pDetail);
if (pIter == NULL) break;
if (pDetail->compactId == pCompact->compactId) {
SSdbRaw *pCommitRaw = mndCompactDetailActionEncode(pDetail);
if (pCommitRaw == NULL || mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) {
mError("trans:%d, failed to append commit log since %s", pTrans->id, terrstr());
mndTransDrop(pTrans);
return -1;
}
(void)sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED);
}
sdbRelease(pMnode->pSdb, pDetail);
}
SSdbRaw *pCommitRaw = mndCompactActionEncode(pCompact);
if (pCommitRaw == NULL || mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) {
mError("trans:%d, failed to append commit log since %s", pTrans->id, terrstr());
mndTransDrop(pTrans);
return -1;
}
(void)sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED);
}
if (mndTransPrepare(pMnode, pTrans) != 0) {
mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
mndTransDrop(pTrans);
sdbRelease(pMnode->pSdb, pCompact);
return -1;
}
sdbRelease(pMnode->pSdb, pCompact);
mndTransDrop(pTrans);
return 0;
}
void mndCompactPullup(SMnode *pMnode) {
SSdb *pSdb = pMnode->pSdb;
SArray *pArray = taosArrayInit(sdbGetSize(pSdb, SDB_COMPACT), sizeof(int32_t));
if (pArray == NULL) return;
void *pIter = NULL;
while (1) {
SCompactObj *pCompact = NULL;
pIter = sdbFetch(pMnode->pSdb, SDB_COMPACT, pIter, (void **)&pCompact);
if (pIter == NULL) break;
taosArrayPush(pArray, &pCompact->compactId);
sdbRelease(pSdb, pCompact);
}
for (int32_t i = 0; i < taosArrayGetSize(pArray); ++i) {
int32_t *pCompactId = taosArrayGet(pArray, i);
SCompactObj *pCompact = mndAcquireCompact(pMnode, *pCompactId);
if (pCompact != NULL) {
mndCompactSendProgressReq(pMnode, pCompact);
mndSaveCompactProgress(pMnode, pCompact->compactId);
}
mndReleaseCompact(pMnode, pCompact);
}
taosArrayDestroy(pArray);
}
static int32_t mndProcessCompactTimer(SRpcMsg *pReq) {
mTrace("start to process compact timer");
mndCompactPullup(pReq->info.node);
return 0;
}

300
source/dnode/mnode/impl/src/mndCompactDetail.c Normal file
View File

@ -0,0 +1,300 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "mndCompactDetail.h"
#include "mndTrans.h"
#include "mndShow.h"
#include "mndDb.h"
#define MND_COMPACT_VER_NUMBER 1
int32_t mndInitCompactDetail(SMnode *pMnode) {
mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_COMPACT_DETAIL, mndRetrieveCompactDetail);
SSdbTable table = {
.sdbType = SDB_COMPACT_DETAIL,
.keyType = SDB_KEY_INT32,
.encodeFp = (SdbEncodeFp)mndCompactDetailActionEncode,
.decodeFp = (SdbDecodeFp)mndCompactDetailActionDecode,
.insertFp = (SdbInsertFp)mndCompactDetailActionInsert,
.updateFp = (SdbUpdateFp)mndCompactDetailActionUpdate,
.deleteFp = (SdbDeleteFp)mndCompactDetailActionDelete,
};
return sdbSetTable(pMnode->pSdb, table);
}
void mndCleanupCompactDetail(SMnode *pMnode) {
mDebug("mnd compact detail cleanup");
}
int32_t mndRetrieveCompactDetail(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows){
SMnode *pMnode = pReq->info.node;
SSdb *pSdb = pMnode->pSdb;
int32_t numOfRows = 0;
SCompactDetailObj *pCompactDetail = NULL;
char *sep = NULL;
SDbObj *pDb = NULL;
if (strlen(pShow->db) > 0) {
sep = strchr(pShow->db, '.');
if (sep && ((0 == strcmp(sep + 1, TSDB_INFORMATION_SCHEMA_DB) || (0 == strcmp(sep + 1, TSDB_PERFORMANCE_SCHEMA_DB))))) {
sep++;
} else {
pDb = mndAcquireDb(pMnode, pShow->db);
if (pDb == NULL) return terrno;
}
}
while(numOfRows < rows){
pShow->pIter = sdbFetch(pSdb, SDB_COMPACT_DETAIL, pShow->pIter, (void **)&pCompactDetail);
if (pShow->pIter == NULL) break;
SColumnInfoData *pColInfo;
SName n;
int32_t cols = 0;
char tmpBuf[TSDB_SHOW_SQL_LEN + VARSTR_HEADER_SIZE] = {0};
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataSetVal(pColInfo, numOfRows, (const char *)&pCompactDetail->compactId, false);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataSetVal(pColInfo, numOfRows, (const char *)&pCompactDetail->vgId, false);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataSetVal(pColInfo, numOfRows, (const char *)&pCompactDetail->dnodeId, false);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataSetVal(pColInfo, numOfRows, (const char *)&pCompactDetail->numberFileset, false);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataSetVal(pColInfo, numOfRows, (const char *)&pCompactDetail->finished, false);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataSetVal(pColInfo, numOfRows, (const char *)&pCompactDetail->startTime, false);
numOfRows++;
sdbRelease(pSdb, pCompactDetail);
}
pShow->numOfRows += numOfRows;
mndReleaseDb(pMnode, pDb);
return numOfRows;
}
void tFreeCompactDetailObj(SCompactDetailObj *pCompact) {
}
int32_t tSerializeSCompactDetailObj(void *buf, int32_t bufLen, const SCompactDetailObj *pObj) {
SEncoder encoder = {0};
tEncoderInit(&encoder, buf, bufLen);
if (tStartEncode(&encoder) < 0) return -1;
if (tEncodeI32(&encoder, pObj->compactDetailId) < 0) return -1;
if (tEncodeI32(&encoder, pObj->compactId) < 0) return -1;
if (tEncodeI32(&encoder, pObj->vgId) < 0) return -1;
if (tEncodeI32(&encoder, pObj->dnodeId) < 0) return -1;
if (tEncodeI32(&encoder, pObj->numberFileset) < 0) return -1;
if (tEncodeI32(&encoder, pObj->finished) < 0) return -1;
if (tEncodeI64(&encoder, pObj->startTime) < 0) return -1;
if (tEncodeI32(&encoder, pObj->newNumberFileset) < 0) return -1;
if (tEncodeI32(&encoder, pObj->newFinished) < 0) return -1;
tEndEncode(&encoder);
int32_t tlen = encoder.pos;
tEncoderClear(&encoder);
return tlen;
}
int32_t tDeserializeSCompactDetailObj(void *buf, int32_t bufLen, SCompactDetailObj *pObj) {
int8_t ex = 0;
SDecoder decoder = {0};
tDecoderInit(&decoder, buf, bufLen);
if (tStartDecode(&decoder) < 0) return -1;
if (tDecodeI32(&decoder, &pObj->compactDetailId) < 0) return -1;
if (tDecodeI32(&decoder, &pObj->compactId) < 0) return -1;
if (tDecodeI32(&decoder, &pObj->vgId) < 0) return -1;
if (tDecodeI32(&decoder, &pObj->dnodeId) < 0) return -1;
if (tDecodeI32(&decoder, &pObj->numberFileset) < 0) return -1;
if (tDecodeI32(&decoder, &pObj->finished) < 0) return -1;
if (tDecodeI64(&decoder, &pObj->startTime) < 0) return -1;
if (tDecodeI32(&decoder, &pObj->newNumberFileset) < 0) return -1;
if (tDecodeI32(&decoder, &pObj->newFinished) < 0) return -1;
tEndDecode(&decoder);
tDecoderClear(&decoder);
return 0;
}
SSdbRaw *mndCompactDetailActionEncode(SCompactDetailObj *pCompact) {
terrno = TSDB_CODE_SUCCESS;
void *buf = NULL;
SSdbRaw *pRaw = NULL;
int32_t tlen = tSerializeSCompactDetailObj(NULL, 0, pCompact);
if (tlen < 0) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
goto OVER;
}
int32_t size = sizeof(int32_t) + tlen;
pRaw = sdbAllocRaw(SDB_COMPACT_DETAIL, MND_COMPACT_VER_NUMBER, size);
if (pRaw == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
goto OVER;
}
buf = taosMemoryMalloc(tlen);
if (buf == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
goto OVER;
}
tlen = tSerializeSCompactDetailObj(buf, tlen, pCompact);
if (tlen < 0) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
goto OVER;
}
int32_t dataPos = 0;
SDB_SET_INT32(pRaw, dataPos, tlen, OVER);
SDB_SET_BINARY(pRaw, dataPos, buf, tlen, OVER);
SDB_SET_DATALEN(pRaw, dataPos, OVER);
OVER:
taosMemoryFreeClear(buf);
if (terrno != TSDB_CODE_SUCCESS) {
mError("compact detail:%" PRId32 ", failed to encode to raw:%p since %s", pCompact->compactId, pRaw, terrstr());
sdbFreeRaw(pRaw);
return NULL;
}
mTrace("compact detail:%" PRId32 ", encode to raw:%p, row:%p", pCompact->compactId, pRaw, pCompact);
return pRaw;
}
SSdbRow *mndCompactDetailActionDecode(SSdbRaw *pRaw) {
SSdbRow *pRow = NULL;
SCompactDetailObj *pCompact = NULL;
void *buf = NULL;
terrno = TSDB_CODE_SUCCESS;
int8_t sver = 0;
if (sdbGetRawSoftVer(pRaw, &sver) != 0) {
goto OVER;
}
if (sver != MND_COMPACT_VER_NUMBER) {
terrno = TSDB_CODE_SDB_INVALID_DATA_VER;
mError("compact detail read invalid ver, data ver: %d, curr ver: %d", sver, MND_COMPACT_VER_NUMBER);
goto OVER;
}
pRow = sdbAllocRow(sizeof(SCompactObj));
if (pRow == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
goto OVER;
}
pCompact = sdbGetRowObj(pRow);
if (pCompact == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
goto OVER;
}
int32_t tlen;
int32_t dataPos = 0;
SDB_GET_INT32(pRaw, dataPos, &tlen, OVER);
buf = taosMemoryMalloc(tlen + 1);
if (buf == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
goto OVER;
}
SDB_GET_BINARY(pRaw, dataPos, buf, tlen, OVER);
if (tDeserializeSCompactDetailObj(buf, tlen, pCompact) < 0) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
goto OVER;
}
//taosInitRWLatch(&pView->lock);
OVER:
taosMemoryFreeClear(buf);
if (terrno != TSDB_CODE_SUCCESS) {
mError("compact detail:%" PRId32 ", failed to decode from raw:%p since %s", pCompact->compactId, pRaw, terrstr());
taosMemoryFreeClear(pRow);
return NULL;
}
mTrace("compact detail:%" PRId32 ", decode from raw:%p, row:%p", pCompact->compactId, pRaw, pCompact);
return pRow;
}
int32_t mndCompactDetailActionInsert(SSdb *pSdb, SCompactDetailObj *pCompact) {
mTrace("compact detail:%" PRId32 ", perform insert action", pCompact->compactId);
return 0;
}
int32_t mndCompactDetailActionDelete(SSdb *pSdb, SCompactDetailObj *pCompact) {
mTrace("compact detail:%" PRId32 ", perform insert action", pCompact->compactId);
tFreeCompactDetailObj(pCompact);
return 0;
}
int32_t mndCompactDetailActionUpdate(SSdb *pSdb, SCompactDetailObj *pOldCompact, SCompactDetailObj *pNewCompact) {
mTrace("compact detail:%" PRId32 ", perform update action, old row:%p new row:%p",
pOldCompact->compactId, pOldCompact, pNewCompact);
pOldCompact->numberFileset = pNewCompact->numberFileset;
pOldCompact->finished = pNewCompact->finished;
return 0;
}
int32_t mndAddCompactDetailToTran(SMnode *pMnode, STrans *pTrans, SCompactObj* pCompact, SVgObj *pVgroup,
SVnodeGid *pVgid, int32_t index){
SCompactDetailObj compactDetail = {0};
compactDetail.compactDetailId = index;
compactDetail.compactId = pCompact->compactId;
compactDetail.vgId = pVgroup->vgId;
compactDetail.dnodeId = pVgid->dnodeId;
compactDetail.startTime = taosGetTimestampMs();
compactDetail.numberFileset = -1;
compactDetail.finished = -1;
compactDetail.newNumberFileset = -1;
compactDetail.newFinished = -1;
mInfo("compact:%d, add compact detail to trans, vgId:%d, dnodeId:%d",
pCompact->compactId, pVgroup->vgId, pVgid->dnodeId);
SSdbRaw *pVgRaw = mndCompactDetailActionEncode(&compactDetail);
if (pVgRaw == NULL) return -1;
if (mndTransAppendRedolog(pTrans, pVgRaw) != 0) {
sdbFreeRaw(pVgRaw);
return -1;
}
(void)sdbSetRawStatus(pVgRaw, SDB_STATUS_READY);
return 0;
}

21
source/dnode/mnode/impl/src/mndMain.c
View File

@ -42,6 +42,8 @@
#include "mndUser.h"
#include "mndVgroup.h"
#include "mndView.h"
#include "mndCompact.h"
#include "mndCompactDetail.h"
static inline int32_t mndAcquireRpc(SMnode *pMnode) {
int32_t code = 0;
@ -112,6 +114,16 @@ static void mndPullupTrans(SMnode *pMnode) {
}
}
static void mndPullupCompacts(SMnode *pMnode) {
mTrace("pullup compact timer msg");
int32_t contLen = 0;
void *pReq = mndBuildTimerMsg(&contLen);
if (pReq != NULL) {
SRpcMsg rpcMsg = {.msgType = TDMT_MND_COMPACT_TIMER, .pCont = pReq, .contLen = contLen};
tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, &rpcMsg);
}
}
static void mndPullupTtl(SMnode *pMnode) {
mTrace("pullup ttl");
int32_t contLen = 0;
@ -287,6 +299,10 @@ static void *mndThreadFp(void *param) {
mndPullupTrans(pMnode);
}
if (sec % tsCompactPullupInterval == 0) {
mndPullupCompacts(pMnode);
}
if (sec % tsMqRebalanceInterval == 0) {
mndCalMqRebalance(pMnode);
}
@ -462,6 +478,8 @@ static int32_t mndInitSteps(SMnode *pMnode) {
if (mndAllocStep(pMnode, "mnode-db", mndInitDb, mndCleanupDb) != 0) return -1;
if (mndAllocStep(pMnode, "mnode-func", mndInitFunc, mndCleanupFunc) != 0) return -1;
if (mndAllocStep(pMnode, "mnode-view", mndInitView, mndCleanupView) != 0) return -1;
if (mndAllocStep(pMnode, "mnode-compact", mndInitCompact, mndCleanupCompact) != 0) return -1;
if (mndAllocStep(pMnode, "mnode-compact-detail", mndInitCompactDetail, mndCleanupCompactDetail) != 0) return -1;
if (mndAllocStep(pMnode, "mnode-sdb", mndOpenSdb, NULL) != 0) return -1;
if (mndAllocStep(pMnode, "mnode-profile", mndInitProfile, mndCleanupProfile) != 0) return -1;
if (mndAllocStep(pMnode, "mnode-show", mndInitShow, mndCleanupShow) != 0) return -1;
@ -693,7 +711,8 @@ static int32_t mndCheckMnodeState(SRpcMsg *pMsg) {
_OVER:
if (pMsg->msgType == TDMT_MND_TMQ_TIMER || pMsg->msgType == TDMT_MND_TELEM_TIMER ||
pMsg->msgType == TDMT_MND_TRANS_TIMER || pMsg->msgType == TDMT_MND_TTL_TIMER ||
pMsg->msgType == TDMT_MND_TRIM_DB_TIMER || pMsg->msgType == TDMT_MND_UPTIME_TIMER) {
pMsg->msgType == TDMT_MND_TRIM_DB_TIMER || pMsg->msgType == TDMT_MND_UPTIME_TIMER ||
pMsg->msgType == TDMT_MND_COMPACT_TIMER) {
mTrace("timer not process since mnode restored:%d stopped:%d, sync restored:%d role:%s ", pMnode->restored,
pMnode->stopped, state.restored, syncStr(state.state));
return -1;

7
source/dnode/mnode/impl/src/mndShow.c
View File

@ -17,6 +17,7 @@
#include "mndShow.h"
#include "mndPrivilege.h"
#include "systable.h"
#include "mndUser.h"
#define SHOW_STEP_SIZE 100
#define SHOW_COLS_STEP_SIZE 4096
@ -118,6 +119,10 @@ static int32_t convertToRetrieveType(char *name, int32_t len) {
type = TSDB_MGMT_TABLE_PRIVILEGES;
} else if (strncasecmp(name, TSDB_INS_TABLE_VIEWS, len) == 0) {
type = TSDB_MGMT_TABLE_VIEWS;
} else if (strncasecmp(name, TSDB_INS_TABLE_COMPACTS, len) == 0) {
type = TSDB_MGMT_TABLE_COMPACT;
} else if (strncasecmp(name, TSDB_INS_TABLE_COMPACT_DETAILS, len) == 0) {
type = TSDB_MGMT_TABLE_COMPACT_DETAIL;
} else {
mError("invalid show name:%s len:%d", name, len);
}
@ -206,7 +211,7 @@ static int32_t mndProcessRetrieveSysTableReq(SRpcMsg *pReq) {
return -1;
}
mDebug("mndProcessRetrieveSysTableReq tb:%s", retrieveReq.tb);
mDebug("process to retrieve systable req db:%s, tb:%s", retrieveReq.db, retrieveReq.tb);
if (retrieveReq.showId == 0) {
STableMetaRsp *pMeta = taosHashGet(pMnode->infosMeta, retrieveReq.tb, strlen(retrieveReq.tb));

22
source/dnode/mnode/sdb/inc/sdb.h
View File

@ -150,7 +150,9 @@ typedef enum {
SDB_IDX = 21,
SDB_VIEW = 22,
SDB_STREAM_SEQ = 23,
SDB_MAX = 24
SDB_COMPACT = 24,
SDB_COMPACT_DETAIL = 25,
SDB_MAX = 26
} ESdbType;
typedef struct SSdbRaw {
@ -170,11 +172,11 @@ typedef struct SSdbRow {
} SSdbRow;
typedef struct SSdb {
SMnode * pMnode;
SWal * pWal;
SMnode *pMnode;
SWal *pWal;
int64_t sync;
char * currDir;
char * tmpDir;
char *currDir;
char *tmpDir;
int64_t commitIndex;
int64_t commitTerm;
int64_t commitConfig;
@ -184,7 +186,7 @@ typedef struct SSdb {
int64_t tableVer[SDB_MAX];
int64_t maxId[SDB_MAX];
EKeyType keyTypes[SDB_MAX];
SHashObj * hashObjs[SDB_MAX];
SHashObj *hashObjs[SDB_MAX];
TdThreadRwlock locks[SDB_MAX];
SdbInsertFp insertFps[SDB_MAX];
SdbUpdateFp updateFps[SDB_MAX];
@ -199,7 +201,7 @@ typedef struct SSdb {
typedef struct SSdbIter {
TdFilePtr file;
int64_t total;
char * name;
char *name;
} SSdbIter;
typedef struct {
@ -216,8 +218,8 @@ typedef struct {
typedef struct SSdbOpt {
const char *path;
SMnode * pMnode;
SWal * pWal;
SMnode *pMnode;
SWal *pWal;
int64_t sync;
} SSdbOpt;
@ -394,7 +396,7 @@ int32_t sdbGetRawSoftVer(SSdbRaw *pRaw, int8_t *sver);
int32_t sdbGetRawTotalSize(SSdbRaw *pRaw);
SSdbRow *sdbAllocRow(int32_t objSize);
void * sdbGetRowObj(SSdbRow *pRow);
void *sdbGetRowObj(SSdbRow *pRow);
void sdbFreeRow(SSdb *pSdb, SSdbRow *pRow, bool callFunc);
int32_t sdbStartRead(SSdb *pSdb, SSdbIter **ppIter, int64_t *index, int64_t *term, int64_t *config);

6
source/dnode/mnode/sdb/src/sdbHash.c
View File

@ -64,7 +64,11 @@ const char *sdbTableName(ESdbType type) {
return "idx";
case SDB_VIEW:
return "view";
default:
case SDB_COMPACT:
return "compact";
case SDB_COMPACT_DETAIL:
return "compact_detail";
default:
return "undefine";
}
}

1
source/dnode/vnode/CMakeLists.txt
View File

@ -93,6 +93,7 @@ IF (TD_VNODE_PLUGINS)
vnode
PRIVATE
${TD_ENTERPRISE_DIR}/src/plugins/vnode/src/tsdbCompact.c
${TD_ENTERPRISE_DIR}/src/plugins/vnode/src/tsdbCompactMonitor.c
${TD_ENTERPRISE_DIR}/src/plugins/vnode/src/vnodeCompact.c
)
ENDIF ()

3
source/dnode/vnode/src/inc/tsdb.h
View File

@ -74,6 +74,7 @@ typedef struct SDiskDataBuilder SDiskDataBuilder;
typedef struct SBlkInfo SBlkInfo;
typedef struct STsdbDataIter2 STsdbDataIter2;
typedef struct STsdbFilterInfo STsdbFilterInfo;
typedef struct STFileSystem STFileSystem;
#define TSDBROW_ROW_FMT ((int8_t)0x0)
#define TSDBROW_COL_FMT ((int8_t)0x1)
@ -392,6 +393,8 @@ struct STsdb {
TdThreadMutex pgMutex;
struct STFileSystem *pFS; // new
SRocksCache rCache;
// compact monitor
struct SCompMonitor *pCompMonitor;
};
struct TSDBKEY {

5
source/dnode/vnode/src/tsdb/tsdbFS2.c
View File

@ -747,6 +747,7 @@ _exit:
}
static int32_t tsdbFSSetBlockCommit(STFileSet *fset, bool block);
extern int32_t tsdbStopAllCompTask(STsdb *tsdb);
int32_t tsdbDisableAndCancelAllBgTask(STsdb *pTsdb) {
STFileSystem *fs = pTsdb->pFS;
@ -774,6 +775,10 @@ int32_t tsdbDisableAndCancelAllBgTask(STsdb *pTsdb) {
int64_t channel;
TARRAY2_FOREACH(&channelArr, channel) { vnodeAChannelDestroy(vnodeAsyncHandle[1], channel, true); }
TARRAY2_DESTROY(&channelArr, NULL);
#ifdef TD_ENTERPRISE
tsdbStopAllCompTask(pTsdb);
#endif
return 0;
}

1
source/dnode/vnode/src/tsdb/tsdbFSet2.h
View File

@ -27,7 +27,6 @@ typedef struct SSttLvl SSttLvl;
typedef TARRAY2(STFileObj *) TFileObjArray;
typedef TARRAY2(SSttLvl *) TSttLvlArray;
typedef TARRAY2(STFileOp) TFileOpArray;
typedef struct STFileSystem STFileSystem;
typedef enum {
TSDB_FOP_NONE = 0,

12
source/dnode/vnode/src/tsdb/tsdbOpen.c
View File

@ -16,6 +16,9 @@
#include "tsdb.h"
#include "tsdbFS2.h"
extern int32_t tsdbOpenCompMonitor(STsdb *tsdb);
extern int32_t tsdbCloseCompMonitor(STsdb *tsdb);
int32_t tsdbSetKeepCfg(STsdb *pTsdb, STsdbCfg *pCfg) {
STsdbKeepCfg *pKeepCfg = &pTsdb->keepCfg;
pKeepCfg->precision = pCfg->precision;
@ -81,6 +84,12 @@ int tsdbOpen(SVnode *pVnode, STsdb **ppTsdb, const char *dir, STsdbKeepCfg *pKee
goto _err;
}
#ifdef TD_ENTERPRISE
if (tsdbOpenCompMonitor(pTsdb) < 0) {
goto _err;
}
#endif
tsdbDebug("vgId:%d, tsdb is opened at %s, days:%d, keep:%d,%d,%d, keepTimeoffset:%d", TD_VID(pVnode), pTsdb->path,
pTsdb->keepCfg.days, pTsdb->keepCfg.keep0, pTsdb->keepCfg.keep1, pTsdb->keepCfg.keep2,
pTsdb->keepCfg.keepTimeOffset);
@ -108,6 +117,9 @@ int tsdbClose(STsdb **pTsdb) {
tsdbCloseFS(&(*pTsdb)->pFS);
tsdbCloseCache(*pTsdb);
#ifdef TD_ENTERPRISE
tsdbCloseCompMonitor(*pTsdb);
#endif
taosThreadMutexDestroy(&(*pTsdb)->mutex);
taosMemoryFreeClear(*pTsdb);
}

2
source/dnode/vnode/src/vnd/vnodeAsync.c
View File

@ -591,7 +591,7 @@ int32_t vnodeACancel(SVAsync *async, int64_t taskId) {
task->prev->next = task->next;
vnodeAsyncTaskDone(async, task);
} else {
ret = 0; // task is running, should return code TSDB_CODE_BUSY ??
ret = TSDB_CODE_FAILED;
}
}

8
source/dnode/vnode/src/vnd/vnodeHash.c
View File

@ -24,14 +24,6 @@ struct SVHashEntry {
void* obj;
};
struct SVHashTable {
uint32_t (*hash)(const void*);
int32_t (*compare)(const void*, const void*);
int32_t numEntries;
uint32_t numBuckets;
SVHashEntry** buckets;
};
static int32_t vHashRehash(SVHashTable* ht, uint32_t newNumBuckets) {
SVHashEntry** newBuckets = (SVHashEntry**)taosMemoryCalloc(newNumBuckets, sizeof(SVHashEntry*));
if (newBuckets == NULL) {

9
source/dnode/vnode/src/vnd/vnodeHash.h
View File

@ -24,6 +24,15 @@ extern "C" {
typedef struct SVHashTable SVHashTable;
struct SVHashTable {
uint32_t (*hash)(const void*);
int32_t (*compare)(const void*, const void*);
int32_t numEntries;
uint32_t numBuckets;
struct SVHashEntry** buckets;
};
#define vHashNumEntries(ht) ((ht)->numEntries)
int32_t vHashInit(SVHashTable** ht, uint32_t (*hash)(const void*), int32_t (*compare)(const void*, const void*));
int32_t vHashDestroy(SVHashTable** ht);
int32_t vHashPut(SVHashTable* ht, void* obj);

12
source/dnode/vnode/src/vnd/vnodeQuery.c
View File

@ -423,9 +423,9 @@ void vnodeResetLoad(SVnode *pVnode, SVnodeLoad *pLoad) {
"nBatchInsertSuccess");
}
void vnodeGetInfo(void *pVnode, const char **dbname, int32_t *vgId, int64_t* numOfTables, int64_t* numOfNormalTables) {
SVnode* pVnodeObj = pVnode;
SVnodeCfg* pConf = &pVnodeObj->config;
void vnodeGetInfo(void *pVnode, const char **dbname, int32_t *vgId, int64_t *numOfTables, int64_t *numOfNormalTables) {
SVnode *pVnodeObj = pVnode;
SVnodeCfg *pConf = &pVnodeObj->config;
if (dbname) {
*dbname = pConf->dbname;
@ -444,7 +444,7 @@ void vnodeGetInfo(void *pVnode, const char **dbname, int32_t *vgId, int64_t* num
}
}
int32_t vnodeGetTableList(void* pVnode, int8_t type, SArray* pList) {
int32_t vnodeGetTableList(void *pVnode, int8_t type, SArray *pList) {
if (type == TSDB_SUPER_TABLE) {
return vnodeGetStbIdList(pVnode, 0, pList);
} else {
@ -694,12 +694,12 @@ void *vnodeGetIdx(void *pVnode) {
return NULL;
}
return metaGetIdx(((SVnode*)pVnode)->pMeta);
return metaGetIdx(((SVnode *)pVnode)->pMeta);
}
void *vnodeGetIvtIdx(void *pVnode) {
if (pVnode == NULL) {
return NULL;
}
return metaGetIvtIdx(((SVnode*)pVnode)->pMeta);
return metaGetIvtIdx(((SVnode *)pVnode)->pMeta);
}

54
source/dnode/vnode/src/vnd/vnodeSvr.c
View File

@ -14,24 +14,22 @@
*/
#include "audit.h"
#include "cos.h"
#include "tencode.h"
#include "tmsg.h"
#include "tstrbuild.h"
#include "vnd.h"
#include "cos.h"
#include "vnode.h"
#include "vnodeInt.h"
#include "audit.h"
#include "tstrbuild.h"
static int32_t vnodeProcessCreateStbReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp);
static int32_t vnodeProcessAlterStbReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp);
static int32_t vnodeProcessDropStbReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp);
static int32_t vnodeProcessCreateTbReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp,
SRpcMsg *pOriginRpc);
static int32_t vnodeProcessCreateTbReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp,
SRpcMsg *pOriginRpc);
static int32_t vnodeProcessAlterTbReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp);
static int32_t vnodeProcessDropTbReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp,
SRpcMsg *pOriginRpc);
static int32_t vnodeProcessDropTbReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp,
SRpcMsg *pOriginRpc);
static int32_t vnodeProcessSubmitReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp);
static int32_t vnodeProcessCreateTSmaReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp);
static int32_t vnodeProcessAlterConfirmReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp);
@ -45,6 +43,9 @@ static int32_t vnodeProcessDropIndexReq(SVnode *pVnode, int64_t ver, void *pReq,
static int32_t vnodeProcessCompactVnodeReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp);
static int32_t vnodeProcessConfigChangeReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp);
extern int32_t vnodeProcessKillCompactReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp);
extern int32_t vnodeQueryCompactProgress(SVnode *pVnode, SRpcMsg *pMsg);
static int32_t vnodePreprocessCreateTableReq(SVnode *pVnode, SDecoder *pCoder, int64_t btime, int64_t *pUid) {
int32_t code = 0;
int32_t lino = 0;
@ -629,6 +630,11 @@ int32_t vnodeProcessWriteMsg(SVnode *pVnode, SRpcMsg *pMsg, int64_t ver, SRpcMsg
case TDMT_SYNC_CONFIG_CHANGE:
vnodeProcessConfigChangeReq(pVnode, ver, pReq, len, pRsp);
break;
#ifdef TD_ENTERPRISE
case TDMT_VND_KILL_COMPACT:
vnodeProcessKillCompactReq(pVnode, ver, pReq, len, pRsp);
break;
#endif
default:
vError("vgId:%d, unprocessed msg, %d", TD_VID(pVnode), pMsg->msgType);
return -1;
@ -738,6 +744,10 @@ int32_t vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo) {
return vnodeGetTableCfg(pVnode, pMsg, true);
case TDMT_VND_BATCH_META:
return vnodeGetBatchMeta(pVnode, pMsg);
#ifdef TD_ENTERPRISE
case TDMT_VND_QUERY_COMPACT_PROGRESS:
return vnodeQueryCompactProgress(pVnode, pMsg);
#endif
// case TDMT_VND_TMQ_CONSUME:
// return tqProcessPollReq(pVnode->pTq, pMsg);
case TDMT_VND_TMQ_VG_WALINFO:
@ -886,8 +896,8 @@ _err:
return -1;
}
static int32_t vnodeProcessCreateTbReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp,
SRpcMsg *pOriginRpc) {
static int32_t vnodeProcessCreateTbReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp,
SRpcMsg *pOriginRpc) {
SDecoder decoder = {0};
SEncoder encoder = {0};
int32_t rcode = 0;
@ -937,8 +947,8 @@ static int32_t vnodeProcessCreateTbReq(SVnode *pVnode, int64_t ver, void *pReq,
goto _exit;
}
if(tsEnableAudit && tsEnableAuditCreateTable){
char* str = taosMemoryCalloc(1, TSDB_TABLE_FNAME_LEN);
if (tsEnableAudit && tsEnableAuditCreateTable) {
char *str = taosMemoryCalloc(1, TSDB_TABLE_FNAME_LEN);
if (str == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
rcode = -1;
@ -992,26 +1002,26 @@ static int32_t vnodeProcessCreateTbReq(SVnode *pVnode, int64_t ver, void *pReq,
tEncoderInit(&encoder, pRsp->pCont, pRsp->contLen);
tEncodeSVCreateTbBatchRsp(&encoder, &rsp);
if(tsEnableAudit && tsEnableAuditCreateTable){
if (tsEnableAudit && tsEnableAuditCreateTable) {
int64_t clusterId = pVnode->config.syncCfg.nodeInfo[0].clusterId;
SName name = {0};
tNameFromString(&name, pVnode->config.dbname, T_NAME_ACCT | T_NAME_DB);
SStringBuilder sb = {0};
for(int32_t i = 0; i < tbNames->size; i++){
char** key = (char**)taosArrayGet(tbNames, i);
for (int32_t i = 0; i < tbNames->size; i++) {
char **key = (char **)taosArrayGet(tbNames, i);
taosStringBuilderAppendStringLen(&sb, *key, strlen(*key));
if(i < tbNames->size - 1){
if (i < tbNames->size - 1) {
taosStringBuilderAppendChar(&sb, ',');
}
taosMemoryFreeClear(*key);
}
size_t len = 0;
char* keyJoined = taosStringBuilderGetResult(&sb, &len);
size_t len = 0;
char *keyJoined = taosStringBuilderGetResult(&sb, &len);
if(pOriginRpc->info.conn.user != NULL && strlen(pOriginRpc->info.conn.user) > 0){
if (pOriginRpc->info.conn.user != NULL && strlen(pOriginRpc->info.conn.user) > 0) {
auditAddRecord(pOriginRpc, clusterId, "createTable", name.dbname, "", keyJoined, len);
}
@ -1023,7 +1033,7 @@ _exit:
pCreateReq = req.pReqs + iReq;
taosMemoryFree(pCreateReq->sql);
taosMemoryFree(pCreateReq->comment);
taosArrayDestroy(pCreateReq->ctb.tagName);
taosArrayDestroy(pCreateReq->ctb.tagName);
}
taosArrayDestroyEx(rsp.pArray, tFreeSVCreateTbRsp);
taosArrayDestroy(tbUids);
@ -1156,7 +1166,7 @@ _exit:
}
static int32_t vnodeProcessDropTbReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp,
SRpcMsg *pOriginRpc) {
SRpcMsg *pOriginRpc) {
SVDropTbBatchReq req = {0};
SVDropTbBatchRsp rsp = {0};
SDecoder decoder = {0};
@ -1235,13 +1245,13 @@ static int32_t vnodeProcessDropTbReq(SVnode *pVnode, int64_t ver, void *pReq, in
size_t len = 0;
char *keyJoined = taosStringBuilderGetResult(&sb, &len);
if(pOriginRpc->info.conn.user != NULL && strlen(pOriginRpc->info.conn.user) > 0){
if (pOriginRpc->info.conn.user != NULL && strlen(pOriginRpc->info.conn.user) > 0) {
auditAddRecord(pOriginRpc, clusterId, "dropTable", name.dbname, "", keyJoined, len);
}
taosStringBuilderDestroy(&sb);
}
_exit:
taosArrayDestroy(tbUids);
tdUidStoreFree(pStore);

5
source/libs/executor/src/sysscanoperator.c
View File

@ -1610,7 +1610,10 @@ static SSDataBlock* doSysTableScan(SOperatorInfo* pOperator) {
const char* name = tNameGetTableName(&pInfo->name);
if (pInfo->showRewrite) {
getDBNameFromCondition(pInfo->pCondition, dbName);
sprintf(pInfo->req.db, "%d.%s", pInfo->accountId, dbName);
if (strncasecmp(name, TSDB_INS_TABLE_COMPACTS, TSDB_TABLE_FNAME_LEN) != 0 &&
strncasecmp(name, TSDB_INS_TABLE_COMPACT_DETAILS, TSDB_TABLE_FNAME_LEN) != 0) {
sprintf(pInfo->req.db, "%d.%s", pInfo->accountId, dbName);
}
} else if (strncasecmp(name, TSDB_INS_TABLE_COLS, TSDB_TABLE_FNAME_LEN) == 0) {
getDBNameFromCondition(pInfo->pCondition, dbName);
if (dbName[0]) sprintf(pInfo->req.db, "%d.%s", pInfo->accountId, dbName);

4
source/libs/nodes/src/nodesCodeFuncs.c
View File

@ -259,6 +259,10 @@ const char* nodesNodeName(ENodeType type) {
return "ShowLocalVariablesStmt";
case QUERY_NODE_SHOW_TABLE_TAGS_STMT:
return "ShowTableTagsStmt";
case QUERY_NODE_SHOW_COMPACTS_STMT:
return "ShowCompactsStmt";
case QUERY_NODE_SHOW_COMPACT_DETAILS_STMT:
return "ShowCompactDetailsStmt";
case QUERY_NODE_DELETE_STMT:
return "DeleteStmt";
case QUERY_NODE_INSERT_STMT:

13
source/libs/nodes/src/nodesUtilFuncs.c
View File

@ -456,10 +456,15 @@ SNode* nodesMakeNode(ENodeType type) {
return makeNode(type, sizeof(SShowCreateViewStmt));
case QUERY_NODE_SHOW_TABLE_DISTRIBUTED_STMT:
return makeNode(type, sizeof(SShowTableDistributedStmt));
case QUERY_NODE_SHOW_COMPACTS_STMT:
return makeNode(type, sizeof(SShowCompactsStmt));
case QUERY_NODE_SHOW_COMPACT_DETAILS_STMT:
return makeNode(type, sizeof(SShowCompactDetailsStmt));
case QUERY_NODE_KILL_QUERY_STMT:
return makeNode(type, sizeof(SKillQueryStmt));
case QUERY_NODE_KILL_TRANSACTION_STMT:
case QUERY_NODE_KILL_CONNECTION_STMT:
case QUERY_NODE_KILL_COMPACT_STMT:
return makeNode(type, sizeof(SKillStmt));
case QUERY_NODE_DELETE_STMT:
return makeNode(type, sizeof(SDeleteStmt));
@ -1085,6 +1090,13 @@ void nodesDestroyNode(SNode* pNode) {
nodesDestroyNode(((SShowDnodeVariablesStmt*)pNode)->pDnodeId);
nodesDestroyNode(((SShowDnodeVariablesStmt*)pNode)->pLikePattern);
break;
case QUERY_NODE_SHOW_COMPACTS_STMT:
break;
case QUERY_NODE_SHOW_COMPACT_DETAILS_STMT: {
SShowCompactDetailsStmt* pStmt = (SShowCompactDetailsStmt*)pNode;
nodesDestroyNode(pStmt->pCompactId);
break;
}
case QUERY_NODE_SHOW_CREATE_DATABASE_STMT:
taosMemoryFreeClear(((SShowCreateDatabaseStmt*)pNode)->pCfg);
break;
@ -1098,6 +1110,7 @@ void nodesDestroyNode(SNode* pNode) {
case QUERY_NODE_KILL_CONNECTION_STMT: // no pointer field
case QUERY_NODE_KILL_QUERY_STMT: // no pointer field
case QUERY_NODE_KILL_TRANSACTION_STMT: // no pointer field
case QUERY_NODE_KILL_COMPACT_STMT: // no pointer field
break;
case QUERY_NODE_DELETE_STMT: {
SDeleteStmt* pStmt = (SDeleteStmt*)pNode;

3
source/libs/parser/inc/parAst.h
View File

@ -260,7 +260,8 @@ SNode* createDeleteStmt(SAstCreateContext* pCxt, SNode* pTable, SNode* pWhere);
SNode* createInsertStmt(SAstCreateContext* pCxt, SNode* pTable, SNodeList* pCols, SNode* pQuery);
SNode* createCreateViewStmt(SAstCreateContext* pCxt, bool orReplace, SNode* pView, const SToken* pAs, SNode* pQuery);
SNode* createDropViewStmt(SAstCreateContext* pCxt, bool ignoreNotExists, SNode* pView);
SNode* createShowCompactDetailsStmt(SAstCreateContext* pCxt, SNode* pCompactIdNode);
SNode* createShowCompactsStmt(SAstCreateContext* pCxt, ENodeType type);
#ifdef __cplusplus
}
#endif

3
source/libs/parser/inc/sql.y
View File

@ -511,6 +511,8 @@ cmd ::= SHOW db_name_cond_opt(A) ALIVE.
cmd ::= SHOW CLUSTER ALIVE. { pCxt->pRootNode = createShowAliveStmt(pCxt, NULL, QUERY_NODE_SHOW_CLUSTER_ALIVE_STMT); }
cmd ::= SHOW db_name_cond_opt(A) VIEWS. { pCxt->pRootNode = createShowStmtWithCond(pCxt, QUERY_NODE_SHOW_VIEWS_STMT, A, NULL, OP_TYPE_LIKE); }
cmd ::= SHOW CREATE VIEW full_table_name(A). { pCxt->pRootNode = createShowCreateViewStmt(pCxt, QUERY_NODE_SHOW_CREATE_VIEW_STMT, A); }
cmd ::= SHOW COMPACTS. { pCxt->pRootNode = createShowCompactsStmt(pCxt, QUERY_NODE_SHOW_COMPACTS_STMT); }
cmd ::= SHOW COMPACT NK_INTEGER(A). { pCxt->pRootNode = createShowCompactDetailsStmt(pCxt, createValueNode(pCxt, TSDB_DATA_TYPE_BIGINT, &A)); }
%type table_kind_db_name_cond_opt { SShowTablesOption }
%destructor table_kind_db_name_cond_opt { }
@ -700,6 +702,7 @@ ignore_opt(A) ::= IGNORE UNTREATED.
cmd ::= KILL CONNECTION NK_INTEGER(A). { pCxt->pRootNode = createKillStmt(pCxt, QUERY_NODE_KILL_CONNECTION_STMT, &A); }
cmd ::= KILL QUERY NK_STRING(A). { pCxt->pRootNode = createKillQueryStmt(pCxt, &A); }
cmd ::= KILL TRANSACTION NK_INTEGER(A). { pCxt->pRootNode = createKillStmt(pCxt, QUERY_NODE_KILL_TRANSACTION_STMT, &A); }
cmd ::= KILL COMPACT NK_INTEGER(A). { pCxt->pRootNode = createKillStmt(pCxt, QUERY_NODE_KILL_COMPACT_STMT, &A); }
/************************************************ merge/redistribute/ vgroup ******************************************/
cmd ::= BALANCE VGROUP. { pCxt->pRootNode = createBalanceVgroupStmt(pCxt); }

14
source/libs/parser/src/parAstCreater.c
View File

@ -1638,6 +1638,13 @@ SNode* createShowStmt(SAstCreateContext* pCxt, ENodeType type) {
return (SNode*)pStmt;
}
SNode* createShowCompactsStmt(SAstCreateContext* pCxt, ENodeType type) {
CHECK_PARSER_STATUS(pCxt);
SShowCompactsStmt* pStmt = (SShowCompactsStmt*)nodesMakeNode(type);
CHECK_OUT_OF_MEM(pStmt);
return (SNode*)pStmt;
}
SNode* setShowKind(SAstCreateContext* pCxt, SNode* pStmt, EShowKind showKind) {
if (pStmt == NULL) {
return NULL;
@ -1783,6 +1790,13 @@ SNode* createShowTableTagsStmt(SAstCreateContext* pCxt, SNode* pTbName, SNode* p
return (SNode*)pStmt;
}
SNode* createShowCompactDetailsStmt(SAstCreateContext* pCxt, SNode* pCompactId) {
CHECK_PARSER_STATUS(pCxt);
SShowCompactDetailsStmt* pStmt = (SShowCompactDetailsStmt*)nodesMakeNode(QUERY_NODE_SHOW_COMPACT_DETAILS_STMT);
pStmt->pCompactId = pCompactId;
return (SNode*)pStmt;
}
static int32_t getIpV4RangeFromWhitelistItem(char* ipRange, SIpV4Range* pIpRange) {
int32_t code = TSDB_CODE_SUCCESS;
char* ipCopy = taosStrdup(ipRange);

15
source/libs/parser/src/parAstParser.c
View File

@ -607,6 +607,17 @@ static int32_t collectMetaKeyFromShowViews(SCollectMetaKeyCxt* pCxt, SShowStmt*
return code;
}
static int32_t collectMetaKeyFromShowCompacts(SCollectMetaKeyCxt* pCxt, SShowStmt* pStmt) {
int32_t code = reserveTableMetaInCache(pCxt->pParseCxt->acctId, TSDB_INFORMATION_SCHEMA_DB, TSDB_INS_TABLE_COMPACTS,
pCxt->pMetaCache);
return code;
}
static int32_t collectMetaKeyFromShowCompactDetails(SCollectMetaKeyCxt* pCxt, SShowStmt* pStmt) {
int32_t code = reserveTableMetaInCache(pCxt->pParseCxt->acctId, TSDB_INFORMATION_SCHEMA_DB, TSDB_INS_TABLE_COMPACT_DETAILS,
pCxt->pMetaCache);
return code;
}
static int32_t collectMetaKeyFromShowCreateDatabase(SCollectMetaKeyCxt* pCxt, SShowCreateDatabaseStmt* pStmt) {
return reserveDbCfgInCache(pCxt->pParseCxt->acctId, pStmt->dbName, pCxt->pMetaCache);
@ -824,6 +835,10 @@ static int32_t collectMetaKeyFromQuery(SCollectMetaKeyCxt* pCxt, SNode* pStmt) {
return collectMetaKeyFromShowUserPrivileges(pCxt, (SShowStmt*)pStmt);
case QUERY_NODE_SHOW_VIEWS_STMT:
return collectMetaKeyFromShowViews(pCxt, (SShowStmt*)pStmt);
case QUERY_NODE_SHOW_COMPACTS_STMT:
return collectMetaKeyFromShowCompacts(pCxt, (SShowStmt*)pStmt);
case QUERY_NODE_SHOW_COMPACT_DETAILS_STMT:
return collectMetaKeyFromShowCompactDetails(pCxt, (SShowStmt*)pStmt);
case QUERY_NODE_SHOW_CREATE_DATABASE_STMT:
return collectMetaKeyFromShowCreateDatabase(pCxt, (SShowCreateDatabaseStmt*)pStmt);
case QUERY_NODE_SHOW_CREATE_TABLE_STMT:

1
source/libs/parser/src/parTokenizer.c
View File

@ -64,6 +64,7 @@ static SKeyword keywordTable[] = {
{"COMMENT", TK_COMMENT},
{"COMP", TK_COMP},
{"COMPACT", TK_COMPACT},
{"COMPACTS", TK_COMPACTS},
{"CONNECTION", TK_CONNECTION},
{"CONNECTIONS", TK_CONNECTIONS},
{"CONNS", TK_CONNS},

83
source/libs/parser/src/parTranslater.c
View File

@ -257,6 +257,18 @@ static const SSysTableShowAdapter sysTableShowAdapter[] = {
.numOfShowCols = 1,
.pShowCols = {"*"}
},
{ .showType = QUERY_NODE_SHOW_COMPACTS_STMT,
.pDbName = TSDB_INFORMATION_SCHEMA_DB,
.pTableName = TSDB_INS_TABLE_COMPACTS,
.numOfShowCols = 1,
.pShowCols = {"*"}
},
{ .showType = QUERY_NODE_SHOW_COMPACT_DETAILS_STMT,
.pDbName = TSDB_INFORMATION_SCHEMA_DB,
.pTableName = TSDB_INS_TABLE_COMPACT_DETAILS,
.numOfShowCols = 1,
.pShowCols = {"*"}
},
};
// clang-format on
@ -7118,6 +7130,12 @@ static int32_t translateKillConnection(STranslateContext* pCxt, SKillStmt* pStmt
return buildCmdMsg(pCxt, TDMT_MND_KILL_CONN, (FSerializeFunc)tSerializeSKillConnReq, &killReq);
}
static int32_t translateKillCompact(STranslateContext* pCxt, SKillStmt* pStmt) {
SKillCompactReq killReq = {0};
killReq.compactId = pStmt->targetId;
return buildCmdMsg(pCxt, TDMT_MND_KILL_COMPACT, (FSerializeFunc)tSerializeSKillCompactReq, &killReq);
}
static int32_t translateKillQuery(STranslateContext* pCxt, SKillQueryStmt* pStmt) {
SKillQueryReq killReq = {0};
strcpy(killReq.queryStrId, pStmt->queryId);
@ -8511,6 +8529,9 @@ static int32_t translateQuery(STranslateContext* pCxt, SNode* pNode) {
case QUERY_NODE_KILL_CONNECTION_STMT:
code = translateKillConnection(pCxt, (SKillStmt*)pNode);
break;
case QUERY_NODE_KILL_COMPACT_STMT:
code = translateKillCompact(pCxt, (SKillStmt*)pNode);
break;
case QUERY_NODE_KILL_QUERY_STMT:
code = translateKillQuery(pCxt, (SKillQueryStmt*)pNode);
break;
@ -8761,6 +8782,28 @@ static int32_t extractShowVariablesResultSchema(int32_t* numOfCols, SSchema** pS
return TSDB_CODE_SUCCESS;
}
static int32_t extractCompactDbResultSchema(int32_t* numOfCols, SSchema** pSchema) {
*numOfCols = COMPACT_DB_RESULT_COLS;
*pSchema = taosMemoryCalloc((*numOfCols), sizeof(SSchema));
if (NULL == (*pSchema)) {
return TSDB_CODE_OUT_OF_MEMORY;
}
(*pSchema)[0].type = TSDB_DATA_TYPE_BINARY;
(*pSchema)[0].bytes = COMPACT_DB_RESULT_FIELD1_LEN;
strcpy((*pSchema)[0].name, "name");
(*pSchema)[1].type = TSDB_DATA_TYPE_INT;
(*pSchema)[1].bytes = tDataTypes[TSDB_DATA_TYPE_INT].bytes;
strcpy((*pSchema)[1].name, "id");
(*pSchema)[2].type = TSDB_DATA_TYPE_BINARY;
(*pSchema)[2].bytes = COMPACT_DB_RESULT_FIELD3_LEN;
strcpy((*pSchema)[2].name, "reason");
return TSDB_CODE_SUCCESS;
}
int32_t extractResultSchema(const SNode* pRoot, int32_t* numOfCols, SSchema** pSchema) {
if (NULL == pRoot) {
return TSDB_CODE_SUCCESS;
@ -8787,6 +8830,8 @@ int32_t extractResultSchema(const SNode* pRoot, int32_t* numOfCols, SSchema** pS
case QUERY_NODE_SHOW_LOCAL_VARIABLES_STMT:
case QUERY_NODE_SHOW_VARIABLES_STMT:
return extractShowVariablesResultSchema(numOfCols, pSchema);
case QUERY_NODE_COMPACT_DATABASE_STMT:
return extractCompactDbResultSchema(numOfCols, pSchema);
default:
break;
}
@ -10268,6 +10313,37 @@ static int32_t rewriteFlushDatabase(STranslateContext* pCxt, SQuery* pQuery) {
return code;
}
static int32_t rewriteShowCompacts(STranslateContext* pCxt, SQuery* pQuery) {
SShowCompactsStmt* pShow = (SShowCompactsStmt*)(pQuery->pRoot);
SSelectStmt* pStmt = NULL;
int32_t code = createSelectStmtForShow(QUERY_NODE_SHOW_COMPACTS_STMT, &pStmt);
if (TSDB_CODE_SUCCESS == code) {
pCxt->showRewrite = true;
pQuery->showRewrite = true;
nodesDestroyNode(pQuery->pRoot);
pQuery->pRoot = (SNode*)pStmt;
}
return code;
}
static int32_t rewriteShowCompactDetailsStmt(STranslateContext* pCxt, SQuery* pQuery) {
SShowCompactDetailsStmt* pShow = (SShowCompactDetailsStmt*)(pQuery->pRoot);
SSelectStmt* pStmt = NULL;
int32_t code = createSelectStmtForShow(QUERY_NODE_SHOW_COMPACT_DETAILS_STMT, &pStmt);
if (TSDB_CODE_SUCCESS == code) {
if (NULL != pShow->pCompactId) {
code = createOperatorNode(OP_TYPE_EQUAL, "compact_id", pShow->pCompactId, &pStmt->pWhere);
}
}
if (TSDB_CODE_SUCCESS == code) {
pCxt->showRewrite = true;
pQuery->showRewrite = true;
nodesDestroyNode(pQuery->pRoot);
pQuery->pRoot = (SNode*)pStmt;
}
return code;
}
static int32_t rewriteQuery(STranslateContext* pCxt, SQuery* pQuery) {
int32_t code = TSDB_CODE_SUCCESS;
switch (nodeType(pQuery->pRoot)) {
@ -10330,6 +10406,12 @@ static int32_t rewriteQuery(STranslateContext* pCxt, SQuery* pQuery) {
case QUERY_NODE_FLUSH_DATABASE_STMT:
code = rewriteFlushDatabase(pCxt, pQuery);
break;
case QUERY_NODE_SHOW_COMPACTS_STMT:
code = rewriteShowCompacts(pCxt, pQuery);
break;
case QUERY_NODE_SHOW_COMPACT_DETAILS_STMT:
code = rewriteShowCompactDetailsStmt(pCxt, pQuery);
break;
default:
break;
}
@ -10435,6 +10517,7 @@ static int32_t setQuery(STranslateContext* pCxt, SQuery* pQuery) {
pQuery->execMode = QUERY_EXEC_MODE_LOCAL;
break;
case QUERY_NODE_SHOW_VARIABLES_STMT:
case QUERY_NODE_COMPACT_DATABASE_STMT:
pQuery->haveResultSet = true;
pQuery->execMode = QUERY_EXEC_MODE_RPC;
if (NULL != pCxt->pCmdMsg) {

8885
source/libs/parser/src/sql.c
View File

File diff suppressed because it is too large Load Diff

16
tests/develop-test/2-query/table_count_scan.py
View File

@ -65,7 +65,7 @@ class TDTestCase:
tdSql.query('select count(*),db_name, stable_name from information_schema.ins_tables group by db_name, stable_name;')
tdSql.checkRows(3)
tdSql.checkData(0, 0, 24)
tdSql.checkData(0, 0, 26)
tdSql.checkData(0, 1, 'information_schema')
tdSql.checkData(0, 2, None)
tdSql.checkData(1, 0, 3)
@ -77,7 +77,7 @@ class TDTestCase:
tdSql.query('select count(1) v,db_name, stable_name from information_schema.ins_tables group by db_name, stable_name order by v desc;')
tdSql.checkRows(3)
tdSql.checkData(0, 0, 24)
tdSql.checkData(0, 0, 26)
tdSql.checkData(0, 1, 'information_schema')
tdSql.checkData(0, 2, None)
tdSql.checkData(1, 0, 5)
@ -93,7 +93,7 @@ class TDTestCase:
tdSql.checkData(1, 1, 'performance_schema')
tdSql.checkData(0, 0, 3)
tdSql.checkData(0, 1, 'tbl_count')
tdSql.checkData(2, 0, 24)
tdSql.checkData(2, 0, 26)
tdSql.checkData(2, 1, 'information_schema')
tdSql.query("select count(*) from information_schema.ins_tables where db_name='tbl_count'")
@ -106,7 +106,7 @@ class TDTestCase:
tdSql.query('select count(*) from information_schema.ins_tables')
tdSql.checkRows(1)
tdSql.checkData(0, 0, 32)
tdSql.checkData(0, 0, 34)
tdSql.execute('create table stba (ts timestamp, c1 bool, c2 tinyint, c3 smallint, c4 int, c5 bigint, c6 float, c7 double, c8 binary(10), c9 nchar(10), c10 tinyint unsigned, c11 smallint unsigned, c12 int unsigned, c13 bigint unsigned) TAGS(t1 int, t2 binary(10), t3 double);')
@ -189,7 +189,7 @@ class TDTestCase:
tdSql.checkData(2, 0, 5)
tdSql.checkData(2, 1, 'performance_schema')
tdSql.checkData(2, 2, None)
tdSql.checkData(3, 0, 24)
tdSql.checkData(3, 0, 26)
tdSql.checkData(3, 1, 'information_schema')
tdSql.checkData(3, 2, None)
@ -204,7 +204,7 @@ class TDTestCase:
tdSql.checkData(2, 0, 5)
tdSql.checkData(2, 1, 'performance_schema')
tdSql.checkData(2, 2, None)
tdSql.checkData(3, 0, 24)
tdSql.checkData(3, 0, 26)
tdSql.checkData(3, 1, 'information_schema')
tdSql.checkData(3, 2, None)
@ -215,7 +215,7 @@ class TDTestCase:
tdSql.checkData(0, 1, 'tbl_count')
tdSql.checkData(1, 0, 5)
tdSql.checkData(1, 1, 'performance_schema')
tdSql.checkData(2, 0, 24)
tdSql.checkData(2, 0, 26)
tdSql.checkData(2, 1, 'information_schema')
tdSql.query("select count(*) from information_schema.ins_tables where db_name='tbl_count'")
@ -228,7 +228,7 @@ class TDTestCase:
tdSql.query('select count(*) from information_schema.ins_tables')
tdSql.checkRows(1)
tdSql.checkData(0, 0, 33)
tdSql.checkData(0, 0, 35)
tdSql.execute('drop database tbl_count')

2
tests/script/tsim/query/sys_tbname.sim
View File

@ -58,7 +58,7 @@ endi
sql select tbname from information_schema.ins_tables;
print $rows $data00
if $rows != 33 then
if $rows != 35 then
return -1
endi
if $data00 != @ins_tables@ then

6
tests/script/tsim/query/tableCount.sim
View File

@ -53,7 +53,7 @@ sql select stable_name,count(table_name) from information_schema.ins_tables grou
if $rows != 3 then
return -1
endi
if $data01 != 30 then
if $data01 != 32 then
return -1
endi
if $data11 != 10 then
@ -72,7 +72,7 @@ endi
if $data11 != 5 then
return -1
endi
if $data21 != 24 then
if $data21 != 26 then
return -1
endi
if $data31 != 5 then
@ -97,7 +97,7 @@ endi
if $data42 != 3 then
return -1
endi
if $data52 != 24 then
if $data52 != 26 then
return -1
endi
if $data62 != 5 then

5
tests/system-test/0-others/information_schema.py
View File

@ -57,7 +57,8 @@ class TDTestCase:
self.nchar_str = '涛思数据'
self.ins_list = ['ins_dnodes','ins_mnodes','ins_qnodes','ins_snodes','ins_cluster','ins_databases','ins_functions',\
'ins_indexes','ins_stables','ins_tables','ins_tags','ins_columns','ins_users','ins_grants','ins_vgroups','ins_configs','ins_dnode_variables',\
'ins_topics','ins_subscriptions','ins_streams','ins_stream_tasks','ins_vnodes','ins_user_privileges','ins_views']
'ins_topics','ins_subscriptions','ins_streams','ins_stream_tasks','ins_vnodes','ins_user_privileges','ins_views',
'ins_compacts', 'ins_compact_details']
self.perf_list = ['perf_connections','perf_queries','perf_consumers','perf_trans','perf_apps']
def insert_data(self,column_dict,tbname,row_num):
insert_sql = self.setsql.set_insertsql(column_dict,tbname,self.binary_str,self.nchar_str)
@ -217,7 +218,7 @@ class TDTestCase:
tdSql.checkEqual(20470,len(tdSql.queryResult))
tdSql.query("select * from information_schema.ins_columns where db_name ='information_schema'")
tdSql.checkEqual(210, len(tdSql.queryResult))
tdSql.checkEqual(219, len(tdSql.queryResult))
tdSql.query("select * from information_schema.ins_columns where db_name ='performance_schema'")
tdSql.checkEqual(54, len(tdSql.queryResult))