Merge branch '3.0' of https://github.com/taosdata/TDengine into refact/submit_req

include/common/systable.h

@@ -47,6 +47,7 @@ extern "C" {
 #define TSDB_INS_TABLE_TOPICS            "ins_topics"
 #define TSDB_INS_TABLE_STREAMS           "ins_streams"
 #define TSDB_INS_TABLE_STREAM_TASKS      "ins_stream_tasks"
+#define TSDB_INS_TABLE_USER_PRIVILEGES   "ins_user_privileges"
 
 #define TSDB_PERFORMANCE_SCHEMA_DB   "performance_schema"
 #define TSDB_PERFS_TABLE_SMAS        "perf_smas"

include/common/tmsg.h

@@ -151,6 +151,8 @@ typedef enum _mgmt_table {
 #define TSDB_ALTER_USER_REMOVE_ALL_DB          0x8
 #define TSDB_ALTER_USER_ENABLE                 0x9
 #define TSDB_ALTER_USER_SYSINFO                0xA
+#define TSDB_ALTER_USER_ADD_SUBSCRIBE_TOPIC    0xB
+#define TSDB_ALTER_USER_REMOVE_SUBSCRIBE_TOPIC 0xC
 
 #define TSDB_ALTER_USER_PRIVILEGES 0x2
 
@@ -618,7 +620,7 @@ typedef struct {
   int8_t enable;
   char   user[TSDB_USER_LEN];
   char   pass[TSDB_USET_PASSWORD_LEN];
-  char   dbname[TSDB_DB_FNAME_LEN];
+  char   objname[TSDB_DB_FNAME_LEN];  // db or topic
 } SAlterUserReq;
 
 int32_t tSerializeSAlterUserReq(void* buf, int32_t bufLen, SAlterUserReq* pReq);

include/common/ttokendef.h

@@ -58,282 +58,284 @@
 #define TK_TO                   40
 #define TK_REVOKE               41
 #define TK_FROM                 42
-#define TK_NK_COMMA             43
-#define TK_READ                 44
-#define TK_WRITE                45
-#define TK_NK_DOT               46
-#define TK_DNODE                47
-#define TK_PORT                 48
-#define TK_DNODES               49
-#define TK_NK_IPTOKEN           50
-#define TK_FORCE                51
-#define TK_LOCAL                52
-#define TK_QNODE                53
-#define TK_BNODE                54
-#define TK_SNODE                55
-#define TK_MNODE                56
-#define TK_DATABASE             57
-#define TK_USE                  58
-#define TK_FLUSH                59
-#define TK_TRIM                 60
-#define TK_IF                   61
-#define TK_NOT                  62
-#define TK_EXISTS               63
-#define TK_BUFFER               64
-#define TK_CACHEMODEL           65
-#define TK_CACHESIZE            66
-#define TK_COMP                 67
-#define TK_DURATION             68
-#define TK_NK_VARIABLE          69
-#define TK_MAXROWS              70
-#define TK_MINROWS              71
-#define TK_KEEP                 72
-#define TK_PAGES                73
-#define TK_PAGESIZE             74
-#define TK_TSDB_PAGESIZE        75
-#define TK_PRECISION            76
-#define TK_REPLICA              77
-#define TK_STRICT               78
-#define TK_VGROUPS              79
-#define TK_SINGLE_STABLE        80
-#define TK_RETENTIONS           81
-#define TK_SCHEMALESS           82
-#define TK_WAL_LEVEL            83
-#define TK_WAL_FSYNC_PERIOD     84
-#define TK_WAL_RETENTION_PERIOD 85
-#define TK_WAL_RETENTION_SIZE   86
-#define TK_WAL_ROLL_PERIOD      87
-#define TK_WAL_SEGMENT_SIZE     88
-#define TK_STT_TRIGGER          89
-#define TK_TABLE_PREFIX         90
-#define TK_TABLE_SUFFIX         91
-#define TK_NK_COLON             92
-#define TK_MAX_SPEED            93
-#define TK_TABLE                94
-#define TK_NK_LP                95
-#define TK_NK_RP                96
-#define TK_STABLE               97
-#define TK_ADD                  98
-#define TK_COLUMN               99
-#define TK_MODIFY               100
-#define TK_RENAME               101
-#define TK_TAG                  102
-#define TK_SET                  103
-#define TK_NK_EQ                104
-#define TK_USING                105
-#define TK_TAGS                 106
-#define TK_COMMENT              107
-#define TK_BOOL                 108
-#define TK_TINYINT              109
-#define TK_SMALLINT             110
-#define TK_INT                  111
-#define TK_INTEGER              112
-#define TK_BIGINT               113
-#define TK_FLOAT                114
-#define TK_DOUBLE               115
-#define TK_BINARY               116
-#define TK_TIMESTAMP            117
-#define TK_NCHAR                118
-#define TK_UNSIGNED             119
-#define TK_JSON                 120
-#define TK_VARCHAR              121
-#define TK_MEDIUMBLOB           122
-#define TK_BLOB                 123
-#define TK_VARBINARY            124
-#define TK_DECIMAL              125
-#define TK_MAX_DELAY            126
-#define TK_WATERMARK            127
-#define TK_ROLLUP               128
-#define TK_TTL                  129
-#define TK_SMA                  130
-#define TK_FIRST                131
-#define TK_LAST                 132
-#define TK_SHOW                 133
-#define TK_DATABASES            134
-#define TK_TABLES               135
-#define TK_STABLES              136
-#define TK_MNODES               137
-#define TK_QNODES               138
-#define TK_FUNCTIONS            139
-#define TK_INDEXES              140
-#define TK_ACCOUNTS             141
-#define TK_APPS                 142
-#define TK_CONNECTIONS          143
-#define TK_LICENCES             144
-#define TK_GRANTS               145
-#define TK_QUERIES              146
-#define TK_SCORES               147
-#define TK_TOPICS               148
-#define TK_VARIABLES            149
-#define TK_CLUSTER              150
-#define TK_BNODES               151
-#define TK_SNODES               152
-#define TK_TRANSACTIONS         153
-#define TK_DISTRIBUTED          154
-#define TK_CONSUMERS            155
-#define TK_SUBSCRIPTIONS        156
-#define TK_VNODES               157
-#define TK_LIKE                 158
-#define TK_TBNAME               159
-#define TK_QTAGS                160
-#define TK_AS                   161
-#define TK_INDEX                162
-#define TK_FUNCTION             163
-#define TK_INTERVAL             164
-#define TK_TOPIC                165
-#define TK_WITH                 166
-#define TK_META                 167
-#define TK_CONSUMER             168
-#define TK_GROUP                169
-#define TK_DESC                 170
-#define TK_DESCRIBE             171
-#define TK_RESET                172
-#define TK_QUERY                173
-#define TK_CACHE                174
-#define TK_EXPLAIN              175
-#define TK_ANALYZE              176
-#define TK_VERBOSE              177
-#define TK_NK_BOOL              178
-#define TK_RATIO                179
-#define TK_NK_FLOAT             180
-#define TK_OUTPUTTYPE           181
-#define TK_AGGREGATE            182
-#define TK_BUFSIZE              183
-#define TK_STREAM               184
-#define TK_INTO                 185
-#define TK_TRIGGER              186
-#define TK_AT_ONCE              187
-#define TK_WINDOW_CLOSE         188
-#define TK_IGNORE               189
-#define TK_EXPIRED              190
-#define TK_FILL_HISTORY         191
-#define TK_SUBTABLE             192
-#define TK_KILL                 193
-#define TK_CONNECTION           194
-#define TK_TRANSACTION          195
-#define TK_BALANCE              196
-#define TK_VGROUP               197
-#define TK_MERGE                198
-#define TK_REDISTRIBUTE         199
-#define TK_SPLIT                200
-#define TK_DELETE               201
-#define TK_INSERT               202
-#define TK_NULL                 203
-#define TK_NK_QUESTION          204
-#define TK_NK_ARROW             205
-#define TK_ROWTS                206
-#define TK_QSTART               207
-#define TK_QEND                 208
-#define TK_QDURATION            209
-#define TK_WSTART               210
-#define TK_WEND                 211
-#define TK_WDURATION            212
-#define TK_IROWTS               213
-#define TK_CAST                 214
-#define TK_NOW                  215
-#define TK_TODAY                216
-#define TK_TIMEZONE             217
-#define TK_CLIENT_VERSION       218
-#define TK_SERVER_VERSION       219
-#define TK_SERVER_STATUS        220
-#define TK_CURRENT_USER         221
-#define TK_COUNT                222
-#define TK_LAST_ROW             223
-#define TK_CASE                 224
-#define TK_END                  225
-#define TK_WHEN                 226
-#define TK_THEN                 227
-#define TK_ELSE                 228
-#define TK_BETWEEN              229
-#define TK_IS                   230
-#define TK_NK_LT                231
-#define TK_NK_GT                232
-#define TK_NK_LE                233
-#define TK_NK_GE                234
-#define TK_NK_NE                235
-#define TK_MATCH                236
-#define TK_NMATCH               237
-#define TK_CONTAINS             238
-#define TK_IN                   239
-#define TK_JOIN                 240
-#define TK_INNER                241
-#define TK_SELECT               242
-#define TK_DISTINCT             243
-#define TK_WHERE                244
-#define TK_PARTITION            245
-#define TK_BY                   246
-#define TK_SESSION              247
-#define TK_STATE_WINDOW         248
-#define TK_SLIDING              249
-#define TK_FILL                 250
-#define TK_VALUE                251
-#define TK_NONE                 252
-#define TK_PREV                 253
-#define TK_LINEAR               254
-#define TK_NEXT                 255
-#define TK_HAVING               256
-#define TK_RANGE                257
-#define TK_EVERY                258
-#define TK_ORDER                259
-#define TK_SLIMIT               260
-#define TK_SOFFSET              261
-#define TK_LIMIT                262
-#define TK_OFFSET               263
-#define TK_ASC                  264
-#define TK_NULLS                265
-#define TK_ABORT                266
-#define TK_AFTER                267
-#define TK_ATTACH               268
-#define TK_BEFORE               269
-#define TK_BEGIN                270
-#define TK_BITAND               271
-#define TK_BITNOT               272
-#define TK_BITOR                273
-#define TK_BLOCKS               274
-#define TK_CHANGE               275
-#define TK_COMMA                276
-#define TK_COMPACT              277
-#define TK_CONCAT               278
-#define TK_CONFLICT             279
-#define TK_COPY                 280
-#define TK_DEFERRED             281
-#define TK_DELIMITERS           282
-#define TK_DETACH               283
-#define TK_DIVIDE               284
-#define TK_DOT                  285
-#define TK_EACH                 286
-#define TK_FAIL                 287
-#define TK_FILE                 288
-#define TK_FOR                  289
-#define TK_GLOB                 290
-#define TK_ID                   291
-#define TK_IMMEDIATE            292
-#define TK_IMPORT               293
-#define TK_INITIALLY            294
-#define TK_INSTEAD              295
-#define TK_ISNULL               296
-#define TK_KEY                  297
-#define TK_MODULES              298
-#define TK_NK_BITNOT            299
-#define TK_NK_SEMI              300
-#define TK_NOTNULL              301
-#define TK_OF                   302
-#define TK_PLUS                 303
-#define TK_PRIVILEGE            304
-#define TK_RAISE                305
-#define TK_REPLACE              306
-#define TK_RESTRICT             307
-#define TK_ROW                  308
-#define TK_SEMI                 309
-#define TK_STAR                 310
-#define TK_STATEMENT            311
-#define TK_STRING               312
-#define TK_TIMES                313
-#define TK_UPDATE               314
-#define TK_VALUES               315
-#define TK_VARIABLE             316
-#define TK_VIEW                 317
-#define TK_WAL                  318
+#define TK_SUBSCRIBE            43
+#define TK_NK_COMMA             44
+#define TK_READ                 45
+#define TK_WRITE                46
+#define TK_NK_DOT               47
+#define TK_DNODE                48
+#define TK_PORT                 49
+#define TK_DNODES               50
+#define TK_NK_IPTOKEN           51
+#define TK_FORCE                52
+#define TK_LOCAL                53
+#define TK_QNODE                54
+#define TK_BNODE                55
+#define TK_SNODE                56
+#define TK_MNODE                57
+#define TK_DATABASE             58
+#define TK_USE                  59
+#define TK_FLUSH                60
+#define TK_TRIM                 61
+#define TK_IF                   62
+#define TK_NOT                  63
+#define TK_EXISTS               64
+#define TK_BUFFER               65
+#define TK_CACHEMODEL           66
+#define TK_CACHESIZE            67
+#define TK_COMP                 68
+#define TK_DURATION             69
+#define TK_NK_VARIABLE          70
+#define TK_MAXROWS              71
+#define TK_MINROWS              72
+#define TK_KEEP                 73
+#define TK_PAGES                74
+#define TK_PAGESIZE             75
+#define TK_TSDB_PAGESIZE        76
+#define TK_PRECISION            77
+#define TK_REPLICA              78
+#define TK_STRICT               79
+#define TK_VGROUPS              80
+#define TK_SINGLE_STABLE        81
+#define TK_RETENTIONS           82
+#define TK_SCHEMALESS           83
+#define TK_WAL_LEVEL            84
+#define TK_WAL_FSYNC_PERIOD     85
+#define TK_WAL_RETENTION_PERIOD 86
+#define TK_WAL_RETENTION_SIZE   87
+#define TK_WAL_ROLL_PERIOD      88
+#define TK_WAL_SEGMENT_SIZE     89
+#define TK_STT_TRIGGER          90
+#define TK_TABLE_PREFIX         91
+#define TK_TABLE_SUFFIX         92
+#define TK_NK_COLON             93
+#define TK_MAX_SPEED            94
+#define TK_TABLE                95
+#define TK_NK_LP                96
+#define TK_NK_RP                97
+#define TK_STABLE               98
+#define TK_ADD                  99
+#define TK_COLUMN               100
+#define TK_MODIFY               101
+#define TK_RENAME               102
+#define TK_TAG                  103
+#define TK_SET                  104
+#define TK_NK_EQ                105
+#define TK_USING                106
+#define TK_TAGS                 107
+#define TK_COMMENT              108
+#define TK_BOOL                 109
+#define TK_TINYINT              110
+#define TK_SMALLINT             111
+#define TK_INT                  112
+#define TK_INTEGER              113
+#define TK_BIGINT               114
+#define TK_FLOAT                115
+#define TK_DOUBLE               116
+#define TK_BINARY               117
+#define TK_TIMESTAMP            118
+#define TK_NCHAR                119
+#define TK_UNSIGNED             120
+#define TK_JSON                 121
+#define TK_VARCHAR              122
+#define TK_MEDIUMBLOB           123
+#define TK_BLOB                 124
+#define TK_VARBINARY            125
+#define TK_DECIMAL              126
+#define TK_MAX_DELAY            127
+#define TK_WATERMARK            128
+#define TK_ROLLUP               129
+#define TK_TTL                  130
+#define TK_SMA                  131
+#define TK_FIRST                132
+#define TK_LAST                 133
+#define TK_SHOW                 134
+#define TK_PRIVILEGES           135
+#define TK_DATABASES            136
+#define TK_TABLES               137
+#define TK_STABLES              138
+#define TK_MNODES               139
+#define TK_QNODES               140
+#define TK_FUNCTIONS            141
+#define TK_INDEXES              142
+#define TK_ACCOUNTS             143
+#define TK_APPS                 144
+#define TK_CONNECTIONS          145
+#define TK_LICENCES             146
+#define TK_GRANTS               147
+#define TK_QUERIES              148
+#define TK_SCORES               149
+#define TK_TOPICS               150
+#define TK_VARIABLES            151
+#define TK_CLUSTER              152
+#define TK_BNODES               153
+#define TK_SNODES               154
+#define TK_TRANSACTIONS         155
+#define TK_DISTRIBUTED          156
+#define TK_CONSUMERS            157
+#define TK_SUBSCRIPTIONS        158
+#define TK_VNODES               159
+#define TK_LIKE                 160
+#define TK_TBNAME               161
+#define TK_QTAGS                162
+#define TK_AS                   163
+#define TK_INDEX                164
+#define TK_FUNCTION             165
+#define TK_INTERVAL             166
+#define TK_TOPIC                167
+#define TK_WITH                 168
+#define TK_META                 169
+#define TK_CONSUMER             170
+#define TK_GROUP                171
+#define TK_DESC                 172
+#define TK_DESCRIBE             173
+#define TK_RESET                174
+#define TK_QUERY                175
+#define TK_CACHE                176
+#define TK_EXPLAIN              177
+#define TK_ANALYZE              178
+#define TK_VERBOSE              179
+#define TK_NK_BOOL              180
+#define TK_RATIO                181
+#define TK_NK_FLOAT             182
+#define TK_OUTPUTTYPE           183
+#define TK_AGGREGATE            184
+#define TK_BUFSIZE              185
+#define TK_STREAM               186
+#define TK_INTO                 187
+#define TK_TRIGGER              188
+#define TK_AT_ONCE              189
+#define TK_WINDOW_CLOSE         190
+#define TK_IGNORE               191
+#define TK_EXPIRED              192
+#define TK_FILL_HISTORY         193
+#define TK_SUBTABLE             194
+#define TK_KILL                 195
+#define TK_CONNECTION           196
+#define TK_TRANSACTION          197
+#define TK_BALANCE              198
+#define TK_VGROUP               199
+#define TK_MERGE                200
+#define TK_REDISTRIBUTE         201
+#define TK_SPLIT                202
+#define TK_DELETE               203
+#define TK_INSERT               204
+#define TK_NULL                 205
+#define TK_NK_QUESTION          206
+#define TK_NK_ARROW             207
+#define TK_ROWTS                208
+#define TK_QSTART               209
+#define TK_QEND                 210
+#define TK_QDURATION            211
+#define TK_WSTART               212
+#define TK_WEND                 213
+#define TK_WDURATION            214
+#define TK_IROWTS               215
+#define TK_CAST                 216
+#define TK_NOW                  217
+#define TK_TODAY                218
+#define TK_TIMEZONE             219
+#define TK_CLIENT_VERSION       220
+#define TK_SERVER_VERSION       221
+#define TK_SERVER_STATUS        222
+#define TK_CURRENT_USER         223
+#define TK_COUNT                224
+#define TK_LAST_ROW             225
+#define TK_CASE                 226
+#define TK_END                  227
+#define TK_WHEN                 228
+#define TK_THEN                 229
+#define TK_ELSE                 230
+#define TK_BETWEEN              231
+#define TK_IS                   232
+#define TK_NK_LT                233
+#define TK_NK_GT                234
+#define TK_NK_LE                235
+#define TK_NK_GE                236
+#define TK_NK_NE                237
+#define TK_MATCH                238
+#define TK_NMATCH               239
+#define TK_CONTAINS             240
+#define TK_IN                   241
+#define TK_JOIN                 242
+#define TK_INNER                243
+#define TK_SELECT               244
+#define TK_DISTINCT             245
+#define TK_WHERE                246
+#define TK_PARTITION            247
+#define TK_BY                   248
+#define TK_SESSION              249
+#define TK_STATE_WINDOW         250
+#define TK_SLIDING              251
+#define TK_FILL                 252
+#define TK_VALUE                253
+#define TK_NONE                 254
+#define TK_PREV                 255
+#define TK_LINEAR               256
+#define TK_NEXT                 257
+#define TK_HAVING               258
+#define TK_RANGE                259
+#define TK_EVERY                260
+#define TK_ORDER                261
+#define TK_SLIMIT               262
+#define TK_SOFFSET              263
+#define TK_LIMIT                264
+#define TK_OFFSET               265
+#define TK_ASC                  266
+#define TK_NULLS                267
+#define TK_ABORT                268
+#define TK_AFTER                269
+#define TK_ATTACH               270
+#define TK_BEFORE               271
+#define TK_BEGIN                272
+#define TK_BITAND               273
+#define TK_BITNOT               274
+#define TK_BITOR                275
+#define TK_BLOCKS               276
+#define TK_CHANGE               277
+#define TK_COMMA                278
+#define TK_COMPACT              279
+#define TK_CONCAT               280
+#define TK_CONFLICT             281
+#define TK_COPY                 282
+#define TK_DEFERRED             283
+#define TK_DELIMITERS           284
+#define TK_DETACH               285
+#define TK_DIVIDE               286
+#define TK_DOT                  287
+#define TK_EACH                 288
+#define TK_FAIL                 289
+#define TK_FILE                 290
+#define TK_FOR                  291
+#define TK_GLOB                 292
+#define TK_ID                   293
+#define TK_IMMEDIATE            294
+#define TK_IMPORT               295
+#define TK_INITIALLY            296
+#define TK_INSTEAD              297
+#define TK_ISNULL               298
+#define TK_KEY                  299
+#define TK_MODULES              300
+#define TK_NK_BITNOT            301
+#define TK_NK_SEMI              302
+#define TK_NOTNULL              303
+#define TK_OF                   304
+#define TK_PLUS                 305
+#define TK_PRIVILEGE            306
+#define TK_RAISE                307
+#define TK_REPLACE              308
+#define TK_RESTRICT             309
+#define TK_ROW                  310
+#define TK_SEMI                 311
+#define TK_STAR                 312
+#define TK_STATEMENT            313
+#define TK_STRING               314
+#define TK_TIMES                315
+#define TK_UPDATE               316
+#define TK_VALUES               317
+#define TK_VARIABLE             318
+#define TK_VIEW                 319
+#define TK_WAL                  320
 
 #define TK_NK_SPACE   600
 #define TK_NK_COMMENT 601

include/libs/nodes/cmdnodes.h

@@ -45,6 +45,7 @@ extern "C" {
 #define PRIVILEGE_TYPE_ALL       PRIVILEGE_TYPE_MASK(0)
 #define PRIVILEGE_TYPE_READ      PRIVILEGE_TYPE_MASK(1)
 #define PRIVILEGE_TYPE_WRITE     PRIVILEGE_TYPE_MASK(2)
+#define PRIVILEGE_TYPE_SUBSCRIBE PRIVILEGE_TYPE_MASK(3)
 
 #define PRIVILEGE_TYPE_TEST_MASK(val, mask) (((val) & (mask)) != 0)
 
@@ -423,7 +424,7 @@ typedef struct SDropFunctionStmt {
 typedef struct SGrantStmt {
   ENodeType type;
   char      userName[TSDB_USER_LEN];
-  char      dbName[TSDB_DB_NAME_LEN];
+  char      objName[TSDB_DB_NAME_LEN];  // db or topic
   int64_t   privileges;
 } SGrantStmt;
 

include/libs/nodes/nodes.h

@@ -187,6 +187,7 @@ typedef enum ENodeType {
   QUERY_NODE_SHOW_TRANSACTIONS_STMT,
   QUERY_NODE_SHOW_SUBSCRIPTIONS_STMT,
   QUERY_NODE_SHOW_VNODES_STMT,
+  QUERY_NODE_SHOW_USER_PRIVILEGES_STMT,
   QUERY_NODE_SHOW_CREATE_DATABASE_STMT,
   QUERY_NODE_SHOW_CREATE_TABLE_STMT,
   QUERY_NODE_SHOW_CREATE_STABLE_STMT,

include/libs/sync/sync.h

@@ -211,9 +211,14 @@ typedef struct SSyncInfo {
   int32_t (*syncEqCtrlMsg)(const SMsgCb* msgcb, SRpcMsg* pMsg);
 } SSyncInfo;
 
+// if state == leader
+//     if restored, display "leader"
+//     if !restored && canRead, display "leader*"
+//     if !restored && !canRead, display "leader**"
 typedef struct SSyncState {
   ESyncState state;
   bool       restored;
+  bool       canRead;
 } SSyncState;
 
 int32_t syncInit();

source/common/src/systable.c

@@ -273,6 +273,12 @@ static const SSysDbTableSchema vnodesSchema[] = {
     {.name = "dnode_ep", .bytes = TSDB_EP_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = true},
 };
 
+static const SSysDbTableSchema userUserPrivilegesSchema[] = {
+    {.name = "user_name", .bytes = TSDB_USER_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = false},
+    {.name = "privilege", .bytes = 10 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = false},
+    {.name = "object_name", .bytes = TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = false},
+};
+
 static const SSysTableMeta infosMeta[] = {
     {TSDB_INS_TABLE_DNODES, dnodesSchema, tListLen(dnodesSchema), true},
     {TSDB_INS_TABLE_MNODES, mnodesSchema, tListLen(mnodesSchema), true},
@@ -297,6 +303,7 @@ static const SSysTableMeta infosMeta[] = {
     {TSDB_INS_TABLE_STREAMS, streamSchema, tListLen(streamSchema), false},
     {TSDB_INS_TABLE_STREAM_TASKS, streamTaskSchema, tListLen(streamTaskSchema), false},
     {TSDB_INS_TABLE_VNODES, vnodesSchema, tListLen(vnodesSchema), true},
+    {TSDB_INS_TABLE_USER_PRIVILEGES, userUserPrivilegesSchema, tListLen(userUserPrivilegesSchema), false},
 };
 
 static const SSysDbTableSchema connectionsSchema[] = {

source/common/src/tmsg.c

@@ -1288,7 +1288,7 @@ int32_t tSerializeSAlterUserReq(void *buf, int32_t bufLen, SAlterUserReq *pReq)
   if (tEncodeI8(&encoder, pReq->enable) < 0) return -1;
   if (tEncodeCStr(&encoder, pReq->user) < 0) return -1;
   if (tEncodeCStr(&encoder, pReq->pass) < 0) return -1;
-  if (tEncodeCStr(&encoder, pReq->dbname) < 0) return -1;
+  if (tEncodeCStr(&encoder, pReq->objname) < 0) return -1;
   tEndEncode(&encoder);
 
   int32_t tlen = encoder.pos;
@@ -1307,7 +1307,7 @@ int32_t tDeserializeSAlterUserReq(void *buf, int32_t bufLen, SAlterUserReq *pReq
   if (tDecodeI8(&decoder, &pReq->enable) < 0) return -1;
   if (tDecodeCStrTo(&decoder, pReq->user) < 0) return -1;
   if (tDecodeCStrTo(&decoder, pReq->pass) < 0) return -1;
-  if (tDecodeCStrTo(&decoder, pReq->dbname) < 0) return -1;
+  if (tDecodeCStrTo(&decoder, pReq->objname) < 0) return -1;
   tEndDecode(&decoder);
 
   tDecoderClear(&decoder);

source/dnode/mnode/impl/src/mndUser.c

@@ -507,14 +507,14 @@ static int32_t mndProcessAlterUserReq(SRpcMsg *pReq) {
   }
 
   if (alterReq.alterType == TSDB_ALTER_USER_ADD_READ_DB || alterReq.alterType == TSDB_ALTER_USER_ADD_ALL_DB) {
-    if (strcmp(alterReq.dbname, "1.*") != 0) {
-      int32_t len = strlen(alterReq.dbname) + 1;
-      SDbObj *pDb = mndAcquireDb(pMnode, alterReq.dbname);
+    if (strcmp(alterReq.objname, "1.*") != 0) {
+      int32_t len = strlen(alterReq.objname) + 1;
+      SDbObj *pDb = mndAcquireDb(pMnode, alterReq.objname);
       if (pDb == NULL) {
         mndReleaseDb(pMnode, pDb);
         goto _OVER;
       }
-      if (taosHashPut(newUser.readDbs, alterReq.dbname, len, alterReq.dbname, TSDB_DB_FNAME_LEN) != 0) {
+      if (taosHashPut(newUser.readDbs, alterReq.objname, len, alterReq.objname, TSDB_DB_FNAME_LEN) != 0) {
         mndReleaseDb(pMnode, pDb);
         goto _OVER;
       }
@@ -531,14 +531,14 @@ static int32_t mndProcessAlterUserReq(SRpcMsg *pReq) {
   }
 
   if (alterReq.alterType == TSDB_ALTER_USER_ADD_WRITE_DB || alterReq.alterType == TSDB_ALTER_USER_ADD_ALL_DB) {
-    if (strcmp(alterReq.dbname, "1.*") != 0) {
-      int32_t len = strlen(alterReq.dbname) + 1;
-      SDbObj *pDb = mndAcquireDb(pMnode, alterReq.dbname);
+    if (strcmp(alterReq.objname, "1.*") != 0) {
+      int32_t len = strlen(alterReq.objname) + 1;
+      SDbObj *pDb = mndAcquireDb(pMnode, alterReq.objname);
       if (pDb == NULL) {
         mndReleaseDb(pMnode, pDb);
         goto _OVER;
       }
-      if (taosHashPut(newUser.writeDbs, alterReq.dbname, len, alterReq.dbname, TSDB_DB_FNAME_LEN) != 0) {
+      if (taosHashPut(newUser.writeDbs, alterReq.objname, len, alterReq.objname, TSDB_DB_FNAME_LEN) != 0) {
         mndReleaseDb(pMnode, pDb);
         goto _OVER;
       }
@@ -555,28 +555,28 @@ static int32_t mndProcessAlterUserReq(SRpcMsg *pReq) {
   }
 
   if (alterReq.alterType == TSDB_ALTER_USER_REMOVE_READ_DB || alterReq.alterType == TSDB_ALTER_USER_REMOVE_ALL_DB) {
-    if (strcmp(alterReq.dbname, "1.*") != 0) {
-      int32_t len = strlen(alterReq.dbname) + 1;
-      SDbObj *pDb = mndAcquireDb(pMnode, alterReq.dbname);
+    if (strcmp(alterReq.objname, "1.*") != 0) {
+      int32_t len = strlen(alterReq.objname) + 1;
+      SDbObj *pDb = mndAcquireDb(pMnode, alterReq.objname);
       if (pDb == NULL) {
         mndReleaseDb(pMnode, pDb);
         goto _OVER;
       }
-      taosHashRemove(newUser.readDbs, alterReq.dbname, len);
+      taosHashRemove(newUser.readDbs, alterReq.objname, len);
     } else {
       taosHashClear(newUser.readDbs);
     }
   }
 
   if (alterReq.alterType == TSDB_ALTER_USER_REMOVE_WRITE_DB || alterReq.alterType == TSDB_ALTER_USER_REMOVE_ALL_DB) {
-    if (strcmp(alterReq.dbname, "1.*") != 0) {
-      int32_t len = strlen(alterReq.dbname) + 1;
-      SDbObj *pDb = mndAcquireDb(pMnode, alterReq.dbname);
+    if (strcmp(alterReq.objname, "1.*") != 0) {
+      int32_t len = strlen(alterReq.objname) + 1;
+      SDbObj *pDb = mndAcquireDb(pMnode, alterReq.objname);
       if (pDb == NULL) {
         mndReleaseDb(pMnode, pDb);
         goto _OVER;
       }
-      taosHashRemove(newUser.writeDbs, alterReq.dbname, len);
+      taosHashRemove(newUser.writeDbs, alterReq.objname, len);
     } else {
       taosHashClear(newUser.writeDbs);
     }

source/libs/nodes/src/nodesUtilFuncs.c

@@ -424,6 +424,7 @@ SNode* nodesMakeNode(ENodeType type) {
     case QUERY_NODE_SHOW_TRANSACTIONS_STMT:
     case QUERY_NODE_SHOW_SUBSCRIPTIONS_STMT:
     case QUERY_NODE_SHOW_TAGS_STMT:
+    case QUERY_NODE_SHOW_USER_PRIVILEGES_STMT:
       return makeNode(type, sizeof(SShowStmt));
     case QUERY_NODE_SHOW_TABLE_TAGS_STMT:
       return makeNode(type, sizeof(SShowTableTagsStmt));
@@ -944,7 +945,8 @@ void nodesDestroyNode(SNode* pNode) {
     case QUERY_NODE_SHOW_LOCAL_VARIABLES_STMT:
     case QUERY_NODE_SHOW_TRANSACTIONS_STMT:
     case QUERY_NODE_SHOW_SUBSCRIPTIONS_STMT:
-    case QUERY_NODE_SHOW_TAGS_STMT: {
+    case QUERY_NODE_SHOW_TAGS_STMT:
+    case QUERY_NODE_SHOW_USER_PRIVILEGES_STMT: {
       SShowStmt* pStmt = (SShowStmt*)pNode;
       nodesDestroyNode(pStmt->pDbName);
       nodesDestroyNode(pStmt->pTbName);

source/libs/parser/inc/sql.y

@@ -101,6 +101,7 @@ cmd ::= REVOKE privileges(A) ON priv_level(B) FROM user_name(C).
 %destructor privileges                                                            { }
 privileges(A) ::= ALL.                                                            { A = PRIVILEGE_TYPE_ALL; }
 privileges(A) ::= priv_type_list(B).                                              { A = B; }
+privileges(A) ::= SUBSCRIBE.                                                      { A = PRIVILEGE_TYPE_SUBSCRIBE; }
 
 %type priv_type_list                                                              { int64_t }
 %destructor priv_type_list                                                        { }
@@ -116,6 +117,7 @@ priv_type(A) ::= WRITE.
 %destructor priv_level                                                            { }
 priv_level(A) ::= NK_STAR(B) NK_DOT NK_STAR.                                      { A = B; }
 priv_level(A) ::= db_name(B) NK_DOT NK_STAR.                                      { A = B; }
+priv_level(A) ::= topic_name(B).                                                  { A = B; }
 
 /************************************************ create/drop/alter dnode *********************************************/
 cmd ::= CREATE DNODE dnode_endpoint(A).                                           { pCxt->pRootNode = createCreateDnodeStmt(pCxt, &A, NULL); }
@@ -393,6 +395,7 @@ col_name(A) ::= column_name(B).
 /************************************************ show ****************************************************************/
 cmd ::= SHOW DNODES.                                                              { pCxt->pRootNode = createShowStmt(pCxt, QUERY_NODE_SHOW_DNODES_STMT); }
 cmd ::= SHOW USERS.                                                               { pCxt->pRootNode = createShowStmt(pCxt, QUERY_NODE_SHOW_USERS_STMT); }
+cmd ::= SHOW USER PRIVILEGES.                                                     { pCxt->pRootNode = createShowStmt(pCxt, QUERY_NODE_SHOW_USER_PRIVILEGES_STMT); }
 cmd ::= SHOW DATABASES.                                                           { pCxt->pRootNode = createShowStmt(pCxt, QUERY_NODE_SHOW_DATABASES_STMT); }
 cmd ::= SHOW db_name_cond_opt(A) TABLES like_pattern_opt(B).                      { pCxt->pRootNode = createShowStmtWithCond(pCxt, QUERY_NODE_SHOW_TABLES_STMT, A, B, OP_TYPE_LIKE); }
 cmd ::= SHOW db_name_cond_opt(A) STABLES like_pattern_opt(B).                     { pCxt->pRootNode = createShowStmtWithCond(pCxt, QUERY_NODE_SHOW_STABLES_STMT, A, B, OP_TYPE_LIKE); }

source/libs/parser/src/parAstCreater.c

@@ -1815,7 +1815,7 @@ SNode* createGrantStmt(SAstCreateContext* pCxt, int64_t privileges, SToken* pDbN
   SGrantStmt* pStmt = (SGrantStmt*)nodesMakeNode(QUERY_NODE_GRANT_STMT);
   CHECK_OUT_OF_MEM(pStmt);
   pStmt->privileges = privileges;
-  COPY_STRING_FORM_ID_TOKEN(pStmt->dbName, pDbName);
+  COPY_STRING_FORM_ID_TOKEN(pStmt->objName, pDbName);
   COPY_STRING_FORM_ID_TOKEN(pStmt->userName, pUserName);
   return (SNode*)pStmt;
 }
@@ -1828,7 +1828,7 @@ SNode* createRevokeStmt(SAstCreateContext* pCxt, int64_t privileges, SToken* pDb
   SRevokeStmt* pStmt = (SRevokeStmt*)nodesMakeNode(QUERY_NODE_REVOKE_STMT);
   CHECK_OUT_OF_MEM(pStmt);
   pStmt->privileges = privileges;
-  COPY_STRING_FORM_ID_TOKEN(pStmt->dbName, pDbName);
+  COPY_STRING_FORM_ID_TOKEN(pStmt->objName, pDbName);
   COPY_STRING_FORM_ID_TOKEN(pStmt->userName, pUserName);
   return (SNode*)pStmt;
 }

source/libs/parser/src/parAstParser.c

@@ -504,6 +504,11 @@ static int32_t collectMetaKeyFromShowVnodes(SCollectMetaKeyCxt* pCxt, SShowVnode
                                  pCxt->pMetaCache);
 }
 
+static int32_t collectMetaKeyFromShowUserPrivileges(SCollectMetaKeyCxt* pCxt, SShowStmt* pStmt) {
+  return reserveTableMetaInCache(pCxt->pParseCxt->acctId, TSDB_INFORMATION_SCHEMA_DB, TSDB_INS_TABLE_USER_PRIVILEGES,
+                                 pCxt->pMetaCache);
+}
+
 static int32_t collectMetaKeyFromShowCreateDatabase(SCollectMetaKeyCxt* pCxt, SShowCreateDatabaseStmt* pStmt) {
   return reserveDbCfgInCache(pCxt->pParseCxt->acctId, pStmt->dbName, pCxt->pMetaCache);
 }
@@ -648,6 +653,8 @@ static int32_t collectMetaKeyFromQuery(SCollectMetaKeyCxt* pCxt, SNode* pStmt) {
       return collectMetaKeyFromShowDnodeVariables(pCxt, (SShowDnodeVariablesStmt*)pStmt);
     case QUERY_NODE_SHOW_VNODES_STMT:
       return collectMetaKeyFromShowVnodes(pCxt, (SShowVnodesStmt*)pStmt);
+    case QUERY_NODE_SHOW_USER_PRIVILEGES_STMT:
+      return collectMetaKeyFromShowUserPrivileges(pCxt, (SShowStmt*)pStmt);
     case QUERY_NODE_SHOW_CREATE_DATABASE_STMT:
       return collectMetaKeyFromShowCreateDatabase(pCxt, (SShowCreateDatabaseStmt*)pStmt);
     case QUERY_NODE_SHOW_CREATE_TABLE_STMT:

source/libs/parser/src/parTokenizer.c

@@ -161,6 +161,7 @@ static SKeyword keywordTable[] = {
     {"PPS",                  TK_PPS},
     {"PRECISION",            TK_PRECISION},
     {"PREV",                 TK_PREV},
+    {"PRIVILEGES",           TK_PRIVILEGES},
     {"QNODE",                TK_QNODE},
     {"QNODES",               TK_QNODES},
     {"QTIME",                TK_QTIME},
@@ -202,6 +203,7 @@ static SKeyword keywordTable[] = {
     {"STREAM",               TK_STREAM},
     {"STREAMS",              TK_STREAMS},
     {"STRICT",               TK_STRICT},
+    {"SUBSCRIBE",            TK_SUBSCRIBE},
     {"SUBSCRIPTIONS",        TK_SUBSCRIPTIONS},
     {"SUBTABLE",             TK_SUBTABLE},
     {"SYSINFO",              TK_SYSINFO},

source/libs/parser/src/parTranslater.c

@@ -251,6 +251,12 @@ static const SSysTableShowAdapter sysTableShowAdapter[] = {
     .numOfShowCols = 1,
     .pShowCols = {"*"}
   },
+  { .showType = QUERY_NODE_SHOW_USER_PRIVILEGES_STMT,
+    .pDbName = TSDB_INFORMATION_SCHEMA_DB,
+    .pTableName = TSDB_INS_TABLE_USER_PRIVILEGES,
+    .numOfShowCols = 1,
+    .pShowCols = {"*"}
+  },
 };
 // clang-format on
 
@@ -5031,7 +5037,7 @@ static int32_t translateAlterUser(STranslateContext* pCxt, SAlterUserStmt* pStmt
   alterReq.sysInfo = pStmt->sysinfo;
   snprintf(alterReq.pass, sizeof(alterReq.pass), "%s", pStmt->password);
   if (NULL != pCxt->pParseCxt->db) {
-    snprintf(alterReq.dbname, sizeof(alterReq.dbname), "%s", pCxt->pParseCxt->db);
+    snprintf(alterReq.objname, sizeof(alterReq.objname), "%s", pCxt->pParseCxt->db);
   }
 
   return buildCmdMsg(pCxt, TDMT_MND_ALTER_USER, (FSerializeFunc)tSerializeSAlterUserReq, &alterReq);
@@ -5710,9 +5716,11 @@ static int32_t translateGrant(STranslateContext* pCxt, SGrantStmt* pStmt) {
     req.alterType = TSDB_ALTER_USER_ADD_READ_DB;
   } else if (PRIVILEGE_TYPE_TEST_MASK(pStmt->privileges, PRIVILEGE_TYPE_WRITE)) {
     req.alterType = TSDB_ALTER_USER_ADD_WRITE_DB;
+  } else if (PRIVILEGE_TYPE_TEST_MASK(pStmt->privileges, PRIVILEGE_TYPE_SUBSCRIBE)) {
+    req.alterType = TSDB_ALTER_USER_ADD_SUBSCRIBE_TOPIC;
   }
   strcpy(req.user, pStmt->userName);
-  sprintf(req.dbname, "%d.%s", pCxt->pParseCxt->acctId, pStmt->dbName);
+  sprintf(req.objname, "%d.%s", pCxt->pParseCxt->acctId, pStmt->objName);
   return buildCmdMsg(pCxt, TDMT_MND_ALTER_USER, (FSerializeFunc)tSerializeSAlterUserReq, &req);
 }
 
@@ -5726,9 +5734,11 @@ static int32_t translateRevoke(STranslateContext* pCxt, SRevokeStmt* pStmt) {
     req.alterType = TSDB_ALTER_USER_REMOVE_READ_DB;
   } else if (PRIVILEGE_TYPE_TEST_MASK(pStmt->privileges, PRIVILEGE_TYPE_WRITE)) {
     req.alterType = TSDB_ALTER_USER_REMOVE_WRITE_DB;
+  } else if (PRIVILEGE_TYPE_TEST_MASK(pStmt->privileges, PRIVILEGE_TYPE_SUBSCRIBE)) {
+    req.alterType = TSDB_ALTER_USER_REMOVE_SUBSCRIBE_TOPIC;
   }
   strcpy(req.user, pStmt->userName);
-  sprintf(req.dbname, "%d.%s", pCxt->pParseCxt->acctId, pStmt->dbName);
+  sprintf(req.objname, "%d.%s", pCxt->pParseCxt->acctId, pStmt->objName);
   return buildCmdMsg(pCxt, TDMT_MND_ALTER_USER, (FSerializeFunc)tSerializeSAlterUserReq, &req);
 }
 
@@ -7504,6 +7514,7 @@ static int32_t rewriteQuery(STranslateContext* pCxt, SQuery* pQuery) {
     case QUERY_NODE_SHOW_CONSUMERS_STMT:
     case QUERY_NODE_SHOW_SUBSCRIPTIONS_STMT:
     case QUERY_NODE_SHOW_TAGS_STMT:
+    case QUERY_NODE_SHOW_USER_PRIVILEGES_STMT:
       code = rewriteShow(pCxt, pQuery);
       break;
     case QUERY_NODE_SHOW_VGROUPS_STMT:

source/libs/parser/test/mockCatalog.cpp

@@ -101,6 +101,10 @@ void generateInformationSchema(MockCatalogService* mcs) {
       .addColumn("table_name", TSDB_DATA_TYPE_BINARY, TSDB_TABLE_NAME_LEN)
       .addColumn("db_name", TSDB_DATA_TYPE_BINARY, TSDB_DB_NAME_LEN)
       .done();
+  mcs->createTableBuilder(TSDB_INFORMATION_SCHEMA_DB, TSDB_INS_TABLE_USER_PRIVILEGES, TSDB_SYSTEM_TABLE, 2)
+      .addColumn("user_name", TSDB_DATA_TYPE_BINARY, TSDB_USER_LEN)
+      .addColumn("privilege", TSDB_DATA_TYPE_BINARY, 10)
+      .done();
 }
 
 void generatePerformanceSchema(MockCatalogService* mcs) {
@@ -248,8 +252,8 @@ int32_t __catalogGetTableDistVgInfo(SCatalog* pCtg, SRequestConnInfo* pConn, con
   return g_mockCatalogService->catalogGetTableDistVgInfo(pTableName, pVgList);
 }
 
-int32_t __catalogGetDBVgVersion(SCatalog* pCtg, const char* dbFName, int32_t* version, int64_t* dbId,
-                                int32_t* tableNum, int64_t* stateTs) {
+int32_t __catalogGetDBVgVersion(SCatalog* pCtg, const char* dbFName, int32_t* version, int64_t* dbId, int32_t* tableNum,
+                                int64_t* stateTs) {
   return 0;
 }
 

source/libs/parser/test/parAlterToBalanceTest.cpp

@@ -613,7 +613,7 @@ TEST_F(ParserInitialATest, alterUser) {
     if (nullptr != pPass) {
       strcpy(expect.pass, pPass);
     }
-    strcpy(expect.dbname, "test");
+    strcpy(expect.objname, "test");
   };
 
   setCheckDdlFunc([&](const SQuery* pQuery, ParserStage stage) {
@@ -627,7 +627,7 @@ TEST_F(ParserInitialATest, alterUser) {
     ASSERT_EQ(req.enable, expect.enable);
     ASSERT_EQ(std::string(req.user), std::string(expect.user));
     ASSERT_EQ(std::string(req.pass), std::string(expect.pass));
-    ASSERT_EQ(std::string(req.dbname), std::string(expect.dbname));
+    ASSERT_EQ(std::string(req.objname), std::string(expect.objname));
   });
 
   setAlterUserReq("wxy", TSDB_ALTER_USER_PASSWD, "123456");

source/libs/parser/test/parExplainToSyncdbTest.cpp

@@ -34,10 +34,38 @@ TEST_F(ParserExplainToSyncdbTest, explain) {
 TEST_F(ParserExplainToSyncdbTest, grant) {
   useDb("root", "test");
 
+  SAlterUserReq expect = {0};
+
+  auto setAlterUserReq = [&](int8_t alterType, const string& user, const string& obj) {
+    expect.alterType = alterType;
+    snprintf(expect.user, sizeof(expect.user), "%s", user.c_str());
+    snprintf(expect.objname, sizeof(expect.objname), "%s", obj.c_str());
+  };
+
+  setCheckDdlFunc([&](const SQuery* pQuery, ParserStage stage) {
+    ASSERT_EQ(nodeType(pQuery->pRoot), QUERY_NODE_GRANT_STMT);
+    ASSERT_EQ(pQuery->pCmdMsg->msgType, TDMT_MND_ALTER_USER);
+    SAlterUserReq req = {0};
+    ASSERT_EQ(tDeserializeSAlterUserReq(pQuery->pCmdMsg->pMsg, pQuery->pCmdMsg->msgLen, &req), TSDB_CODE_SUCCESS);
+    ASSERT_EQ(req.alterType, expect.alterType);
+    ASSERT_EQ(string(req.user), string(expect.user));
+    ASSERT_EQ(string(req.objname), string(expect.objname));
+  });
+
+  setAlterUserReq(TSDB_ALTER_USER_ADD_ALL_DB, "wxy", "0.test");
   run("GRANT ALL ON test.* TO wxy");
+
+  setAlterUserReq(TSDB_ALTER_USER_ADD_READ_DB, "wxy", "0.test");
   run("GRANT READ ON test.* TO wxy");
+
+  setAlterUserReq(TSDB_ALTER_USER_ADD_WRITE_DB, "wxy", "0.test");
   run("GRANT WRITE ON test.* TO wxy");
+
+  setAlterUserReq(TSDB_ALTER_USER_ADD_ALL_DB, "wxy", "0.test");
   run("GRANT READ, WRITE ON test.* TO wxy");
+
+  setAlterUserReq(TSDB_ALTER_USER_ADD_SUBSCRIBE_TOPIC, "wxy", "0.tp1");
+  run("GRANT SUBSCRIBE ON tp1 TO wxy");
 }
 
 TEST_F(ParserExplainToSyncdbTest, insert) {

source/libs/parser/test/parShowToUse.cpp

@@ -213,7 +213,13 @@ TEST_F(ParserShowToUseTest, showTags) {
 TEST_F(ParserShowToUseTest, showUsers) {
   useDb("root", "test");
 
-  run("SHOW users");
+  run("SHOW USERS");
+}
+
+TEST_F(ParserShowToUseTest, showUserPrivileges) {
+  useDb("root", "test");
+
+  run("SHOW USER PRIVILEGES");
 }
 
 TEST_F(ParserShowToUseTest, showVariables) {

source/libs/sync/inc/syncInt.h

@@ -244,6 +244,7 @@ int32_t   syncCacheEntry(SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry, LRUHa
 bool      syncNodeHeartbeatReplyTimeout(SSyncNode* pSyncNode);
 bool      syncNodeSnapshotSending(SSyncNode* pSyncNode);
 bool      syncNodeSnapshotRecving(SSyncNode* pSyncNode);
+bool      syncNodeIsReadyForRead(SSyncNode* pSyncNode);
 
 // raft state change --------------
 void syncNodeUpdateTerm(SSyncNode* pSyncNode, SyncTerm term);

source/libs/sync/inc/syncUtil.h

@@ -94,11 +94,12 @@ void syncLogRecvLocalCmd(SSyncNode* pSyncNode, const SyncLocalCmd* pMsg, const c
 void syncLogSendAppendEntriesReply(SSyncNode* pSyncNode, const SyncAppendEntriesReply* pMsg, const char* s);
 void syncLogRecvAppendEntriesReply(SSyncNode* pSyncNode, const SyncAppendEntriesReply* pMsg, const char* s);
 
-void syncLogSendHeartbeat(SSyncNode* pSyncNode, const SyncHeartbeat* pMsg, bool printX, int64_t timerElapsed, int64_t execTime);
-void syncLogRecvHeartbeat(SSyncNode* pSyncNode, const SyncHeartbeat* pMsg, int64_t timeDiff);
+void syncLogSendHeartbeat(SSyncNode* pSyncNode, const SyncHeartbeat* pMsg, bool printX, int64_t timerElapsed,
+                          int64_t execTime);
+void syncLogRecvHeartbeat(SSyncNode* pSyncNode, const SyncHeartbeat* pMsg, int64_t timeDiff, const char* s);
 
 void syncLogSendHeartbeatReply(SSyncNode* pSyncNode, const SyncHeartbeatReply* pMsg, const char* s);
-void syncLogRecvHeartbeatReply(SSyncNode* pSyncNode, const SyncHeartbeatReply* pMsg, int64_t timeDiff);
+void syncLogRecvHeartbeatReply(SSyncNode* pSyncNode, const SyncHeartbeatReply* pMsg, int64_t timeDiff, const char* s);
 
 void syncLogSendSyncPreSnapshot(SSyncNode* pSyncNode, const SyncPreSnapshot* pMsg, const char* s);
 void syncLogRecvSyncPreSnapshot(SSyncNode* pSyncNode, const SyncPreSnapshot* pMsg, const char* s);

source/libs/sync/src/syncMain.c

@@ -381,15 +381,13 @@ int32_t syncStepDown(int64_t rid, SyncTerm newTerm) {
   return 0;
 }
 
-bool syncIsReadyForRead(int64_t rid) {
-  SSyncNode* pSyncNode = syncNodeAcquire(rid);
+bool syncNodeIsReadyForRead(SSyncNode* pSyncNode) {
   if (pSyncNode == NULL) {
     sError("sync ready for read error");
     return false;
   }
 
   if (pSyncNode->state == TAOS_SYNC_STATE_LEADER && pSyncNode->restoreFinish) {
-    syncNodeRelease(pSyncNode);
     return true;
   }
 
@@ -443,6 +441,18 @@ bool syncIsReadyForRead(int64_t rid) {
     }
   }
 
+  return ready;
+}
+
+bool syncIsReadyForRead(int64_t rid) {
+  SSyncNode* pSyncNode = syncNodeAcquire(rid);
+  if (pSyncNode == NULL) {
+    sError("sync ready for read error");
+    return false;
+  }
+
+  bool ready = syncNodeIsReadyForRead(pSyncNode);
+
   syncNodeRelease(pSyncNode);
   return ready;
 }
@@ -521,6 +531,7 @@ SSyncState syncGetState(int64_t rid) {
   if (pSyncNode != NULL) {
     state.state = pSyncNode->state;
     state.restored = pSyncNode->restoreFinish;
+    state.canRead = syncNodeIsReadyForRead(pSyncNode);
     syncNodeRelease(pSyncNode);
   }
 
@@ -2255,9 +2266,13 @@ static int32_t syncNodeAppendNoop(SSyncNode* ths) {
 int32_t syncNodeOnHeartbeat(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
   SyncHeartbeat* pMsg = pRpcMsg->pCont;
 
+  const STraceId* trace = &pRpcMsg->info.traceId;
+  char            tbuf[40] = {0};
+  TRACE_TO_STR(trace, tbuf);
+
   int64_t tsMs = taosGetTimestampMs();
   int64_t timeDiff = tsMs - pMsg->timeStamp;
-  syncLogRecvHeartbeat(ths, pMsg, timeDiff);
+  syncLogRecvHeartbeat(ths, pMsg, timeDiff, tbuf);
 
   SRpcMsg rpcMsg = {0};
   (void)syncBuildHeartbeatReply(&rpcMsg, ths->vgId);
@@ -2332,9 +2347,13 @@ int32_t syncNodeOnHeartbeat(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
 int32_t syncNodeOnHeartbeatReply(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
   SyncHeartbeatReply* pMsg = pRpcMsg->pCont;
 
+  const STraceId* trace = &pRpcMsg->info.traceId;
+  char            tbuf[40] = {0};
+  TRACE_TO_STR(trace, tbuf);
+
   int64_t tsMs = taosGetTimestampMs();
   int64_t timeDiff = tsMs - pMsg->timeStamp;
-  syncLogRecvHeartbeatReply(ths, pMsg, timeDiff);
+  syncLogRecvHeartbeatReply(ths, pMsg, timeDiff, tbuf);
 
   // update last reply time, make decision whether the other node is alive or not
   syncIndexMgrSetRecvTime(ths->pMatchIndex, &pMsg->srcId, tsMs);

source/libs/sync/src/syncUtil.c

@@ -464,7 +464,7 @@ void syncLogSendHeartbeat(SSyncNode* pSyncNode, const SyncHeartbeat* pMsg, bool
   }
 }
 
-void syncLogRecvHeartbeat(SSyncNode* pSyncNode, const SyncHeartbeat* pMsg, int64_t timeDiff) {
+void syncLogRecvHeartbeat(SSyncNode* pSyncNode, const SyncHeartbeat* pMsg, int64_t timeDiff, const char* s) {
   if (timeDiff > SYNC_HEARTBEAT_SLOW_MS) {
     pSyncNode->hbSlowNum++;
 
@@ -473,8 +473,8 @@ void syncLogRecvHeartbeat(SSyncNode* pSyncNode, const SyncHeartbeat* pMsg, int64
     syncUtilU642Addr(pMsg->srcId.addr, host, sizeof(host), &port);
     sNInfo(pSyncNode,
            "recv sync-heartbeat from %s:%d slow {term:%" PRId64 ", cmt:%" PRId64 ", min-match:%" PRId64 ", ts:%" PRId64
-           "}, net elapsed:%" PRId64,
-           host, port, pMsg->term, pMsg->commitIndex, pMsg->minMatchIndex, pMsg->timeStamp, timeDiff);
+           "}, %s, net elapsed:%" PRId64,
+           host, port, pMsg->term, pMsg->commitIndex, pMsg->minMatchIndex, pMsg->timeStamp, s, timeDiff);
   }
 
   if (!(sDebugFlag & DEBUG_TRACE)) return;
@@ -484,8 +484,8 @@ void syncLogRecvHeartbeat(SSyncNode* pSyncNode, const SyncHeartbeat* pMsg, int64
   syncUtilU642Addr(pMsg->srcId.addr, host, sizeof(host), &port);
   sNTrace(pSyncNode,
           "recv sync-heartbeat from %s:%d {term:%" PRId64 ", cmt:%" PRId64 ", min-match:%" PRId64 ", ts:%" PRId64
-          "}, net elapsed:%" PRId64,
-          host, port, pMsg->term, pMsg->commitIndex, pMsg->minMatchIndex, pMsg->timeStamp, timeDiff);
+          "}, %s, net elapsed:%" PRId64,
+          host, port, pMsg->term, pMsg->commitIndex, pMsg->minMatchIndex, pMsg->timeStamp, s, timeDiff);
 }
 
 void syncLogSendHeartbeatReply(SSyncNode* pSyncNode, const SyncHeartbeatReply* pMsg, const char* s) {
@@ -499,7 +499,7 @@ void syncLogSendHeartbeatReply(SSyncNode* pSyncNode, const SyncHeartbeatReply* p
           pMsg->term, pMsg->timeStamp, s);
 }
 
-void syncLogRecvHeartbeatReply(SSyncNode* pSyncNode, const SyncHeartbeatReply* pMsg, int64_t timeDiff) {
+void syncLogRecvHeartbeatReply(SSyncNode* pSyncNode, const SyncHeartbeatReply* pMsg, int64_t timeDiff, const char* s) {
   if (timeDiff > SYNC_HEARTBEAT_REPLY_SLOW_MS) {
     pSyncNode->hbrSlowNum++;
 
@@ -507,8 +507,8 @@ void syncLogRecvHeartbeatReply(SSyncNode* pSyncNode, const SyncHeartbeatReply* p
     uint16_t port;
     syncUtilU642Addr(pMsg->srcId.addr, host, sizeof(host), &port);
     sNTrace(pSyncNode,
-            "recv sync-heartbeat-reply from %s:%d slow {term:%" PRId64 ", ts:%" PRId64 "}, net elapsed:%" PRId64, host,
-            port, pMsg->term, pMsg->timeStamp, timeDiff);
+            "recv sync-heartbeat-reply from %s:%d slow {term:%" PRId64 ", ts:%" PRId64 "}, %s, net elapsed:%" PRId64,
+            host, port, pMsg->term, pMsg->timeStamp, s, timeDiff);
   }
 
   if (!(sDebugFlag & DEBUG_TRACE)) return;
@@ -516,8 +516,9 @@ void syncLogRecvHeartbeatReply(SSyncNode* pSyncNode, const SyncHeartbeatReply* p
   char     host[64];
   uint16_t port;
   syncUtilU642Addr(pMsg->srcId.addr, host, sizeof(host), &port);
-  sNTrace(pSyncNode, "recv sync-heartbeat-reply from %s:%d {term:%" PRId64 ", ts:%" PRId64 "}, net elapsed:%" PRId64,
-          host, port, pMsg->term, pMsg->timeStamp, timeDiff);
+  sNTrace(pSyncNode,
+          "recv sync-heartbeat-reply from %s:%d {term:%" PRId64 ", ts:%" PRId64 "}, %s, net elapsed:%" PRId64, host,
+          port, pMsg->term, pMsg->timeStamp, s, timeDiff);
 }
 
 void syncLogSendSyncPreSnapshot(SSyncNode* pSyncNode, const SyncPreSnapshot* pMsg, const char* s) {

tests/parallel_test/cases.task

@@ -418,7 +418,7 @@
 ,,y,system-test,./pytest.sh python3 ./test.py -f 0-others/fsync.py
 ,,n,system-test,python3 ./test.py -f 0-others/compatibility.py
 ,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/alter_database.py
-,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/influxdb_line_taosc_insert.py
+#,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/influxdb_line_taosc_insert.py
 ,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/opentsdb_telnet_line_taosc_insert.py
 ,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/opentsdb_json_taosc_insert.py
 ,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/test_stmt_muti_insert_query.py

tests/system-test/0-others/compatibility.py

@@ -4,7 +4,7 @@ import sys
 import os
 import time
 
-
+from pathlib import Path
 from util.log import *
 from util.sql import *
 from util.cases import *
@@ -13,12 +13,12 @@ from util.dnodes import TDDnodes
 from util.dnodes import TDDnode
 from util.cluster import *
 
-
+BASEVERSION = "3.0.1.8"
 class TDTestCase:
     def caseDescription(self):
         '''
         3.0 data compatibility test 
-		case1: basedata version is 3.0.1.0
+        case1: basedata version is 3.0.1.8
         '''
         return
 
@@ -61,13 +61,21 @@ class TDTestCase:
         # os.system(f"mv {self.getBuildPath()}/build/lib/libtaos.so  {self.getBuildPath()}/build/lib/libtaos.so_bak ")
         # os.system(f"mv {self.getBuildPath()}/build/lib/libtaos.so.1  {self.getBuildPath()}/build/lib/libtaos.so.1_bak ")
         
-		packagePath="/usr/local/src/"
-		packageName="TDengine-server-3.0.1.0-Linux-x64.tar.gz"
-		os.system(f"cd {packagePath} &&  tar xvf  TDengine-server-3.0.1.0-Linux-x64.tar.gz && cd TDengine-server-3.0.1.0 && ./install.sh  -e no  " )
+        packagePath = "/usr/local/src/"
+        
+        packageName = "TDengine-server-"+  BASEVERSION + "-Linux-x64.tar.gz"
+        packageTPath = packageName.split("-Linux-")[0]
+        my_file = Path(f"{packagePath}/{packageName}")
+        if not  my_file.exists():
+            print(f"{packageName} is not exists")
+            os.system(f"cd {packagePath} &&  wget https://www.tdengine.com/assets-download/3.0/{packageName}")
+        else: 
+            print(f"{packageName} has been exists")
+        os.system(f"cd {packagePath} &&  tar xvf  {packageName} && cd {packageTPath} &&  ./install.sh  -e no  " )
         tdDnodes.stop(1)
         print(f"start taosd: nohup taosd -c {cPath} & ")
         os.system(f" nohup taosd -c {cPath} & " )
-		sleep(1)
+        sleep(5)
 
 
         
@@ -77,33 +85,27 @@ class TDTestCase:
 
 
     def run(self):
-		print(f"start taosd run")
         bPath=self.getBuildPath()
         cPath=self.getCfgPath()
         dbname = "test"
         stb = f"{dbname}.meters"
         self.installTaosd(bPath,cPath)
         os.system("echo 'debugFlag 143' > /etc/taos/taos.cfg ")
-		os.system("echo ' supportVnodes 256' > /etc/taos/taos.cfg ")
         tableNumbers=100
         recordNumbers1=100
         recordNumbers2=1000
-		#tdsqlF=tdCom.newTdSql()
-		#print(tdsqlF)
-
-		oldServerVersion = '3.0.1.0' 
-		#tdsqlF.query(f"SELECT SERVER_VERSION();")
-		#print(tdsqlF.query(f"SELECT SERVER_VERSION();"))
-		#oldServerVersion=tdsqlF.queryResult[0][0]
-		#tdLog.info(f"Base server version is {oldServerVersion}")
-		#tdsqlF.query(f"SELECT CLIENT_VERSION();")
-		#	
-		## the oldClientVersion can't be updated in the same python process,so the version is new compiled verison
-		#oldClientVersion=tdsqlF.queryResult[0][0]
-
-		#tdLog.info(f"Base client version is {oldClientVersion}")
-
-		tdLog.printNoPrefix(f"==========step1:prepare and check data in old version-{oldServerVersion}")
+        # tdsqlF=tdCom.newTdSql()
+        # print(tdsqlF)
+        # tdsqlF.query(f"SELECT SERVER_VERSION();")
+        # print(tdsqlF.query(f"SELECT SERVER_VERSION();"))
+        # oldServerVersion=tdsqlF.queryResult[0][0]
+        # tdLog.info(f"Base server version is {oldServerVersion}")
+        # tdsqlF.query(f"SELECT CLIENT_VERSION();")
+        # # the oldClientVersion can't be updated in the same python process,so the version is new compiled verison
+        # oldClientVersion=tdsqlF.queryResult[0][0]
+        # tdLog.info(f"Base client version is {oldClientVersion}")
+        # baseVersion = "3.0.1.8"
+        tdLog.printNoPrefix(f"==========step1:prepare and check data in old version-{BASEVERSION}")
         tdLog.info(f" LD_LIBRARY_PATH=/usr/lib  taosBenchmark -t {tableNumbers} -n {recordNumbers1} -y  ")
         os.system(f"LD_LIBRARY_PATH=/usr/lib taosBenchmark -t {tableNumbers} -n {recordNumbers1} -y  ")
         sleep(3)
@@ -143,6 +145,8 @@ class TDTestCase:
         os.system(f"taosBenchmark -t {tableNumbers} -n {recordNumbers2} -y  ")
         tdsql.query(f"select count(*) from {stb}")
         tdsql.checkData(0,0,tableNumbers*recordNumbers2)
+        tdsql.query(f"select count(*) from db4096.stb0")
+        tdsql.checkData(0,0,50000)
 
         tdsql=tdCom.newTdSql()
         tdLog.printNoPrefix(f"==========step4:verify backticks in taos Sql-TD18542")
@@ -170,5 +174,6 @@ class TDTestCase:
         tdSql.close()
         tdLog.success(f"{__file__} successfully executed")
 
+
 tdCases.addLinux(__file__, TDTestCase())
 tdCases.addWindows(__file__, TDTestCase())