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

Merge remote-tracking branch 'origin/feature/3_liaohj' into feature/3_liaohj

This commit is contained in:
Haojun Liao 2022-06-23 16:35:05 +08:00
parent e071649619 bd64bebb63
commit 7b75a8e441
36 changed files with 1099 additions and 236 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
examples/c/tmq.c
View File

@ -76,19 +76,41 @@ int32_t init_env() {
} }
taos_free_result(pRes); taos_free_result(pRes);
pRes = taos_query(pConn, "create table if not exists ct1 using st1 tags(2000)"); pRes = taos_query(pConn, "insert into ct0 values(now, 1, 2, 'a')");
if (taos_errno(pRes) != 0) { if (taos_errno(pRes) != 0) {
printf("failed to create child table tu2, reason:%s\n", taos_errstr(pRes)); printf("failed to insert into ct0, reason:%s\n", taos_errstr(pRes));
return -1; return -1;
} }
taos_free_result(pRes);
pRes = taos_query(pConn, "create table if not exists ct1 using st1 tags(2000)");
if (taos_errno(pRes) != 0) {
printf("failed to create child table ct1, reason:%s\n", taos_errstr(pRes));
return -1;
}
taos_free_result(pRes);
pRes = taos_query(pConn, "insert into ct1 values(now, 3, 4, 'b')");
if (taos_errno(pRes) != 0) {
printf("failed to insert into ct1, reason:%s\n", taos_errstr(pRes));
return -1;
}
taos_free_result(pRes);
pRes = taos_query(pConn, "create table if not exists ct3 using st1 tags(3000)"); pRes = taos_query(pConn, "create table if not exists ct3 using st1 tags(3000)");
if (taos_errno(pRes) != 0) { if (taos_errno(pRes) != 0) {
printf("failed to create child table tu3, reason:%s\n", taos_errstr(pRes)); printf("failed to create child table ct3, reason:%s\n", taos_errstr(pRes));
return -1; return -1;
} }
taos_free_result(pRes); taos_free_result(pRes);
pRes = taos_query(pConn, "insert into ct3 values(now, 5, 6, 'c')");
if (taos_errno(pRes) != 0) {
printf("failed to insert into ct3, reason:%s\n", taos_errstr(pRes));
return -1;
}
taos_free_result(pRes);
return 0; return 0;
} }
@ -168,6 +190,9 @@ tmq_t* build_consumer() {
tmq_conf_set(conf, "td.connect.pass", "taosdata"); tmq_conf_set(conf, "td.connect.pass", "taosdata");
tmq_conf_set(conf, "msg.with.table.name", "true"); tmq_conf_set(conf, "msg.with.table.name", "true");
tmq_conf_set(conf, "enable.auto.commit", "true"); tmq_conf_set(conf, "enable.auto.commit", "true");
tmq_conf_set(conf, "experiment.use.snapshot", "true");
tmq_conf_set_auto_commit_cb(conf, tmq_commit_cb_print, NULL); tmq_conf_set_auto_commit_cb(conf, tmq_commit_cb_print, NULL);
tmq_t* tmq = tmq_consumer_new(conf, NULL, 0); tmq_t* tmq = tmq_consumer_new(conf, NULL, 0);
assert(tmq); assert(tmq);

5
include/common/tcommon.h
View File

@ -54,6 +54,7 @@ typedef struct {
SHashObj* map; // speedup acquire the tableQueryInfo by table uid SHashObj* map; // speedup acquire the tableQueryInfo by table uid
void* pTagCond; void* pTagCond;
void* pTagIndexCond; void* pTagIndexCond;
uint64_t suid;
} STableListInfo; } STableListInfo;
typedef struct SColumnDataAgg { typedef struct SColumnDataAgg {
@ -114,8 +115,8 @@ typedef struct SQueryTableDataCond {
int32_t type; // data block load type: int32_t type; // data block load type:
int32_t numOfTWindows; int32_t numOfTWindows;
STimeWindow* twindows; STimeWindow* twindows;
int32_t startVersion; int64_t startVersion;
int32_t endVersion; int64_t endVersion;
} SQueryTableDataCond; } SQueryTableDataCond;
void* blockDataDestroy(SSDataBlock* pBlock); void* blockDataDestroy(SSDataBlock* pBlock);

11
include/common/tmsg.h
View File

@ -691,12 +691,12 @@ typedef struct {
typedef STableCfg STableCfgRsp; typedef STableCfg STableCfgRsp;
int32_t tSerializeSTableCfgReq(void *buf, int32_t bufLen, STableCfgReq *pReq); int32_t tSerializeSTableCfgReq(void* buf, int32_t bufLen, STableCfgReq* pReq);
int32_t tDeserializeSTableCfgReq(void *buf, int32_t bufLen, STableCfgReq *pReq); int32_t tDeserializeSTableCfgReq(void* buf, int32_t bufLen, STableCfgReq* pReq);
int32_t tSerializeSTableCfgRsp(void *buf, int32_t bufLen, STableCfgRsp *pRsp); int32_t tSerializeSTableCfgRsp(void* buf, int32_t bufLen, STableCfgRsp* pRsp);
int32_t tDeserializeSTableCfgRsp(void *buf, int32_t bufLen, STableCfgRsp *pRsp); int32_t tDeserializeSTableCfgRsp(void* buf, int32_t bufLen, STableCfgRsp* pRsp);
void tFreeSTableCfgRsp(STableCfgRsp *pRsp); void tFreeSTableCfgRsp(STableCfgRsp* pRsp);
typedef struct { typedef struct {
char db[TSDB_DB_FNAME_LEN]; char db[TSDB_DB_FNAME_LEN];
@ -2664,6 +2664,7 @@ typedef struct {
SMsgHead head; SMsgHead head;
char subKey[TSDB_SUBSCRIBE_KEY_LEN]; char subKey[TSDB_SUBSCRIBE_KEY_LEN];
int8_t withTbName; int8_t withTbName;
int8_t useSnapshot;
int32_t epoch; int32_t epoch;
uint64_t reqId; uint64_t reqId;
int64_t consumerId; int64_t consumerId;

5
include/libs/executor/dataSinkMgt.h
View File

@ -33,7 +33,7 @@ struct SDataSink;
struct SSDataBlock; struct SSDataBlock;
typedef struct SDeleterRes { typedef struct SDeleterRes {
uint64_t uid; uint64_t suid;
SArray* uidList; SArray* uidList;
int64_t skey; int64_t skey;
int64_t ekey; int64_t ekey;
@ -41,7 +41,8 @@ typedef struct SDeleterRes {
} SDeleterRes; } SDeleterRes;
typedef struct SDeleterParam { typedef struct SDeleterParam {
SArray* pUidList; uint64_t suid;
SArray* pUidList;
} SDeleterParam; } SDeleterParam;
typedef struct SDataSinkStat { typedef struct SDataSinkStat {

9
include/libs/executor/executor.h
View File

@ -36,11 +36,13 @@ typedef struct SReadHandle {
void* vnode; void* vnode;
void* mnd; void* mnd;
SMsgCb* pMsgCb; SMsgCb* pMsgCb;
int8_t initTsdbReader;
} SReadHandle; } SReadHandle;
enum { enum {
STREAM_DATA_TYPE_SUBMIT_BLOCK = 1, STREAM_DATA_TYPE_SUBMIT_BLOCK = 1,
STREAM_DATA_TYPE_SSDATA_BLOCK = 2, STREAM_DATA_TYPE_SSDATA_BLOCK = 2,
STREAM_DATA_TYPE_FROM_SNAPSHOT = 3,
}; };
typedef enum { typedef enum {
@ -56,6 +58,13 @@ typedef enum {
*/ */
qTaskInfo_t qCreateStreamExecTaskInfo(void* msg, void* streamReadHandle); qTaskInfo_t qCreateStreamExecTaskInfo(void* msg, void* streamReadHandle);
/**
* Switch the stream scan to snapshot mode
* @param tinfo
* @return
*/
int32_t qStreamScanSnapshot(qTaskInfo_t tinfo);
/** /**
* Set the input data block for the stream scan. * Set the input data block for the stream scan.
* @param tinfo * @param tinfo

2
include/libs/qworker/qworker.h
View File

@ -32,7 +32,7 @@ enum {
}; };
typedef struct SDeleteRes { typedef struct SDeleteRes {
uint64_t uid; uint64_t suid;
SArray* uidList; SArray* uidList;
int64_t skey; int64_t skey;
int64_t ekey; int64_t ekey;

1
include/util/taoserror.h
View File

@ -71,6 +71,7 @@ int32_t* taosGetErrno();
#define TSDB_CODE_OUT_OF_RPC_MEMORY_QUEUE TAOS_DEF_ERROR_CODE(0, 0x0029) #define TSDB_CODE_OUT_OF_RPC_MEMORY_QUEUE TAOS_DEF_ERROR_CODE(0, 0x0029)
#define TSDB_CODE_INVALID_TIMESTAMP TAOS_DEF_ERROR_CODE(0, 0x0030) #define TSDB_CODE_INVALID_TIMESTAMP TAOS_DEF_ERROR_CODE(0, 0x0030)
#define TSDB_CODE_MSG_DECODE_ERROR TAOS_DEF_ERROR_CODE(0, 0x0031) #define TSDB_CODE_MSG_DECODE_ERROR TAOS_DEF_ERROR_CODE(0, 0x0031)
#define TSDB_CODE_NO_AVAIL_DISK TAOS_DEF_ERROR_CODE(0, 0x0032)
#define TSDB_CODE_REF_NO_MEMORY TAOS_DEF_ERROR_CODE(0, 0x0040) #define TSDB_CODE_REF_NO_MEMORY TAOS_DEF_ERROR_CODE(0, 0x0040)
#define TSDB_CODE_REF_FULL TAOS_DEF_ERROR_CODE(0, 0x0041) #define TSDB_CODE_REF_FULL TAOS_DEF_ERROR_CODE(0, 0x0041)

17
source/client/src/tmq.c
View File

@ -54,6 +54,7 @@ struct tmq_conf_t {
int8_t autoCommit; int8_t autoCommit;
int8_t resetOffset; int8_t resetOffset;
int8_t withTbName; int8_t withTbName;
int8_t useSnapshot;
uint16_t port; uint16_t port;
int32_t autoCommitInterval; int32_t autoCommitInterval;
char* ip; char* ip;
@ -69,6 +70,7 @@ struct tmq_t {
char groupId[TSDB_CGROUP_LEN]; char groupId[TSDB_CGROUP_LEN];
char clientId[256]; char clientId[256];
int8_t withTbName; int8_t withTbName;
int8_t useSnapshot;
int8_t autoCommit; int8_t autoCommit;
int32_t autoCommitInterval; int32_t autoCommitInterval;
int32_t resetOffsetCfg; int32_t resetOffsetCfg;
@ -282,6 +284,18 @@ tmq_conf_res_t tmq_conf_set(tmq_conf_t* conf, const char* key, const char* value
} }
} }
if (strcmp(key, "experiment.use.snapshot") == 0) {
if (strcmp(value, "true") == 0) {
conf->useSnapshot = true;
return TMQ_CONF_OK;
} else if (strcmp(value, "false") == 0) {
conf->useSnapshot = false;
return TMQ_CONF_OK;
} else {
return TMQ_CONF_INVALID;
}
}
if (strcmp(key, "td.connect.ip") == 0) { if (strcmp(key, "td.connect.ip") == 0) {
conf->ip = strdup(value); conf->ip = strdup(value);
return TMQ_CONF_OK; return TMQ_CONF_OK;
@ -953,6 +967,7 @@ tmq_t* tmq_consumer_new(tmq_conf_t* conf, char* errstr, int32_t errstrLen) {
strcpy(pTmq->clientId, conf->clientId); strcpy(pTmq->clientId, conf->clientId);
strcpy(pTmq->groupId, conf->groupId); strcpy(pTmq->groupId, conf->groupId);
pTmq->withTbName = conf->withTbName; pTmq->withTbName = conf->withTbName;
pTmq->useSnapshot = conf->useSnapshot;
pTmq->autoCommit = conf->autoCommit; pTmq->autoCommit = conf->autoCommit;
pTmq->autoCommitInterval = conf->autoCommitInterval; pTmq->autoCommitInterval = conf->autoCommitInterval;
pTmq->commitCb = conf->commitCb; pTmq->commitCb = conf->commitCb;
@ -1534,6 +1549,8 @@ SMqPollReq* tmqBuildConsumeReqImpl(tmq_t* tmq, int64_t timeout, SMqClientTopic*
pReq->currentOffset = reqOffset; pReq->currentOffset = reqOffset;
pReq->reqId = generateRequestId(); pReq->reqId = generateRequestId();
pReq->useSnapshot = tmq->useSnapshot;
pReq->head.vgId = htonl(pVg->vgId); pReq->head.vgId = htonl(pVg->vgId);
pReq->head.contLen = htonl(sizeof(SMqPollReq)); pReq->head.contLen = htonl(sizeof(SMqPollReq));
return pReq; return pReq;

35
source/dnode/mnode/impl/src/mndDnode.c
View File

@ -648,30 +648,30 @@ _OVER:
} }
static int32_t mndProcessDropDnodeReq(SRpcMsg *pReq) { static int32_t mndProcessDropDnodeReq(SRpcMsg *pReq) {
SMnode *pMnode = pReq->info.node; SMnode *pMnode = pReq->info.node;
int32_t code = -1; int32_t code = -1;
SDnodeObj *pDnode = NULL; SDnodeObj *pDnode = NULL;
SMnodeObj *pMObj = NULL; SMnodeObj *pMObj = NULL;
SQnodeObj *pQObj = NULL; SQnodeObj *pQObj = NULL;
SSnodeObj *pSObj = NULL; SSnodeObj *pSObj = NULL;
SMDropMnodeReq dropReq = {0}; SDropDnodeReq dropReq = {0};
if (tDeserializeSCreateDropMQSBNodeReq(pReq->pCont, pReq->contLen, &dropReq) != 0) { if (tDeserializeSDropDnodeReq(pReq->pCont, pReq->contLen, &dropReq) != 0) {
terrno = TSDB_CODE_INVALID_MSG; terrno = TSDB_CODE_INVALID_MSG;
goto _OVER; goto _OVER;
} }
mInfo("dnode:%d, start to drop", dropReq.dnodeId); mInfo("dnode:%d, start to drop, ep:%s:%d", dropReq.dnodeId, dropReq.fqdn, dropReq.port);
if (dropReq.dnodeId <= 0) {
terrno = TSDB_CODE_MND_INVALID_DNODE_ID;
goto _OVER;
}
pDnode = mndAcquireDnode(pMnode, dropReq.dnodeId); pDnode = mndAcquireDnode(pMnode, dropReq.dnodeId);
if (pDnode == NULL) { if (pDnode == NULL) {
terrno = TSDB_CODE_MND_DNODE_NOT_EXIST; char ep[TSDB_EP_LEN + 1] = {0};
goto _OVER; snprintf(ep, sizeof(ep), dropReq.fqdn, dropReq.port);
pDnode = mndAcquireDnodeByEp(pMnode, ep);
if (pDnode == NULL) {
terrno = TSDB_CODE_MND_DNODE_NOT_EXIST;
goto _OVER;
}
} }
pQObj = mndAcquireQnode(pMnode, dropReq.dnodeId); pQObj = mndAcquireQnode(pMnode, dropReq.dnodeId);
@ -726,6 +726,7 @@ static int32_t mndProcessConfigDnodeReq(SRpcMsg *pReq) {
return -1; return -1;
} }
mInfo("dnode:%d, start to config, option:%s, value:%s", cfgReq.dnodeId, cfgReq.config, cfgReq.value);
SDnodeObj *pDnode = mndAcquireDnode(pMnode, cfgReq.dnodeId); SDnodeObj *pDnode = mndAcquireDnode(pMnode, cfgReq.dnodeId);
if (pDnode == NULL) { if (pDnode == NULL) {
mError("dnode:%d, failed to config since %s ", cfgReq.dnodeId, terrstr()); mError("dnode:%d, failed to config since %s ", cfgReq.dnodeId, terrstr());
@ -742,7 +743,7 @@ static int32_t mndProcessConfigDnodeReq(SRpcMsg *pReq) {
tSerializeSMCfgDnodeReq(pBuf, bufLen, &cfgReq); tSerializeSMCfgDnodeReq(pBuf, bufLen, &cfgReq);
mDebug("dnode:%d, send config req to dnode, app:%p", cfgReq.dnodeId, pReq->info.ahandle); mDebug("dnode:%d, send config req to dnode, app:%p", cfgReq.dnodeId, pReq->info.ahandle);
SRpcMsg rpcMsg = {.msgType = TDMT_DND_CONFIG_DNODE, .pCont = pBuf, .contLen = bufLen, .info = pReq->info}; SRpcMsg rpcMsg = {.msgType = TDMT_DND_CONFIG_DNODE, .pCont = pBuf, .contLen = bufLen};
return tmsgSendReq(&epSet, &rpcMsg); return tmsgSendReq(&epSet, &rpcMsg);
} }

2
source/dnode/mnode/impl/src/mndScheduler.c
View File

@ -509,7 +509,7 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
SStreamTask* pTask = tNewSStreamTask(pStream->uid); SStreamTask* pTask = tNewSStreamTask(pStream->uid);
mndAddTaskToTaskSet(taskOneLevel, pTask); mndAddTaskToTaskSet(taskOneLevel, pTask);
// input // source
pTask->isDataScan = 1; pTask->isDataScan = 1;
// trigger // trigger

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

@ -29,6 +29,7 @@ target_sources(
# sma # sma
"src/sma/sma.c" "src/sma/sma.c"
"src/sma/smaEnv.c" "src/sma/smaEnv.c"
"src/sma/smaUtil.c"
"src/sma/smaOpen.c" "src/sma/smaOpen.c"
"src/sma/smaRollup.c" "src/sma/smaRollup.c"
"src/sma/smaTimeRange.c" "src/sma/smaTimeRange.c"

120
source/dnode/vnode/src/inc/sma.h
View File

@ -34,7 +34,8 @@ extern "C" {
typedef struct SSmaEnv SSmaEnv; typedef struct SSmaEnv SSmaEnv;
typedef struct SSmaStat SSmaStat; typedef struct SSmaStat SSmaStat;
typedef struct SSmaStatItem SSmaStatItem; typedef struct STSmaStat STSmaStat;
typedef struct SRSmaStat SRSmaStat;
typedef struct SSmaKey SSmaKey; typedef struct SSmaKey SSmaKey;
typedef struct SRSmaInfo SRSmaInfo; typedef struct SRSmaInfo SRSmaInfo;
typedef struct SRSmaInfoItem SRSmaInfoItem; typedef struct SRSmaInfoItem SRSmaInfoItem;
@ -45,26 +46,38 @@ struct SSmaEnv {
SSmaStat *pStat; SSmaStat *pStat;
}; };
#define SMA_ENV_LOCK(env) ((env)->lock) #define SMA_ENV_LOCK(env) ((env)->lock)
#define SMA_ENV_TYPE(env) ((env)->type) #define SMA_ENV_TYPE(env) ((env)->type)
#define SMA_ENV_STAT(env) ((env)->pStat) #define SMA_ENV_STAT(env) ((env)->pStat)
#define SMA_ENV_STAT_ITEM(env) ((env)->pStat->tsmaStatItem)
struct SSmaStatItem { struct STSmaStat {
int8_t state; // ETsdbSmaStat int8_t state; // ETsdbSmaStat
STSma *pTSma; // cache schema STSma *pTSma; // cache schema
STSchema *pTSchema; STSchema *pTSchema;
}; };
struct SRSmaStat {
SSma *pSma;
void *tmrHandle;
tmr_h tmrId;
int8_t tmrStat;
int32_t tmrSeconds;
SHashObj *rsmaInfoHash; // key: stbUid, value: SRSmaInfo;
};
struct SSmaStat { struct SSmaStat {
union { union {
SSmaStatItem tsmaStatItem; STSmaStat tsmaStat; // time-range-wise sma
SHashObj *rsmaInfoHash; // key: stbUid, value: SRSmaInfo; SRSmaStat rsmaStat; // rollup sma
}; };
T_REF_DECLARE() T_REF_DECLARE()
}; };
#define SMA_STAT_ITEM(s) ((s)->tsmaStatItem) #define SMA_TSMA_STAT(s) (&(s)->tsmaStat)
#define SMA_STAT_INFO_HASH(s) ((s)->rsmaInfoHash) #define SMA_RSMA_STAT(s) (&(s)->rsmaStat)
#define SMA_RSMA_INFO_HASH(s) ((s)->rsmaStat.rsmaInfoHash)
#define SMA_RSMA_TMR_HANDLE(s) ((s)->rsmaStat.tmrHandle)
#define SMA_RSMA_TMR_STAT(s) ((s)->rsmaStat.tmrStat)
#define RSMA_INFO_HASH(r) ((r)->rsmaInfoHash)
void tdDestroySmaEnv(SSmaEnv *pSmaEnv); void tdDestroySmaEnv(SSmaEnv *pSmaEnv);
void *tdFreeSmaEnv(SSmaEnv *pSmaEnv); void *tdFreeSmaEnv(SSmaEnv *pSmaEnv);
@ -107,53 +120,51 @@ static FORCE_INLINE int32_t tdUnLockSmaEnv(SSmaEnv *pEnv) {
return 0; return 0;
} }
static FORCE_INLINE int8_t tdSmaStat(SSmaStatItem *pStatItem) { static FORCE_INLINE int8_t tdSmaStat(STSmaStat *pTStat) {
if (pStatItem) { if (pTStat) {
return atomic_load_8(&pStatItem->state); return atomic_load_8(&pTStat->state);
} }
return TSDB_SMA_STAT_UNKNOWN; return TSDB_SMA_STAT_UNKNOWN;
} }
static FORCE_INLINE bool tdSmaStatIsOK(SSmaStatItem *pStatItem, int8_t *state) { static FORCE_INLINE bool tdSmaStatIsOK(STSmaStat *pTStat, int8_t *state) {
if (!pStatItem) { if (!pTStat) {
return false; return false;
} }
if (state) { if (state) {
*state = atomic_load_8(&pStatItem->state); *state = atomic_load_8(&pTStat->state);
return *state == TSDB_SMA_STAT_OK; return *state == TSDB_SMA_STAT_OK;
} }
return atomic_load_8(&pStatItem->state) == TSDB_SMA_STAT_OK; return atomic_load_8(&pTStat->state) == TSDB_SMA_STAT_OK;
} }
static FORCE_INLINE bool tdSmaStatIsExpired(SSmaStatItem *pStatItem) { static FORCE_INLINE bool tdSmaStatIsExpired(STSmaStat *pTStat) {
return pStatItem ? (atomic_load_8(&pStatItem->state) & TSDB_SMA_STAT_EXPIRED) : true; return pTStat ? (atomic_load_8(&pTStat->state) & TSDB_SMA_STAT_EXPIRED) : true;
} }
static FORCE_INLINE bool tdSmaStatIsDropped(SSmaStatItem *pStatItem) { static FORCE_INLINE bool tdSmaStatIsDropped(STSmaStat *pTStat) {
return pStatItem ? (atomic_load_8(&pStatItem->state) & TSDB_SMA_STAT_DROPPED) : true; return pTStat ? (atomic_load_8(&pTStat->state) & TSDB_SMA_STAT_DROPPED) : true;
} }
static FORCE_INLINE void tdSmaStatSetOK(SSmaStatItem *pStatItem) { static FORCE_INLINE void tdSmaStatSetOK(STSmaStat *pTStat) {
if (pStatItem) { if (pTStat) {
atomic_store_8(&pStatItem->state, TSDB_SMA_STAT_OK); atomic_store_8(&pTStat->state, TSDB_SMA_STAT_OK);
} }
} }
static FORCE_INLINE void tdSmaStatSetExpired(SSmaStatItem *pStatItem) { static FORCE_INLINE void tdSmaStatSetExpired(STSmaStat *pTStat) {
if (pStatItem) { if (pTStat) {
atomic_or_fetch_8(&pStatItem->state, TSDB_SMA_STAT_EXPIRED); atomic_or_fetch_8(&pTStat->state, TSDB_SMA_STAT_EXPIRED);
} }
} }
static FORCE_INLINE void tdSmaStatSetDropped(SSmaStatItem *pStatItem) { static FORCE_INLINE void tdSmaStatSetDropped(STSmaStat *pTStat) {
if (pStatItem) { if (pTStat) {
atomic_or_fetch_8(&pStatItem->state, TSDB_SMA_STAT_DROPPED); atomic_or_fetch_8(&pTStat->state, TSDB_SMA_STAT_DROPPED);
} }
} }
static int32_t tdInitSmaStat(SSmaStat **pSmaStat, int8_t smaType);
void *tdFreeSmaStatItem(SSmaStatItem *pSmaStatItem);
static int32_t tdDestroySmaState(SSmaStat *pSmaStat, int8_t smaType); static int32_t tdDestroySmaState(SSmaStat *pSmaStat, int8_t smaType);
void *tdFreeSmaState(SSmaStat *pSmaStat, int8_t smaType); void *tdFreeSmaState(SSmaStat *pSmaStat, int8_t smaType);
@ -163,6 +174,51 @@ int32_t tdProcessTSmaCreateImpl(SSma *pSma, int64_t version, const char *pMsg);
int32_t tdProcessTSmaInsertImpl(SSma *pSma, int64_t indexUid, const char *msg); int32_t tdProcessTSmaInsertImpl(SSma *pSma, int64_t indexUid, const char *msg);
int32_t tdProcessTSmaGetDaysImpl(SVnodeCfg *pCfg, void *pCont, uint32_t contLen, int32_t *days); int32_t tdProcessTSmaGetDaysImpl(SVnodeCfg *pCfg, void *pCont, uint32_t contLen, int32_t *days);
typedef struct STFInfo STFInfo;
typedef struct STFile STFile;
struct STFInfo {
uint32_t magic;
uint32_t ftype;
uint32_t fver;
uint64_t fsize;
};
struct STFile {
STFInfo info;
STfsFile f;
TdFilePtr pFile;
uint8_t state;
};
#define TD_FILE_F(tf) (&((tf)->f))
#define TD_FILE_PFILE(tf) ((tf)->pFile)
#define TD_FILE_OPENED(tf) (TD_FILE_PFILE(tf) != NULL)
#define TD_FILE_FULL_NAME(tf) (TD_FILE_F(tf)->aname)
#define TD_FILE_REL_NAME(tf) (TD_FILE_F(tf)->rname)
#define TD_FILE_OPENED(tf) (TD_FILE_PFILE(tf) != NULL)
#define TD_FILE_CLOSED(tf) (!TD_FILE_OPENED(tf))
#define TD_FILE_SET_CLOSED(f) (TD_FILE_PFILE(f) = NULL)
#define TD_FILE_STATE(tf) ((tf)->state)
#define TD_FILE_SET_STATE(tf, s) ((tf)->state = (s))
#define TD_FILE_DID(tf) (TD_FILE_F(tf)->did)
#define TD_FILE_IS_OK(tf) (TD_FILE_STATE(tf) == TD_FILE_STATE_OK)
#define TD_FILE_IS_BAD(tf) (TD_FILE_STATE(tf) == TD_FILE_STATE_BAD)
int32_t tdInitTFile(STFile *pTFile, STfs *pTfs, const char *fname);
int32_t tdCreateTFile(STFile *pTFile, STfs *pTfs, bool updateHeader, int8_t fType);
int32_t tdOpenTFile(STFile *pTFile, int flags);
int64_t tdReadTFile(STFile *pTFile, void *buf, int64_t nbyte);
int64_t tdSeekTFile(STFile *pTFile, int64_t offset, int whence);
int64_t tdWriteTFile(STFile *pTFile, void *buf, int64_t nbyte);
int64_t tdAppendTFile(STFile *pTFile, void *buf, int64_t nbyte, int64_t *offset);
int32_t tdRemoveTFile(STFile *pTFile);
int32_t tdLoadTFileHeader(STFile *pTFile, STFInfo *pInfo);
int32_t tdUpdateTFileHeader(STFile *pTFile);
void tdUpdateTFileMagic(STFile *pTFile, void *pCksm);
void tdCloseTFile(STFile *pTFile);
void tdGetVndFileName(int32_t vid, const char *dname, const char *fname, char *outputName);
#ifdef __cplusplus #ifdef __cplusplus
} }
#endif #endif

1
source/dnode/vnode/src/inc/tq.h
View File

@ -150,6 +150,7 @@ int64_t tqFetchLog(STQ* pTq, STqHandle* pHandle, int64_t* fetchOffset, SWalHead*
// tqExec // tqExec
int32_t tqDataExec(STQ* pTq, STqExecHandle* pExec, SSubmitReq* pReq, SMqDataBlkRsp* pRsp, int32_t workerId); int32_t tqDataExec(STQ* pTq, STqExecHandle* pExec, SSubmitReq* pReq, SMqDataBlkRsp* pRsp, int32_t workerId);
int32_t tqScanSnapshot(STQ* pTq, const STqExecHandle* pExec, SMqDataBlkRsp* pRsp, int32_t workerId);
int32_t tqSendPollRsp(STQ* pTq, const SRpcMsg* pMsg, const SMqPollReq* pReq, const SMqDataBlkRsp* pRsp); int32_t tqSendPollRsp(STQ* pTq, const SRpcMsg* pMsg, const SMqPollReq* pReq, const SMqDataBlkRsp* pRsp);
// tqMeta // tqMeta

1
source/dnode/vnode/src/inc/vnodeInt.h
View File

@ -247,7 +247,6 @@ struct SVnode {
struct STbUidStore { struct STbUidStore {
tb_uid_t suid; tb_uid_t suid;
tb_uid_t uid; // TODO: just for debugging, remove when uid provided in SSDataBlock
SArray* tbUids; SArray* tbUids;
SHashObj* uidHash; SHashObj* uidHash;
}; };

4
source/dnode/vnode/src/meta/metaQuery.c
View File

@ -264,8 +264,8 @@ struct SMCtbCursor {
SMCtbCursor *metaOpenCtbCursor(SMeta *pMeta, tb_uid_t uid) { SMCtbCursor *metaOpenCtbCursor(SMeta *pMeta, tb_uid_t uid) {
SMCtbCursor *pCtbCur = NULL; SMCtbCursor *pCtbCur = NULL;
SCtbIdxKey ctbIdxKey; SCtbIdxKey ctbIdxKey;
int ret; int ret = 0;
int c; int c = 0;
pCtbCur = (SMCtbCursor *)taosMemoryCalloc(1, sizeof(*pCtbCur)); pCtbCur = (SMCtbCursor *)taosMemoryCalloc(1, sizeof(*pCtbCur));
if (pCtbCur == NULL) { if (pCtbCur == NULL) {

49
source/dnode/vnode/src/sma/smaEnv.c
View File

@ -21,9 +21,10 @@ typedef struct SSmaStat SSmaStat;
// declaration of static functions // declaration of static functions
static int32_t tdInitSmaStat(SSmaStat **pSmaStat, int8_t smaType); static int32_t tdInitSmaStat(SSmaStat **pSmaStat, int8_t smaType, const SSma *pSma);
static SSmaEnv *tdNewSmaEnv(const SSma *pSma, int8_t smaType, const char *path); static SSmaEnv *tdNewSmaEnv(const SSma *pSma, int8_t smaType, const char *path);
static int32_t tdInitSmaEnv(SSma *pSma, int8_t smaType, const char *path, SSmaEnv **pEnv); static int32_t tdInitSmaEnv(SSma *pSma, int8_t smaType, const char *path, SSmaEnv **pEnv);
static void *tdFreeTSmaStat(STSmaStat *pStat);
// implementation // implementation
@ -45,7 +46,7 @@ static SSmaEnv *tdNewSmaEnv(const SSma *pSma, int8_t smaType, const char *path)
return NULL; return NULL;
} }
if (tdInitSmaStat(&SMA_ENV_STAT(pEnv), smaType) != TSDB_CODE_SUCCESS) { if (tdInitSmaStat(&SMA_ENV_STAT(pEnv), smaType, pSma) != TSDB_CODE_SUCCESS) {
tdFreeSmaEnv(pEnv); tdFreeSmaEnv(pEnv);
return NULL; return NULL;
} }
@ -105,7 +106,7 @@ int32_t tdUnRefSmaStat(SSma *pSma, SSmaStat *pStat) {
return 0; return 0;
} }
static int32_t tdInitSmaStat(SSmaStat **pSmaStat, int8_t smaType) { static int32_t tdInitSmaStat(SSmaStat **pSmaStat, int8_t smaType, const SSma *pSma) {
ASSERT(pSmaStat != NULL); ASSERT(pSmaStat != NULL);
if (*pSmaStat) { // no lock if (*pSmaStat) { // no lock
@ -125,10 +126,23 @@ static int32_t tdInitSmaStat(SSmaStat **pSmaStat, int8_t smaType) {
} }
if (smaType == TSDB_SMA_TYPE_ROLLUP) { if (smaType == TSDB_SMA_TYPE_ROLLUP) {
SMA_STAT_INFO_HASH(*pSmaStat) = taosHashInit( SMA_RSMA_STAT(*pSmaStat)->pSma = (SSma*)pSma;
RSMA_TASK_INFO_HASH_SLOT, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_ENTRY_LOCK); // init timer
SMA_RSMA_TMR_HANDLE(*pSmaStat) = taosTmrInit(10000, 100, 10000, "RSMA_G");
if (!SMA_RSMA_TMR_HANDLE(*pSmaStat)) {
taosMemoryFreeClear(*pSmaStat);
return TSDB_CODE_FAILED;
}
if (!SMA_STAT_INFO_HASH(*pSmaStat)) { atomic_store_8(&SMA_RSMA_TMR_STAT(*pSmaStat), TASK_TRIGGER_STATUS__ACTIVE);
// init hash
SMA_RSMA_INFO_HASH(*pSmaStat) = taosHashInit(
RSMA_TASK_INFO_HASH_SLOT, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_ENTRY_LOCK);
if (!SMA_RSMA_INFO_HASH(*pSmaStat)) {
if (SMA_RSMA_TMR_HANDLE(*pSmaStat)) {
taosTmrCleanUp(SMA_RSMA_TMR_HANDLE(*pSmaStat));
}
taosMemoryFreeClear(*pSmaStat); taosMemoryFreeClear(*pSmaStat);
return TSDB_CODE_FAILED; return TSDB_CODE_FAILED;
} }
@ -141,16 +155,16 @@ static int32_t tdInitSmaStat(SSmaStat **pSmaStat, int8_t smaType) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
void *tdFreeSmaStatItem(SSmaStatItem *pSmaStatItem) { static void *tdFreeTSmaStat(STSmaStat *pStat) {
if (pSmaStatItem) { if (pStat) {
tDestroyTSma(pSmaStatItem->pTSma); tDestroyTSma(pStat->pTSma);
taosMemoryFreeClear(pSmaStatItem->pTSma); taosMemoryFreeClear(pStat->pTSma);
taosMemoryFreeClear(pSmaStatItem); taosMemoryFreeClear(pStat);
} }
return NULL; return NULL;
} }
void* tdFreeSmaState(SSmaStat *pSmaStat, int8_t smaType) { void *tdFreeSmaState(SSmaStat *pSmaStat, int8_t smaType) {
tdDestroySmaState(pSmaStat, smaType); tdDestroySmaState(pSmaStat, smaType);
taosMemoryFreeClear(pSmaStat); taosMemoryFreeClear(pSmaStat);
return NULL; return NULL;
@ -165,16 +179,19 @@ void* tdFreeSmaState(SSmaStat *pSmaStat, int8_t smaType) {
int32_t tdDestroySmaState(SSmaStat *pSmaStat, int8_t smaType) { int32_t tdDestroySmaState(SSmaStat *pSmaStat, int8_t smaType) {
if (pSmaStat) { if (pSmaStat) {
if (smaType == TSDB_SMA_TYPE_TIME_RANGE) { if (smaType == TSDB_SMA_TYPE_TIME_RANGE) {
tdFreeSmaStatItem(&pSmaStat->tsmaStatItem); tdFreeTSmaStat(&pSmaStat->tsmaStat);
} else if (smaType == TSDB_SMA_TYPE_ROLLUP) { } else if (smaType == TSDB_SMA_TYPE_ROLLUP) {
if (SMA_RSMA_TMR_HANDLE(pSmaStat)) {
taosTmrCleanUp(SMA_RSMA_TMR_HANDLE(pSmaStat));
}
// TODO: use taosHashSetFreeFp when taosHashSetFreeFp is ready. // TODO: use taosHashSetFreeFp when taosHashSetFreeFp is ready.
void *infoHash = taosHashIterate(SMA_STAT_INFO_HASH(pSmaStat), NULL); void *infoHash = taosHashIterate(SMA_RSMA_INFO_HASH(pSmaStat), NULL);
while (infoHash) { while (infoHash) {
SRSmaInfo *pInfoHash = *(SRSmaInfo **)infoHash; SRSmaInfo *pInfoHash = *(SRSmaInfo **)infoHash;
tdFreeRSmaInfo(pInfoHash); tdFreeRSmaInfo(pInfoHash);
infoHash = taosHashIterate(SMA_STAT_INFO_HASH(pSmaStat), infoHash); infoHash = taosHashIterate(SMA_RSMA_INFO_HASH(pSmaStat), infoHash);
} }
taosHashCleanup(SMA_STAT_INFO_HASH(pSmaStat)); taosHashCleanup(SMA_RSMA_INFO_HASH(pSmaStat));
} else { } else {
ASSERT(0); ASSERT(0);
} }

13
source/dnode/vnode/src/sma/smaOpen.c
View File

@ -138,3 +138,16 @@ int32_t smaClose(SSma *pSma) {
} }
return 0; return 0;
} }
/**
* @brief rsma env restore
*
* @param pSma
* @return int32_t
*/
int32_t smaRestore(SSma *pSma) {
if (!pSma) return 0;
// iterate all stables to restore the rsma env
return TSDB_CODE_SUCCESS;
}

248
source/dnode/vnode/src/sma/smaRollup.c
View File

@ -16,35 +16,17 @@
#include "sma.h" #include "sma.h"
#include "tstream.h" #include "tstream.h"
static FORCE_INLINE int32_t tdUidStorePut(STbUidStore *pStore, tb_uid_t suid, tb_uid_t *uid); typedef enum { TD_QTASK_TMP_FILE = 0, TD_QTASK_CUR_FILE } TD_QTASK_FILE_T;
static FORCE_INLINE int32_t tdUpdateTbUidListImpl(SSma *pSma, tb_uid_t *suid, SArray *tbUids); static const char *tdQTaskInfoFname[] = {"qtaskinfo.t", "qtaskinfo"};
static FORCE_INLINE int32_t tdExecuteRSmaImpl(SSma *pSma, const void *pMsg, int32_t inputType, SRSmaInfoItem *rsmaItem,
tb_uid_t suid, int8_t level);
#define SET_RSMA_INFO_ITEM_PARAMS(__idx, __level) \ static int32_t tdUidStorePut(STbUidStore *pStore, tb_uid_t suid, tb_uid_t *uid);
if (param->qmsg[__idx]) { \ static int32_t tdUpdateTbUidListImpl(SSma *pSma, tb_uid_t *suid, SArray *tbUids);
pRSmaInfo->items[__idx].pRsmaInfo = pRSmaInfo; \ static int32_t tdSetRSmaInfoItemParams(SSma *pSma, SRSmaParam *param, SRSmaInfo *pRSmaInfo, SReadHandle *handle,
pRSmaInfo->items[__idx].taskInfo = qCreateStreamExecTaskInfo(param->qmsg[0], &handle); \ int8_t idx);
if (!pRSmaInfo->items[__idx].taskInfo) { \ static int32_t tdExecuteRSmaImpl(SSma *pSma, const void *pMsg, int32_t inputType, SRSmaInfoItem *rsmaItem,
goto _err; \ tb_uid_t suid, int8_t level);
} \ static void tdRSmaFetchTrigger(void *param, void *tmrId);
pRSmaInfo->items[__idx].triggerStatus = TASK_TRIGGER_STATUS__IN_ACTIVE; \ static void tdRSmaPersistTrigger(void *param, void *tmrId);
if (param->maxdelay[__idx] < 1) { \
int64_t msInterval = \
convertTimeFromPrecisionToUnit(pRetention[__level].freq, pTsdbCfg->precision, TIME_UNIT_MILLISECOND); \
pRSmaInfo->items[__idx].maxDelay = msInterval; \
} else { \
pRSmaInfo->items[__idx].maxDelay = param->maxdelay[__idx]; \
} \
if (pRSmaInfo->items[__idx].maxDelay > TSDB_MAX_ROLLUP_MAX_DELAY) { \
pRSmaInfo->items[__idx].maxDelay = TSDB_MAX_ROLLUP_MAX_DELAY; \
} \
pRSmaInfo->items[__idx].level = TSDB_RETENTION_L##__level; \
pRSmaInfo->items[__idx].tmrHandle = taosTmrInit(10000, 100, 10000, "RSMA"); \
if (!pRSmaInfo->items[__idx].tmrHandle) { \
goto _err; \
} \
}
struct SRSmaInfoItem { struct SRSmaInfoItem {
SRSmaInfo *pRsmaInfo; SRSmaInfo *pRsmaInfo;
@ -56,14 +38,6 @@ struct SRSmaInfoItem {
int8_t triggerStatus; // TASK_TRIGGER_STATUS__IN_ACTIVE/TASK_TRIGGER_STATUS__ACTIVE int8_t triggerStatus; // TASK_TRIGGER_STATUS__IN_ACTIVE/TASK_TRIGGER_STATUS__ACTIVE
int32_t maxDelay; int32_t maxDelay;
}; };
typedef struct {
int64_t suid;
SRSmaInfoItem *pItem;
SSma *pSma;
STSchema *pTSchema;
} SRSmaTriggerParam;
struct SRSmaInfo { struct SRSmaInfo {
STSchema *pTSchema; STSchema *pTSchema;
SSma *pSma; SSma *pSma;
@ -81,7 +55,7 @@ static FORCE_INLINE void tdFreeTaskHandle(qTaskInfo_t *taskHandle) {
void *tdFreeRSmaInfo(SRSmaInfo *pInfo) { void *tdFreeRSmaInfo(SRSmaInfo *pInfo) {
if (pInfo) { if (pInfo) {
for (int32_t i = 0; i < TSDB_RETENTION_MAX; ++i) { for (int32_t i = 0; i < TSDB_RETENTION_L2; ++i) {
SRSmaInfoItem *pItem = &pInfo->items[i]; SRSmaInfoItem *pItem = &pInfo->items[i];
if (pItem->taskInfo) { if (pItem->taskInfo) {
tdFreeTaskHandle(pItem->taskInfo); tdFreeTaskHandle(pItem->taskInfo);
@ -118,7 +92,7 @@ static FORCE_INLINE int32_t tdUpdateTbUidListImpl(SSma *pSma, tb_uid_t *suid, SA
return TSDB_CODE_FAILED; return TSDB_CODE_FAILED;
} }
pRSmaInfo = taosHashGet(SMA_STAT_INFO_HASH(pStat), suid, sizeof(tb_uid_t)); pRSmaInfo = taosHashGet(SMA_RSMA_INFO_HASH(pStat), suid, sizeof(tb_uid_t));
if (!pRSmaInfo || !(pRSmaInfo = *(SRSmaInfo **)pRSmaInfo)) { if (!pRSmaInfo || !(pRSmaInfo = *(SRSmaInfo **)pRSmaInfo)) {
smaError("vgId:%d, failed to get rsma info for uid:%" PRIi64, SMA_VID(pSma), *suid); smaError("vgId:%d, failed to get rsma info for uid:%" PRIi64, SMA_VID(pSma), *suid);
terrno = TSDB_CODE_RSMA_INVALID_STAT; terrno = TSDB_CODE_RSMA_INVALID_STAT;
@ -187,7 +161,7 @@ int32_t tdFetchTbUidList(SSma *pSma, STbUidStore **ppStore, tb_uid_t suid, tb_ui
SSmaStat *pStat = SMA_ENV_STAT(pEnv); SSmaStat *pStat = SMA_ENV_STAT(pEnv);
SHashObj *infoHash = NULL; SHashObj *infoHash = NULL;
if (!pStat || !(infoHash = SMA_STAT_INFO_HASH(pStat))) { if (!pStat || !(infoHash = SMA_RSMA_INFO_HASH(pStat))) {
terrno = TSDB_CODE_RSMA_INVALID_STAT; terrno = TSDB_CODE_RSMA_INVALID_STAT;
return TSDB_CODE_FAILED; return TSDB_CODE_FAILED;
} }
@ -213,6 +187,40 @@ int32_t tdFetchTbUidList(SSma *pSma, STbUidStore **ppStore, tb_uid_t suid, tb_ui
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static int32_t tdSetRSmaInfoItemParams(SSma *pSma, SRSmaParam *param, SRSmaInfo *pRSmaInfo, SReadHandle *pReadHandle,
int8_t idx) {
SRetention *pRetention = SMA_RETENTION(pSma);
STsdbCfg *pTsdbCfg = SMA_TSDB_CFG(pSma);
if (param->qmsg[idx]) {
SRSmaInfoItem *pItem = &(pRSmaInfo->items[idx]);
pItem->pRsmaInfo = pRSmaInfo;
pItem->taskInfo = qCreateStreamExecTaskInfo(param->qmsg[0], pReadHandle);
if (!pItem->taskInfo) {
goto _err;
}
pItem->triggerStatus = TASK_TRIGGER_STATUS__IN_ACTIVE;
if (param->maxdelay[idx] < TSDB_MIN_ROLLUP_MAX_DELAY) {
int64_t msInterval =
convertTimeFromPrecisionToUnit(pRetention[idx + 1].freq, pTsdbCfg->precision, TIME_UNIT_MILLISECOND);
pItem->maxDelay = (int32_t)msInterval;
} else {
pItem->maxDelay = (int32_t)param->maxdelay[idx];
}
if (pItem->maxDelay > TSDB_MAX_ROLLUP_MAX_DELAY) {
pItem->maxDelay = TSDB_MAX_ROLLUP_MAX_DELAY;
}
pItem->level = (idx == 0 ? TSDB_RETENTION_L1 : TSDB_RETENTION_L2);
pItem->tmrHandle = taosTmrInit(10000, 100, 10000, "RSMA");
if (!pItem->tmrHandle) {
goto _err;
}
}
return TSDB_CODE_SUCCESS;
_err:
return TSDB_CODE_FAILED;
}
/** /**
* @brief Check and init qTaskInfo_t, only applicable to stable with SRSmaParam. * @brief Check and init qTaskInfo_t, only applicable to stable with SRSmaParam.
* *
@ -246,7 +254,7 @@ int32_t tdProcessRSmaCreate(SVnode *pVnode, SVCreateStbReq *pReq) {
SSmaStat *pStat = SMA_ENV_STAT(pEnv); SSmaStat *pStat = SMA_ENV_STAT(pEnv);
SRSmaInfo *pRSmaInfo = NULL; SRSmaInfo *pRSmaInfo = NULL;
pRSmaInfo = taosHashGet(SMA_STAT_INFO_HASH(pStat), &pReq->suid, sizeof(tb_uid_t)); pRSmaInfo = taosHashGet(SMA_RSMA_INFO_HASH(pStat), &pReq->suid, sizeof(tb_uid_t));
if (pRSmaInfo) { if (pRSmaInfo) {
ASSERT(0); // TODO: free original pRSmaInfo is exists abnormally ASSERT(0); // TODO: free original pRSmaInfo is exists abnormally
smaWarn("vgId:%d, rsma info already exists for stb: %s, %" PRIi64, SMA_VID(pSma), pReq->name, pReq->suid); smaWarn("vgId:%d, rsma info already exists for stb: %s, %" PRIi64, SMA_VID(pSma), pReq->name, pReq->suid);
@ -282,14 +290,14 @@ int32_t tdProcessRSmaCreate(SVnode *pVnode, SVCreateStbReq *pReq) {
pRSmaInfo->pSma = pSma; pRSmaInfo->pSma = pSma;
pRSmaInfo->suid = pReq->suid; pRSmaInfo->suid = pReq->suid;
SRetention *pRetention = SMA_RETENTION(pSma); if (tdSetRSmaInfoItemParams(pSma, param, pRSmaInfo, &handle, 0) < 0) {
STsdbCfg *pTsdbCfg = SMA_TSDB_CFG(pSma); goto _err;
}
if (tdSetRSmaInfoItemParams(pSma, param, pRSmaInfo, &handle, 1) < 0) {
goto _err;
}
SET_RSMA_INFO_ITEM_PARAMS(0, 1); if (taosHashPut(SMA_RSMA_INFO_HASH(pStat), &pReq->suid, sizeof(tb_uid_t), &pRSmaInfo, sizeof(pRSmaInfo)) < 0) {
SET_RSMA_INFO_ITEM_PARAMS(1, 2);
if (taosHashPut(SMA_STAT_INFO_HASH(pStat), &pReq->suid, sizeof(tb_uid_t), &pRSmaInfo, sizeof(pRSmaInfo)) !=
TSDB_CODE_SUCCESS) {
goto _err; goto _err;
} else { } else {
smaDebug("vgId:%d, register rsma info succeed for suid:%" PRIi64, SMA_VID(pSma), pReq->suid); smaDebug("vgId:%d, register rsma info succeed for suid:%" PRIi64, SMA_VID(pSma), pReq->suid);
@ -418,7 +426,6 @@ static int32_t tdFetchSubmitReqSuids(SSubmitReq *pMsg, STbUidStore *pStore) {
if (!pBlock) break; if (!pBlock) break;
tdUidStorePut(pStore, msgIter.suid, NULL); tdUidStorePut(pStore, msgIter.suid, NULL);
pStore->uid = msgIter.uid; // TODO: remove, just for debugging
} }
if (terrno != TSDB_CODE_SUCCESS) return -1; if (terrno != TSDB_CODE_SUCCESS) return -1;
@ -439,8 +446,9 @@ static int32_t tdFetchAndSubmitRSmaResult(SRSmaInfoItem *pItem, int8_t blkType)
if (!output) { if (!output) {
break; break;
} }
if (!pResult) { if (!pResult) {
pResult = taosArrayInit(0, sizeof(SSDataBlock)); pResult = taosArrayInit(1, sizeof(SSDataBlock));
if (!pResult) { if (!pResult) {
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
return TSDB_CODE_FAILED; return TSDB_CODE_FAILED;
@ -451,7 +459,7 @@ static int32_t tdFetchAndSubmitRSmaResult(SRSmaInfoItem *pItem, int8_t blkType)
} }
if (taosArrayGetSize(pResult) > 0) { if (taosArrayGetSize(pResult) > 0) {
#if 0 #if 1
char flag[10] = {0}; char flag[10] = {0};
snprintf(flag, 10, "level %" PRIi8, pItem->level); snprintf(flag, 10, "level %" PRIi8, pItem->level);
blockDebugShowData(pResult, flag); blockDebugShowData(pResult, flag);
@ -459,14 +467,12 @@ static int32_t tdFetchAndSubmitRSmaResult(SRSmaInfoItem *pItem, int8_t blkType)
STsdb *sinkTsdb = (pItem->level == TSDB_RETENTION_L1 ? pSma->pRSmaTsdb1 : pSma->pRSmaTsdb2); STsdb *sinkTsdb = (pItem->level == TSDB_RETENTION_L1 ? pSma->pRSmaTsdb1 : pSma->pRSmaTsdb2);
SSubmitReq *pReq = NULL; SSubmitReq *pReq = NULL;
if (buildSubmitReqFromDataBlock(&pReq, pResult, pRSmaInfo->pTSchema, SMA_VID(pSma), pRSmaInfo->suid) < 0) { if (buildSubmitReqFromDataBlock(&pReq, pResult, pRSmaInfo->pTSchema, SMA_VID(pSma), pRSmaInfo->suid) < 0) {
taosArrayDestroy(pResult); goto _err;
return TSDB_CODE_FAILED;
} }
if (pReq && tdProcessSubmitReq(sinkTsdb, INT64_MAX, pReq) < 0) { if (pReq && tdProcessSubmitReq(sinkTsdb, INT64_MAX, pReq) < 0) {
taosArrayDestroy(pResult);
taosMemoryFreeClear(pReq); taosMemoryFreeClear(pReq);
return TSDB_CODE_FAILED; goto _err;
} }
taosMemoryFreeClear(pReq); taosMemoryFreeClear(pReq);
@ -479,7 +485,10 @@ static int32_t tdFetchAndSubmitRSmaResult(SRSmaInfoItem *pItem, int8_t blkType)
} }
taosArrayDestroy(pResult); taosArrayDestroy(pResult);
return 0; return TSDB_CODE_SUCCESS;
_err:
taosArrayDestroy(pResult);
return TSDB_CODE_FAILED;
} }
/** /**
@ -488,13 +497,12 @@ static int32_t tdFetchAndSubmitRSmaResult(SRSmaInfoItem *pItem, int8_t blkType)
* @param param * @param param
* @param tmrId * @param tmrId
*/ */
static void rsmaTriggerByTimer(void *param, void *tmrId) { static void tdRSmaFetchTrigger(void *param, void *tmrId) {
// SRSmaTriggerParam *pParam = (SRSmaTriggerParam *)param;
// SRSmaInfoItem *pItem = pParam->pItem;
SRSmaInfoItem *pItem = param; SRSmaInfoItem *pItem = param;
if (atomic_load_8(&pItem->triggerStatus) == TASK_TRIGGER_STATUS__ACTIVE) { if (atomic_load_8(&pItem->triggerStatus) == TASK_TRIGGER_STATUS__ACTIVE) {
smaTrace("level %" PRIi8 " status is active for tb suid:%" PRIi64, pItem->level, pItem->pRsmaInfo->suid); smaWarn("%s:%d THREAD:%" PRIi64 " level %" PRIi8 " status is active for tb suid:%" PRIi64, __func__, __LINE__,
taosGetSelfPthreadId(), pItem->level, pItem->pRsmaInfo->suid);
SSDataBlock dataBlock = {.info.type = STREAM_GET_ALL}; SSDataBlock dataBlock = {.info.type = STREAM_GET_ALL};
atomic_store_8(&pItem->triggerStatus, TASK_TRIGGER_STATUS__IN_ACTIVE); atomic_store_8(&pItem->triggerStatus, TASK_TRIGGER_STATUS__IN_ACTIVE);
@ -502,10 +510,11 @@ static void rsmaTriggerByTimer(void *param, void *tmrId) {
tdFetchAndSubmitRSmaResult(pItem, STREAM_DATA_TYPE_SSDATA_BLOCK); tdFetchAndSubmitRSmaResult(pItem, STREAM_DATA_TYPE_SSDATA_BLOCK);
} else { } else {
smaTrace("level %" PRIi8 " status is inactive for tb suid:%" PRIi64, pItem->level, pItem->pRsmaInfo->suid); smaWarn("%s:%d THREAD:%" PRIi64 " level %" PRIi8 " status is inactive for tb suid:%" PRIi64, __func__, __LINE__,
taosGetSelfPthreadId(), pItem->level, pItem->pRsmaInfo->suid);
} }
// taosTmrReset(rsmaTriggerByTimer, pItem->maxDelay, pItem, pItem->tmrHandle, &pItem->tmrId); // taosTmrReset(tdRSmaFetchTrigger, pItem->maxDelay, pItem, pItem->tmrHandle, &pItem->tmrId);
} }
static FORCE_INLINE int32_t tdExecuteRSmaImpl(SSma *pSma, const void *pMsg, int32_t inputType, SRSmaInfoItem *pItem, static FORCE_INLINE int32_t tdExecuteRSmaImpl(SSma *pSma, const void *pMsg, int32_t inputType, SRSmaInfoItem *pItem,
@ -518,16 +527,20 @@ static FORCE_INLINE int32_t tdExecuteRSmaImpl(SSma *pSma, const void *pMsg, int3
smaDebug("vgId:%d, execute rsma %" PRIi8 " task for qTaskInfo:%p suid:%" PRIu64, SMA_VID(pSma), level, smaDebug("vgId:%d, execute rsma %" PRIi8 " task for qTaskInfo:%p suid:%" PRIu64, SMA_VID(pSma), level,
pItem->taskInfo, suid); pItem->taskInfo, suid);
// inputType = STREAM_DATA_TYPE_SUBMIT_BLOCK(1) if (qSetStreamInput(pItem->taskInfo, pMsg, inputType, true) < 0) { // STREAM_DATA_TYPE_SUBMIT_BLOCK
if (qSetStreamInput(pItem->taskInfo, pMsg, inputType, true) < 0) {
smaError("vgId:%d, rsma % " PRIi8 " qSetStreamInput failed since %s", SMA_VID(pSma), level, tstrerror(terrno)); smaError("vgId:%d, rsma % " PRIi8 " qSetStreamInput failed since %s", SMA_VID(pSma), level, tstrerror(terrno));
return TSDB_CODE_FAILED; return TSDB_CODE_FAILED;
} }
// SRSmaTriggerParam triggerParam = {.suid = suid, .pItem = pItem, .pSma = pSma, .pTSchema = pTSchema};
tdFetchAndSubmitRSmaResult(pItem, STREAM_DATA_TYPE_SUBMIT_BLOCK); tdFetchAndSubmitRSmaResult(pItem, STREAM_DATA_TYPE_SUBMIT_BLOCK);
atomic_store_8(&pItem->triggerStatus, TASK_TRIGGER_STATUS__ACTIVE); atomic_store_8(&pItem->triggerStatus, TASK_TRIGGER_STATUS__ACTIVE);
taosTmrReset(rsmaTriggerByTimer, pItem->maxDelay, pItem, pItem->tmrHandle, &pItem->tmrId); smaWarn("%s:%d THREAD:%" PRIi64 " process rsma insert", __func__, __LINE__, taosGetSelfPthreadId());
SSmaEnv *pEnv = SMA_RSMA_ENV(pSma);
SRSmaStat *pStat = SMA_RSMA_STAT(pEnv->pStat);
taosTmrStart(tdRSmaPersistTrigger, 5000, pStat, pStat->tmrHandle);
taosTmrReset(tdRSmaFetchTrigger, pItem->maxDelay, pItem, pItem->tmrHandle, &pItem->tmrId);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -542,7 +555,7 @@ static int32_t tdExecuteRSma(SSma *pSma, const void *pMsg, int32_t inputType, tb
SSmaStat *pStat = SMA_ENV_STAT(pEnv); SSmaStat *pStat = SMA_ENV_STAT(pEnv);
SRSmaInfo *pRSmaInfo = NULL; SRSmaInfo *pRSmaInfo = NULL;
pRSmaInfo = taosHashGet(SMA_STAT_INFO_HASH(pStat), &suid, sizeof(tb_uid_t)); pRSmaInfo = taosHashGet(SMA_RSMA_INFO_HASH(pStat), &suid, sizeof(tb_uid_t));
if (!pRSmaInfo || !(pRSmaInfo = *(SRSmaInfo **)pRSmaInfo)) { if (!pRSmaInfo || !(pRSmaInfo = *(SRSmaInfo **)pRSmaInfo)) {
smaDebug("vgId:%d, return as no rsma info for suid:%" PRIu64, SMA_VID(pSma), suid); smaDebug("vgId:%d, return as no rsma info for suid:%" PRIu64, SMA_VID(pSma), suid);
@ -594,3 +607,106 @@ int32_t tdProcessRSmaSubmit(SSma *pSma, void *pMsg, int32_t inputType) {
} }
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
void tdRSmaQTaskGetFName(int32_t vid, int8_t ftype, char* outputName) {
tdGetVndFileName(vid, "rsma", tdQTaskInfoFname[ftype], outputName);
}
static void *tdRSmaPersistExec(void *param) {
setThreadName("rsma-task-persist");
SRSmaStat *pRSmaStat = param;
SSma *pSma = pRSmaStat->pSma;
STfs *pTfs = pSma->pVnode->pTfs;
int64_t toffset = 0;
void *infoHash = taosHashIterate(RSMA_INFO_HASH(pRSmaStat), NULL);
if (!infoHash) {
goto _end;
}
STFile tFile = {0};
int32_t vid = 2;
char qTaskInfoFName[TSDB_FILENAME_LEN];
tdRSmaQTaskGetFName(vid, TD_QTASK_TMP_FILE, qTaskInfoFName);
tdInitTFile(&tFile, pTfs, qTaskInfoFName);
tdCreateTFile(&tFile, pTfs, true, -1);
while (infoHash) {
SRSmaInfo *pRSmaInfo = *(SRSmaInfo **)infoHash;
char *pOutput = NULL;
int32_t len = 0;
if (qSerializeTaskStatus(pRSmaInfo->items[0].taskInfo, &pOutput, &len) < 0) {
smaError("serialize rsma task for table %" PRIi64 " failed since %s", pRSmaInfo->items[0].pRsmaInfo->suid,
terrstr(terrno));
} else {
smaWarn("serialize rsma task for table %" PRIi64 " success and len is %d", pRSmaInfo->items[0].pRsmaInfo->suid,
len);
}
tdAppendTFile(&tFile, &len, sizeof(len), &toffset);
tdAppendTFile(&tFile, pOutput, len, &toffset);
taosMemoryFree(pOutput);
infoHash = taosHashIterate(RSMA_INFO_HASH(pRSmaStat), infoHash);
}
_end:
if (tdUpdateTFileHeader(&tFile) < 0) {
smaError("vgId:%d, failed to update tfile %s header since %s", vid, TD_FILE_FULL_NAME(&tFile), tstrerror(terrno));
tdCloseTFile(&tFile);
tdRemoveTFile(&tFile);
return NULL;
}
tdCloseTFile(&tFile);
char newFName[TSDB_FILENAME_LEN];
strncpy(newFName, TD_FILE_FULL_NAME(&tFile), TSDB_FILENAME_LEN);
char *pos = strstr(newFName, tdQTaskInfoFname[TD_QTASK_TMP_FILE]);
strncpy(pos, tdQTaskInfoFname[TD_QTASK_CUR_FILE], TSDB_FILENAME_LEN - POINTER_DISTANCE(pos, newFName));
taosRenameFile(TD_FILE_FULL_NAME(&tFile), newFName);
atomic_store_8(&pRSmaStat->tmrStat, TASK_TRIGGER_STATUS__ACTIVE);
return NULL;
_err:
atomic_store_8(&pRSmaStat->tmrStat, TASK_TRIGGER_STATUS__ACTIVE);
// remove the .tmp file
return NULL;
}
static void tdRSmaPersistTask(SRSmaStat *pRSmaStat) {
smaWarn("%s:%d entry ", __func__, __LINE__);
TdThread threadId;
TdThreadAttr thAttr;
taosThreadAttrInit(&thAttr);
taosThreadAttrSetDetachState(&thAttr, PTHREAD_CREATE_DETACHED);
if (taosThreadCreate(&threadId, &thAttr, tdRSmaPersistExec, pRSmaStat) != 0) {
smaError("failed to create thread to persist rsma qTaskInfo since %s", strerror(errno));
}
taosThreadAttrDestroy(&thAttr);
smaWarn("%s:%d end ", __func__, __LINE__);
}
/**
* @brief trigger to persist rsma qTaskInfo
*
* @param param
* @param tmrId
*/
static void tdRSmaPersistTrigger(void *param, void *tmrId) {
SRSmaStat *pRSmaStat = param;
if (atomic_load_8(&pRSmaStat->tmrStat) == TASK_TRIGGER_STATUS__ACTIVE) {
smaWarn("%s:%d THREAD:%" PRIi64 " rsma persistence start since active", __func__, __LINE__, taosGetSelfPthreadId());
atomic_store_8(&pRSmaStat->tmrStat, TASK_TRIGGER_STATUS__IN_ACTIVE);
// execution
tdRSmaPersistTask(pRSmaStat);
} else {
smaWarn("%s:%d THREAD:%" PRIi64 " rsma persistence not start since inactive", __func__, __LINE__,
taosGetSelfPthreadId());
}
taosTmrReset(tdRSmaPersistTrigger, 3600000, pRSmaStat, pRSmaStat->tmrHandle, &pRSmaStat->tmrId);
}

4
source/dnode/vnode/src/sma/smaTimeRange.c
View File

@ -129,7 +129,7 @@ int32_t tdProcessTSmaInsertImpl(SSma *pSma, int64_t indexUid, const char *msg) {
SSmaEnv *pEnv = SMA_TSMA_ENV(pSma); SSmaEnv *pEnv = SMA_TSMA_ENV(pSma);
SSmaStat *pStat = NULL; SSmaStat *pStat = NULL;
SSmaStatItem *pItem = NULL; STSmaStat *pItem = NULL;
if (!pEnv || !(pStat = SMA_ENV_STAT(pEnv))) { if (!pEnv || !(pStat = SMA_ENV_STAT(pEnv))) {
terrno = TSDB_CODE_TSMA_INVALID_STAT; terrno = TSDB_CODE_TSMA_INVALID_STAT;
@ -137,7 +137,7 @@ int32_t tdProcessTSmaInsertImpl(SSma *pSma, int64_t indexUid, const char *msg) {
} }
tdRefSmaStat(pSma, pStat); tdRefSmaStat(pSma, pStat);
pItem = &pStat->tsmaStatItem; pItem = &pStat->tsmaStat;
ASSERT(pItem); ASSERT(pItem);

238
source/dnode/vnode/src/sma/smaUtil.c Normal file
View File

@ -0,0 +1,238 @@
/*
* 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 "sma.h"
#define TD_FILE_HEAD_SIZE 512
#define TD_FILE_STATE_OK 0
#define TD_FILE_STATE_BAD 1
#define TD_FILE_INIT_MAGIC 0xFFFFFFFF
static int32_t tdEncodeTFInfo(void **buf, STFInfo *pInfo);
static void *tdDecodeTFInfo(void *buf, STFInfo *pInfo);
static int32_t tdEncodeTFInfo(void **buf, STFInfo *pInfo) {
int32_t tlen = 0;
tlen += taosEncodeFixedU32(buf, pInfo->magic);
tlen += taosEncodeFixedU32(buf, pInfo->ftype);
tlen += taosEncodeFixedU32(buf, pInfo->fver);
tlen += taosEncodeFixedU64(buf, pInfo->fsize);
return tlen;
}
static void *tdDecodeTFInfo(void *buf, STFInfo *pInfo) {
buf = taosDecodeFixedU32(buf, &(pInfo->magic));
buf = taosDecodeFixedU32(buf, &(pInfo->ftype));
buf = taosDecodeFixedU32(buf, &(pInfo->fver));
buf = taosDecodeFixedU64(buf, &(pInfo->fsize));
return buf;
}
int64_t tdWriteTFile(STFile *pTFile, void *buf, int64_t nbyte) {
ASSERT(TD_FILE_OPENED(pTFile));
int64_t nwrite = taosWriteFile(pTFile->pFile, buf, nbyte);
if (nwrite < nbyte) {
terrno = TAOS_SYSTEM_ERROR(errno);
return -1;
}
return nwrite;
}
int64_t tdSeekTFile(STFile *pTFile, int64_t offset, int whence) {
ASSERT(TD_FILE_OPENED(pTFile));
int64_t loffset = taosLSeekFile(TD_FILE_PFILE(pTFile), offset, whence);
if (loffset < 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
return -1;
}
return loffset;
}
int64_t tdReadTFile(STFile *pTFile, void *buf, int64_t nbyte) {
ASSERT(TD_FILE_OPENED(pTFile));
int64_t nread = taosReadFile(pTFile->pFile, buf, nbyte);
if (nread < 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
return -1;
}
return nread;
}
int32_t tdUpdateTFileHeader(STFile *pTFile) {
char buf[TD_FILE_HEAD_SIZE] = "\0";
if (tdSeekTFile(pTFile, 0, SEEK_SET) < 0) {
return -1;
}
void *ptr = buf;
tdEncodeTFInfo(&ptr, &(pTFile->info));
taosCalcChecksumAppend(0, (uint8_t *)buf, TD_FILE_HEAD_SIZE);
if (tdWriteTFile(pTFile, buf, TD_FILE_HEAD_SIZE) < 0) {
return -1;
}
return 0;
}
int32_t tdLoadTFileHeader(STFile *pTFile, STFInfo *pInfo) {
char buf[TD_FILE_HEAD_SIZE] = "\0";
uint32_t _version;
ASSERT(TD_FILE_OPENED(pTFile));
if (tdSeekTFile(pTFile, 0, SEEK_SET) < 0) {
return -1;
}
if (tdReadTFile(pTFile, buf, TD_FILE_HEAD_SIZE) < 0) {
return -1;
}
if (!taosCheckChecksumWhole((uint8_t *)buf, TD_FILE_HEAD_SIZE)) {
terrno = TSDB_CODE_FILE_CORRUPTED;
return -1;
}
void *pBuf = buf;
pBuf = tdDecodeTFInfo(pBuf, pInfo);
return 0;
}
void tdUpdateTFileMagic(STFile *pTFile, void *pCksm) {
pTFile->info.magic = taosCalcChecksum(pTFile->info.magic, (uint8_t *)(pCksm), sizeof(TSCKSUM));
}
int64_t tdAppendTFile(STFile *pTFile, void *buf, int64_t nbyte, int64_t *offset) {
ASSERT(TD_FILE_OPENED(pTFile));
int64_t toffset;
if ((toffset = tdSeekTFile(pTFile, 0, SEEK_END)) < 0) {
return -1;
}
ASSERT(pTFile->info.fsize == toffset);
if (offset) {
*offset = toffset;
}
if (tdWriteTFile(pTFile, buf, nbyte) < 0) {
return -1;
}
pTFile->info.fsize += nbyte;
return nbyte;
}
int32_t tdOpenTFile(STFile *pTFile, int flags) {
ASSERT(!TD_FILE_OPENED(pTFile));
pTFile->pFile = taosOpenFile(TD_FILE_FULL_NAME(pTFile), flags);
if (pTFile->pFile == NULL) {
terrno = TAOS_SYSTEM_ERROR(errno);
return -1;
}
return 0;
}
void tdCloseTFile(STFile *pTFile) {
if (TD_FILE_OPENED(pTFile)) {
taosCloseFile(&pTFile->pFile);
TD_FILE_SET_CLOSED(pTFile);
}
}
void tdGetVndFileName(int32_t vid, const char *dname, const char *fname, char *outputName) {
snprintf(outputName, TSDB_FILENAME_LEN, "vnode/vnode%d/%s/%s", vid, dname, fname);
}
int32_t tdInitTFile(STFile *pTFile, STfs *pTfs, const char *fname) {
char fullname[TSDB_FILENAME_LEN];
SDiskID did = {0};
TD_FILE_SET_STATE(pTFile, TD_FILE_STATE_OK);
TD_FILE_SET_CLOSED(pTFile);
memset(&(pTFile->info), 0, sizeof(pTFile->info));
pTFile->info.magic = TD_FILE_INIT_MAGIC;
if (tfsAllocDisk(pTfs, 0, &did) < 0) {
terrno = TSDB_CODE_NO_AVAIL_DISK;
return -1;
}
tfsInitFile(pTfs, &(pTFile->f), did, fname);
return 0;
}
int32_t tdCreateTFile(STFile *pTFile, STfs *pTfs, bool updateHeader, int8_t fType) {
ASSERT(pTFile->info.fsize == 0 && pTFile->info.magic == TD_FILE_INIT_MAGIC);
pTFile->pFile = taosOpenFile(TD_FILE_FULL_NAME(pTFile), TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC);
if (pTFile->pFile == NULL) {
if (errno == ENOENT) {
// Try to create directory recursively
char *s = strdup(TD_FILE_REL_NAME(pTFile));
if (tfsMkdirRecurAt(pTfs, taosDirName(s), TD_FILE_DID(pTFile)) < 0) {
taosMemoryFreeClear(s);
return -1;
}
taosMemoryFreeClear(s);
pTFile->pFile = taosOpenFile(TD_FILE_FULL_NAME(pTFile), TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC);
if (pTFile->pFile == NULL) {
terrno = TAOS_SYSTEM_ERROR(errno);
return -1;
}
} else {
terrno = TAOS_SYSTEM_ERROR(errno);
return -1;
}
}
if (!updateHeader) {
return 0;
}
pTFile->info.fsize += TD_FILE_HEAD_SIZE;
pTFile->info.fver = 0;
if (tdUpdateTFileHeader(pTFile) < 0) {
tdCloseTFile(pTFile);
tdRemoveTFile(pTFile);
return -1;
}
return 0;
}
int32_t tdRemoveTFile(STFile *pTFile) { return tfsRemoveFile(TD_FILE_F(pTFile)); }

40
source/dnode/vnode/src/tq/tq.c
View File

@ -227,19 +227,16 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
consumerEpoch = atomic_val_compare_exchange_32(&pHandle->epoch, consumerEpoch, reqEpoch); consumerEpoch = atomic_val_compare_exchange_32(&pHandle->epoch, consumerEpoch, reqEpoch);
} }
SWalHead* pHeadWithCkSum = taosMemoryMalloc(sizeof(SWalHead) + 2048);
if (pHeadWithCkSum == NULL) {
return -1;
}
walSetReaderCapacity(pHandle->pWalReader, 2048);
SMqDataBlkRsp rsp = {0}; SMqDataBlkRsp rsp = {0};
rsp.reqOffset = pReq->currentOffset; rsp.reqOffset = pReq->currentOffset;
rsp.blockData = taosArrayInit(0, sizeof(void*)); rsp.blockData = taosArrayInit(0, sizeof(void*));
rsp.blockDataLen = taosArrayInit(0, sizeof(int32_t)); rsp.blockDataLen = taosArrayInit(0, sizeof(int32_t));
if (rsp.blockData == NULL || rsp.blockDataLen == NULL) {
return -1;
}
rsp.withTbName = pReq->withTbName; rsp.withTbName = pReq->withTbName;
if (rsp.withTbName) { if (rsp.withTbName) {
rsp.blockTbName = taosArrayInit(0, sizeof(void*)); rsp.blockTbName = taosArrayInit(0, sizeof(void*));
@ -253,6 +250,30 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
rsp.blockSchema = taosArrayInit(0, sizeof(void*)); rsp.blockSchema = taosArrayInit(0, sizeof(void*));
} }
#if 1
if (pReq->useSnapshot) {
tqInfo("retrieve using snapshot");
int64_t lastVer = walGetCommittedVer(pTq->pWal);
if (rsp.reqOffset < lastVer) {
tqScanSnapshot(pTq, &pHandle->execHandle, &rsp, workerId);
if (rsp.blockNum != 0) {
rsp.withTbName = false;
rsp.rspOffset = lastVer;
tqInfo("direct send by snapshot rsp offset %ld", lastVer);
goto SEND_RSP;
}
}
}
#endif
SWalHead* pHeadWithCkSum = taosMemoryMalloc(sizeof(SWalHead) + 2048);
if (pHeadWithCkSum == NULL) {
return -1;
}
walSetReaderCapacity(pHandle->pWalReader, 2048);
while (1) { while (1) {
consumerEpoch = atomic_load_32(&pHandle->epoch); consumerEpoch = atomic_load_32(&pHandle->epoch);
if (consumerEpoch > reqEpoch) { if (consumerEpoch > reqEpoch) {
@ -292,6 +313,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
metaRsp.metaRsp = pHead->body; metaRsp.metaRsp = pHead->body;
if (tqSendMetaPollRsp(pTq, pMsg, pReq, &metaRsp) < 0) { if (tqSendMetaPollRsp(pTq, pMsg, pReq, &metaRsp) < 0) {
code = -1; code = -1;
goto OVER;
} }
code = 0; code = 0;
goto OVER; goto OVER;
@ -308,6 +330,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
taosMemoryFree(pHeadWithCkSum); taosMemoryFree(pHeadWithCkSum);
SEND_RSP:
ASSERT(taosArrayGetSize(rsp.blockData) == rsp.blockNum); ASSERT(taosArrayGetSize(rsp.blockData) == rsp.blockNum);
ASSERT(taosArrayGetSize(rsp.blockDataLen) == rsp.blockNum); ASSERT(taosArrayGetSize(rsp.blockDataLen) == rsp.blockNum);
if (rsp.withSchema) { if (rsp.withSchema) {
@ -376,6 +399,8 @@ int32_t tqProcessVgChangeReq(STQ* pTq, char* msg, int32_t msgLen) {
SReadHandle handle = { SReadHandle handle = {
.reader = pHandle->execHandle.pExecReader[i], .reader = pHandle->execHandle.pExecReader[i],
.meta = pTq->pVnode->pMeta, .meta = pTq->pVnode->pMeta,
.vnode = pTq->pVnode,
.initTsdbReader = 1,
}; };
pHandle->execHandle.execCol.task[i] = qCreateStreamExecTaskInfo(pHandle->execHandle.execCol.qmsg, &handle); pHandle->execHandle.execCol.task[i] = qCreateStreamExecTaskInfo(pHandle->execHandle.execCol.qmsg, &handle);
ASSERT(pHandle->execHandle.execCol.task[i]); ASSERT(pHandle->execHandle.execCol.task[i]);
@ -448,6 +473,7 @@ int32_t tqProcessTaskDeployReq(STQ* pTq, char* msg, int32_t msgLen) {
.reader = pStreamReader, .reader = pStreamReader,
.meta = pTq->pVnode->pMeta, .meta = pTq->pVnode->pMeta,
.vnode = pTq->pVnode, .vnode = pTq->pVnode,
.initTsdbReader = 1,
}; };
/*pTask->exec.inputHandle = pStreamReader;*/ /*pTask->exec.inputHandle = pStreamReader;*/
pTask->exec.executor = qCreateStreamExecTaskInfo(pTask->exec.qmsg, &handle); pTask->exec.executor = qCreateStreamExecTaskInfo(pTask->exec.qmsg, &handle);

24
source/dnode/vnode/src/tq/tqExec.c
View File

@ -60,6 +60,30 @@ static int32_t tqAddTbNameToRsp(const STQ* pTq, const STqExecHandle* pExec, SMqD
return 0; return 0;
} }
int32_t tqScanSnapshot(STQ* pTq, const STqExecHandle* pExec, SMqDataBlkRsp* pRsp, int32_t workerId) {
ASSERT(pExec->subType == TOPIC_SUB_TYPE__COLUMN);
qTaskInfo_t task = pExec->execCol.task[workerId];
if (qStreamScanSnapshot(task) < 0) {
ASSERT(0);
}
while (1) {
SSDataBlock* pDataBlock = NULL;
uint64_t ts = 0;
if (qExecTask(task, &pDataBlock, &ts) < 0) {
ASSERT(0);
}
if (pDataBlock == NULL) break;
ASSERT(pDataBlock->info.rows != 0);
ASSERT(pDataBlock->info.numOfCols != 0);
tqAddBlockDataToRsp(pDataBlock, pRsp);
pRsp->blockNum++;
}
return 0;
}
int32_t tqDataExec(STQ* pTq, STqExecHandle* pExec, SSubmitReq* pReq, SMqDataBlkRsp* pRsp, int32_t workerId) { int32_t tqDataExec(STQ* pTq, STqExecHandle* pExec, SSubmitReq* pReq, SMqDataBlkRsp* pRsp, int32_t workerId) {
if (pExec->subType == TOPIC_SUB_TYPE__COLUMN) { if (pExec->subType == TOPIC_SUB_TYPE__COLUMN) {
qTaskInfo_t task = pExec->execCol.task[workerId]; qTaskInfo_t task = pExec->execCol.task[workerId];

2
source/libs/executor/src/dataDeleter.c
View File

@ -86,7 +86,7 @@ static void toDataCacheEntry(SDataDeleterHandle* pHandle, const SInputData* pInp
SColumnInfoData* pColRes = (SColumnInfoData*)taosArrayGet(pInput->pData->pDataBlock, 0); SColumnInfoData* pColRes = (SColumnInfoData*)taosArrayGet(pInput->pData->pDataBlock, 0);
SDeleterRes* pRes = (SDeleterRes*)pEntry->data; SDeleterRes* pRes = (SDeleterRes*)pEntry->data;
pRes->uid = pHandle->pDeleter->tableId; pRes->suid = pHandle->pParam->suid;
pRes->uidList = pHandle->pParam->pUidList; pRes->uidList = pHandle->pParam->pUidList;
pRes->skey = pHandle->pDeleter->deleteTimeRange.skey; pRes->skey = pHandle->pDeleter->deleteTimeRange.skey;
pRes->ekey = pHandle->pDeleter->deleteTimeRange.ekey; pRes->ekey = pHandle->pDeleter->deleteTimeRange.ekey;

2
source/libs/executor/src/executil.c
View File

@ -290,6 +290,8 @@ int32_t getTableList(void* metaHandle, SScanPhysiNode* pScanNode, STableListInfo
uint64_t tableUid = pScanNode->uid; uint64_t tableUid = pScanNode->uid;
pListInfo->suid = pScanNode->suid;
SNode* pTagCond = (SNode*)pListInfo->pTagCond; SNode* pTagCond = (SNode*)pListInfo->pTagCond;
SNode* pTagIndexCond = (SNode*)pListInfo->pTagIndexCond; SNode* pTagIndexCond = (SNode*)pListInfo->pTagIndexCond;
if (pScanNode->tableType == TSDB_SUPER_TABLE) { if (pScanNode->tableType == TSDB_SUPER_TABLE) {

19
source/libs/executor/src/executor.c
View File

@ -67,6 +67,9 @@ static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, size_t nu
taosArrayAddAll(p->pDataBlock, pDataBlock->pDataBlock); taosArrayAddAll(p->pDataBlock, pDataBlock->pDataBlock);
taosArrayPush(pInfo->pBlockLists, &p); taosArrayPush(pInfo->pBlockLists, &p);
} }
} else if (type == STREAM_DATA_TYPE_FROM_SNAPSHOT) {
// do nothing
ASSERT(pInfo->blockType == STREAM_DATA_TYPE_FROM_SNAPSHOT);
} else { } else {
ASSERT(0); ASSERT(0);
} }
@ -75,6 +78,14 @@ static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, size_t nu
} }
} }
int32_t qStreamScanSnapshot(qTaskInfo_t tinfo) {
if (tinfo == NULL) {
return TSDB_CODE_QRY_APP_ERROR;
}
SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;
return doSetStreamBlock(pTaskInfo->pRoot, NULL, 0, STREAM_DATA_TYPE_FROM_SNAPSHOT, 0, NULL);
}
int32_t qSetStreamInput(qTaskInfo_t tinfo, const void* input, int32_t type, bool assignUid) { int32_t qSetStreamInput(qTaskInfo_t tinfo, const void* input, int32_t type, bool assignUid) {
return qSetMultiStreamInput(tinfo, input, 1, type, assignUid); return qSetMultiStreamInput(tinfo, input, 1, type, assignUid);
} }
@ -106,14 +117,6 @@ qTaskInfo_t qCreateStreamExecTaskInfo(void* msg, void* streamReadHandle) {
return NULL; return NULL;
} }
// print those info into log
#if 0
pMsg->sId = pMsg->sId;
pMsg->queryId = pMsg->queryId;
pMsg->taskId = pMsg->taskId;
pMsg->contentLen = pMsg->contentLen;
#endif
/*qDebugL("stream task string %s", (const char*)msg);*/ /*qDebugL("stream task string %s", (const char*)msg);*/
struct SSubplan* plan = NULL; struct SSubplan* plan = NULL;

8
source/libs/executor/src/executorimpl.c
View File

@ -4028,16 +4028,17 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
tsdbReaderT pDataReader = NULL; tsdbReaderT pDataReader = NULL;
if (pHandle) { if (pHandle) {
if (pHandle->vnode) { if (pHandle->initTsdbReader) {
// for stram // for stream
ASSERT(pHandle->vnode);
pDataReader = pDataReader =
doCreateDataReader(pTableScanNode, pHandle, pTableListInfo, (uint64_t)queryId, taskId); doCreateDataReader(pTableScanNode, pHandle, pTableListInfo, (uint64_t)queryId, taskId);
} else { } else {
// for tq // for tq
ASSERT(pHandle->meta);
getTableList(pHandle->meta, pScanPhyNode, pTableListInfo); getTableList(pHandle->meta, pScanPhyNode, pTableListInfo);
} }
} }
if (pDataReader == NULL && terrno != 0) { if (pDataReader == NULL && terrno != 0) {
qDebug("%s pDataReader is NULL", GET_TASKID(pTaskInfo)); qDebug("%s pDataReader is NULL", GET_TASKID(pTaskInfo));
// return NULL; // return NULL;
@ -4451,6 +4452,7 @@ int32_t createDataSinkParam(SDataSinkNode* pNode, void** pParam, qTaskInfo_t* pT
return TSDB_CODE_OUT_OF_MEMORY; return TSDB_CODE_OUT_OF_MEMORY;
} }
int32_t tbNum = taosArrayGetSize(pTask->tableqinfoList.pTableList); int32_t tbNum = taosArrayGetSize(pTask->tableqinfoList.pTableList);
pDeleterParam->suid = pTask->tableqinfoList.suid;
pDeleterParam->pUidList = taosArrayInit(tbNum, sizeof(uint64_t)); pDeleterParam->pUidList = taosArrayInit(tbNum, sizeof(uint64_t));
if (NULL == pDeleterParam->pUidList) { if (NULL == pDeleterParam->pUidList) {
taosMemoryFree(pDeleterParam); taosMemoryFree(pDeleterParam);

11
source/libs/executor/src/scanoperator.c
View File

@ -925,7 +925,7 @@ static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator) {
SSDataBlock* pBlock = taosArrayGetP(pInfo->pBlockLists, current); SSDataBlock* pBlock = taosArrayGetP(pInfo->pBlockLists, current);
blockDataUpdateTsWindow(pBlock, 0); blockDataUpdateTsWindow(pBlock, 0);
return pBlock; return pBlock;
} else { } else if (pInfo->blockType == STREAM_DATA_TYPE_SUBMIT_BLOCK) {
if (pInfo->scanMode == STREAM_SCAN_FROM_RES) { if (pInfo->scanMode == STREAM_SCAN_FROM_RES) {
blockDataDestroy(pInfo->pUpdateRes); blockDataDestroy(pInfo->pUpdateRes);
pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE; pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE;
@ -1060,6 +1060,15 @@ static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator) {
} }
return (pBlockInfo->rows == 0) ? NULL : pInfo->pRes; return (pBlockInfo->rows == 0) ? NULL : pInfo->pRes;
} else if (pInfo->blockType == STREAM_DATA_TYPE_FROM_SNAPSHOT) {
SSDataBlock* pResult = doTableScan(pInfo->pOperatorDumy);
if (pResult) {
return pResult->info.rows > 0 ? pResult : NULL;
}
return NULL;
} else {
ASSERT(0);
return NULL;
} }
} }

2
source/libs/qworker/src/qworker.c
View File

@ -271,7 +271,7 @@ int32_t qwGetDeleteResFromSink(QW_FPARAMS_DEF, SQWTaskCtx *ctx, int32_t *dataLen
SDeleterRes* pDelRes = (SDeleterRes*)output.pData; SDeleterRes* pDelRes = (SDeleterRes*)output.pData;
rsp.affectedRows = pDelRes->affectedRows; rsp.affectedRows = pDelRes->affectedRows;
pRes->uid = pDelRes->uid; pRes->suid = pDelRes->suid;
pRes->uidList = pDelRes->uidList; pRes->uidList = pDelRes->uidList;
pRes->skey = pDelRes->skey; pRes->skey = pDelRes->skey;
pRes->ekey = pDelRes->ekey; pRes->ekey = pDelRes->ekey;

1
source/libs/stream/src/streamExec.c
View File

@ -80,7 +80,6 @@ static SArray* streamExecForQall(SStreamTask* pTask, SArray* pRes) {
} }
qRes->type = STREAM_INPUT__DATA_BLOCK; qRes->type = STREAM_INPUT__DATA_BLOCK;
qRes->blocks = pRes; qRes->blocks = pRes;
/*qRes->sourceVg = pTask->nodeId;*/
if (streamTaskOutput(pTask, qRes) < 0) { if (streamTaskOutput(pTask, qRes) < 0) {
streamQueueProcessFail(pTask->inputQueue); streamQueueProcessFail(pTask->inputQueue);
taosArrayDestroy(pRes); taosArrayDestroy(pRes);

3
source/util/src/terror.c
View File

@ -76,6 +76,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_NEED_RETRY, "Retry needed")
TAOS_DEFINE_ERROR(TSDB_CODE_OUT_OF_RPC_MEMORY_QUEUE, "Out of memory in rpc queue") TAOS_DEFINE_ERROR(TSDB_CODE_OUT_OF_RPC_MEMORY_QUEUE, "Out of memory in rpc queue")
TAOS_DEFINE_ERROR(TSDB_CODE_INVALID_TIMESTAMP, "Invalid timestamp format") TAOS_DEFINE_ERROR(TSDB_CODE_INVALID_TIMESTAMP, "Invalid timestamp format")
TAOS_DEFINE_ERROR(TSDB_CODE_MSG_DECODE_ERROR, "Msg decode error") TAOS_DEFINE_ERROR(TSDB_CODE_MSG_DECODE_ERROR, "Msg decode error")
TAOS_DEFINE_ERROR(TSDB_CODE_NO_AVAIL_DISK, "No available disk")
TAOS_DEFINE_ERROR(TSDB_CODE_REF_NO_MEMORY, "Ref out of memory") TAOS_DEFINE_ERROR(TSDB_CODE_REF_NO_MEMORY, "Ref out of memory")
TAOS_DEFINE_ERROR(TSDB_CODE_REF_FULL, "too many Ref Objs") TAOS_DEFINE_ERROR(TSDB_CODE_REF_FULL, "too many Ref Objs")
@ -356,7 +357,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_TDB_NO_TABLE_DATA_IN_MEM, "No table data in memo
TAOS_DEFINE_ERROR(TSDB_CODE_TDB_FILE_ALREADY_EXISTS, "File already exists") TAOS_DEFINE_ERROR(TSDB_CODE_TDB_FILE_ALREADY_EXISTS, "File already exists")
TAOS_DEFINE_ERROR(TSDB_CODE_TDB_TABLE_RECONFIGURE, "Need to reconfigure table") TAOS_DEFINE_ERROR(TSDB_CODE_TDB_TABLE_RECONFIGURE, "Need to reconfigure table")
TAOS_DEFINE_ERROR(TSDB_CODE_TDB_IVD_CREATE_TABLE_INFO, "Invalid information to create table") TAOS_DEFINE_ERROR(TSDB_CODE_TDB_IVD_CREATE_TABLE_INFO, "Invalid information to create table")
TAOS_DEFINE_ERROR(TSDB_CODE_TDB_NO_AVAIL_DISK, "No available disk") TAOS_DEFINE_ERROR(TSDB_CODE_TDB_NO_AVAIL_DISK, "TSDB no available disk")
TAOS_DEFINE_ERROR(TSDB_CODE_TDB_MESSED_MSG, "TSDB messed message") TAOS_DEFINE_ERROR(TSDB_CODE_TDB_MESSED_MSG, "TSDB messed message")
TAOS_DEFINE_ERROR(TSDB_CODE_TDB_IVLD_TAG_VAL, "TSDB invalid tag value") TAOS_DEFINE_ERROR(TSDB_CODE_TDB_IVLD_TAG_VAL, "TSDB invalid tag value")
TAOS_DEFINE_ERROR(TSDB_CODE_TDB_NO_CACHE_LAST_ROW, "TSDB no cache last row data") TAOS_DEFINE_ERROR(TSDB_CODE_TDB_NO_CACHE_LAST_ROW, "TSDB no cache last row data")

1
tests/script/jenkins/basic.txt
View File

@ -109,6 +109,7 @@
./test.sh -f tsim/tmq/basic4Of2Cons.sim ./test.sh -f tsim/tmq/basic4Of2Cons.sim
./test.sh -f tsim/tmq/basic2Of2ConsOverlap.sim ./test.sh -f tsim/tmq/basic2Of2ConsOverlap.sim
./test.sh -f tsim/tmq/topic.sim ./test.sh -f tsim/tmq/topic.sim
./test.sh -f tsim/tmq/snapshot.sim
# --- stable # --- stable
./test.sh -f tsim/stable/disk.sim ./test.sh -f tsim/stable/disk.sim

1
tests/script/tsim/mnode/basic5.sim
View File

@ -233,6 +233,7 @@ if $data(1)[3] != dropping then
endi endi
print =============== step9: start mnode1 and wait it dropped print =============== step9: start mnode1 and wait it dropped
sleep 3000
system sh/exec.sh -n dnode1 -s start system sh/exec.sh -n dnode1 -s start
$x = 0 $x = 0

4
tests/script/tsim/show/basic.sim
View File

@ -233,12 +233,12 @@ if $rows != 4 then
endi endi
sql show dnode 1 variables; sql show dnode 1 variables;
if $rows != 116 then if $rows <= 0 then
return -1 return -1
endi endi
sql show local variables; sql show local variables;
if $rows != 50 then if $rows <= 0 then
return -1 return -1
endi endi

10
tests/script/tsim/tmq/consume.sh
View File

@ -17,8 +17,9 @@ VALGRIND=0
SIGNAL=SIGINT SIGNAL=SIGINT
SHOW_MSG=0 SHOW_MSG=0
SHOW_ROW=0 SHOW_ROW=0
EXP_USE_SNAPSHOT=0
while getopts "d:s:v:y:x:g:r:w:" arg while getopts "d:s:v:y:x:g:r:w:e:" arg
do do
case $arg in case $arg in
d) d)
@ -45,6 +46,9 @@ do
w) w)
CDB_NAME=$OPTARG CDB_NAME=$OPTARG
;; ;;
e)
EXP_USE_SNAPSHOT=$OPTARG
;;
?) ?)
echo "unkown argument" echo "unkown argument"
;; ;;
@ -91,8 +95,8 @@ if [ "$EXEC_OPTON" = "start" ]; then
echo nohup valgrind --tool=memcheck --leak-check=full --show-reachable=no --track-origins=yes --show-leak-kinds=all -v --workaround-gcc296-bugs=yes --log-file=${LOG_DIR}/valgrind-tmq_sim.log $PROGRAM -c $CFG_DIR -y $POLL_DELAY -d $DB_NAME -g $SHOW_MSG -r $SHOW_ROW > /dev/null 2>&1 & echo nohup valgrind --tool=memcheck --leak-check=full --show-reachable=no --track-origins=yes --show-leak-kinds=all -v --workaround-gcc296-bugs=yes --log-file=${LOG_DIR}/valgrind-tmq_sim.log $PROGRAM -c $CFG_DIR -y $POLL_DELAY -d $DB_NAME -g $SHOW_MSG -r $SHOW_ROW > /dev/null 2>&1 &
nohup valgrind --tool=memcheck --leak-check=full --show-reachable=no --track-origins=yes --show-leak-kinds=all -v --workaround-gcc296-bugs=yes --log-file=${LOG_DIR}/valgrind-tmq_sim.log $PROGRAM -c $CFG_DIR -y $POLL_DELAY -d $DB_NAME -g $SHOW_MSG -r $SHOW_ROW > /dev/null 2>&1 & nohup valgrind --tool=memcheck --leak-check=full --show-reachable=no --track-origins=yes --show-leak-kinds=all -v --workaround-gcc296-bugs=yes --log-file=${LOG_DIR}/valgrind-tmq_sim.log $PROGRAM -c $CFG_DIR -y $POLL_DELAY -d $DB_NAME -g $SHOW_MSG -r $SHOW_ROW > /dev/null 2>&1 &
else else
echo "nohup $PROGRAM -c $CFG_DIR -y $POLL_DELAY -d $DB_NAME -g $SHOW_MSG -r $SHOW_ROW -w $CDB_NAME > /dev/null 2>&1 &" echo "nohup $PROGRAM -c $CFG_DIR -y $POLL_DELAY -d $DB_NAME -g $SHOW_MSG -r $SHOW_ROW -w $CDB_NAME -e $EXP_USE_SNAPSHOT > /dev/null 2>&1 &"
nohup $PROGRAM -c $CFG_DIR -y $POLL_DELAY -d $DB_NAME -g $SHOW_MSG -r $SHOW_ROW -w $CDB_NAME > /dev/null 2>&1 & nohup $PROGRAM -c $CFG_DIR -y $POLL_DELAY -d $DB_NAME -g $SHOW_MSG -r $SHOW_ROW -w $CDB_NAME -e $EXP_USE_SNAPSHOT > /dev/null 2>&1 &
fi fi
else else
PID=`ps -ef|grep tmq_sim | grep -v grep | awk '{print $2}'` PID=`ps -ef|grep tmq_sim | grep -v grep | awk '{print $2}'`

289
tests/script/tsim/tmq/snapshot.sim Normal file
View File

@ -0,0 +1,289 @@
#### test scenario, please refer to https://jira.taosdata.com:18090/pages/viewpage.action?pageId=135120406
#basic1.sim: vgroups=1, one topic for one consumer, firstly insert data, then start consume. Include six topics
#basic2.sim: vgroups=1, multi topics for one consumer, firstly insert data, then start consume. Include six topics
#basic3.sim: vgroups=4, one topic for one consumer, firstly insert data, then start consume. Include six topics
#basic4.sim: vgroups=4, multi topics for one consumer, firstly insert data, then start consume. Include six topics
# notes1: Scalar function: ABS/ACOS/ASIN/ATAN/CEIL/COS/FLOOR/LOG/POW/ROUND/SIN/SQRT/TAN
# The above use cases are combined with where filter conditions, such as: where ts > "2017-08-12 18:25:58.128Z" and sin(a) > 0.5;
#
# notes2: not support aggregate functions(such as sum/count/min/max) and time-windows(interval).
#
run tsim/tmq/prepareBasicEnv-1vgrp.sim
#---- global parameters start ----#
$dbName = db
$vgroups = 1
$stbPrefix = stb
$ctbPrefix = ctb
$ntbPrefix = ntb
$stbNum = 1
$ctbNum = 10
$ntbNum = 10
$rowsPerCtb = 10
$tstart = 1640966400000 # 2022-01-01 00:00:00.000
#---- global parameters end ----#
$pullDelay = 3
$ifcheckdata = 1
$ifmanualcommit = 1
$showMsg = 1
$showRow = 0
sql connect
sql use $dbName
print == create topics from super table
sql create topic topic_stb_column as select ts, c3 from stb
sql create topic topic_stb_all as select ts, c1, c2, c3 from stb
sql create topic topic_stb_function as select ts, abs(c1), sin(c2) from stb
print == create topics from child table
sql create topic topic_ctb_column as select ts, c3 from ctb0
sql create topic topic_ctb_all as select * from ctb0
sql create topic topic_ctb_function as select ts, abs(c1), sin(c2) from ctb0
print == create topics from normal table
sql create topic topic_ntb_column as select ts, c3 from ntb0
sql create topic topic_ntb_all as select * from ntb0
sql create topic topic_ntb_function as select ts, abs(c1), sin(c2) from ntb0
#sql show topics
#if $rows != 9 then
# return -1
#endi
#'group.id:cgrp1,enable.auto.commit:false,auto.commit.interval.ms:6000,auto.offset.reset:earliest'
$keyList = ' . group.id:cgrp1
$keyList = $keyList . ,
$keyList = $keyList . enable.auto.commit:false
#$keyList = $keyList . ,
#$keyList = $keyList . auto.commit.interval.ms:6000
#$keyList = $keyList . ,
#$keyList = $keyList . auto.offset.reset:earliest
$keyList = $keyList . '
print ========== key list: $keyList
$cdb_index = 0
#=============================== start consume =============================#
print ================ test consume from stb
$loop_cnt = 0
loop_consume_diff_topic_from_stb:
#######################################################################################
# clear consume info and consume result
#run tsim/tmq/clearConsume.sim
# because drop table function no stable, so by create new db for consume info and result. Modify it later
$cdb_index = $cdb_index + 1
$cdbName = cdb . $cdb_index
sql create database $cdbName vgroups 1
sleep 500
sql use $cdbName
print == create consume info table and consume result table
sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int, ifmanualcommit int)
sql create table consumeresult (ts timestamp, consumerid int, consummsgcnt bigint, consumrowcnt bigint, checkresult int)
sql show tables
if $rows != 2 then
return -1
endi
#######################################################################################
if $loop_cnt == 0 then
print == scenario 1: topic_stb_column
$topicList = ' . topic_stb_column
$topicList = $topicList . '
elif $loop_cnt == 1 then
print == scenario 2: topic_stb_all
$topicList = ' . topic_stb_all
$topicList = $topicList . '
elif $loop_cnt == 2 then
print == scenario 3: topic_stb_function
$topicList = ' . topic_stb_function
$topicList = $topicList . '
else
goto loop_consume_diff_topic_from_stb_end
endi
$consumerId = 0
$totalMsgOfStb = $ctbNum * $rowsPerCtb
$expectmsgcnt = 1
$expectrowcnt = 100
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
print == start consumer to pull msgs from stb
print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -w $cdbName -e 1 -s start
system tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -w $cdbName -e 1 -s start
print == check consume result
wait_consumer_end_from_stb:
sql select * from consumeresult
print ==> rows: $rows
print ==> rows[0]: $data[0][0] $data[0][1] $data[0][2] $data[0][3] $data[0][4] $data[0][5] $data[0][6]
if $rows != 1 then
sleep 1000
goto wait_consumer_end_from_stb
endi
if $data[0][1] != $consumerId then
return -1
endi
if $data[0][2] != $expectmsgcnt then
return -1
endi
if $data[0][3] != $expectrowcnt then
return -1
endi
$loop_cnt = $loop_cnt + 1
goto loop_consume_diff_topic_from_stb
loop_consume_diff_topic_from_stb_end:
print ================ test consume from ctb
$loop_cnt = 0
loop_consume_diff_topic_from_ctb:
#######################################################################################
# clear consume info and consume result
#run tsim/tmq/clearConsume.sim
# because drop table function no stable, so by create new db for consume info and result. Modify it later
$cdb_index = $cdb_index + 1
$cdbName = cdb . $cdb_index
sql create database $cdbName vgroups 1
sleep 500
sql use $cdbName
print == create consume info table and consume result table
sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int, ifmanualcommit int)
sql create table consumeresult (ts timestamp, consumerid int, consummsgcnt bigint, consumrowcnt bigint, checkresult int)
sql show tables
if $rows != 2 then
return -1
endi
#######################################################################################
if $loop_cnt == 0 then
print == scenario 1: topic_ctb_column
$topicList = ' . topic_ctb_column
$topicList = $topicList . '
elif $loop_cnt == 1 then
print == scenario 2: topic_ctb_all
$topicList = ' . topic_ctb_all
$topicList = $topicList . '
elif $loop_cnt == 2 then
print == scenario 3: topic_ctb_function
$topicList = ' . topic_ctb_function
$topicList = $topicList . '
else
goto loop_consume_diff_topic_from_ctb_end
endi
$consumerId = 0
$totalMsgOfCtb = $rowsPerCtb
$expectmsgcnt = 1
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
print == start consumer to pull msgs from ctb
print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -s start -e 1
system tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -w $cdbName -s start -e 1
print == check consume result
wait_consumer_end_from_ctb:
sql select * from consumeresult
print ==> rows: $rows
print ==> rows[0]: $data[0][0] $data[0][1] $data[0][2] $data[0][3] $data[0][4] $data[0][5] $data[0][6]
if $rows != 1 then
sleep 1000
goto wait_consumer_end_from_ctb
endi
if $data[0][1] != $consumerId then
return -1
endi
if $data[0][2] != 1 then
return -1
endi
if $data[0][3] != 10 then
return -1
endi
$loop_cnt = $loop_cnt + 1
goto loop_consume_diff_topic_from_ctb
loop_consume_diff_topic_from_ctb_end:
print ================ test consume from ntb
$loop_cnt = 0
loop_consume_diff_topic_from_ntb:
#######################################################################################
# clear consume info and consume result
#run tsim/tmq/clearConsume.sim
# because drop table function no stable, so by create new db for consume info and result. Modify it later
$cdb_index = $cdb_index + 1
$cdbName = cdb . $cdb_index
sql create database $cdbName vgroups 1
sleep 500
sql use $cdbName
print == create consume info table and consume result table
sql create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int, ifmanualcommit int)
sql create table consumeresult (ts timestamp, consumerid int, consummsgcnt bigint, consumrowcnt bigint, checkresult int)
sql show tables
if $rows != 2 then
return -1
endi
#######################################################################################
if $loop_cnt == 0 then
print == scenario 1: topic_ntb_column
$topicList = ' . topic_ntb_column
$topicList = $topicList . '
elif $loop_cnt == 1 then
print == scenario 2: topic_ntb_all
$topicList = ' . topic_ntb_all
$topicList = $topicList . '
elif $loop_cnt == 2 then
print == scenario 3: topic_ntb_function
$topicList = ' . topic_ntb_function
$topicList = $topicList . '
else
goto loop_consume_diff_topic_from_ntb_end
endi
$consumerId = 0
$totalMsgOfNtb = $rowsPerCtb
$expectmsgcnt = $totalMsgOfNtb
sql insert into consumeinfo values (now , $consumerId , $topicList , $keyList , $expectmsgcnt , $ifcheckdata , $ifmanualcommit )
print == start consumer to pull msgs from ntb
print == tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -s start -e 1
system tsim/tmq/consume.sh -d $dbName -y $pullDelay -g $showMsg -r $showRow -w $cdbName -s start -e 1
print == check consume result from ntb
wait_consumer_end_from_ntb:
sql select * from consumeresult
print ==> rows: $rows
print ==> rows[0]: $data[0][0] $data[0][1] $data[0][2] $data[0][3] $data[0][4] $data[0][5] $data[0][6]
if $rows != 1 then
sleep 1000
goto wait_consumer_end_from_ntb
endi
if $data[0][1] != $consumerId then
return -1
endi
if $data[0][2] != 1 then
return -1
endi
if $data[0][3] != $totalMsgOfNtb then
return -1
endi
$loop_cnt = $loop_cnt + 1
goto loop_consume_diff_topic_from_ntb
loop_consume_diff_topic_from_ntb_end:
#------ not need stop consumer, because it exit after pull msg overthan expect msg
#system tsim/tmq/consume.sh -s stop -x SIGINT
system sh/exec.sh -n dnode1 -s stop -x SIGINT

113
tests/test/c/tmqSim.c
View File

@ -22,9 +22,9 @@
#include <time.h> #include <time.h>
#include "taos.h" #include "taos.h"
#include "taosdef.h"
#include "taoserror.h" #include "taoserror.h"
#include "tlog.h" #include "tlog.h"
#include "taosdef.h"
#include "types.h" #include "types.h"
#define GREEN "\033[1;32m" #define GREEN "\033[1;32m"
@ -36,11 +36,7 @@
#define MAX_CONSUMER_THREAD_CNT (16) #define MAX_CONSUMER_THREAD_CNT (16)
#define MAX_VGROUP_CNT (32) #define MAX_VGROUP_CNT (32)
typedef enum { typedef enum { NOTIFY_CMD_START_CONSUM, NOTIFY_CMD_START_COMMIT, NOTIFY_CMD_ID_BUTT } NOTIFY_CMD_ID;
NOTIFY_CMD_START_CONSUM,
NOTIFY_CMD_START_COMMIT,
NOTIFY_CMD_ID_BUTT
}NOTIFY_CMD_ID;
typedef struct { typedef struct {
TdThread thread; TdThread thread;
@ -52,8 +48,8 @@ typedef struct {
// char autoOffsetRest[16]; // none, earliest, latest // char autoOffsetRest[16]; // none, earliest, latest
TdFilePtr pConsumeRowsFile; TdFilePtr pConsumeRowsFile;
int32_t ifCheckData; int32_t ifCheckData;
int64_t expectMsgCnt; int64_t expectMsgCnt;
int64_t consumeMsgCnt; int64_t consumeMsgCnt;
int64_t consumeRowCnt; int64_t consumeRowCnt;
@ -89,6 +85,7 @@ typedef struct {
int32_t saveRowFlag; int32_t saveRowFlag;
int32_t consumeDelay; // unit s int32_t consumeDelay; // unit s
int32_t numOfThread; int32_t numOfThread;
int32_t useSnapshot;
SThreadInfo stThreads[MAX_CONSUMER_THREAD_CNT]; SThreadInfo stThreads[MAX_CONSUMER_THREAD_CNT];
} SConfInfo; } SConfInfo;
@ -96,6 +93,8 @@ static SConfInfo g_stConfInfo;
TdFilePtr g_fp = NULL; TdFilePtr g_fp = NULL;
static int running = 1; static int running = 1;
int8_t useSnapshot = 0;
// char* g_pRowValue = NULL; // char* g_pRowValue = NULL;
// TdFilePtr g_fp = NULL; // TdFilePtr g_fp = NULL;
@ -205,6 +204,8 @@ void parseArgument(int32_t argc, char* argv[]) {
g_stConfInfo.saveRowFlag = atol(argv[++i]); g_stConfInfo.saveRowFlag = atol(argv[++i]);
} else if (strcmp(argv[i], "-y") == 0) { } else if (strcmp(argv[i], "-y") == 0) {
g_stConfInfo.consumeDelay = atol(argv[++i]); g_stConfInfo.consumeDelay = atol(argv[++i]);
} else if (strcmp(argv[i], "-e") == 0) {
useSnapshot = (int8_t)atol(argv[++i]);
} else { } else {
pError("%s unknow para: %s %s", GREEN, argv[++i], NC); pError("%s unknow para: %s %s", GREEN, argv[++i], NC);
exit(-1); exit(-1);
@ -298,11 +299,11 @@ int32_t saveConsumeContentToTbl(SThreadInfo* pInfo, char* buf) {
return 0; return 0;
} }
static char *shellFormatTimestamp(char *buf, int64_t val, int32_t precision) { static char* shellFormatTimestamp(char* buf, int64_t val, int32_t precision) {
//if (shell.args.is_raw_time) { // if (shell.args.is_raw_time) {
// sprintf(buf, "%" PRId64, val); // sprintf(buf, "%" PRId64, val);
// return buf; // return buf;
//} // }
time_t tt; time_t tt;
int32_t ms = 0; int32_t ms = 0;
@ -340,7 +341,7 @@ static char *shellFormatTimestamp(char *buf, int64_t val, int32_t precision) {
} }
} }
struct tm *ptm = taosLocalTime(&tt, NULL); struct tm* ptm = taosLocalTime(&tt, NULL);
size_t pos = strftime(buf, 35, "%Y-%m-%d %H:%M:%S", ptm); size_t pos = strftime(buf, 35, "%Y-%m-%d %H:%M:%S", ptm);
if (precision == TSDB_TIME_PRECISION_NANO) { if (precision == TSDB_TIME_PRECISION_NANO) {
@ -354,7 +355,8 @@ static char *shellFormatTimestamp(char *buf, int64_t val, int32_t precision) {
return buf; return buf;
} }
static void shellDumpFieldToFile(TdFilePtr pFile, const char *val, TAOS_FIELD *field, int32_t length, int32_t precision) { static void shellDumpFieldToFile(TdFilePtr pFile, const char* val, TAOS_FIELD* field, int32_t length,
int32_t precision) {
if (val == NULL) { if (val == NULL) {
taosFprintfFile(pFile, "%s", TSDB_DATA_NULL_STR); taosFprintfFile(pFile, "%s", TSDB_DATA_NULL_STR);
return; return;
@ -364,31 +366,31 @@ static void shellDumpFieldToFile(TdFilePtr pFile, const char *val, TAOS_FIELD *f
char buf[TSDB_MAX_BYTES_PER_ROW]; char buf[TSDB_MAX_BYTES_PER_ROW];
switch (field->type) { switch (field->type) {
case TSDB_DATA_TYPE_BOOL: case TSDB_DATA_TYPE_BOOL:
taosFprintfFile(pFile, "%d", ((((int32_t)(*((char *)val))) == 1) ? 1 : 0)); taosFprintfFile(pFile, "%d", ((((int32_t)(*((char*)val))) == 1) ? 1 : 0));
break; break;
case TSDB_DATA_TYPE_TINYINT: case TSDB_DATA_TYPE_TINYINT:
taosFprintfFile(pFile, "%d", *((int8_t *)val)); taosFprintfFile(pFile, "%d", *((int8_t*)val));
break; break;
case TSDB_DATA_TYPE_UTINYINT: case TSDB_DATA_TYPE_UTINYINT:
taosFprintfFile(pFile, "%u", *((uint8_t *)val)); taosFprintfFile(pFile, "%u", *((uint8_t*)val));
break; break;
case TSDB_DATA_TYPE_SMALLINT: case TSDB_DATA_TYPE_SMALLINT:
taosFprintfFile(pFile, "%d", *((int16_t *)val)); taosFprintfFile(pFile, "%d", *((int16_t*)val));
break; break;
case TSDB_DATA_TYPE_USMALLINT: case TSDB_DATA_TYPE_USMALLINT:
taosFprintfFile(pFile, "%u", *((uint16_t *)val)); taosFprintfFile(pFile, "%u", *((uint16_t*)val));
break; break;
case TSDB_DATA_TYPE_INT: case TSDB_DATA_TYPE_INT:
taosFprintfFile(pFile, "%d", *((int32_t *)val)); taosFprintfFile(pFile, "%d", *((int32_t*)val));
break; break;
case TSDB_DATA_TYPE_UINT: case TSDB_DATA_TYPE_UINT:
taosFprintfFile(pFile, "%u", *((uint32_t *)val)); taosFprintfFile(pFile, "%u", *((uint32_t*)val));
break; break;
case TSDB_DATA_TYPE_BIGINT: case TSDB_DATA_TYPE_BIGINT:
taosFprintfFile(pFile, "%" PRId64, *((int64_t *)val)); taosFprintfFile(pFile, "%" PRId64, *((int64_t*)val));
break; break;
case TSDB_DATA_TYPE_UBIGINT: case TSDB_DATA_TYPE_UBIGINT:
taosFprintfFile(pFile, "%" PRIu64, *((uint64_t *)val)); taosFprintfFile(pFile, "%" PRIu64, *((uint64_t*)val));
break; break;
case TSDB_DATA_TYPE_FLOAT: case TSDB_DATA_TYPE_FLOAT:
taosFprintfFile(pFile, "%.5f", GET_FLOAT_VAL(val)); taosFprintfFile(pFile, "%.5f", GET_FLOAT_VAL(val));
@ -409,7 +411,7 @@ static void shellDumpFieldToFile(TdFilePtr pFile, const char *val, TAOS_FIELD *f
taosFprintfFile(pFile, "\'%s\'", buf); taosFprintfFile(pFile, "\'%s\'", buf);
break; break;
case TSDB_DATA_TYPE_TIMESTAMP: case TSDB_DATA_TYPE_TIMESTAMP:
shellFormatTimestamp(buf, *(int64_t *)val, precision); shellFormatTimestamp(buf, *(int64_t*)val, precision);
taosFprintfFile(pFile, "'%s'", buf); taosFprintfFile(pFile, "'%s'", buf);
break; break;
default: default:
@ -417,12 +419,13 @@ static void shellDumpFieldToFile(TdFilePtr pFile, const char *val, TAOS_FIELD *f
} }
} }
static void dumpToFileForCheck(TdFilePtr pFile, TAOS_ROW row, TAOS_FIELD* fields, int32_t* length, int32_t num_fields, int32_t precision) { static void dumpToFileForCheck(TdFilePtr pFile, TAOS_ROW row, TAOS_FIELD* fields, int32_t* length, int32_t num_fields,
int32_t precision) {
for (int32_t i = 0; i < num_fields; i++) { for (int32_t i = 0; i < num_fields; i++) {
if (i > 0) { if (i > 0) {
taosFprintfFile(pFile, "\n"); taosFprintfFile(pFile, "\n");
} }
shellDumpFieldToFile(pFile, (const char *)row[i], fields + i, length[i], precision); shellDumpFieldToFile(pFile, (const char*)row[i], fields + i, length[i], precision);
} }
taosFprintfFile(pFile, "\n"); taosFprintfFile(pFile, "\n");
} }
@ -432,31 +435,32 @@ static int32_t msg_process(TAOS_RES* msg, SThreadInfo* pInfo, int32_t msgIndex)
int32_t totalRows = 0; int32_t totalRows = 0;
// printf("topic: %s\n", tmq_get_topic_name(msg)); // printf("topic: %s\n", tmq_get_topic_name(msg));
int32_t vgroupId = tmq_get_vgroup_id(msg); int32_t vgroupId = tmq_get_vgroup_id(msg);
const char* dbName = tmq_get_db_name(msg); const char* dbName = tmq_get_db_name(msg);
taosFprintfFile(g_fp, "consumerId: %d, msg index:%" PRId64 "\n", pInfo->consumerId, msgIndex); taosFprintfFile(g_fp, "consumerId: %d, msg index:%" PRId64 "\n", pInfo->consumerId, msgIndex);
taosFprintfFile(g_fp, "dbName: %s, topic: %s, vgroupId: %d\n", dbName != NULL ? dbName : "invalid table", tmq_get_topic_name(msg), vgroupId); taosFprintfFile(g_fp, "dbName: %s, topic: %s, vgroupId: %d\n", dbName != NULL ? dbName : "invalid table",
tmq_get_topic_name(msg), vgroupId);
while (1) { while (1) {
TAOS_ROW row = taos_fetch_row(msg); TAOS_ROW row = taos_fetch_row(msg);
if (row == NULL) break; if (row == NULL) break;
TAOS_FIELD* fields = taos_fetch_fields(msg); TAOS_FIELD* fields = taos_fetch_fields(msg);
int32_t numOfFields = taos_field_count(msg); int32_t numOfFields = taos_field_count(msg);
int32_t* length = taos_fetch_lengths(msg); int32_t* length = taos_fetch_lengths(msg);
int32_t precision = taos_result_precision(msg); int32_t precision = taos_result_precision(msg);
const char* tbName = tmq_get_table_name(msg); const char* tbName = tmq_get_table_name(msg);
dumpToFileForCheck(pInfo->pConsumeRowsFile, row, fields, length, numOfFields, precision); dumpToFileForCheck(pInfo->pConsumeRowsFile, row, fields, length, numOfFields, precision);
taos_print_row(buf, row, fields, numOfFields); taos_print_row(buf, row, fields, numOfFields);
if (0 != g_stConfInfo.showRowFlag) { if (0 != g_stConfInfo.showRowFlag) {
taosFprintfFile(g_fp, "tbname:%s, rows[%d]: %s\n", (tbName != NULL ? tbName : "null table"), totalRows, buf); taosFprintfFile(g_fp, "tbname:%s, rows[%d]: %s\n", (tbName != NULL ? tbName : "null table"), totalRows, buf);
//if (0 != g_stConfInfo.saveRowFlag) { // if (0 != g_stConfInfo.saveRowFlag) {
// saveConsumeContentToTbl(pInfo, buf); // saveConsumeContentToTbl(pInfo, buf);
//} // }
} }
totalRows++; totalRows++;
@ -479,8 +483,7 @@ int queryDB(TAOS* taos, char* command) {
return 0; return 0;
} }
static void appNothing(void* param, TAOS_RES* res, int32_t numOfRows) { static void appNothing(void* param, TAOS_RES* res, int32_t numOfRows) {}
}
int32_t notifyMainScript(SThreadInfo* pInfo, int32_t cmdId) { int32_t notifyMainScript(SThreadInfo* pInfo, int32_t cmdId) {
char sqlStr[1024] = {0}; char sqlStr[1024] = {0};
@ -488,11 +491,8 @@ int32_t notifyMainScript(SThreadInfo* pInfo, int32_t cmdId) {
int64_t now = taosGetTimestampMs(); int64_t now = taosGetTimestampMs();
// schema: ts timestamp, consumerid int, consummsgcnt bigint, checkresult int // schema: ts timestamp, consumerid int, consummsgcnt bigint, checkresult int
sprintf(sqlStr, "insert into %s.notifyinfo values (%"PRId64", %d, %d)", sprintf(sqlStr, "insert into %s.notifyinfo values (%" PRId64 ", %d, %d)", g_stConfInfo.cdbName, now, cmdId,
g_stConfInfo.cdbName, pInfo->consumerId);
now,
cmdId,
pInfo->consumerId);
taos_query_a(pInfo->taos, sqlStr, appNothing, NULL); taos_query_a(pInfo->taos, sqlStr, appNothing, NULL);
@ -502,12 +502,12 @@ int32_t notifyMainScript(SThreadInfo* pInfo, int32_t cmdId) {
} }
static int32_t g_once_commit_flag = 0; static int32_t g_once_commit_flag = 0;
static void tmq_commit_cb_print(tmq_t* tmq, int32_t code, void* param) { static void tmq_commit_cb_print(tmq_t* tmq, int32_t code, void* param) {
pError("tmq_commit_cb_print() commit %d\n", code); pError("tmq_commit_cb_print() commit %d\n", code);
if (0 == g_once_commit_flag) { if (0 == g_once_commit_flag) {
g_once_commit_flag = 1; g_once_commit_flag = 1;
notifyMainScript((SThreadInfo*)param, (int32_t)NOTIFY_CMD_START_COMMIT); notifyMainScript((SThreadInfo*)param, (int32_t)NOTIFY_CMD_START_COMMIT);
} }
taosFprintfFile(g_fp, "tmq_commit_cb_print() be called\n"); taosFprintfFile(g_fp, "tmq_commit_cb_print() be called\n");
} }
@ -541,6 +541,10 @@ void build_consumer(SThreadInfo* pInfo) {
// tmq_conf_set(conf, "auto.offset.reset", "none"); // tmq_conf_set(conf, "auto.offset.reset", "none");
// tmq_conf_set(conf, "auto.offset.reset", "earliest"); // tmq_conf_set(conf, "auto.offset.reset", "earliest");
// tmq_conf_set(conf, "auto.offset.reset", "latest"); // tmq_conf_set(conf, "auto.offset.reset", "latest");
//
if (useSnapshot) {
tmq_conf_set(conf, "experiment.use.snapshot", "true");
}
pInfo->tmq = tmq_consumer_new(conf, NULL, 0); pInfo->tmq = tmq_consumer_new(conf, NULL, 0);
@ -597,10 +601,11 @@ void loop_consume(SThreadInfo* pInfo) {
pInfo->ts = taosGetTimestampMs(); pInfo->ts = taosGetTimestampMs();
if (pInfo->ifCheckData) { if (pInfo->ifCheckData) {
char filename[256] = {0}; char filename[256] = {0};
char tmpString[128]; char tmpString[128];
//sprintf(filename, "%s/../log/consumerid_%d_%s.txt", configDir, pInfo->consumerId, getCurrentTimeString(tmpString)); // sprintf(filename, "%s/../log/consumerid_%d_%s.txt", configDir, pInfo->consumerId,
sprintf(filename, "%s/../log/consumerid_%d.txt", configDir, pInfo->consumerId); // getCurrentTimeString(tmpString));
sprintf(filename, "%s/../log/consumerid_%d.txt", configDir, pInfo->consumerId);
pInfo->pConsumeRowsFile = taosOpenFile(filename, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC | TD_FILE_STREAM); pInfo->pConsumeRowsFile = taosOpenFile(filename, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC | TD_FILE_STREAM);
if (pInfo->pConsumeRowsFile == NULL) { if (pInfo->pConsumeRowsFile == NULL) {
taosFprintfFile(g_fp, "%s create file fail for save rows content\n", getCurrentTimeString(tmpString)); taosFprintfFile(g_fp, "%s create file fail for save rows content\n", getCurrentTimeString(tmpString));
@ -619,10 +624,10 @@ void loop_consume(SThreadInfo* pInfo) {
totalMsgs++; totalMsgs++;
if (0 == once_flag) { if (0 == once_flag) {
once_flag = 1; once_flag = 1;
notifyMainScript(pInfo, NOTIFY_CMD_START_CONSUM); notifyMainScript(pInfo, NOTIFY_CMD_START_CONSUM);
} }
if (totalRows >= pInfo->expectMsgCnt) { if (totalRows >= pInfo->expectMsgCnt) {
char tmpString[128]; char tmpString[128];
@ -651,7 +656,7 @@ void* consumeThreadFunc(void* param) {
pInfo->taos = taos_connect(NULL, "root", "taosdata", NULL, 0); pInfo->taos = taos_connect(NULL, "root", "taosdata", NULL, 0);
if (pInfo->taos == NULL) { if (pInfo->taos == NULL) {
taosFprintfFile(g_fp, "taos_connect() fail, can not notify and save consume result to main scripte\n"); taosFprintfFile(g_fp, "taos_connect() fail, can not notify and save consume result to main scripte\n");
exit(-1); exit(-1);
} }
build_consumer(pInfo); build_consumer(pInfo);