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

other: merge 3.0

This commit is contained in:
Haojun Liao 2022-06-23 10:37:07 +08:00
parent ce7e7fe23d 9012834531
commit 2d31d3cb43
59 changed files with 1313 additions and 475 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
include/common/tcommon.h
View File

@ -52,6 +52,8 @@ typedef enum EStreamType {
typedef struct {
SArray* pTableList;
SHashObj* map; // speedup acquire the tableQueryInfo by table uid
void* pTagCond;
void* pTagIndexCond;
} STableListInfo;
typedef struct SColumnDataAgg {

3
include/common/tdatablock.h
View File

@ -71,7 +71,8 @@ SEpSet getEpSet_s(SCorEpSet* pEpSet);
#define colDataGetData(p1_, r_) \
((IS_VAR_DATA_TYPE((p1_)->info.type)) ? colDataGetVarData(p1_, r_) : colDataGetNumData(p1_, r_))
#define IS_JSON_NULL(type, data) ((type) == TSDB_DATA_TYPE_JSON && *(data) == TSDB_DATA_TYPE_NULL)
#define IS_JSON_NULL(type, data) ((type) == TSDB_DATA_TYPE_JSON && \
(*(data) == TSDB_DATA_TYPE_NULL || tTagIsJsonNull(data)))
static FORCE_INLINE bool colDataIsNull_s(const SColumnInfoData* pColumnInfoData, uint32_t row) {
if (!pColumnInfoData->hasNull) {

2
include/common/tdataformat.h
View File

@ -70,6 +70,8 @@ int32_t tGetTSRow(uint8_t *p, STSRow2 *pRow);
// STag
int32_t tTagNew(SArray *pArray, int32_t version, int8_t isJson, STag **ppTag);
void tTagFree(STag *pTag);
bool tTagIsJson(const void *pTag);
bool tTagIsJsonNull(void *tagVal);
bool tTagGet(const STag *pTag, STagVal *pTagVal);
char *tTagValToData(const STagVal *pTagVal, bool isJson);
int32_t tEncodeTag(SEncoder *pEncoder, const STag *pTag);

2
include/common/tglobal.h
View File

@ -137,6 +137,8 @@ extern bool tsSmlDataFormat;
// internal
extern int32_t tsTransPullupInterval;
extern int32_t tsMqRebalanceInterval;
extern int32_t tsTtlUnit;
extern int32_t tsTtlPushInterval;
#define NEEDTO_COMPRESSS_MSG(size) (tsCompressMsgSize != -1 && (size) > tsCompressMsgSize)

1
include/common/tmsgdef.h
View File

@ -145,6 +145,7 @@ enum {
TD_DEF_MSG_TYPE(TDMT_MND_MQ_TIMER, "mq-tmr", SMTimerReq, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_TELEM_TIMER, "telem-tmr", SMTimerReq, SMTimerReq)
TD_DEF_MSG_TYPE(TDMT_MND_TRANS_TIMER, "trans-tmr", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_TTL_TIMER, "ttl-tmr", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_KILL_TRANS, "kill-trans", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_KILL_QUERY, "kill-query", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_KILL_CONN, "kill-conn", NULL, NULL)

1
include/libs/nodes/nodes.h
View File

@ -226,6 +226,7 @@ typedef enum ENodeType {
QUERY_NODE_PHYSICAL_PLAN_EXCHANGE,
QUERY_NODE_PHYSICAL_PLAN_MERGE,
QUERY_NODE_PHYSICAL_PLAN_SORT,
QUERY_NODE_PHYSICAL_PLAN_GROUP_SORT,
QUERY_NODE_PHYSICAL_PLAN_HASH_INTERVAL,
QUERY_NODE_PHYSICAL_PLAN_MERGE_INTERVAL,
QUERY_NODE_PHYSICAL_PLAN_MERGE_ALIGNED_INTERVAL,

4
include/libs/nodes/plannodes.h
View File

@ -66,6 +66,7 @@ typedef struct SScanLogicNode {
int8_t intervalUnit;
int8_t slidingUnit;
SNode* pTagCond;
SNode* pTagIndexCond;
int8_t triggerType;
int64_t watermark;
int16_t tsColId;
@ -420,6 +421,8 @@ typedef struct SSortPhysiNode {
SNodeList* pTargets;
} SSortPhysiNode;
typedef SSortPhysiNode SGroupSortPhysiNode;
typedef struct SPartitionPhysiNode {
SPhysiNode node;
SNodeList* pExprs; // these are expression list of partition_by_clause
@ -466,6 +469,7 @@ typedef struct SSubplan {
SPhysiNode* pNode; // physical plan of current subplan
SDataSinkNode* pDataSink; // data of the subplan flow into the datasink
SNode* pTagCond;
SNode* pTagIndexCond;
} SSubplan;
typedef enum EExplainMode { EXPLAIN_MODE_DISABLE = 1, EXPLAIN_MODE_STATIC, EXPLAIN_MODE_ANALYZE } EExplainMode;

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

@ -288,11 +288,11 @@ typedef enum ESqlClause {
} ESqlClause;
typedef struct SDeleteStmt {
ENodeType type; // QUERY_NODE_DELETE_STMT
SNode* pFromTable; // FROM clause
SNode* pWhere; // WHERE clause
SNode* pCountFunc; // count the number of rows affected
SNode* pTagIndexCond; // pWhere divided into pTagIndexCond and timeRange
ENodeType type; // QUERY_NODE_DELETE_STMT
SNode* pFromTable; // FROM clause
SNode* pWhere; // WHERE clause
SNode* pCountFunc; // count the number of rows affected
SNode* pTagCond; // pWhere divided into pTagCond and timeRange
STimeWindow timeRange;
uint8_t precision;
bool deleteZeroRows;
@ -397,7 +397,8 @@ void nodesValueNodeToVariant(const SValueNode* pNode, SVariant* pVal);
char* nodesGetFillModeString(EFillMode mode);
int32_t nodesMergeConds(SNode** pDst, SNodeList** pSrc);
int32_t nodesPartitionCond(SNode** pCondition, SNode** pPrimaryKeyCond, SNode** pTagCond, SNode** pOtherCond);
int32_t nodesPartitionCond(SNode** pCondition, SNode** pPrimaryKeyCond, SNode** pTagIndexCond, SNode** pTagCond,
SNode** pOtherCond);
#ifdef __cplusplus
}

9
include/libs/stream/tstream.h
View File

@ -239,17 +239,14 @@ typedef struct {
struct SStreamTask {
int64_t streamId;
int32_t taskId;
int8_t inputType;
int8_t taskStatus;
int8_t execStatus;
int8_t isDataScan;
int8_t execType;
int8_t sinkType;
int8_t dispatchType;
int16_t dispatchMsgType;
int8_t isDataScan;
int8_t taskStatus;
int8_t execStatus;
// node info
int32_t selfChildId;

2
include/os/os.h
View File

@ -104,8 +104,6 @@ extern "C" {
#include "osTimezone.h"
#include "osEnv.h"
void osDefaultInit();
#ifdef __cplusplus
}
#endif

4
source/client/src/clientImpl.c
View File

@ -1507,7 +1507,7 @@ static int32_t estimateJsonLen(SReqResultInfo* pResultInfo, int32_t numOfCols, i
char* jsonInnerData = data + CHAR_BYTES;
if (jsonInnerType == TSDB_DATA_TYPE_NULL) {
len += (VARSTR_HEADER_SIZE + strlen(TSDB_DATA_NULL_STR_L));
} else if (jsonInnerType & TD_TAG_JSON) {
} else if (tTagIsJson(data)) {
len += (VARSTR_HEADER_SIZE + ((const STag*)(data))->len);
} else if (jsonInnerType == TSDB_DATA_TYPE_NCHAR) { // value -> "value"
len += varDataTLen(jsonInnerData) + CHAR_BYTES * 2;
@ -1592,7 +1592,7 @@ static int32_t doConvertJson(SReqResultInfo* pResultInfo, int32_t numOfCols, int
if (jsonInnerType == TSDB_DATA_TYPE_NULL) {
sprintf(varDataVal(dst), "%s", TSDB_DATA_NULL_STR_L);
varDataSetLen(dst, strlen(varDataVal(dst)));
} else if (jsonInnerType & TD_TAG_JSON) {
} else if (tTagIsJson(data)) {
char* jsonString = parseTagDatatoJson(data);
STR_TO_VARSTR(dst, jsonString);
taosMemoryFree(jsonString);

2
source/common/src/tdatablock.c
View File

@ -110,7 +110,7 @@ int32_t getJsonValueLen(const char* data) {
dataLen = DOUBLE_BYTES + CHAR_BYTES;
} else if (*data == TSDB_DATA_TYPE_BOOL) {
dataLen = CHAR_BYTES + CHAR_BYTES;
} else if (*data & TD_TAG_JSON) { // json string
} else if (tTagIsJson(data)) { // json string
dataLen = ((STag*)(data))->len;
} else {
ASSERT(0);

12
source/common/src/tdataformat.c
View File

@ -924,6 +924,18 @@ static int32_t tGetTagVal(uint8_t *p, STagVal *pTagVal, int8_t isJson) {
return n;
}
bool tTagIsJson(const void *pTag){
return (((const STag *)pTag)->flags & TD_TAG_JSON);
}
bool tTagIsJsonNull(void *data){
STag *pTag = (STag*)data;
int8_t isJson = tTagIsJson(pTag);
if(!isJson) return false;
return ((STag*)data)->nTag == 0;
}
int32_t tTagNew(SArray *pArray, int32_t version, int8_t isJson, STag **ppTag) {
int32_t code = 0;
uint8_t *p = NULL;

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

@ -187,6 +187,9 @@ bool tsStartUdfd = true;
// internal
int32_t tsTransPullupInterval = 2;
int32_t tsMqRebalanceInterval = 2;
int32_t tsTtlUnit = 86400;
int32_t tsTtlPushInterval = 60;
void taosAddDataDir(int32_t index, char *v1, int32_t level, int32_t primary) {
tstrncpy(tsDiskCfg[index].dir, v1, TSDB_FILENAME_LEN);
@ -467,6 +470,8 @@ static int32_t taosAddServerCfg(SConfig *pCfg) {
if (cfgAddInt32(pCfg, "transPullupInterval", tsTransPullupInterval, 1, 10000, 1) != 0) return -1;
if (cfgAddInt32(pCfg, "mqRebalanceInterval", tsMqRebalanceInterval, 1, 10000, 1) != 0) return -1;
if (cfgAddInt32(pCfg, "ttlUnit", tsTtlUnit, 1, 86400*365, 1) != 0) return -1;
if (cfgAddInt32(pCfg, "ttlPushInterval", tsTtlPushInterval, 1, 10000, 1) != 0) return -1;
if (cfgAddBool(pCfg, "udf", tsStartUdfd, 0) != 0) return -1;
return 0;
@ -619,6 +624,8 @@ static int32_t taosSetServerCfg(SConfig *pCfg) {
tsTransPullupInterval = cfgGetItem(pCfg, "transPullupInterval")->i32;
tsMqRebalanceInterval = cfgGetItem(pCfg, "mqRebalanceInterval")->i32;
tsTtlUnit = cfgGetItem(pCfg, "ttlUnit")->i32;
tsTtlPushInterval = cfgGetItem(pCfg, "ttlPushInterval")->i32;
tsStartUdfd = cfgGetItem(pCfg, "udf")->bval;
@ -631,7 +638,7 @@ static int32_t taosSetServerCfg(SConfig *pCfg) {
int32_t taosCreateLog(const char *logname, int32_t logFileNum, const char *cfgDir, const char **envCmd,
const char *envFile, char *apolloUrl, SArray *pArgs, bool tsc) {
osDefaultInit();
if (tsCfg == NULL) osDefaultInit();
SConfig *pCfg = cfgInit();
if (pCfg == NULL) return -1;

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

@ -580,6 +580,7 @@ static int32_t mndDropDnode(SMnode *pMnode, SRpcMsg *pReq, SDnodeObj *pDnode, SM
mInfo("trans:%d, %d vnodes on dnode:%d will be dropped", pTrans->id, numOfVnodes, pDnode->id);
if (mndSetMoveVgroupsInfoToTrans(pMnode, pTrans, pDnode->id) != 0) goto _OVER;
}
if (mndTransPrepare(pMnode, pTrans) != 0) goto _OVER;
code = 0;

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

@ -65,6 +65,13 @@ static void mndPullupTrans(SMnode *pMnode) {
}
}
static void mndTtlTimer(SMnode *pMnode) {
int32_t contLen = 0;
void *pReq = mndBuildTimerMsg(&contLen);
SRpcMsg rpcMsg = {.msgType = TDMT_MND_TTL_TIMER, .pCont = pReq, .contLen = contLen};
tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, &rpcMsg);
}
static void mndCalMqRebalance(SMnode *pMnode) {
int32_t contLen = 0;
void *pReq = mndBuildTimerMsg(&contLen);
@ -83,41 +90,6 @@ static void mndPullupTelem(SMnode *pMnode) {
}
}
static void mndPushTtlTime(SMnode *pMnode) {
SSdb *pSdb = pMnode->pSdb;
SVgObj *pVgroup = NULL;
void *pIter = NULL;
while (1) {
pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
if (pIter == NULL) break;
int32_t contLen = sizeof(SMsgHead) + sizeof(int32_t);
SMsgHead *pHead = rpcMallocCont(contLen);
if (pHead == NULL) {
sdbCancelFetch(pSdb, pIter);
sdbRelease(pSdb, pVgroup);
continue;
}
pHead->contLen = htonl(contLen);
pHead->vgId = htonl(pVgroup->vgId);
int32_t t = taosGetTimestampSec();
*(int32_t *)(POINTER_SHIFT(pHead, sizeof(SMsgHead))) = htonl(t);
SRpcMsg rpcMsg = {.msgType = TDMT_VND_DROP_TTL_TABLE, .pCont = pHead, .contLen = contLen};
SEpSet epSet = mndGetVgroupEpset(pMnode, pVgroup);
int32_t code = tmsgSendReq(&epSet, &rpcMsg);
if (code != 0) {
mError("failed to send ttl time seed msg, code:0x%x", code);
} else {
mInfo("send ttl time seed msg, time:%d", t);
}
sdbRelease(pSdb, pVgroup);
}
}
static void *mndThreadFp(void *param) {
SMnode *pMnode = param;
int64_t lastTime = 0;
@ -125,14 +97,13 @@ static void *mndThreadFp(void *param) {
while (1) {
lastTime++;
if (lastTime % (864000) == 0) { // sleep 1 day for ttl
mndPushTtlTime(pMnode);
}
taosMsleep(100);
if (mndGetStop(pMnode)) break;
if (lastTime % (tsTransPullupInterval * 10) == 1) {
mndTtlTimer(pMnode);
}
if (lastTime % (tsTransPullupInterval * 10) == 0) {
mndPullupTrans(pMnode);
}
@ -558,12 +529,12 @@ static int32_t mndCheckMnodeState(SRpcMsg *pMsg) {
if (!IsReq(pMsg)) return 0;
if (mndAcquireRpcRef(pMsg->info.node) == 0) return 0;
if (pMsg->msgType == TDMT_MND_MQ_TIMER || pMsg->msgType == TDMT_MND_TELEM_TIMER ||
pMsg->msgType == TDMT_MND_TRANS_TIMER) {
pMsg->msgType == TDMT_MND_TRANS_TIMER || TDMT_MND_TTL_TIMER) {
return -1;
}
const STraceId *trace = &pMsg->info.traceId;
mGError("msg:%p, failed to check mnode state since %s, type:%s", pMsg, terrstr(), TMSG_INFO(pMsg->msgType));
mError("msg:%p, failed to check mnode state since %s, type:%s", pMsg, terrstr(), TMSG_INFO(pMsg->msgType));
SEpSet epSet = {0};
mndGetMnodeEpSet(pMsg->info.node, &epSet);

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

@ -270,10 +270,8 @@ int32_t mndAddShuffleSinkTasksToStream(SMnode* pMnode, STrans* pTrans, SStreamOb
pTask->nodeId = pVgroup->vgId;
pTask->epSet = mndGetVgroupEpset(pMnode, pVgroup);
pTask->isDataScan = 0;
// source
pTask->inputType = TASK_INPUT_TYPE__DATA_BLOCK;
pTask->isDataScan = 0;
// exec
pTask->execType = TASK_EXEC__NONE;
@ -320,10 +318,8 @@ int32_t mndAddFixedSinkTaskToStream(SMnode* pMnode, STrans* pTrans, SStreamObj*
#endif
pTask->epSet = mndGetVgroupEpset(pMnode, &pStream->fixedSinkVg);
pTask->isDataScan = 0;
// source
pTask->inputType = TASK_INPUT_TYPE__DATA_BLOCK;
pTask->isDataScan = 0;
// exec
pTask->execType = TASK_EXEC__NONE;
@ -392,12 +388,10 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
pInnerTask = tNewSStreamTask(pStream->uid);
mndAddTaskToTaskSet(taskInnerLevel, pInnerTask);
pInnerTask->isDataScan = 0;
pInnerTask->childEpInfo = taosArrayInit(0, sizeof(void*));
// input
pInnerTask->inputType = TASK_INPUT_TYPE__DATA_BLOCK;
// source
pInnerTask->isDataScan = 0;
// trigger
pInnerTask->triggerParam = pStream->triggerParam;
@ -458,11 +452,9 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
SStreamTask* pTask = tNewSStreamTask(pStream->uid);
mndAddTaskToTaskSet(taskSourceLevel, pTask);
// source
pTask->isDataScan = 1;
// input
pTask->inputType = TASK_INPUT_TYPE__SUMBIT_BLOCK;
// add fixed vg dispatch
pTask->sinkType = TASK_SINK__NONE;
pTask->dispatchMsgType = TDMT_STREAM_TASK_DISPATCH;
@ -517,10 +509,8 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
SStreamTask* pTask = tNewSStreamTask(pStream->uid);
mndAddTaskToTaskSet(taskOneLevel, pTask);
pTask->isDataScan = 1;
// input
pTask->inputType = TASK_INPUT_TYPE__SUMBIT_BLOCK;
pTask->isDataScan = 1;
// trigger
pTask->triggerParam = pStream->triggerParam;

39
source/dnode/mnode/impl/src/mndStb.c
View File

@ -37,6 +37,7 @@ static SSdbRow *mndStbActionDecode(SSdbRaw *pRaw);
static int32_t mndStbActionInsert(SSdb *pSdb, SStbObj *pStb);
static int32_t mndStbActionDelete(SSdb *pSdb, SStbObj *pStb);
static int32_t mndStbActionUpdate(SSdb *pSdb, SStbObj *pOld, SStbObj *pNew);
static int32_t mndProcessTtlTimer(SRpcMsg *pReq);
static int32_t mndProcessCreateStbReq(SRpcMsg *pReq);
static int32_t mndProcessAlterStbReq(SRpcMsg *pReq);
static int32_t mndProcessDropStbReq(SRpcMsg *pReq);
@ -63,6 +64,7 @@ int32_t mndInitStb(SMnode *pMnode) {
mndSetMsgHandle(pMnode, TDMT_VND_ALTER_STB_RSP, mndTransProcessRsp);
mndSetMsgHandle(pMnode, TDMT_VND_DROP_STB_RSP, mndTransProcessRsp);
mndSetMsgHandle(pMnode, TDMT_MND_TABLE_META, mndProcessTableMetaReq);
mndSetMsgHandle(pMnode, TDMT_MND_TTL_TIMER, mndProcessTtlTimer);
mndSetMsgHandle(pMnode, TDMT_MND_TABLE_CFG, mndProcessTableCfgReq);
mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_STB, mndRetrieveStb);
@ -799,6 +801,43 @@ int32_t mndAddStbToTrans(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *p
return 0;
}
static int32_t mndProcessTtlTimer(SRpcMsg *pReq) {
SMnode *pMnode = pReq->info.node;
SSdb *pSdb = pMnode->pSdb;
SVgObj *pVgroup = NULL;
void *pIter = NULL;
while (1) {
pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
if (pIter == NULL) break;
int32_t contLen = sizeof(SMsgHead) + sizeof(int32_t);
SMsgHead *pHead = rpcMallocCont(contLen);
if (pHead == NULL) {
sdbCancelFetch(pSdb, pVgroup);
sdbRelease(pSdb, pVgroup);
continue;
}
pHead->contLen = htonl(contLen);
pHead->vgId = htonl(pVgroup->vgId);
int32_t t = taosGetTimestampSec();
*(int32_t *)((char *)pHead + sizeof(SMsgHead)) = htonl(t);
SRpcMsg rpcMsg = {.msgType = TDMT_VND_DROP_TTL_TABLE, .pCont = pHead, .contLen = contLen};
SEpSet epSet = mndGetVgroupEpset(pMnode, pVgroup);
int32_t code = tmsgSendReq(&epSet, &rpcMsg);
if (code != 0) {
mError("failed to send ttl time seed, code:0x%x", code);
} else {
mDebug("send ttl time seed success, time:%d", t);
}
sdbRelease(pSdb, pVgroup);
}
return 0;
}
static int32_t mndProcessCreateStbReq(SRpcMsg *pReq) {
SMnode *pMnode = pReq->info.node;
int32_t code = -1;

1
source/dnode/mnode/impl/src/mndTrans.c
View File

@ -56,6 +56,7 @@ static bool mndCannotExecuteTransAction(SMnode *pMnode) { return !pMnode->dep
static void mndTransSendRpcRsp(SMnode *pMnode, STrans *pTrans);
static int32_t mndProcessTransReq(SRpcMsg *pReq);
static int32_t mndProcessTtl(SRpcMsg *pReq);
static int32_t mndProcessKillTransReq(SRpcMsg *pReq);
static int32_t mndRetrieveTrans(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows);

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

@ -1553,10 +1553,11 @@ static int32_t mndSetBalanceVgroupInfoToTrans(SMnode *pMnode, STrans *pTrans, SD
static int32_t mndBalanceVgroupBetweenDnode(SMnode *pMnode, STrans *pTrans, SDnodeObj *pSrc, SDnodeObj *pDst) {
void *pIter = NULL;
int32_t code = -1;
SSdb *pSdb = pMnode->pSdb;
while (1) {
SVgObj *pVgroup = NULL;
pIter = sdbFetch(pMnode->pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
if (pIter == NULL) break;
bool existInSrc = false;
@ -1568,13 +1569,15 @@ static int32_t mndBalanceVgroupBetweenDnode(SMnode *pMnode, STrans *pTrans, SDno
}
if (!existInSrc || existInDst) {
sdbRelease(pMnode->pSdb, pVgroup);
sdbRelease(pSdb, pVgroup);
continue;
}
SDbObj *pDb = mndAcquireDb(pMnode, pVgroup->dbName);
code = mndSetBalanceVgroupInfoToTrans(pMnode, pTrans, pDb, pVgroup, pSrc, pDst);
mndReleaseDb(pMnode, pDb);
sdbRelease(pMnode->pSdb, pVgroup);
sdbRelease(pSdb, pVgroup);
sdbCancelFetch(pSdb, pIter);
break;
}
@ -1593,15 +1596,25 @@ static int32_t mndBalanceVgroup(SMnode *pMnode, SRpcMsg *pReq, SArray *pArray) {
while (1) {
taosArraySort(pArray, (__compar_fn_t)mndCompareDnodeVnodes);
SDnodeObj *pSrc = taosArrayGet(pArray, 0);
SDnodeObj *pDst = taosArrayGet(pArray, taosArrayGetSize(pArray) - 1);
for (int32_t i = 0; i < taosArrayGetSize(pArray); ++i) {
SDnodeObj *pDnode = taosArrayGet(pArray, i);
mDebug("dnode:%d, equivalent vnodes:%d support:%d, score:%f", pDnode->id, pDnode->numOfVnodes,
pDnode->numOfSupportVnodes, (float)pDnode->numOfVnodes / pDnode->numOfSupportVnodes);
}
SDnodeObj *pSrc = taosArrayGet(pArray, taosArrayGetSize(pArray) - 1);
SDnodeObj *pDst = taosArrayGet(pArray, 0);
float srcScore = (float)(pSrc->numOfVnodes - 1) / pSrc->numOfSupportVnodes;
float dstScore = (float)(pDst->numOfVnodes + 1) / pDst->numOfSupportVnodes;
if (srcScore + 0.0001 < dstScore) {
mDebug("trans:%d, balance vgroup from dnode:%d to dnode:%d", pTrans->id, pSrc->id, pDst->id);
mDebug("trans:%d, after balance, src dnode:%d score:%f, dst dnode:%d score:%f", pTrans->id, pSrc->id, srcScore,
pDst->id, dstScore);
if (srcScore > dstScore - 0.000001) {
code = mndBalanceVgroupBetweenDnode(pMnode, pTrans, pSrc, pDst);
if (code == 0) {
pSrc->numOfVnodes--;
pDst->numOfVnodes++;
numOfVgroups++;
continue;
} else {
@ -1635,7 +1648,13 @@ static int32_t mndProcessBalanceVgroupMsg(SRpcMsg *pReq) {
void *pIter = NULL;
int64_t curMs = taosGetTimestampMs();
mDebug("start to balance vgroup");
SBalanceVgroupReq req = {0};
if (tDeserializeSBalanceVgroupReq(pReq->pCont, pReq->contLen, &req) != 0) {
terrno = TSDB_CODE_INVALID_MSG;
goto _OVER;
}
mInfo("start to balance vgroup");
if (mndCheckOperAuth(pMnode, pReq->info.conn.user, MND_OPER_BALANCE_VGROUP) != 0) goto _OVER;

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

@ -613,9 +613,6 @@ const void *metaGetTableTagVal(SMetaEntry *pEntry, int16_t type, STagVal *val) {
ASSERT(pEntry->type == TSDB_CHILD_TABLE);
STag *tag = (STag *)pEntry->ctbEntry.pTags;
if (type == TSDB_DATA_TYPE_JSON) {
if (tag->nTag == 0) {
return NULL;
}
return tag;
}
bool find = tTagGet(tag, val);

3
source/dnode/vnode/src/meta/metaTable.c
View File

@ -400,8 +400,7 @@ static void metaBuildTtlIdxKey(STtlIdxKey *ttlKey, const SMetaEntry *pME){
if (ttlDays <= 0) return;
ttlKey->dtime = ctime / 1000 + ttlDays * 24 * 60 * 60;
// ttlKey->dtime = ctime / 1000 + ttlDays;
ttlKey->dtime = ctime / 1000 + ttlDays * tsTtlUnit;
ttlKey->uid = pME->uid;
}

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

@ -426,6 +426,7 @@ int32_t tqProcessTaskDeployReq(STQ* pTq, char* msg, int32_t msgLen) {
ASSERT(0);
}
tDecoderClear(&decoder);
ASSERT(pTask->isDataScan == 0 || pTask->isDataScan == 1);
pTask->execStatus = TASK_EXEC_STATUS__IDLE;
@ -505,7 +506,7 @@ int32_t tqProcessStreamTrigger(STQ* pTq, SSubmitReq* pReq) {
if (atomic_load_8(&pTask->taskStatus) == TASK_STATUS__DROPPING) {
continue;
}
if (pTask->inputType != STREAM_INPUT__DATA_SUBMIT) continue;
if (!pTask->isDataScan) continue;
if (!failed) {
if (streamTaskInput(pTask, (SStreamQueueItem*)pSubmit) < 0) {

4
source/dnode/vnode/src/tq/tqRead.c
View File

@ -331,8 +331,8 @@ int32_t tqUpdateTbUidList(STQ* pTq, const SArray* tbUidList, bool isAdd) {
while (1) {
pIter = taosHashIterate(pTq->pStreamTasks, pIter);
if (pIter == NULL) break;
SStreamTask* pTask = (SStreamTask*)pIter;
if (pTask->inputType == STREAM_INPUT__DATA_SUBMIT) {
SStreamTask* pTask = *(SStreamTask**)pIter;
if (pTask->isDataScan) {
int32_t code = qUpdateQualifiedTableId(pTask->exec.executor, tbUidList, isAdd);
ASSERT(code == 0);
}

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

@ -315,7 +315,7 @@ static int32_t vnodeProcessDropTtlTbReq(SVnode *pVnode, int64_t version, void *p
if (tbUids == NULL) return TSDB_CODE_OUT_OF_MEMORY;
int32_t t = ntohl(*(int32_t *)pReq);
vError("rec ttl time:%d", t);
vDebug("vgId:%d, recv ttl msg, time:%d", pVnode->config.vgId, t);
int32_t ret = metaTtlDropTable(pVnode->pMeta, t, tbUids);
if (ret != 0) {
goto end;

3
source/libs/executor/inc/executil.h
View File

@ -106,7 +106,8 @@ int32_t getNumOfTotalRes(SGroupResInfo* pGroupResInfo);
SSDataBlock* createResDataBlock(SDataBlockDescNode* pNode);
int32_t getTableList(void* metaHandle, SScanPhysiNode* pScanNode, STableListInfo* pListInfo, SNode* pTagCond);
EDealRes doTranslateTagExpr(SNode** pNode, void* pContext);
int32_t getTableList(void* metaHandle, SScanPhysiNode* pScanNode, STableListInfo* pListInfo);
SArray* createSortInfo(SNodeList* pNodeList);
SArray* extractPartitionColInfo(SNodeList* pNodeList);
SArray* extractColMatchInfo(SNodeList* pNodeList, SDataBlockDescNode* pOutputNodeList, int32_t* numOfOutputCols, int32_t type);

10
source/libs/executor/inc/executorimpl.h
View File

@ -282,7 +282,6 @@ typedef struct STagScanInfo {
int32_t curPos;
SReadHandle readHandle;
STableListInfo *pTableList;
SNode* pFilterNode; // filter info,
} STagScanInfo;
typedef enum EStreamScanMode {
@ -841,11 +840,10 @@ int32_t finalizeResultRowIntoResultDataBlock(SDiskbasedBuf* pBuf, SResultRowPosi
int32_t createScanTableListInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* pHandle,
STableListInfo* pTableListInfo, uint64_t queryId, uint64_t taskId,
SNode* pTagCond);
int32_t doCreateMultipleDataReaders(STableScanPhysiNode* pTableScanNode, SReadHandle* pHandle,
STableListInfo* pTableListInfo, SArray* arrayReader, uint64_t queryId,
uint64_t taskId);
SOperatorInfo* createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SArray* dataReaders,
SReadHandle* readHandle, SExecTaskInfo* pTaskInfo);
SOperatorInfo* createGroupSortOperatorInfo(SOperatorInfo* downstream, SGroupSortPhysiNode* pSortPhyNode,
SExecTaskInfo* pTaskInfo);
SOperatorInfo* createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanNode, STableListInfo *pTableListInfo,
SReadHandle* readHandle, SExecTaskInfo* pTaskInfo, uint64_t queryId, uint64_t taskId);
void copyUpdateDataBlock(SSDataBlock* pDest, SSDataBlock* pSource, int32_t tsColIndex);

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

@ -204,28 +204,111 @@ SSDataBlock* createResDataBlock(SDataBlockDescNode* pNode) {
return pBlock;
}
int32_t getTableList(void* metaHandle, SScanPhysiNode* pScanNode, STableListInfo* pListInfo, SNode* pTagCond) {
EDealRes doTranslateTagExpr(SNode** pNode, void* pContext) {
SMetaReader* mr = (SMetaReader*)pContext;
if(nodeType(*pNode) == QUERY_NODE_COLUMN){
SColumnNode* pSColumnNode = *(SColumnNode**)pNode;
SValueNode *res = (SValueNode *)nodesMakeNode(QUERY_NODE_VALUE);
if (NULL == res) {
return DEAL_RES_ERROR;
}
res->translate = true;
res->node.resType = pSColumnNode->node.resType;
STagVal tagVal = {0};
tagVal.cid = pSColumnNode->colId;
const char* p = metaGetTableTagVal(&mr->me, pSColumnNode->node.resType.type, &tagVal);
if (p == NULL) {
res->node.resType.type = TSDB_DATA_TYPE_NULL;
}else if (pSColumnNode->node.resType.type == TSDB_DATA_TYPE_JSON) {
int32_t len = ((const STag*)p) -> len;
res->datum.p = taosMemoryCalloc(len + 1, 1);
memcpy(res->datum.p, p, len);
} else if (IS_VAR_DATA_TYPE(pSColumnNode->node.resType.type)) {
res->datum.p = taosMemoryCalloc(tagVal.nData + VARSTR_HEADER_SIZE + 1, 1);
memcpy(varDataVal(res->datum.p), tagVal.pData, tagVal.nData);
varDataSetLen(res->datum.p, tagVal.nData);
} else {
nodesSetValueNodeValue(res, &(tagVal.i64));
}
nodesDestroyNode(*pNode);
*pNode = (SNode*)res;
}else if (nodeType(*pNode) == QUERY_NODE_FUNCTION){
SFunctionNode * pFuncNode = *(SFunctionNode**)pNode;
if(pFuncNode->funcType == FUNCTION_TYPE_TBNAME){
SValueNode *res = (SValueNode *)nodesMakeNode(QUERY_NODE_VALUE);
if (NULL == res) {
return DEAL_RES_ERROR;
}
res->translate = true;
res->node.resType = pFuncNode->node.resType;
int32_t len = strlen(mr->me.name);
res->datum.p = taosMemoryCalloc(len + VARSTR_HEADER_SIZE + 1, 1);
memcpy(varDataVal(res->datum.p), mr->me.name, len);
varDataSetLen(res->datum.p, len);
nodesDestroyNode(*pNode);
*pNode = (SNode*)res;
}
}
return DEAL_RES_CONTINUE;
}
static bool isTableOk(STableKeyInfo* info, SNode *pTagCond, SMeta *metaHandle){
SMetaReader mr = {0};
metaReaderInit(&mr, metaHandle, 0);
metaGetTableEntryByUid(&mr, info->uid);
SNode *pTagCondTmp = nodesCloneNode(pTagCond);
nodesRewriteExprPostOrder(&pTagCondTmp, doTranslateTagExpr, &mr);
metaReaderClear(&mr);
SNode* pNew = NULL;
int32_t code = scalarCalculateConstants(pTagCondTmp, &pNew);
if (TSDB_CODE_SUCCESS != code) {
nodesDestroyNode(pTagCondTmp);
return false;
}
ASSERT(nodeType(pNew) == QUERY_NODE_VALUE);
SValueNode *pValue = (SValueNode *)pNew;
ASSERT(pValue->node.resType.type == TSDB_DATA_TYPE_BOOL);
bool result = pValue->datum.b;
nodesDestroyNode(pNew);
return result;
}
int32_t getTableList(void* metaHandle, SScanPhysiNode* pScanNode, STableListInfo* pListInfo) {
int32_t code = TSDB_CODE_SUCCESS;
pListInfo->pTableList = taosArrayInit(8, sizeof(STableKeyInfo));
uint64_t tableUid = pScanNode->uid;
SNode* pTagCond = (SNode*)pListInfo->pTagCond;
SNode* pTagIndexCond = (SNode*)pListInfo->pTagIndexCond;
if (pScanNode->tableType == TSDB_SUPER_TABLE) {
if (pTagCond) {
if (pTagIndexCond) {
SIndexMetaArg metaArg = {
.metaEx = metaHandle, .idx = tsdbGetIdx(metaHandle), .ivtIdx = tsdbGetIvtIdx(metaHandle), .suid = tableUid};
SArray* res = taosArrayInit(8, sizeof(uint64_t));
code = doFilterTag(pTagCond, &metaArg, res);
if (code == TSDB_CODE_INDEX_REBUILDING) { // todo
// doFilter();
//code = doFilterTag(pTagIndexCond, &metaArg, res);
code = TSDB_CODE_INDEX_REBUILDING;
if (code == TSDB_CODE_INDEX_REBUILDING) {
code = tsdbGetAllTableList(metaHandle, tableUid, pListInfo->pTableList);
} else if (code != TSDB_CODE_SUCCESS) {
qError("failed to get tableIds, reason: %s, suid: %" PRIu64 "", tstrerror(code), tableUid);
taosArrayDestroy(res);
terrno = code;
return code;
} else {
qDebug("success to get tableIds, size: %d, suid: %" PRIu64 "", (int)taosArrayGetSize(res), tableUid);
qDebug("sucess to get tableIds, size: %d, suid: %" PRIu64 "", (int)taosArrayGetSize(res), tableUid);
}
for (int i = 0; i < taosArrayGetSize(res); i++) {
@ -236,7 +319,20 @@ int32_t getTableList(void* metaHandle, SScanPhysiNode* pScanNode, STableListInfo
} else {
code = tsdbGetAllTableList(metaHandle, tableUid, pListInfo->pTableList);
}
} else { // Create one table group.
if(pTagCond){
int32_t i = 0;
while(i < taosArrayGetSize(pListInfo->pTableList)) {
STableKeyInfo* info = taosArrayGet(pListInfo->pTableList, i);
bool isOk = isTableOk(info, pTagCond, metaHandle);
if(!isOk){
taosArrayRemove(pListInfo->pTableList, i);
continue;
}
i++;
}
}
}else { // Create one table group.
STableKeyInfo info = {.lastKey = 0, .uid = tableUid, .groupId = 0};
taosArrayPush(pListInfo->pTableList, &info);
}

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

@ -40,7 +40,7 @@ static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, size_t nu
SStreamBlockScanInfo* pInfo = pOperator->info;
pInfo->assignBlockUid = assignUid;
// the block type can not be changed in the streamscan operators
// no need to check
#if 0
if (pInfo->blockType == 0) {
pInfo->blockType = type;
@ -49,10 +49,7 @@ static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, size_t nu
return TSDB_CODE_QRY_APP_ERROR;
}
#endif
// rollup sma, the same qTaskInfo is used to insert data by SubmitReq and fetch result by SSDataBlock
if (pInfo->blockType != type) {
pInfo->blockType = type;
}
pInfo->blockType = type;
if (type == STREAM_DATA_TYPE_SUBMIT_BLOCK) {
if (tqReadHandleSetMsg(pInfo->streamBlockReader, input, 0) < 0) {

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

@ -3859,8 +3859,7 @@ static SExecTaskInfo* createExecTaskInfo(uint64_t queryId, uint64_t taskId, EOPT
}
static tsdbReaderT doCreateDataReader(STableScanPhysiNode* pTableScanNode, SReadHandle* pHandle,
STableListInfo* pTableListInfo, uint64_t queryId, uint64_t taskId,
SNode* pTagCond);
STableListInfo* pTableListInfo, uint64_t queryId, uint64_t taskId);
static SArray* extractColumnInfo(SNodeList* pNodeList);
@ -3974,7 +3973,7 @@ int32_t generateGroupIdMap(STableListInfo* pTableListInfo, SReadHandle* pHandle,
}
SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo, SReadHandle* pHandle,
uint64_t queryId, uint64_t taskId, STableListInfo* pTableListInfo, SNode* pTagCond) {
uint64_t queryId, uint64_t taskId, STableListInfo* pTableListInfo) {
int32_t type = nodeType(pPhyNode);
if (pPhyNode->pChildren == NULL || LIST_LENGTH(pPhyNode->pChildren) == 0) {
@ -3982,7 +3981,7 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
STableScanPhysiNode* pTableScanNode = (STableScanPhysiNode*)pPhyNode;
tsdbReaderT pDataReader =
doCreateDataReader(pTableScanNode, pHandle, pTableListInfo, (uint64_t)queryId, taskId, pTagCond);
doCreateDataReader(pTableScanNode, pHandle, pTableListInfo, (uint64_t)queryId, taskId);
if (pDataReader == NULL && terrno != 0) {
pTaskInfo->code = terrno;
return NULL;
@ -4010,14 +4009,9 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
return pOperator;
} else if (QUERY_NODE_PHYSICAL_PLAN_TABLE_MERGE_SCAN == type) {
STableMergeScanPhysiNode* pTableScanNode = (STableMergeScanPhysiNode*)pPhyNode;
SArray* dataReaders = taosArrayInit(8, POINTER_BYTES);
createScanTableListInfo(pTableScanNode, pHandle, pTableListInfo, queryId, taskId, pTagCond);
doCreateMultipleDataReaders(pTableScanNode, pHandle, pTableListInfo, dataReaders, queryId, taskId);
extractTableSchemaVersion(pHandle, pTableScanNode->scan.uid, pTaskInfo);
SOperatorInfo* pOperator = createTableMergeScanOperatorInfo(pTableScanNode, dataReaders, pHandle, pTaskInfo);
SOperatorInfo* pOperator = createTableMergeScanOperatorInfo(pTableScanNode, pTableListInfo, pHandle, pTaskInfo, queryId, taskId);
STableScanInfo* pScanInfo = pOperator->info;
pTaskInfo->cost.pRecoder = &pScanInfo->readRecorder;
return pOperator;
@ -4037,10 +4031,10 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
if (pHandle->vnode) {
// for stram
pDataReader =
doCreateDataReader(pTableScanNode, pHandle, pTableListInfo, (uint64_t)queryId, taskId, pTagCond);
doCreateDataReader(pTableScanNode, pHandle, pTableListInfo, (uint64_t)queryId, taskId);
} else {
// for tq
getTableList(pHandle->meta, pScanPhyNode, pTableListInfo, pTagCond);
getTableList(pHandle->meta, pScanPhyNode, pTableListInfo);
}
}
@ -4068,7 +4062,7 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
} else if (QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN == type) {
STagScanPhysiNode* pScanPhyNode = (STagScanPhysiNode*)pPhyNode;
int32_t code = getTableList(pHandle->meta, pScanPhyNode, pTableListInfo, pScanPhyNode->node.pConditions);
int32_t code = getTableList(pHandle->meta, pScanPhyNode, pTableListInfo);
if (code != TSDB_CODE_SUCCESS) {
pTaskInfo->code = terrno;
return NULL;
@ -4126,7 +4120,7 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
SOperatorInfo** ops = taosMemoryCalloc(size, POINTER_BYTES);
for (int32_t i = 0; i < size; ++i) {
SPhysiNode* pChildNode = (SPhysiNode*)nodesListGetNode(pPhyNode->pChildren, i);
ops[i] = createOperatorTree(pChildNode, pTaskInfo, pHandle, queryId, taskId, pTableListInfo, pTagCond);
ops[i] = createOperatorTree(pChildNode, pTaskInfo, pHandle, queryId, taskId, pTableListInfo);
if (ops[i] == NULL) {
return NULL;
}
@ -4217,6 +4211,8 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
pOptr = createStreamFinalIntervalOperatorInfo(ops[0], pPhyNode, pTaskInfo, children);
} else if (QUERY_NODE_PHYSICAL_PLAN_SORT == type) {
pOptr = createSortOperatorInfo(ops[0], (SSortPhysiNode*)pPhyNode, pTaskInfo);
} else if (QUERY_NODE_PHYSICAL_PLAN_GROUP_SORT == type) {
pOptr = createGroupSortOperatorInfo(ops[0], (SGroupSortPhysiNode*)pPhyNode, pTaskInfo);
} else if (QUERY_NODE_PHYSICAL_PLAN_MERGE == type) {
SMergePhysiNode* pMergePhyNode = (SMergePhysiNode*)pPhyNode;
@ -4328,8 +4324,8 @@ SArray* extractColumnInfo(SNodeList* pNodeList) {
}
tsdbReaderT doCreateDataReader(STableScanPhysiNode* pTableScanNode, SReadHandle* pHandle,
STableListInfo* pTableListInfo, uint64_t queryId, uint64_t taskId, SNode* pTagCond) {
int32_t code = getTableList(pHandle->meta, &pTableScanNode->scan, pTableListInfo, pTagCond);
STableListInfo* pTableListInfo, uint64_t queryId, uint64_t taskId) {
int32_t code = getTableList(pHandle->meta, &pTableScanNode->scan, pTableListInfo);
if (code != TSDB_CODE_SUCCESS) {
goto _error;
}
@ -4487,8 +4483,10 @@ int32_t createExecTaskInfoImpl(SSubplan* pPlan, SExecTaskInfo** pTaskInfo, SRead
}
(*pTaskInfo)->sql = sql;
(*pTaskInfo)->tableqinfoList.pTagCond = pPlan->pTagCond;
(*pTaskInfo)->tableqinfoList.pTagIndexCond = pPlan->pTagIndexCond;
(*pTaskInfo)->pRoot = createOperatorTree(pPlan->pNode, *pTaskInfo, pHandle, queryId, taskId,
&(*pTaskInfo)->tableqinfoList, pPlan->pTagCond);
&(*pTaskInfo)->tableqinfoList);
if (NULL == (*pTaskInfo)->pRoot) {
code = (*pTaskInfo)->code;
goto _complete;

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

@ -337,7 +337,7 @@ void addTagPseudoColumnData(SReadHandle* pHandle, SExprInfo* pPseudoExpr, int32_
}
for (int32_t i = 0; i < pBlock->info.rows; ++i) {
colDataAppend(pColInfoData, i, data, (data == NULL));
colDataAppend(pColInfoData, i, data, (data == NULL) || (pColInfoData->info.type == TSDB_DATA_TYPE_JSON && tTagIsJsonNull(data)));
}
if (data && (pColInfoData->info.type != TSDB_DATA_TYPE_JSON) && p != NULL &&
@ -535,7 +535,7 @@ SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode,
// taosSsleep(20);
SDataBlockDescNode* pDescNode = pTableScanNode->scan.node.pOutputDataBlockDesc;
int32_t numOfCols = 0;
int32_t numOfCols = 0;
SArray* pColList = extractColMatchInfo(pTableScanNode->scan.pScanCols, pDescNode, &numOfCols, COL_MATCH_FROM_COL_ID);
int32_t code = initQueryTableDataCond(&pInfo->cond, pTableScanNode);
@ -913,6 +913,7 @@ static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator) {
}
size_t total = taosArrayGetSize(pInfo->pBlockLists);
// TODO: refactor
if (pInfo->blockType == STREAM_DATA_TYPE_SSDATA_BLOCK) {
if (pInfo->validBlockIndex >= total) {
/*doClearBufferedBlocks(pInfo);*/
@ -1816,7 +1817,7 @@ static SSDataBlock* doTagScan(SOperatorInfo* pOperator) {
} else {
data = (char*)p;
}
colDataAppend(pDst, count, data, (data == NULL));
colDataAppend(pDst, count, data, (data == NULL) || (pDst->info.type == TSDB_DATA_TYPE_JSON && tTagIsJsonNull(data)));
if (pDst->info.type != TSDB_DATA_TYPE_JSON && p != NULL && IS_VAR_DATA_TYPE(((const STagVal*)p)->type) &&
data != NULL) {
@ -1839,9 +1840,7 @@ static SSDataBlock* doTagScan(SOperatorInfo* pOperator) {
}
pRes->info.rows = count;
doFilter(pInfo->pFilterNode, pRes);
pOperator->resultInfo.totalRows += pRes->info.rows;
pOperator->resultInfo.totalRows += count;
return (pRes->info.rows == 0) ? NULL : pInfo->pRes;
}
@ -1871,13 +1870,11 @@ SOperatorInfo* createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysi
goto _error;
}
pInfo->pTableList = pTableListInfo;
pInfo->pColMatchInfo = colList;
pInfo->pRes = createResDataBlock(pDescNode);
pInfo->readHandle = *pReadHandle;
pInfo->curPos = 0;
pInfo->pFilterNode = pPhyNode->node.pConditions;
pInfo->pTableList = pTableListInfo;
pInfo->pColMatchInfo = colList;
pInfo->pRes = createResDataBlock(pDescNode);
pInfo->readHandle = *pReadHandle;
pInfo->curPos = 0;
pOperator->name = "TagScanOperator";
pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN;
@ -1902,6 +1899,12 @@ _error:
}
typedef struct STableMergeScanInfo {
STableListInfo* tableListInfo;
int32_t tableStartIndex;
int32_t tableEndIndex;
bool hasGroupId;
uint64_t groupId;
SArray* dataReaders; // array of tsdbReaderT*
SReadHandle readHandle;
@ -1914,11 +1917,9 @@ typedef struct STableMergeScanInfo {
SSDataBlock* pSortInputBlock;
int64_t startTs; // sort start time
bool hasGroupId;
uint64_t groupId;
STupleHandle* prefetchedTuple;
SArray* sortSourceParams;
SArray* sortSourceParams;
uint64_t queryId;
uint64_t taskId;
SFileBlockLoadRecorder readRecorder;
int64_t numOfRows;
@ -1946,8 +1947,6 @@ typedef struct STableMergeScanInfo {
// window to check if current data block needs to be loaded.
SSampleExecInfo sample; // sample execution info
int32_t curTWinIdx;
} STableMergeScanInfo;
int32_t compareTableKeyInfoByGid(const void* p1, const void* p2) {
@ -1957,9 +1956,8 @@ int32_t compareTableKeyInfoByGid(const void* p1, const void* p2) {
}
int32_t createScanTableListInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* pHandle,
STableListInfo* pTableListInfo, uint64_t queryId, uint64_t taskId,
SNode* pTagCond) {
int32_t code = getTableList(pHandle->meta, &pTableScanNode->scan, pTableListInfo, pTagCond);
STableListInfo* pTableListInfo, uint64_t queryId, uint64_t taskId, SNode* pTagCond) {
int32_t code = getTableList(pHandle->meta, &pTableScanNode->scan, pTableListInfo);
if (code != TSDB_CODE_SUCCESS) {
return code;
}
@ -1978,11 +1976,10 @@ int32_t createScanTableListInfo(STableScanPhysiNode* pTableScanNode, SReadHandle
}
int32_t doCreateMultipleDataReaders(STableScanPhysiNode* pTableScanNode, SReadHandle* pHandle,
STableListInfo* pTableListInfo, SArray* arrayReader, uint64_t queryId,
uint64_t taskId) {
STableListInfo* pTableListInfo, SArray* arrayReader, uint64_t queryId,
uint64_t taskId) {
SQueryTableDataCond cond = {0};
int32_t code = initQueryTableDataCond(&cond, pTableScanNode);
int32_t code = initQueryTableDataCond(&cond, pTableScanNode);
if (code != TSDB_CODE_SUCCESS) {
goto _error;
}
@ -2005,7 +2002,25 @@ _error:
return code;
}
// todo refactor or remove it
int32_t createMultipleDataReaders(SQueryTableDataCond* pQueryCond, SReadHandle* pHandle, STableListInfo* pTableListInfo,
int32_t tableStartIdx, int32_t tableEndIdx, SArray* arrayReader, uint64_t queryId,
uint64_t taskId) {
for (int32_t i = tableStartIdx; i <= tableEndIdx; ++i) {
STableListInfo* subListInfo = taosMemoryCalloc(1, sizeof(subListInfo));
subListInfo->pTableList = taosArrayInit(1, sizeof(STableKeyInfo));
taosArrayPush(subListInfo->pTableList, taosArrayGet(pTableListInfo->pTableList, i));
tsdbReaderT* pReader = tsdbReaderOpen(pHandle->vnode, pQueryCond, subListInfo, queryId, taskId);
taosArrayPush(arrayReader, &pReader);
taosArrayDestroy(subListInfo->pTableList);
taosMemoryFree(subListInfo);
}
return TSDB_CODE_SUCCESS;
}
// todo refactor
static int32_t loadDataBlockFromOneTable(SOperatorInfo* pOperator, STableMergeScanInfo* pTableScanInfo,
int32_t readerIdx, SSDataBlock* pBlock, uint32_t* status) {
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
@ -2191,22 +2206,47 @@ SArray* generateSortByTsInfo(int32_t order) {
return pList;
}
int32_t doOpenTableMergeScanOperator(SOperatorInfo* pOperator) {
int32_t startGroupTableMergeScan(SOperatorInfo* pOperator) {
STableMergeScanInfo* pInfo = pOperator->info;
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
if (OPTR_IS_OPENED(pOperator)) {
return TSDB_CODE_SUCCESS;
{
size_t tableListSize = taosArrayGetSize(pInfo->tableListInfo->pTableList);
int32_t i = pInfo->tableStartIndex + 1;
for (; i < tableListSize; ++i) {
STableKeyInfo* tableKeyInfo = taosArrayGet(pInfo->tableListInfo->pTableList, i);
if (tableKeyInfo->groupId != pInfo->groupId) {
break;
}
}
pInfo->tableEndIndex = i - 1;
}
int32_t numOfBufPage = pInfo->sortBufSize / pInfo->bufPageSize;
int32_t tableStartIdx = pInfo->tableStartIndex;
int32_t tableEndIdx = pInfo->tableEndIndex;
STableListInfo* tableListInfo = pInfo->tableListInfo;
createMultipleDataReaders(&pInfo->cond, &pInfo->readHandle, tableListInfo, tableStartIdx, tableEndIdx,
pInfo->dataReaders, pInfo->queryId, pInfo->taskId);
// todo the total available buffer should be determined by total capacity of buffer of this task.
// the additional one is reserved for merge result
pInfo->sortBufSize = pInfo->bufPageSize * (tableEndIdx - tableStartIdx + 1 + 1);
int32_t numOfBufPage = pInfo->sortBufSize / pInfo->bufPageSize;
pInfo->pSortHandle = tsortCreateSortHandle(pInfo->pSortInfo, SORT_MULTISOURCE_MERGE, pInfo->bufPageSize, numOfBufPage,
pInfo->pSortInputBlock, pTaskInfo->id.str);
tsortSetFetchRawDataFp(pInfo->pSortHandle, getTableDataBlock, NULL, NULL);
size_t numReaders = taosArrayGetSize(pInfo->dataReaders);
for (int32_t i = 0; i < numReaders; ++i) {
STableMergeScanSortSourceParam param = {0};
param.readerIdx = i;
param.pOperator = pOperator;
param.inputBlock = createOneDataBlock(pInfo->pResBlock, false);
taosArrayPush(pInfo->sortSourceParams, &param);
}
for (int32_t i = 0; i < numReaders; ++i) {
SSortSource* ps = taosMemoryCalloc(1, sizeof(SSortSource));
STableMergeScanSortSourceParam* param = taosArrayGet(pInfo->sortSourceParams, i);
@ -2220,9 +2260,22 @@ int32_t doOpenTableMergeScanOperator(SOperatorInfo* pOperator) {
longjmp(pTaskInfo->env, terrno);
}
pOperator->status = OP_RES_TO_RETURN;
return TSDB_CODE_SUCCESS;
}
int32_t stopGroupTableMergeScan(SOperatorInfo* pOperator) {
STableMergeScanInfo* pInfo = pOperator->info;
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
tsortDestroySortHandle(pInfo->pSortHandle);
taosArrayClear(pInfo->sortSourceParams);
for (int32_t i = 0; i < taosArrayGetSize(pInfo->dataReaders); ++i) {
tsdbReaderT* reader = taosArrayGetP(pInfo->dataReaders, i);
tsdbCleanupReadHandle(reader);
}
taosArrayDestroy(pInfo->dataReaders);
OPTR_SET_OPENED(pOperator);
return TSDB_CODE_SUCCESS;
}
@ -2265,14 +2318,38 @@ SSDataBlock* doTableMergeScan(SOperatorInfo* pOperator) {
if (code != TSDB_CODE_SUCCESS) {
longjmp(pTaskInfo->env, code);
}
size_t tableListSize = taosArrayGetSize(pInfo->tableListInfo->pTableList);
if (!pInfo->hasGroupId) {
pInfo->hasGroupId = true;
SSDataBlock* pBlock = getSortedTableMergeScanBlockData(pInfo->pSortHandle, pOperator->resultInfo.capacity, pOperator);
if (pBlock != NULL) {
pOperator->resultInfo.totalRows += pBlock->info.rows;
} else {
doSetOperatorCompleted(pOperator);
if (tableListSize == 0) {
doSetOperatorCompleted(pOperator);
return NULL;
}
pInfo->tableStartIndex = 0;
pInfo->groupId = ((STableKeyInfo*)taosArrayGet(pInfo->tableListInfo->pTableList, pInfo->tableStartIndex))->groupId;
startGroupTableMergeScan(pOperator);
}
SSDataBlock* pBlock = NULL;
while (pInfo->tableStartIndex < tableListSize) {
pBlock = getSortedTableMergeScanBlockData(pInfo->pSortHandle, pOperator->resultInfo.capacity, pOperator);
if (pBlock != NULL) {
pBlock->info.groupId = pInfo->groupId;
pOperator->resultInfo.totalRows += pBlock->info.rows;
return pBlock;
} else {
stopGroupTableMergeScan(pOperator);
if (pInfo->tableEndIndex >= tableListSize - 1) {
doSetOperatorCompleted(pOperator);
break;
}
pInfo->tableStartIndex = pInfo->tableEndIndex + 1;
pInfo->groupId =
((STableKeyInfo*)taosArrayGet(pInfo->tableListInfo->pTableList, pInfo->tableStartIndex))->groupId;
startGroupTableMergeScan(pOperator);
}
}
return pBlock;
}
@ -2280,17 +2357,10 @@ void destroyTableMergeScanOperatorInfo(void* param, int32_t numOfOutput) {
STableMergeScanInfo* pTableScanInfo = (STableMergeScanInfo*)param;
cleanupQueryTableDataCond(&pTableScanInfo->cond);
for (int32_t i = 0; i < taosArrayGetSize(pTableScanInfo->dataReaders); ++i) {
tsdbReaderT* reader = taosArrayGetP(pTableScanInfo->dataReaders, i);
tsdbCleanupReadHandle(reader);
}
taosArrayDestroy(pTableScanInfo->dataReaders);
if (pTableScanInfo->pColMatchInfo != NULL) {
taosArrayDestroy(pTableScanInfo->pColMatchInfo);
}
taosArrayDestroy(pTableScanInfo->sortSourceParams);
pTableScanInfo->pResBlock = blockDataDestroy(pTableScanInfo->pResBlock);
pTableScanInfo->pSortInputBlock = blockDataDestroy(pTableScanInfo->pSortInputBlock);
@ -2316,8 +2386,9 @@ int32_t getTableMergeScanExplainExecInfo(SOperatorInfo* pOptr, void** pOptrExpla
return TSDB_CODE_SUCCESS;
}
SOperatorInfo* createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SArray* dataReaders,
SReadHandle* readHandle, SExecTaskInfo* pTaskInfo) {
SOperatorInfo* createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanNode, STableListInfo* pTableListInfo,
SReadHandle* readHandle, SExecTaskInfo* pTaskInfo, uint64_t queryId,
uint64_t taskId) {
STableMergeScanInfo* pInfo = taosMemoryCalloc(1, sizeof(STableMergeScanInfo));
SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
if (pInfo == NULL || pOperator == NULL) {
@ -2347,22 +2418,16 @@ SOperatorInfo* createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanN
pInfo->sample.seed = taosGetTimestampSec();
pInfo->dataBlockLoadFlag = pTableScanNode->dataRequired;
pInfo->pFilterNode = pTableScanNode->scan.node.pConditions;
pInfo->dataReaders = dataReaders;
pInfo->tableListInfo = pTableListInfo;
pInfo->scanFlag = MAIN_SCAN;
pInfo->pColMatchInfo = pColList;
pInfo->curTWinIdx = 0;
pInfo->pResBlock = createResDataBlock(pDescNode);
pInfo->dataReaders = taosArrayInit(64, POINTER_BYTES);
pInfo->queryId = queryId;
pInfo->taskId = taskId;
pInfo->sortSourceParams = taosArrayInit(taosArrayGetSize(dataReaders), sizeof(STableMergeScanSortSourceParam));
for (int32_t i = 0; i < taosArrayGetSize(dataReaders); ++i) {
STableMergeScanSortSourceParam* param = taosMemoryCalloc(1, sizeof(STableMergeScanSortSourceParam));
param->readerIdx = i;
param->pOperator = pOperator;
param->inputBlock = createOneDataBlock(pInfo->pResBlock, false);
taosArrayPush(pInfo->sortSourceParams, param);
taosMemoryFree(param);
}
pInfo->sortSourceParams = taosArrayInit(64, sizeof(STableMergeScanSortSourceParam));
pInfo->pSortInfo = generateSortByTsInfo(pInfo->cond.order);
pInfo->pSortInputBlock = createOneDataBlock(pInfo->pResBlock, false);
@ -2370,14 +2435,7 @@ SOperatorInfo* createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanN
int32_t rowSize = pInfo->pResBlock->info.rowSize;
pInfo->bufPageSize = getProperSortPageSize(rowSize);
// todo the total available buffer should be determined by total capacity of buffer of this task.
// the additional one is reserved for merge result
pInfo->sortBufSize = pInfo->bufPageSize * (taosArrayGetSize(dataReaders) + 1);
pInfo->hasGroupId = false;
pInfo->prefetchedTuple = NULL;
pOperator->name = "TableMergeScanOperator";
// TODO : change it
pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_TABLE_MERGE_SCAN;
pOperator->blocking = false;
pOperator->status = OP_NOT_OPENED;
@ -2387,8 +2445,8 @@ SOperatorInfo* createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanN
initResultSizeInfo(pOperator, 1024);
pOperator->fpSet =
createOperatorFpSet(doOpenTableMergeScanOperator, doTableMergeScan, NULL, NULL, destroyTableMergeScanOperatorInfo,
NULL, NULL, getTableMergeScanExplainExecInfo);
createOperatorFpSet(operatorDummyOpenFn, doTableMergeScan, NULL, NULL, destroyTableMergeScanOperatorInfo, NULL,
NULL, getTableMergeScanExplainExecInfo);
pOperator->cost.openCost = 0;
return pOperator;

30
source/libs/executor/src/sortoperator.c
View File

@ -427,10 +427,17 @@ int32_t getGroupSortExplainExecInfo(SOperatorInfo* pOptr, void** pOptrExplain, u
return TSDB_CODE_SUCCESS;
}
// TODO:
SOperatorInfo* createGroupSortOperatorInfo(SOperatorInfo* downstream, SSortPhysiNode* pSortPhyNode,
void destroyGroupSortOperatorInfo(void* param, int32_t numOfOutput) {
SGroupSortOperatorInfo* pInfo = (SGroupSortOperatorInfo*)param;
pInfo->binfo.pRes = blockDataDestroy(pInfo->binfo.pRes);
taosArrayDestroy(pInfo->pSortInfo);
taosArrayDestroy(pInfo->pColMatchInfo);
}
SOperatorInfo* createGroupSortOperatorInfo(SOperatorInfo* downstream, SGroupSortPhysiNode* pSortPhyNode,
SExecTaskInfo* pTaskInfo) {
SSortOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SSortOperatorInfo));
SGroupSortOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SGroupSortOperatorInfo));
SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
if (pInfo == NULL || pOperator == NULL /* || rowSize > 100 * 1024 * 1024*/) {
goto _error;
@ -455,8 +462,7 @@ SOperatorInfo* createGroupSortOperatorInfo(SOperatorInfo* downstream, SSortPhysi
;
pInfo->pColMatchInfo = pColMatchColInfo;
pOperator->name = "GroupSortOperator";
// TODO
pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_SORT;
pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_GROUP_SORT;
pOperator->blocking = true;
pOperator->status = OP_NOT_OPENED;
pOperator->info = pInfo;
@ -464,7 +470,7 @@ SOperatorInfo* createGroupSortOperatorInfo(SOperatorInfo* downstream, SSortPhysi
pOperator->exprSupp.numOfExprs = numOfCols;
pOperator->pTaskInfo = pTaskInfo;
pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doGroupSort, NULL, NULL, destroyOrderOperatorInfo, NULL,
pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doGroupSort, NULL, NULL, destroyGroupSortOperatorInfo, NULL,
NULL, getGroupSortExplainExecInfo);
int32_t code = appendDownstream(pOperator, &downstream, 1);
@ -481,18 +487,6 @@ _error:
return NULL;
}
void destroyGroupSortOperatorInfo(void* param, int32_t numOfOutput) {
SGroupSortOperatorInfo* pInfo = (SGroupSortOperatorInfo*)param;
pInfo->binfo.pRes = blockDataDestroy(pInfo->binfo.pRes);
taosArrayDestroy(pInfo->pSortInfo);
taosArrayDestroy(pInfo->pColMatchInfo);
}
// TODO: sort group
// TODO: msortCompare compare group id in multiway merge sort.
// TODO: table merge scan, group first, then for each group, multiple readers
//=====================================================================================
// Multiway Sort Merge operator
typedef struct SMultiwaySortMergeOperatorInfo {

3
source/libs/executor/src/timewindowoperator.c
View File

@ -1656,10 +1656,9 @@ static SSDataBlock* doSessionWindowAgg(SOperatorInfo* pOperator) {
doBuildResultDatablock(pOperator, pBInfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf);
if (pBInfo->pRes->info.rows == 0 || !hasDataInGroupInfo(&pInfo->groupResInfo)) {
doSetOperatorCompleted(pOperator);
return NULL;
}
return pBInfo->pRes;
return pBInfo->pRes->info.rows > 0 ? pBInfo->pRes : NULL;
}
int64_t st = taosGetTimestampUs();

1
source/libs/nodes/src/nodesCloneFuncs.c
View File

@ -351,6 +351,7 @@ static SNode* logicScanCopy(const SScanLogicNode* pSrc, SScanLogicNode* pDst) {
COPY_SCALAR_FIELD(intervalUnit);
COPY_SCALAR_FIELD(slidingUnit);
CLONE_NODE_FIELD(pTagCond);
CLONE_NODE_FIELD(pTagIndexCond);
COPY_SCALAR_FIELD(triggerType);
COPY_SCALAR_FIELD(watermark);
COPY_SCALAR_FIELD(tsColId);

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

@ -2326,6 +2326,7 @@ static const char* jkSubplanNodeAddr = "NodeAddr";
static const char* jkSubplanRootNode = "RootNode";
static const char* jkSubplanDataSink = "DataSink";
static const char* jkSubplanTagCond = "TagCond";
static const char* jkSubplanTagIndexCond = "TagIndexCond";
static int32_t subplanToJson(const void* pObj, SJson* pJson) {
const SSubplan* pNode = (const SSubplan*)pObj;
@ -2355,6 +2356,9 @@ static int32_t subplanToJson(const void* pObj, SJson* pJson) {
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddObject(pJson, jkSubplanTagCond, nodeToJson, pNode->pTagCond);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddObject(pJson, jkSubplanTagIndexCond, nodeToJson, pNode->pTagIndexCond);
}
return code;
}
@ -2388,6 +2392,9 @@ static int32_t jsonToSubplan(const SJson* pJson, void* pObj) {
if (TSDB_CODE_SUCCESS == code) {
code = jsonToNodeObject(pJson, jkSubplanTagCond, (SNode**)&pNode->pTagCond);
}
if (TSDB_CODE_SUCCESS == code) {
code = jsonToNodeObject(pJson, jkSubplanTagIndexCond, (SNode**)&pNode->pTagIndexCond);
}
return code;
}
@ -3954,7 +3961,7 @@ static int32_t deleteStmtToJson(const void* pObj, SJson* pJson) {
code = tjsonAddObject(pJson, jkDeleteStmtCountFunc, nodeToJson, pNode->pCountFunc);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddObject(pJson, jkDeleteStmtTagIndexCond, nodeToJson, pNode->pTagIndexCond);
code = tjsonAddObject(pJson, jkDeleteStmtTagIndexCond, nodeToJson, pNode->pTagCond);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkDeleteStmtTimeRangeStartKey, pNode->timeRange.skey);
@ -3983,7 +3990,7 @@ static int32_t jsonToDeleteStmt(const SJson* pJson, void* pObj) {
code = jsonToNodeObject(pJson, jkDeleteStmtCountFunc, &pNode->pCountFunc);
}
if (TSDB_CODE_SUCCESS == code) {
code = jsonToNodeObject(pJson, jkDeleteStmtTagIndexCond, &pNode->pTagIndexCond);
code = jsonToNodeObject(pJson, jkDeleteStmtTagIndexCond, &pNode->pTagCond);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetBigIntValue(pJson, jkDeleteStmtTimeRangeStartKey, &pNode->timeRange.skey);

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

@ -669,7 +669,7 @@ void nodesDestroyNode(SNode* pNode) {
nodesDestroyNode(pStmt->pFromTable);
nodesDestroyNode(pStmt->pWhere);
nodesDestroyNode(pStmt->pCountFunc);
nodesDestroyNode(pStmt->pTagIndexCond);
nodesDestroyNode(pStmt->pTagCond);
break;
}
case QUERY_NODE_QUERY: {
@ -688,7 +688,13 @@ void nodesDestroyNode(SNode* pNode) {
SScanLogicNode* pLogicNode = (SScanLogicNode*)pNode;
destroyLogicNode((SLogicNode*)pLogicNode);
nodesDestroyList(pLogicNode->pScanCols);
nodesDestroyList(pLogicNode->pScanPseudoCols);
taosMemoryFreeClear(pLogicNode->pVgroupList);
nodesDestroyList(pLogicNode->pDynamicScanFuncs);
nodesDestroyNode(pLogicNode->pTagCond);
nodesDestroyNode(pLogicNode->pTagIndexCond);
taosArrayDestroy(pLogicNode->pSmaIndexes);
nodesDestroyList(pLogicNode->pPartTags);
break;
}
case QUERY_NODE_LOGIC_PLAN_JOIN: {
@ -897,6 +903,8 @@ void nodesDestroyNode(SNode* pNode) {
nodesDestroyList(pSubplan->pChildren);
nodesDestroyNode((SNode*)pSubplan->pNode);
nodesDestroyNode((SNode*)pSubplan->pDataSink);
nodesDestroyNode((SNode*)pSubplan->pTagCond);
nodesDestroyNode((SNode*)pSubplan->pTagIndexCond);
nodesClearList(pSubplan->pParents);
break;
}
@ -1130,6 +1138,7 @@ void* nodesGetValueFromNode(SValueNode* pNode) {
case TSDB_DATA_TYPE_NCHAR:
case TSDB_DATA_TYPE_VARCHAR:
case TSDB_DATA_TYPE_VARBINARY:
case TSDB_DATA_TYPE_JSON:
return (void*)pNode->datum.p;
default:
break;
@ -1659,6 +1668,7 @@ int32_t nodesMergeConds(SNode** pDst, SNodeList** pSrc) {
typedef struct SClassifyConditionCxt {
bool hasPrimaryKey;
bool hasTagIndexCol;
bool hasTagCol;
bool hasOtherCol;
} SClassifyConditionCxt;
@ -1670,6 +1680,9 @@ static EDealRes classifyConditionImpl(SNode* pNode, void* pContext) {
pCxt->hasPrimaryKey = true;
} else if (pCol->hasIndex) {
pCxt->hasTagIndexCol = true;
pCxt->hasTagCol = true;
} else if (COLUMN_TYPE_TAG == pCol->colType) {
pCxt->hasTagCol = true;
} else {
pCxt->hasOtherCol = true;
}
@ -1678,23 +1691,31 @@ static EDealRes classifyConditionImpl(SNode* pNode, void* pContext) {
return DEAL_RES_CONTINUE;
}
typedef enum EConditionType { COND_TYPE_PRIMARY_KEY = 1, COND_TYPE_TAG_INDEX, COND_TYPE_NORMAL } EConditionType;
typedef enum EConditionType {
COND_TYPE_PRIMARY_KEY = 1,
COND_TYPE_TAG_INDEX,
COND_TYPE_TAG,
COND_TYPE_NORMAL
} EConditionType;
static EConditionType classifyCondition(SNode* pNode) {
SClassifyConditionCxt cxt = {.hasPrimaryKey = false, .hasTagIndexCol = false, .hasOtherCol = false};
nodesWalkExpr(pNode, classifyConditionImpl, &cxt);
return cxt.hasOtherCol ? COND_TYPE_NORMAL
: (cxt.hasPrimaryKey && cxt.hasTagIndexCol
: (cxt.hasPrimaryKey && cxt.hasTagCol
? COND_TYPE_NORMAL
: (cxt.hasPrimaryKey ? COND_TYPE_PRIMARY_KEY : COND_TYPE_TAG_INDEX));
: (cxt.hasPrimaryKey ? COND_TYPE_PRIMARY_KEY
: (cxt.hasTagIndexCol ? COND_TYPE_TAG_INDEX : COND_TYPE_TAG)));
}
static int32_t partitionLogicCond(SNode** pCondition, SNode** pPrimaryKeyCond, SNode** pTagCond, SNode** pOtherCond) {
static int32_t partitionLogicCond(SNode** pCondition, SNode** pPrimaryKeyCond, SNode** pTagIndexCond, SNode** pTagCond,
SNode** pOtherCond) {
SLogicConditionNode* pLogicCond = (SLogicConditionNode*)(*pCondition);
int32_t code = TSDB_CODE_SUCCESS;
SNodeList* pPrimaryKeyConds = NULL;
SNodeList* pTagIndexConds = NULL;
SNodeList* pTagConds = NULL;
SNodeList* pOtherConds = NULL;
SNode* pCond = NULL;
@ -1706,6 +1727,14 @@ static int32_t partitionLogicCond(SNode** pCondition, SNode** pPrimaryKeyCond, S
}
break;
case COND_TYPE_TAG_INDEX:
if (NULL != pTagIndexCond) {
code = nodesListMakeAppend(&pTagIndexConds, nodesCloneNode(pCond));
}
if (NULL != pTagCond) {
code = nodesListMakeAppend(&pTagConds, nodesCloneNode(pCond));
}
break;
case COND_TYPE_TAG:
if (NULL != pTagCond) {
code = nodesListMakeAppend(&pTagConds, nodesCloneNode(pCond));
}
@ -1723,11 +1752,15 @@ static int32_t partitionLogicCond(SNode** pCondition, SNode** pPrimaryKeyCond, S
}
SNode* pTempPrimaryKeyCond = NULL;
SNode* pTempTagIndexCond = NULL;
SNode* pTempTagCond = NULL;
SNode* pTempOtherCond = NULL;
if (TSDB_CODE_SUCCESS == code) {
code = nodesMergeConds(&pTempPrimaryKeyCond, &pPrimaryKeyConds);
}
if (TSDB_CODE_SUCCESS == code) {
code = nodesMergeConds(&pTempTagIndexCond, &pTagIndexConds);
}
if (TSDB_CODE_SUCCESS == code) {
code = nodesMergeConds(&pTempTagCond, &pTagConds);
}
@ -1739,6 +1772,9 @@ static int32_t partitionLogicCond(SNode** pCondition, SNode** pPrimaryKeyCond, S
if (NULL != pPrimaryKeyCond) {
*pPrimaryKeyCond = pTempPrimaryKeyCond;
}
if (NULL != pTagIndexCond) {
*pTagIndexCond = pTempTagIndexCond;
}
if (NULL != pTagCond) {
*pTagCond = pTempTagCond;
}
@ -1749,9 +1785,11 @@ static int32_t partitionLogicCond(SNode** pCondition, SNode** pPrimaryKeyCond, S
*pCondition = NULL;
} else {
nodesDestroyList(pPrimaryKeyConds);
nodesDestroyList(pTagIndexConds);
nodesDestroyList(pTagConds);
nodesDestroyList(pOtherConds);
nodesDestroyNode(pTempPrimaryKeyCond);
nodesDestroyNode(pTempTagIndexCond);
nodesDestroyNode(pTempTagCond);
nodesDestroyNode(pTempOtherCond);
}
@ -1759,10 +1797,11 @@ static int32_t partitionLogicCond(SNode** pCondition, SNode** pPrimaryKeyCond, S
return code;
}
int32_t nodesPartitionCond(SNode** pCondition, SNode** pPrimaryKeyCond, SNode** pTagCond, SNode** pOtherCond) {
int32_t nodesPartitionCond(SNode** pCondition, SNode** pPrimaryKeyCond, SNode** pTagIndexCond, SNode** pTagCond,
SNode** pOtherCond) {
if (QUERY_NODE_LOGIC_CONDITION == nodeType(*pCondition) &&
LOGIC_COND_TYPE_AND == ((SLogicConditionNode*)*pCondition)->condType) {
return partitionLogicCond(pCondition, pPrimaryKeyCond, pTagCond, pOtherCond);
return partitionLogicCond(pCondition, pPrimaryKeyCond, pTagIndexCond, pTagCond, pOtherCond);
}
switch (classifyCondition(*pCondition)) {
@ -1772,6 +1811,21 @@ int32_t nodesPartitionCond(SNode** pCondition, SNode** pPrimaryKeyCond, SNode**
}
break;
case COND_TYPE_TAG_INDEX:
if (NULL != pTagIndexCond) {
*pTagIndexCond = *pCondition;
}
if (NULL != pTagCond) {
SNode* pTempCond = *pCondition;
if (NULL != pTagIndexCond) {
pTempCond = nodesCloneNode(*pCondition);
if (NULL == pTempCond) {
return TSDB_CODE_OUT_OF_MEMORY;
}
}
*pTagCond = pTempCond;
}
break;
case COND_TYPE_TAG:
if (NULL != pTagCond) {
*pTagCond = *pCondition;
}

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

@ -2013,7 +2013,7 @@ static int32_t getFillTimeRange(STranslateContext* pCxt, SNode* pWhere, STimeWin
}
SNode* pPrimaryKeyCond = NULL;
nodesPartitionCond(&pCond, &pPrimaryKeyCond, NULL, NULL);
nodesPartitionCond(&pCond, &pPrimaryKeyCond, NULL, NULL, NULL);
int32_t code = TSDB_CODE_SUCCESS;
if (NULL != pPrimaryKeyCond) {
@ -2161,9 +2161,6 @@ static EDealRes checkStateExpr(SNode* pNode, void* pContext) {
if (COLUMN_TYPE_TAG == pCol->colType) {
return generateDealNodeErrMsg(pCxt, TSDB_CODE_PAR_INVALID_STATE_WIN_COL);
}
if (TSDB_SUPER_TABLE == pCol->tableType) {
return generateDealNodeErrMsg(pCxt, TSDB_CODE_PAR_INVALID_STATE_WIN_TABLE);
}
}
return DEAL_RES_CONTINUE;
}
@ -2685,7 +2682,7 @@ static int32_t partitionDeleteWhere(STranslateContext* pCxt, SDeleteStmt* pDelet
SNode* pPrimaryKeyCond = NULL;
SNode* pOtherCond = NULL;
int32_t code = nodesPartitionCond(&pDelete->pWhere, &pPrimaryKeyCond, &pDelete->pTagIndexCond, &pOtherCond);
int32_t code = nodesPartitionCond(&pDelete->pWhere, &pPrimaryKeyCond, NULL, &pDelete->pTagCond, &pOtherCond);
if (TSDB_CODE_SUCCESS == code && NULL != pOtherCond) {
code = generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_DELETE_WHERE);
}

4
source/libs/planner/src/planLogicCreater.c
View File

@ -1160,8 +1160,8 @@ static int32_t createDeleteScanLogicNode(SLogicPlanContext* pCxt, SDeleteStmt* p
}
}
if (TSDB_CODE_SUCCESS == code && NULL != pDelete->pTagIndexCond) {
pScan->pTagCond = nodesCloneNode(pDelete->pTagIndexCond);
if (TSDB_CODE_SUCCESS == code && NULL != pDelete->pTagCond) {
pScan->pTagCond = nodesCloneNode(pDelete->pTagCond);
if (NULL == pScan->pTagCond) {
code = TSDB_CODE_OUT_OF_MEMORY;
}

34
source/libs/planner/src/planOptimizer.c
View File

@ -15,7 +15,6 @@
#include "filter.h"
#include "functionMgt.h"
#include "index.h"
#include "planInt.h"
#include "ttime.h"
@ -309,32 +308,6 @@ static int32_t cpdCalcTimeRange(SOptimizeContext* pCxt, SScanLogicNode* pScan, S
return code;
}
static int32_t cpdApplyTagIndex(SScanLogicNode* pScan, SNode** pTagCond, SNode** pOtherCond) {
int32_t code = TSDB_CODE_SUCCESS;
SIdxFltStatus idxStatus = idxGetFltStatus(*pTagCond);
switch (idxStatus) {
case SFLT_NOT_INDEX:
code = cpdCondAppend(pOtherCond, pTagCond);
break;
case SFLT_COARSE_INDEX:
pScan->pTagCond = nodesCloneNode(*pTagCond);
if (NULL == pScan->pTagCond) {
code = TSDB_CODE_OUT_OF_MEMORY;
break;
}
code = cpdCondAppend(pOtherCond, pTagCond);
break;
case SFLT_ACCURATE_INDEX:
pScan->pTagCond = *pTagCond;
*pTagCond = NULL;
break;
default:
code = TSDB_CODE_FAILED;
break;
}
return code;
}
static int32_t cpdOptimizeScanCondition(SOptimizeContext* pCxt, SScanLogicNode* pScan) {
if (NULL == pScan->node.pConditions || OPTIMIZE_FLAG_TEST_MASK(pScan->node.optimizedFlag, OPTIMIZE_FLAG_CPD) ||
TSDB_SYSTEM_TABLE == pScan->tableType) {
@ -342,15 +315,12 @@ static int32_t cpdOptimizeScanCondition(SOptimizeContext* pCxt, SScanLogicNode*
}
SNode* pPrimaryKeyCond = NULL;
SNode* pTagCond = NULL;
SNode* pOtherCond = NULL;
int32_t code = nodesPartitionCond(&pScan->node.pConditions, &pPrimaryKeyCond, &pTagCond, &pOtherCond);
int32_t code = nodesPartitionCond(&pScan->node.pConditions, &pPrimaryKeyCond, &pScan->pTagIndexCond, &pScan->pTagCond,
&pOtherCond);
if (TSDB_CODE_SUCCESS == code && NULL != pPrimaryKeyCond) {
code = cpdCalcTimeRange(pCxt, pScan, &pPrimaryKeyCond, &pOtherCond);
}
if (TSDB_CODE_SUCCESS == code && NULL != pTagCond) {
code = cpdApplyTagIndex(pScan, &pTagCond, &pOtherCond);
}
if (TSDB_CODE_SUCCESS == code) {
pScan->node.pConditions = pOtherCond;
}

9
source/libs/planner/src/planPhysiCreater.c
View File

@ -436,6 +436,15 @@ static int32_t createScanPhysiNodeFinalize(SPhysiPlanContext* pCxt, SSubplan* pS
}
}
if (TSDB_CODE_SUCCESS == code) {
if (NULL != pScanLogicNode->pTagIndexCond) {
pSubplan->pTagIndexCond = nodesCloneNode(pScanLogicNode->pTagIndexCond);
if (NULL == pSubplan->pTagIndexCond) {
code = TSDB_CODE_OUT_OF_MEMORY;
}
}
}
if (TSDB_CODE_SUCCESS == code) {
*pPhyNode = (SPhysiNode*)pScanPhysiNode;
} else {

91
source/libs/planner/src/planSpliter.c
View File

@ -166,6 +166,31 @@ static bool stbSplHasMultiTbScan(bool streamQuery, SLogicNode* pNode) {
return (QUERY_NODE_LOGIC_PLAN_SCAN == nodeType(pChild) && stbSplIsMultiTbScan(streamQuery, (SScanLogicNode*)pChild));
}
static bool stbSplNeedSplitWindow(bool streamQuery, SLogicNode* pNode) {
SWindowLogicNode* pWindow = (SWindowLogicNode*)pNode;
if (WINDOW_TYPE_INTERVAL == pWindow->winType) {
return !stbSplHasGatherExecFunc(pWindow->pFuncs) && stbSplHasMultiTbScan(streamQuery, pNode);
}
if (WINDOW_TYPE_SESSION == pWindow->winType) {
if (!streamQuery) {
return stbSplHasMultiTbScan(streamQuery, pNode);
} else {
return !stbSplHasGatherExecFunc(pWindow->pFuncs) && stbSplHasMultiTbScan(streamQuery, pNode);
}
}
if (WINDOW_TYPE_STATE == pWindow->winType) {
if (!streamQuery) {
return stbSplHasMultiTbScan(streamQuery, pNode);
} else {
return false;
}
}
return false;
}
static bool stbSplNeedSplit(bool streamQuery, SLogicNode* pNode) {
switch (nodeType(pNode)) {
case QUERY_NODE_LOGIC_PLAN_SCAN:
@ -174,13 +199,8 @@ static bool stbSplNeedSplit(bool streamQuery, SLogicNode* pNode) {
return !(((SJoinLogicNode*)pNode)->isSingleTableJoin);
case QUERY_NODE_LOGIC_PLAN_AGG:
return !stbSplHasGatherExecFunc(((SAggLogicNode*)pNode)->pAggFuncs) && stbSplHasMultiTbScan(streamQuery, pNode);
case QUERY_NODE_LOGIC_PLAN_WINDOW: {
SWindowLogicNode* pWindow = (SWindowLogicNode*)pNode;
if (WINDOW_TYPE_STATE == pWindow->winType || (!streamQuery && WINDOW_TYPE_SESSION == pWindow->winType)) {
return false;
}
return !stbSplHasGatherExecFunc(pWindow->pFuncs) && stbSplHasMultiTbScan(streamQuery, pNode);
}
case QUERY_NODE_LOGIC_PLAN_WINDOW:
return stbSplNeedSplitWindow(streamQuery, pNode);
case QUERY_NODE_LOGIC_PLAN_SORT:
return stbSplHasMultiTbScan(streamQuery, pNode);
default:
@ -477,11 +497,64 @@ static int32_t stbSplSplitSessionForStream(SSplitContext* pCxt, SStableSplitInfo
return code;
}
static void splSetTableScanType(SLogicNode* pNode, EScanType scanType) {
if (QUERY_NODE_LOGIC_PLAN_SCAN == nodeType(pNode)) {
((SScanLogicNode*)pNode)->scanType = scanType;
} else {
if (1 == LIST_LENGTH(pNode->pChildren)) {
splSetTableScanType((SLogicNode*)nodesListGetNode(pNode->pChildren, 0), scanType);
}
}
}
static int32_t stbSplSplitSessionOrStateForBatch(SSplitContext* pCxt, SStableSplitInfo* pInfo) {
SLogicNode* pWindow = pInfo->pSplitNode;
SLogicNode* pChild = (SLogicNode*)nodesListGetNode(pWindow->pChildren, 0);
SNodeList* pMergeKeys = NULL;
int32_t code = stbSplCreateMergeKeysByPrimaryKey(((SWindowLogicNode*)pWindow)->pTspk, &pMergeKeys);
if (TSDB_CODE_SUCCESS == code) {
code = stbSplCreateMergeNode(pCxt, pInfo->pSubplan, pChild, pMergeKeys, (SLogicNode*)pChild);
}
if (TSDB_CODE_SUCCESS == code) {
code = nodesListMakeStrictAppend(&pInfo->pSubplan->pChildren,
(SNode*)splCreateScanSubplan(pCxt, pChild, SPLIT_FLAG_STABLE_SPLIT));
}
if (TSDB_CODE_SUCCESS == code) {
splSetTableScanType(pChild, SCAN_TYPE_TABLE_MERGE);
++(pCxt->groupId);
}
if (TSDB_CODE_SUCCESS == code) {
pInfo->pSubplan->subplanType = SUBPLAN_TYPE_MERGE;
SPLIT_FLAG_SET_MASK(pInfo->pSubplan->splitFlag, SPLIT_FLAG_STABLE_SPLIT);
} else {
nodesDestroyList(pMergeKeys);
}
return code;
}
static int32_t stbSplSplitSession(SSplitContext* pCxt, SStableSplitInfo* pInfo) {
if (pCxt->pPlanCxt->streamQuery) {
return stbSplSplitSessionForStream(pCxt, pInfo);
} else {
return TSDB_CODE_PLAN_INTERNAL_ERROR;
return stbSplSplitSessionOrStateForBatch(pCxt, pInfo);
}
}
static int32_t stbSplSplitStateForStream(SSplitContext* pCxt, SStableSplitInfo* pInfo) {
return TSDB_CODE_PLAN_INTERNAL_ERROR;
}
static int32_t stbSplSplitState(SSplitContext* pCxt, SStableSplitInfo* pInfo) {
if (pCxt->pPlanCxt->streamQuery) {
return stbSplSplitStateForStream(pCxt, pInfo);
} else {
return stbSplSplitSessionOrStateForBatch(pCxt, pInfo);
}
}
@ -511,6 +584,8 @@ static int32_t stbSplSplitWindowForMergeTable(SSplitContext* pCxt, SStableSplitI
return stbSplSplitInterval(pCxt, pInfo);
case WINDOW_TYPE_SESSION:
return stbSplSplitSession(pCxt, pInfo);
case WINDOW_TYPE_STATE:
return stbSplSplitState(pCxt, pInfo);
default:
break;
}

2
source/libs/planner/test/planOptimizeTest.cpp
View File

@ -40,6 +40,8 @@ TEST_F(PlanOptimizeTest, ConditionPushDown) {
run("SELECT ts, c1 FROM st1 WHERE tag1 > 4 or tag1 < 2");
run("SELECT ts, c1 FROM st1 WHERE tag1 > 4 AND tag2 = 'hello'");
run("SELECT ts, c1 FROM st1 WHERE tag1 > 4 AND tag2 = 'hello' AND c1 > 10");
}
TEST_F(PlanOptimizeTest, orderByPrimaryKey) {

10
source/libs/planner/test/planSessionTest.cpp
View File

@ -34,3 +34,13 @@ TEST_F(PlanSessionTest, selectFunc) {
// select function along with the columns of select row, and with SESSION clause
run("SELECT MAX(c1), c2 FROM t1 SESSION(ts, 10s)");
}
TEST_F(PlanSessionTest, stable) {
useDb("root", "test");
// select function for SESSION clause
run("SELECT MAX(c1), MIN(c1) FROM st1 SESSION(ts, 10s)");
// select function along with the columns of select row, and with SESSION clause
run("SELECT MAX(c1), c2 FROM st1 SESSION(ts, 10s)");
run("SELECT count(ts) FROM st1 PARTITION BY c1 SESSION(ts, 10s)");
}

9
source/libs/planner/test/planStateTest.cpp
View File

@ -40,3 +40,12 @@ TEST_F(PlanStateTest, selectFunc) {
// select function along with the columns of select row, and with STATE_WINDOW clause
run("SELECT MAX(c1), c2 FROM t1 STATE_WINDOW(c3)");
}
TEST_F(PlanStateTest, stable) {
useDb("root", "test");
// select function for STATE_WINDOW clause
run("SELECT MAX(c1), MIN(c1) FROM st1 STATE_WINDOW(c2)");
// select function along with the columns of select row, and with STATE_WINDOW clause
run("SELECT MAX(c1), c2 FROM st1 STATE_WINDOW(c2)");
}

40
source/libs/scalar/src/sclvector.c
View File

@ -23,6 +23,7 @@
#include "sclvector.h"
#include "tcompare.h"
#include "tdatablock.h"
#include "tdataformat.h"
#include "ttypes.h"
#include "ttime.h"
@ -506,6 +507,16 @@ bool convertJsonValue(__compar_fn_t *fp, int32_t optr, int8_t typeLeft, int8_t t
}
}
// if types can not comparable
if((IS_NUMERIC_TYPE(typeLeft) && !IS_NUMERIC_TYPE(typeRight)) ||
(IS_NUMERIC_TYPE(typeRight) && !IS_NUMERIC_TYPE(typeLeft)) ||
(IS_VAR_DATA_TYPE(typeLeft) && !IS_VAR_DATA_TYPE(typeRight)) ||
(IS_VAR_DATA_TYPE(typeRight) && !IS_VAR_DATA_TYPE(typeLeft)) ||
((typeLeft == TSDB_DATA_TYPE_BOOL) && (typeRight != TSDB_DATA_TYPE_BOOL)) ||
((typeRight == TSDB_DATA_TYPE_BOOL) && (typeLeft != TSDB_DATA_TYPE_BOOL)))
return false;
if(typeLeft == TSDB_DATA_TYPE_NULL || typeRight == TSDB_DATA_TYPE_NULL){
*isNull = true;
return true;
@ -519,24 +530,28 @@ bool convertJsonValue(__compar_fn_t *fp, int32_t optr, int8_t typeLeft, int8_t t
*fp = filterGetCompFunc(type, optr);
if(IS_NUMERIC_TYPE(type) || IS_FLOAT_TYPE(type)){
if(IS_NUMERIC_TYPE(type)){
if(typeLeft == TSDB_DATA_TYPE_NCHAR) {
convertNcharToDouble(*pLeftData, pLeftOut);
*pLeftData = pLeftOut;
ASSERT(0);
// convertNcharToDouble(*pLeftData, pLeftOut);
// *pLeftData = pLeftOut;
} else if(typeLeft == TSDB_DATA_TYPE_BINARY) {
convertBinaryToDouble(*pLeftData, pLeftOut);
*pLeftData = pLeftOut;
ASSERT(0);
// convertBinaryToDouble(*pLeftData, pLeftOut);
// *pLeftData = pLeftOut;
} else if(typeLeft != type) {
convertNumberToNumber(*pLeftData, pLeftOut, typeLeft, type);
*pLeftData = pLeftOut;
}
if(typeRight == TSDB_DATA_TYPE_NCHAR) {
convertNcharToDouble(*pRightData, pRightOut);
*pRightData = pRightOut;
ASSERT(0);
// convertNcharToDouble(*pRightData, pRightOut);
// *pRightData = pRightOut;
} else if(typeRight == TSDB_DATA_TYPE_BINARY) {
convertBinaryToDouble(*pRightData, pRightOut);
*pRightData = pRightOut;
ASSERT(0);
// convertBinaryToDouble(*pRightData, pRightOut);
// *pRightData = pRightOut;
} else if(typeRight != type) {
convertNumberToNumber(*pRightData, pRightOut, typeRight, type);
*pRightData = pRightOut;
@ -1693,6 +1708,13 @@ void vectorIsTrue(SScalarParam* pLeft, SScalarParam* pRight, SScalarParam *pOut,
STagVal getJsonValue(char *json, char *key, bool *isExist) {
STagVal val = {.pKey = key};
if (tTagIsJson((const STag *)json) == false){
if(isExist){
*isExist = false;
}
return val;
}
bool find = tTagGet(((const STag *)json), &val); // json value is null and not exist is different
if(isExist){
*isExist = find;

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

@ -24,13 +24,12 @@ static int32_t streamTaskExecImpl(SStreamTask* pTask, void* data, SArray* pRes)
SStreamTrigger* pTrigger = (SStreamTrigger*)data;
qSetMultiStreamInput(exec, pTrigger->pBlock, 1, STREAM_DATA_TYPE_SSDATA_BLOCK, false);
} else if (pItem->type == STREAM_INPUT__DATA_SUBMIT) {
ASSERT(pTask->isDataScan);
SStreamDataSubmit* pSubmit = (SStreamDataSubmit*)data;
ASSERT(pTask->inputType == STREAM_INPUT__DATA_SUBMIT);
qSetStreamInput(exec, pSubmit->data, STREAM_DATA_TYPE_SUBMIT_BLOCK, false);
} else if (pItem->type == STREAM_INPUT__DATA_BLOCK) {
SStreamDataBlock* pBlock = (SStreamDataBlock*)data;
ASSERT(pTask->inputType == STREAM_INPUT__DATA_BLOCK);
SArray* blocks = pBlock->blocks;
SArray* blocks = pBlock->blocks;
qSetMultiStreamInput(exec, blocks->pData, blocks->size, STREAM_DATA_TYPE_SSDATA_BLOCK, false);
} else if (pItem->type == STREAM_INPUT__DROP) {
// TODO exec drop
@ -89,17 +88,17 @@ static SArray* streamExecForQall(SStreamTask* pTask, SArray* pRes) {
return NULL;
}
if (((SStreamQueueItem*)data)->type == STREAM_INPUT__TRIGGER) {
int8_t type = ((SStreamQueueItem*)data)->type;
if (type == STREAM_INPUT__TRIGGER) {
blockDataDestroy(((SStreamTrigger*)data)->pBlock);
taosFreeQitem(data);
} else {
if (pTask->inputType == STREAM_INPUT__DATA_SUBMIT) {
streamDataSubmitRefDec((SStreamDataSubmit*)data);
taosFreeQitem(data);
} else {
taosArrayDestroyEx(((SStreamDataBlock*)data)->blocks, (FDelete)tDeleteSSDataBlock);
taosFreeQitem(data);
}
} else if (type == STREAM_INPUT__DATA_BLOCK) {
taosArrayDestroyEx(((SStreamDataBlock*)data)->blocks, (FDelete)tDeleteSSDataBlock);
taosFreeQitem(data);
} else if (type == STREAM_INPUT__DATA_SUBMIT) {
ASSERT(pTask->isDataScan);
streamDataSubmitRefDec((SStreamDataSubmit*)data);
taosFreeQitem(data);
}
streamQueueProcessSuccess(pTask->inputQueue);
return taosArrayInit(0, sizeof(SSDataBlock));

16
source/libs/stream/src/streamTask.c
View File

@ -50,14 +50,14 @@ int32_t tEncodeSStreamTask(SEncoder* pEncoder, const SStreamTask* pTask) {
/*if (tStartEncode(pEncoder) < 0) return -1;*/
if (tEncodeI64(pEncoder, pTask->streamId) < 0) return -1;
if (tEncodeI32(pEncoder, pTask->taskId) < 0) return -1;
if (tEncodeI8(pEncoder, pTask->inputType) < 0) return -1;
if (tEncodeI8(pEncoder, pTask->taskStatus) < 0) return -1;
if (tEncodeI8(pEncoder, pTask->execStatus) < 0) return -1;
if (tEncodeI8(pEncoder, pTask->isDataScan) < 0) return -1;
if (tEncodeI8(pEncoder, pTask->execType) < 0) return -1;
if (tEncodeI8(pEncoder, pTask->sinkType) < 0) return -1;
if (tEncodeI8(pEncoder, pTask->dispatchType) < 0) return -1;
if (tEncodeI16(pEncoder, pTask->dispatchMsgType) < 0) return -1;
if (tEncodeI8(pEncoder, pTask->isDataScan) < 0) return -1;
if (tEncodeI8(pEncoder, pTask->taskStatus) < 0) return -1;
if (tEncodeI8(pEncoder, pTask->execStatus) < 0) return -1;
if (tEncodeI32(pEncoder, pTask->selfChildId) < 0) return -1;
if (tEncodeI32(pEncoder, pTask->nodeId) < 0) return -1;
@ -106,14 +106,14 @@ int32_t tDecodeSStreamTask(SDecoder* pDecoder, SStreamTask* pTask) {
/*if (tStartDecode(pDecoder) < 0) return -1;*/
if (tDecodeI64(pDecoder, &pTask->streamId) < 0) return -1;
if (tDecodeI32(pDecoder, &pTask->taskId) < 0) return -1;
if (tDecodeI8(pDecoder, &pTask->inputType) < 0) return -1;
if (tDecodeI8(pDecoder, &pTask->taskStatus) < 0) return -1;
if (tDecodeI8(pDecoder, &pTask->execStatus) < 0) return -1;
if (tDecodeI8(pDecoder, &pTask->isDataScan) < 0) return -1;
if (tDecodeI8(pDecoder, &pTask->execType) < 0) return -1;
if (tDecodeI8(pDecoder, &pTask->sinkType) < 0) return -1;
if (tDecodeI8(pDecoder, &pTask->dispatchType) < 0) return -1;
if (tDecodeI16(pDecoder, &pTask->dispatchMsgType) < 0) return -1;
if (tDecodeI8(pDecoder, &pTask->isDataScan) < 0) return -1;
if (tDecodeI8(pDecoder, &pTask->taskStatus) < 0) return -1;
if (tDecodeI8(pDecoder, &pTask->execStatus) < 0) return -1;
if (tDecodeI32(pDecoder, &pTask->selfChildId) < 0) return -1;
if (tDecodeI32(pDecoder, &pTask->nodeId) < 0) return -1;

4
source/os/src/osLocale.c
View File

@ -88,11 +88,11 @@ void taosSetSystemLocale(const char *inLocale, const char *inCharSet) {
void taosGetSystemLocale(char *outLocale, char *outCharset) {
#ifdef WINDOWS
char *locale = setlocale(LC_CTYPE, "chs");
char *locale = setlocale(LC_CTYPE, "en_US.UTF-8");
if (locale != NULL) {
tstrncpy(outLocale, locale, TD_LOCALE_LEN);
}
strcpy(outCharset, "cp936");
strcpy(outCharset, "UTF-8");
#elif defined(_TD_DARWIN_64)
/*

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

@ -12,6 +12,8 @@
## ---- db
./test.sh -f tsim/db/create_all_options.sim
./test.sh -f tsim/db/alter_option.sim
./test.sh -f tsim/db/alter_replica_13.sim
#./test.sh -f tsim/db/alter_replica_31.sim
./test.sh -f tsim/db/basic1.sim
./test.sh -f tsim/db/basic2.sim
./test.sh -f tsim/db/basic3.sim
@ -21,6 +23,7 @@
./test.sh -f tsim/db/taosdlog.sim
# ---- dnode
./test.sh -f tsim/dnode/balance_replica1.sim
./test.sh -f tsim/dnode/create_dnode.sim
./test.sh -f tsim/dnode/drop_dnode_has_mnode.sim
./test.sh -f tsim/dnode/drop_dnode_has_qnode_snode.sim
@ -28,6 +31,8 @@
#./test.sh -f tsim/dnode/drop_dnode_has_vnode_replica3.sim
#./test.sh -f tsim/dnode/drop_dnode_has_multi_vnode_replica1.sim
#./test.sh -f tsim/dnode/drop_dnode_has_multi_vnode_replica3.sim
#./test.sh -f tsim/dnode/redistribute_vgroup_replica3_v1_leader.sim
./test.sh -f tsim/dnode/redistribute_vgroup_replica3_v1_follower.sim
# ---- insert
./test.sh -f tsim/insert/basic0.sim

123
tests/script/tsim/dnode/balance_replica1.sim Normal file
View File

@ -0,0 +1,123 @@
system sh/stop_dnodes.sh
system sh/deploy.sh -n dnode1 -i 1
system sh/deploy.sh -n dnode2 -i 2
system sh/deploy.sh -n dnode3 -i 3
system sh/cfg.sh -n dnode1 -c supportVnodes -v 0
system sh/exec.sh -n dnode1 -s start
system sh/exec.sh -n dnode2 -s start
sql connect
print =============== step1 create dnode2
# no enough vnodes
sql balance vgroup
sql create dnode $hostname port 7200
sql create dnode $hostname port 7300
$x = 0
step1:
$ = $x + 1
sleep 1000
if $x == 10 then
print ---> dnode not online!
return -1
endi
sql show dnodes
print ---> $data00 $data01 $data02 $data03 $data04 $data05
print ---> $data10 $data11 $data12 $data13 $data14 $data15
if $rows != 3 then
return -1
endi
if $data(1)[4] != ready then
goto step1
endi
if $data(2)[4] != ready then
goto step1
endi
if $data(3)[4] != offline then
goto step1
endi
print =============== step2 create database
sql create database d1 vgroups 2
sql use d1
sql create table d1.st (ts timestamp, i int) tags (j int)
sql create table d1.c1 using st tags(1)
sql create table d1.c2 using st tags(1)
sql create table d1.c3 using st tags(1)
sql create table d1.c4 using st tags(1)
sql create table d1.c5 using st tags(1)
sql create table d1.c6 using st tags(1)
sql show d1.tables
if $rows != 6 then
return -1
endi
sql show d1.vgroups
print $data[0][0] $data[0][1] $data[0][2] $data[0][3] $data[0][4]
if $rows != 2 then
return -1
endi
if $data(2)[3] != 2 then
return -1
endi
if $data(3)[3] != 2 then
return -1
endi
print =============== step3: balance vgroup
# has offline dnode
sql_error balance vgroup
system sh/exec.sh -n dnode3 -s start
$x = 0
step3:
$ = $x + 1
sleep 1000
if $x == 10 then
print ---> dnode not online!
return -1
endi
sql show dnodes
print ---> $data00 $data01 $data02 $data03 $data04 $data05
print ---> $data10 $data11 $data12 $data13 $data14 $data15
if $rows != 3 then
return -1
endi
if $data(1)[4] != ready then
goto step3
endi
if $data(2)[4] != ready then
goto step3
endi
if $data(3)[4] != ready then
goto step3
endi
print =============== step4: balance
sql balance vgroup
print show d1.vgroups
sql show d1.vgroups
print $data[0][0] $data[0][1] $data[0][2] $data[0][3] $data[0][4]
if $rows != 2 then
return -1
endi
if $data(2)[3] != 3 then
return -1
endi
if $data(3)[3] != 2 then
return -1
endi
print =============== step7: select data
sql show d1.tables
print rows $rows
if $rows != 6 then
return -1
endi
return
system sh/exec.sh -n dnode1 -s stop -x SIGINT
system sh/exec.sh -n dnode2 -s stop -x SIGINT
system sh/exec.sh -n dnode3 -s stop -x SIGINT

9
tests/script/tsim/dnode/drop_dnode_has_vnode_replica3.sim
View File

@ -28,7 +28,7 @@ step1:
sql show dnodes
print ===> $data00 $data01 $data02 $data03 $data04 $data05
print ===> $data10 $data11 $data12 $data13 $data14 $data15
if $rows != 3 then
if $rows != 5 then
return -1
endi
if $data(1)[4] != ready then
@ -72,6 +72,13 @@ if $data(2)[7] != 4 then
return -1
endi
system sh/exec.sh -n dnode4 -s stop -x SIGINT
system sh/exec.sh -n dnode3 -s stop -x SIGINT
return
print =============== step4: drop dnode 2
system sh/exec.sh -n dnode5 -s start
$x = 0

2
tests/script/tsim/dnode/redistribute_vgroup_replica3_v1_follower.sim
View File

@ -176,6 +176,8 @@ if $rows != 1 then
return -1
endi
return
print =============== step33: move follower1
print redistribute vgroup 2 dnode $leaderVnode dnode $follower1 dnode 5
sql redistribute vgroup 2 dnode $leaderVnode dnode $follower1 dnode 5

2
tests/script/wtest.bat
View File

@ -52,7 +52,7 @@ echo wal 0 >> %TAOS_CFG%
echo asyncLog 0 >> %TAOS_CFG%
echo locale en_US.UTF-8 >> %TAOS_CFG%
echo enableCoreFile 1 >> %TAOS_CFG%
echo charset cp65001 >> %TAOS_CFG%
echo charset UTF-8 >> %TAOS_CFG%
set "FILE_NAME=testSuite.sim"
if "%1" == "-f" set "FILE_NAME=%2"

278
tests/system-test/2-query/json_tag.py
View File

@ -68,7 +68,7 @@ class TDTestCase:
tdSql.error("CREATE TABLE if not exists jsons1_14 using jsons1 tags('[1,true]')")
tdSql.error("CREATE TABLE if not exists jsons1_14 using jsons1 tags('{222}')")
tdSql.error("CREATE TABLE if not exists jsons1_14 using jsons1 tags('{\"fe\"}')")
#
# test invalidate json key, key must can be printed assic char
tdSql.error("CREATE TABLE if not exists jsons1_14 using jsons1 tags('{\"tag1\":[1,true]}')")
tdSql.error("CREATE TABLE if not exists jsons1_14 using jsons1 tags('{\"tag1\":{}}')")
@ -79,7 +79,7 @@ class TDTestCase:
# test invalidate json value, value number can not be inf,nan TD-12166
tdSql.error("CREATE TABLE if not exists jsons1_14 using jsons1 tags('{\"k\":1.8e308}')")
tdSql.error("CREATE TABLE if not exists jsons1_14 using jsons1 tags('{\"k\":-1.8e308}')")
#
#test length limit
char1= ''.join(['abcd']*64)
char3= ''.join(['abcd']*1021)
@ -87,15 +87,15 @@ class TDTestCase:
tdSql.error("CREATE TABLE if not exists jsons1_15 using jsons1 tags('{\"%s1\":5}')" % char1) # len(key)=257
tdSql.execute("CREATE TABLE if not exists jsons1_15 using jsons1 tags('{\"%s\":5}')" % char1) # len(key)=256
tdSql.error("CREATE TABLE if not exists jsons1_16 using jsons1 tags('{\"TSSSS\":\"%s\"}')" % char3) # len(object)=4096
#tdSql.execute("CREATE TABLE if not exists jsons1_16 using jsons1 tags('{\"TSSS\":\"%s\"}')" % char3) # len(object)=4095
tdSql.execute("CREATE TABLE if not exists jsons1_16 using jsons1 tags('{\"TSSS\":\"%s\"}')" % char3) # len(object)=4095
tdSql.execute("drop table if exists jsons1_15")
tdSql.execute("drop table if exists jsons1_16")
#
print("============== STEP 2 ===== alter table json tag")
tdSql.error("ALTER STABLE jsons1 add tag tag2 nchar(20)")
tdSql.error("ALTER STABLE jsons1 drop tag jtag")
tdSql.error("ALTER TABLE jsons1 MODIFY TAG jtag nchar(128)")
#
tdSql.execute("ALTER TABLE jsons1_1 SET TAG jtag='{\"tag1\":\"femail\",\"tag2\":35,\"tag3\":true}'")
tdSql.query("select jtag from jsons1_1")
tdSql.checkData(0, 0, '{"tag1":"femail","tag2":35,"tag3":true}')
@ -105,9 +105,9 @@ class TDTestCase:
tdSql.execute("create table st(ts timestamp, i int) tags(t int)")
tdSql.error("ALTER STABLE st add tag jtag json")
tdSql.error("ALTER STABLE st add column jtag json")
#
# print("============== STEP 3 ===== query table")
# # test error syntax
print("============== STEP 3 ===== query table")
# test error syntax
tdSql.error("select * from jsons1 where jtag->tag1='beijing'")
tdSql.error("select -> from jsons1")
tdSql.error("select * from jsons1 where contains")
@ -115,17 +115,17 @@ class TDTestCase:
tdSql.error("select jtag->location from jsons1")
tdSql.error("select jtag contains location from jsons1")
tdSql.error("select * from jsons1 where jtag contains location")
#tdSql.error("select * from jsons1 where jtag contains''")
tdSql.query("select * from jsons1 where jtag contains''")
tdSql.error("select * from jsons1 where jtag contains 'location'='beijing'")
#
# # test function error
# test function error
tdSql.error("select avg(jtag->'tag1') from jsons1")
tdSql.error("select avg(jtag) from jsons1")
tdSql.error("select min(jtag->'tag1') from jsons1")
tdSql.error("select min(jtag) from jsons1")
tdSql.error("select ceil(jtag->'tag1') from jsons1")
tdSql.error("select ceil(jtag) from jsons1")
#
#test scalar operation
tdSql.query("select jtag contains 'tag1',jtag->'tag1' from jsons1 order by jtag->'tag1'")
@ -158,10 +158,11 @@ class TDTestCase:
tdSql.checkData(0, 0, None)
tdSql.checkData(0, 1, False)
tdSql.checkData(7, 0, "false")
tdSql.checkData(7, 1, True)
tdSql.checkData(7, 1, False)
tdSql.checkData(8, 1, False)
tdSql.checkData(12, 1, True)
# # test select normal column
# test select normal column
tdSql.query("select dataint from jsons1 order by dataint")
tdSql.checkRows(9)
tdSql.checkData(1, 0, 1)
@ -180,7 +181,7 @@ class TDTestCase:
tdSql.query("select jtag from jsons1_9")
tdSql.checkData(0, 0, None)
# # test select json tag->'key', value is string
# test select json tag->'key', value is string
tdSql.query("select jtag->'tag1' from jsons1_1")
tdSql.checkData(0, 0, '"femail"')
tdSql.query("select jtag->'tag2' from jsons1_6")
@ -200,7 +201,7 @@ class TDTestCase:
# test select json tag->'key', key is not exist
tdSql.query("select jtag->'tag10' from jsons1_4")
tdSql.checkData(0, 0, None)
#
tdSql.query("select jtag->'tag1' from jsons1")
tdSql.checkRows(13)
# test header name
@ -210,24 +211,25 @@ class TDTestCase:
tdSql.checkColNameList(res, cname_list)
# # test where with json tag
# test where with json tag
tdSql.query("select * from jsons1_1 where jtag is not null")
# tdSql.error("select * from jsons1 where jtag='{\"tag1\":11,\"tag2\":\"\"}'")
# tdSql.error("select * from jsons1 where jtag->'tag1'={}")
#
# # where json value is string
# tdSql.query("select * from jsons1 where jtag='{\"tag1\":11,\"tag2\":\"\"}'")
tdSql.error("select * from jsons1 where jtag->'tag1'={}")
# where json value is string
tdSql.query("select * from jsons1 where jtag->'tag2'='beijing'")
tdSql.checkRows(2)
tdSql.query("select dataint,tbname,jtag->'tag1',jtag from jsons1 where jtag->'tag2'='beijing'")
tdSql.query("select dataint,tbname,jtag->'tag1',jtag from jsons1 where jtag->'tag2'='beijing' order by dataint")
tdSql.checkRows(2)
# out of order, cannot compare value
#tdSql.checkData(0, 0, 2)
#tdSql.checkData(0, 1, 'jsons1_2')
#tdSql.checkData(0, 2, 5)
#tdSql.checkData(0, 3, '{"tag1":5,"tag2":"beijing"}')
#tdSql.checkData(1, 0, 3)
#tdSql.checkData(1, 1, 'jsons1_3')
#tdSql.checkData(1, 2, 'false')
tdSql.checkData(0, 0, 2)
tdSql.checkData(0, 1, 'jsons1_2')
tdSql.checkData(0, 2, "5.000000000")
tdSql.checkData(0, 3, '{"tag1":5,"tag2":"beijing"}')
tdSql.checkData(1, 0, 3)
tdSql.checkData(1, 1, 'jsons1_3')
tdSql.checkData(1, 2, 'false')
tdSql.query("select * from jsons1 where jtag->'tag1'='beijing'")
tdSql.checkRows(0)
tdSql.query("select * from jsons1 where jtag->'tag1'='收到货'")
@ -236,72 +238,73 @@ class TDTestCase:
tdSql.checkRows(1)
tdSql.query("select * from jsons1 where jtag->'tag2'>='beijing'")
tdSql.checkRows(3)
# open
#tdSql.query("select * from jsons1 where jtag->'tag2'<'beijing'")
#tdSql.checkRows(2)
tdSql.query("select * from jsons1 where jtag->'tag2'<='beijing'")
tdSql.query("select * from jsons1 where jtag->'tag2'<'beijing'")
tdSql.checkRows(2)
tdSql.query("select * from jsons1 where jtag->'tag2'<='beijing'")
tdSql.checkRows(4)
tdSql.query("select * from jsons1 where jtag->'tag2'!='beijing'")
tdSql.checkRows(5)
tdSql.checkRows(3)
tdSql.query("select * from jsons1 where jtag->'tag2'=''")
tdSql.checkRows(2)
#
# # where json value is int
# where json value is int
tdSql.query("select * from jsons1 where jtag->'tag1'=5")
tdSql.checkRows(1)
tdSql.checkData(0, 1, 2)
tdSql.query("select * from jsons1 where jtag->'tag1'=10")
tdSql.checkRows(0)
tdSql.query("select * from jsons1 where jtag->'tag1'<54")
tdSql.checkRows(4)
tdSql.checkRows(3)
tdSql.query("select * from jsons1 where jtag->'tag1'<=11")
tdSql.checkRows(4)
tdSql.checkRows(3)
tdSql.query("select * from jsons1 where jtag->'tag1'>4")
tdSql.checkRows(2)
tdSql.query("select * from jsons1 where jtag->'tag1'>=5")
tdSql.checkRows(2)
tdSql.query("select * from jsons1 where jtag->'tag1'!=5")
tdSql.checkRows(6)
tdSql.checkRows(2)
tdSql.query("select * from jsons1 where jtag->'tag1'!=55")
tdSql.checkRows(7)
#
# # where json value is double
tdSql.checkRows(3)
# where json value is double
tdSql.query("select * from jsons1 where jtag->'tag1'=1.232")
tdSql.checkRows(1)
tdSql.query("select * from jsons1 where jtag->'tag1'<1.232")
tdSql.checkRows(1)
tdSql.checkRows(0)
tdSql.query("select * from jsons1 where jtag->'tag1'<=1.232")
tdSql.checkRows(2)
tdSql.checkRows(1)
tdSql.query("select * from jsons1 where jtag->'tag1'>1.23")
tdSql.checkRows(3)
tdSql.query("select * from jsons1 where jtag->'tag1'>=1.232")
tdSql.checkRows(3)
tdSql.query("select * from jsons1 where jtag->'tag1'!=1.232")
tdSql.checkRows(6)
tdSql.checkRows(2)
tdSql.query("select * from jsons1 where jtag->'tag1'!=3.232")
tdSql.checkRows(7)
#tdSql.error("select * from jsons1 where jtag->'tag1'/0=3")
#tdSql.error("select * from jsons1 where jtag->'tag1'/5=1")
#
# # where json value is bool
tdSql.checkRows(3)
tdSql.query("select * from jsons1 where jtag->'tag1'/0=3")
tdSql.checkRows(0)
tdSql.query("select * from jsons1 where jtag->'tag1'/5=1")
tdSql.checkRows(1)
# where json value is bool
tdSql.query("select * from jsons1 where jtag->'tag1'=true")
tdSql.checkRows(0)
#tdSql.query("select * from jsons1 where jtag->'tag1'=false")
#tdSql.checkRows(1)
tdSql.query("select * from jsons1 where jtag->'tag1'=false")
tdSql.checkRows(1)
tdSql.query("select * from jsons1 where jtag->'tag1'!=false")
tdSql.checkRows(3)
#tdSql.error("select * from jsons1 where jtag->'tag1'>false")
#
# # where json value is null
# open
#tdSql.query("select * from jsons1 where jtag->'tag1'=null") # only json suport =null. This synatx will change later.
#tdSql.checkRows(1)
#
# # where json key is null
tdSql.checkRows(0)
tdSql.query("select * from jsons1 where jtag->'tag1'>false")
tdSql.checkRows(0)
# where json value is null
tdSql.query("select * from jsons1 where jtag->'tag1'=null")
tdSql.checkRows(0)
# where json key is null
tdSql.query("select * from jsons1 where jtag->'tag_no_exist'=3")
tdSql.checkRows(0)
#
# # where json value is not exist
# where json value is not exist
tdSql.query("select * from jsons1 where jtag->'tag1' is null")
tdSql.checkData(0, 0, 'jsons1_9')
tdSql.checkRows(2)
@ -309,16 +312,16 @@ class TDTestCase:
tdSql.checkRows(9)
tdSql.query("select * from jsons1 where jtag->'tag3' is not null")
tdSql.checkRows(3)
#
# # test contains
# test contains
tdSql.query("select * from jsons1 where jtag contains 'tag1'")
tdSql.checkRows(7)
tdSql.checkRows(8)
tdSql.query("select * from jsons1 where jtag contains 'tag3'")
tdSql.checkRows(3)
tdSql.checkRows(4)
tdSql.query("select * from jsons1 where jtag contains 'tag_no_exist'")
tdSql.checkRows(0)
#
# # test json tag in where condition with and/or
# test json tag in where condition with and/or
tdSql.query("select * from jsons1 where jtag->'tag1'=false and jtag->'tag2'='beijing'")
tdSql.checkRows(1)
tdSql.query("select * from jsons1 where jtag->'tag1'=false or jtag->'tag2'='beijing'")
@ -335,15 +338,15 @@ class TDTestCase:
tdSql.checkRows(3)
tdSql.query("select * from jsons1 where jtag->'tag1'='femail' and jtag contains 'tag3'")
tdSql.checkRows(2)
#
#
# # test with between and
# test with between and
tdSql.query("select * from jsons1 where jtag->'tag1' between 1 and 30")
tdSql.checkRows(3)
tdSql.query("select * from jsons1 where jtag->'tag1' between 'femail' and 'beijing'")
tdSql.checkRows(2)
#
# # test with tbname/normal column
# test with tbname/normal column
tdSql.query("select * from jsons1 where tbname = 'jsons1_1'")
tdSql.checkRows(2)
tdSql.query("select * from jsons1 where tbname = 'jsons1_1' and jtag contains 'tag3'")
@ -352,20 +355,18 @@ class TDTestCase:
tdSql.checkRows(0)
tdSql.query("select * from jsons1 where tbname = 'jsons1_1' and jtag contains 'tag3' and dataint=23")
tdSql.checkRows(1)
#
#
# # test where condition like
# open
# syntax error
#tdSql.query("select *,tbname from jsons1 where jtag->'tag2' like 'bei%'")
#tdSql.checkRows(2)
#tdSql.query("select *,tbname from jsons1 where jtag->'tag1' like 'fe%' and jtag->'tag2' is not null")
#tdSql.checkRows(2)
#
# # test where condition in no support in
# test where condition like
tdSql.query("select * from jsons1 where jtag->'tag2' like 'bei%'")
tdSql.checkRows(2)
tdSql.query("select * from jsons1 where jtag->'tag1' like 'fe%' and jtag->'tag2' is not null")
tdSql.checkRows(2)
# test where condition in no support in
# tdSql.error("select * from jsons1 where jtag->'tag1' in ('beijing')")
#
# # test where condition match/nmath
# test where condition match/nmath
tdSql.query("select * from jsons1 where jtag->'tag1' match 'ma'")
tdSql.checkRows(2)
tdSql.query("select * from jsons1 where jtag->'tag1' match 'ma$'")
@ -376,23 +377,22 @@ class TDTestCase:
tdSql.checkRows(1)
tdSql.query("select * from jsons1 where jtag->'tag1' nmatch 'ma'")
tdSql.checkRows(1)
#
# # test distinct
# test distinct
tdSql.execute("insert into jsons1_14 using jsons1 tags('{\"tag1\":\"收到货\",\"tag2\":\"\",\"tag3\":null}') values(1591062628000, 2, NULL, '你就会', 'dws')")
tdSql.query("select distinct jtag->'tag1' from jsons1")
tdSql.checkRows(8)
tdSql.query("select distinct jtag from jsons1")
tdSql.checkRows(9)
#
# #test dumplicate key with normal colomn
#test dumplicate key with normal colomn
tdSql.execute("INSERT INTO jsons1_15 using jsons1 tags('{\"tbname\":\"tt\",\"databool\":true,\"datastr\":\"是是是\"}') values(1591060828000, 4, false, 'jjsf', \"你就会\")")
#tdSql.query("select *,tbname,jtag from jsons1 where jtag->'datastr' match '是' and datastr match 'js'")
#tdSql.checkRows(1)
# open
#tdSql.query("select tbname,jtag->'tbname' from jsons1 where jtag->'tbname'='tt' and tbname='jsons1_14'")
#tdSql.checkRows(0)
#
# # test join
tdSql.query("select * from jsons1 where jtag->'datastr' match '是' and datastr match 'js'")
tdSql.checkRows(1)
# tdSql.query("select tbname,jtag->'tbname' from jsons1 where jtag->'tbname'='tt' and tbname='jsons1_14'")
# tdSql.checkRows(1)
# test join
tdSql.execute("create table if not exists jsons2(ts timestamp, dataInt int, dataBool bool, dataStr nchar(50), dataStrBin binary(150)) tags(jtag json)")
tdSql.execute("insert into jsons2_1 using jsons2 tags('{\"tag1\":\"fff\",\"tag2\":5, \"tag3\":true}') values(1591060618000, 2, false, 'json2', '你是2')")
tdSql.execute("insert into jsons2_2 using jsons2 tags('{\"tag1\":5,\"tag2\":null}') values (1591060628000, 2, true, 'json2', 'sss')")
@ -460,19 +460,18 @@ class TDTestCase:
tdSql.checkColNameList(res, cname_list)
# test top/bottom with group by json tag
# random failure
#tdSql.query("select top(dataint,2),jtag->'tag1' from jsons1 group by jtag->'tag1' order by jtag->'tag1'")
#tdSql.checkRows(11)
#tdSql.checkData(0, 1, None)
#tdSql.checkData(2, 0, 4)
#tdSql.checkData(3, 0, 3)
#tdSql.checkData(3, 1, "false")
#tdSql.checkData(10, 0, 23)
#tdSql.checkData(10, 1, '"femail"')
# tdSql.query("select top(dataint,2),jtag->'tag1' from jsons1 group by jtag->'tag1' order by jtag->'tag1'")
# tdSql.checkRows(11)
# tdSql.checkData(0, 1, None)
# tdSql.checkData(2, 0, 4)
# tdSql.checkData(3, 0, 3)
# tdSql.checkData(3, 1, "false")
# tdSql.checkData(8, 0, 2)
# tdSql.checkData(10, 1, '"femail"')
# test having
#tdSql.query("select count(*),jtag->'tag1' from jsons1 group by jtag->'tag1' having count(*) > 1")
#tdSql.checkRows(3)
tdSql.query("select count(*),jtag->'tag1' from jsons1 group by jtag->'tag1' having count(*) > 1")
tdSql.checkRows(3)
# subquery with json tag
tdSql.query("select * from (select jtag, dataint from jsons1) order by dataint")
@ -480,22 +479,13 @@ class TDTestCase:
tdSql.checkData(1, 1, 1)
tdSql.checkData(5, 0, '{"tag1":false,"tag2":"beijing"}')
# tdSql.query("select jtag->'tag1' from (select jtag->'tag1', dataint from jsons1)")
# tdSql.checkRows(11)
# tdSql.checkData(1, 0, '"femail"')
# tdSql.checkData(2, 0, 5)
#
# res = tdSql.getColNameList("select jtag->'tag1' from (select jtag->'tag1', dataint from jsons1)")
# cname_list = []
# cname_list.append("jtag->'tag1'")
# tdSql.checkColNameList(res, cname_list)
#
# tdSql.query("select ts,tbname,jtag->'tag1' from (select jtag->'tag1',tbname,ts from jsons1 order by ts)")
tdSql.error("select jtag->'tag1' from (select jtag->'tag1', dataint from jsons1)")
# tdSql.query("select ts,jtag->'tag1' from (select jtag->'tag1',tbname,ts from jsons1 order by ts)")
# tdSql.checkRows(11)
# tdSql.checkData(1, 1, "jsons1_1")
# tdSql.checkData(1, 2, '"femail"')
#
# # union all
# union all
tdSql.query("select jtag->'tag1' from jsons1 union all select jtag->'tag2' from jsons2")
tdSql.checkRows(17)
tdSql.query("select jtag->'tag1' from jsons1_1 union all select jtag->'tag2' from jsons2_1")
@ -508,24 +498,25 @@ class TDTestCase:
tdSql.query("select dataint,jtag,tbname from jsons1 union all select dataint,jtag,tbname from jsons2")
tdSql.checkRows(13)
# #show create table
# tdSql.query("show create table jsons1")
# tdSql.checkData(0, 1, 'CREATE TABLE `jsons1` (`ts` TIMESTAMP,`dataint` INT,`databool` BOOL,`datastr` NCHAR(50),`datastrbin` BINARY(150)) TAGS (`jtag` JSON)')
#
# #test aggregate function:count/avg/twa/irate/sum/stddev/leastsquares
#show create table
tdSql.query("show create table jsons1")
tdSql.checkData(0, 1, 'CREATE STABLE `jsons1` (`ts` TIMESTAMP, `dataint` INT, `databool` BOOL, `datastr` NCHAR(50), `datastrbin` VARCHAR(150)) TAGS (`jtag` JSON) WATERMARK 5000a, 5000a')
#test aggregate function:count/avg/twa/irate/sum/stddev/leastsquares
tdSql.query("select count(*) from jsons1 where jtag is not null")
tdSql.checkData(0, 0, 10)
tdSql.query("select avg(dataint) from jsons1 where jtag is not null")
tdSql.checkData(0, 0, 5.3)
#tdSql.error("select twa(dataint) from jsons1 where jtag is not null")
tdSql.query("select irate(dataint) from jsons1 where jtag is not null")
#tdSql.query("select sum(dataint) from jsons1 where jtag->'tag1' is not null")
#tdSql.checkData(0, 0, 49)
# tdSql.query("select twa(dataint) from jsons1 where jtag is not null")
# tdSql.checkData(0, 0, 36)
tdSql.error("select irate(dataint) from jsons1 where jtag is not null")
tdSql.query("select sum(dataint) from jsons1 where jtag->'tag1' is not null")
tdSql.checkData(0, 0, 45)
tdSql.query("select stddev(dataint) from jsons1 where jtag->'tag1'>1")
tdSql.checkData(0, 0, 4.496912521)
#tdSql.error("SELECT LEASTSQUARES(dataint, 1, 1) from jsons1 where jtag is not null")
#
# #test selection function:min/max/first/last/top/bottom/percentile/apercentile/last_row/interp
tdSql.query("SELECT LEASTSQUARES(dataint, 1, 1) from jsons1 where jtag is not null")
#test selection function:min/max/first/last/top/bottom/percentile/apercentile/last_row/interp
tdSql.query("select min(dataint) from jsons1 where jtag->'tag1'>1")
tdSql.checkData(0, 0, 1)
tdSql.query("select max(dataint) from jsons1 where jtag->'tag1'>1")
@ -541,13 +532,16 @@ class TDTestCase:
tdSql.query("select percentile(dataint,20) from jsons1 where jtag->'tag1'>1")
tdSql.query("select apercentile(dataint, 50) from jsons1 where jtag->'tag1'>1")
tdSql.checkData(0, 0, 1.5)
#tdSql.query("select last_row(dataint) from jsons1 where jtag->'tag1'>1")
#tdSql.checkData(0, 0, 11)
#tdSql.error("select interp(dataint) from jsons1 where ts = '2020-06-02 09:17:08.000' and jtag->'tag1'>1")
#
# #test calculation function:diff/derivative/spread/ceil/floor/round/
#tdSql.error("select diff(dataint) from jsons1 where jtag->'tag1'>1")
#tdSql.error("select derivative(dataint, 10m, 0) from jsons1 where jtag->'tag1'>1")
# tdSql.query("select last_row(dataint) from jsons1 where jtag->'tag1'>1")
# tdSql.query("select interp(dataint) from jsons1 where ts = '2020-06-02 09:17:08.000' and jtag->'tag1'>1")
#test calculation function:diff/derivative/spread/ceil/floor/round/
tdSql.query("select diff(dataint) from jsons1 where jtag->'tag1'>1")
# tdSql.checkRows(2)
# tdSql.checkData(0, 0, -1)
# tdSql.checkData(1, 0, 10)
tdSql.query("select derivative(dataint, 10m, 0) from jsons1 where jtag->'tag1'>1")
tdSql.checkData(0, 0, -2)
tdSql.query("select spread(dataint) from jsons1 where jtag->'tag1'>1")
tdSql.checkData(0, 0, 10)
tdSql.query("select ceil(dataint) from jsons1 where jtag->'tag1'>1")

182
tests/system-test/7-tmq/tmqCheckData.py Normal file
View File

@ -0,0 +1,182 @@
import taos
import sys
import time
import socket
import os
import threading
from util.log import *
from util.sql import *
from util.cases import *
from util.dnodes import *
from util.common import *
sys.path.append("./7-tmq")
from tmqCommon import *
class TDTestCase:
def init(self, conn, logSql):
tdLog.debug(f"start to excute {__file__}")
tdSql.init(conn.cursor())
#tdSql.init(conn.cursor(), logSql) # output sql.txt file
def checkFileContent(self, consumerId, queryString):
buildPath = tdCom.getBuildPath()
cfgPath = tdCom.getClientCfgPath()
dstFile = '%s/../log/dstrows_%d.txt'%(cfgPath, consumerId)
cmdStr = '%s/build/bin/taos -c %s -s "%s >> %s"'%(buildPath, cfgPath, queryString, dstFile)
tdLog.info(cmdStr)
os.system(cmdStr)
consumeRowsFile = '%s/../log/consumerid_%d.txt'%(cfgPath, consumerId)
tdLog.info("rows file: %s, %s"%(consumeRowsFile, dstFile))
consumeFile = open(consumeRowsFile, mode='r')
queryFile = open(dstFile, mode='r')
# skip first line for it is schema
queryFile.readline()
while True:
dst = queryFile.readline()
src = consumeFile.readline()
if dst:
if dst != src:
tdLog.exit("consumerId %d consume rows is not match the rows by direct query"%consumerId)
else:
break
return
def tmqCase1(self):
tdLog.printNoPrefix("======== test case 1: ")
paraDict = {'dbName': 'db1',
'dropFlag': 1,
'event': '',
'vgroups': 4,
'stbName': 'stb',
'colPrefix': 'c',
'tagPrefix': 't',
'colSchema': [{'type': 'INT', 'count':1}, {'type': 'binary', 'len':20, 'count':1}],
'tagSchema': [{'type': 'INT', 'count':1}, {'type': 'binary', 'len':20, 'count':1}],
'ctbPrefix': 'ctb',
'ctbNum': 1,
'rowsPerTbl': 10000,
'batchNum': 10,
'startTs': 1640966400000, # 2022-01-01 00:00:00.000
'pollDelay': 10,
'showMsg': 1,
'showRow': 1}
topicNameList = ['topic1', 'topic2', 'topic3']
expectRowsList = []
tmqCom.initConsumerTable()
tdCom.create_database(tdSql, paraDict["dbName"],paraDict["dropFlag"], vgroups=4,replica=1)
tdLog.info("create stb")
tdCom.create_stable(tdSql, dbname=paraDict["dbName"],stbname=paraDict["stbName"], column_elm_list=paraDict['colSchema'], tag_elm_list=paraDict['tagSchema'])
tdLog.info("create ctb")
tdCom.create_ctable(tdSql, dbname=paraDict["dbName"],stbname=paraDict["stbName"],tag_elm_list=paraDict['tagSchema'],count=paraDict["ctbNum"], default_ctbname_prefix=paraDict['ctbPrefix'])
tdLog.info("insert data")
tmqCom.insert_data(tdSql,paraDict["dbName"],paraDict["ctbPrefix"],paraDict["ctbNum"],paraDict["rowsPerTbl"],paraDict["batchNum"],paraDict["startTs"])
tdLog.info("create topics from stb with filter")
queryString = "select ts, log(c1), ceil(pow(c1,3)) from %s.%s where c1 %% 7 == 0" %(paraDict['dbName'], paraDict['stbName'])
sqlString = "create topic %s as %s" %(topicNameList[0], queryString)
tdLog.info("create topic sql: %s"%sqlString)
tdSql.execute(sqlString)
tdSql.query(queryString)
expectRowsList.append(tdSql.getRows())
# init consume info, and start tmq_sim, then check consume result
tdLog.info("insert consume info to consume processor")
consumerId = 0
expectrowcnt = paraDict["rowsPerTbl"] * paraDict["ctbNum"]
topicList = topicNameList[0]
ifcheckdata = 1
ifManualCommit = 1
keyList = 'group.id:cgrp1, enable.auto.commit:false, auto.commit.interval.ms:6000, auto.offset.reset:earliest'
tmqCom.insertConsumerInfo(consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifManualCommit)
tdLog.info("start consume processor")
tmqCom.startTmqSimProcess(paraDict['pollDelay'],paraDict["dbName"],paraDict['showMsg'], paraDict['showRow'])
tdLog.info("wait the consume result")
expectRows = 1
resultList = tmqCom.selectConsumeResult(expectRows)
if expectRowsList[0] != resultList[0]:
tdLog.info("expect consume rows: %d, act consume rows: %d"%(expectRowsList[0], resultList[0]))
tdLog.exit("0 tmq consume rows error!")
self.checkFileContent(consumerId, queryString)
# reinit consume info, and start tmq_sim, then check consume result
tmqCom.initConsumerTable()
queryString = "select ts, log(c1), cos(c1) from %s.%s where c1 > 3169" %(paraDict['dbName'], paraDict['stbName'])
sqlString = "create topic %s as %s" %(topicNameList[1], queryString)
tdLog.info("create topic sql: %s"%sqlString)
tdSql.execute(sqlString)
tdSql.query(queryString)
expectRowsList.append(tdSql.getRows())
consumerId = 1
topicList = topicNameList[1]
tmqCom.insertConsumerInfo(consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifManualCommit)
tdLog.info("start consume processor")
tmqCom.startTmqSimProcess(paraDict['pollDelay'],paraDict["dbName"],paraDict['showMsg'], paraDict['showRow'])
tdLog.info("wait the consume result")
expectRows = 1
resultList = tmqCom.selectConsumeResult(expectRows)
if expectRowsList[1] != resultList[0]:
tdLog.info("expect consume rows: %d, act consume rows: %d"%(expectRowsList[1], resultList[0]))
tdLog.exit("1 tmq consume rows error!")
self.checkFileContent(consumerId, queryString)
# reinit consume info, and start tmq_sim, then check consume result
tmqCom.initConsumerTable()
queryString = "select ts, log(c1), atan(c1) from %s.%s where ts >= %d" %(paraDict['dbName'], paraDict['stbName'], paraDict["startTs"]+6137)
sqlString = "create topic %s as %s" %(topicNameList[2], queryString)
tdLog.info("create topic sql: %s"%sqlString)
tdSql.execute(sqlString)
tdSql.query(queryString)
expectRowsList.append(tdSql.getRows())
consumerId = 2
topicList = topicNameList[2]
tmqCom.insertConsumerInfo(consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifManualCommit)
tdLog.info("start consume processor")
tmqCom.startTmqSimProcess(paraDict['pollDelay'],paraDict["dbName"],paraDict['showMsg'], paraDict['showRow'])
tdLog.info("wait the consume result")
expectRows = 1
resultList = tmqCom.selectConsumeResult(expectRows)
# if expectRowsList[2] != resultList[0]:
# tdLog.info("expect consume rows: %d, act consume rows: %d"%(expectRowsList[2], resultList[0]))
# tdLog.exit("2 tmq consume rows error!")
# self.checkFileContent(consumerId, queryString)
time.sleep(10)
for i in range(len(topicNameList)):
tdSql.query("drop topic %s"%topicNameList[i])
tdLog.printNoPrefix("======== test case 1 end ...... ")
def run(self):
tdSql.prepare()
self.tmqCase1()
def stop(self):
tdSql.close()
tdLog.success(f"{__file__} successfully executed")
event = threading.Event()
tdCases.addLinux(__file__, TDTestCase())
tdCases.addWindows(__file__, TDTestCase())

1
tests/system-test/fulltest.sh
View File

@ -132,3 +132,4 @@ python3 ./test.py -f 7-tmq/db.py
python3 ./test.py -f 7-tmq/tmqError.py
python3 ./test.py -f 7-tmq/schema.py
python3 ./test.py -f 7-tmq/stbFilter.py
python3 ./test.py -f 7-tmq/tmqCheckData.py

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

@ -24,6 +24,8 @@
#include "taos.h"
#include "taoserror.h"
#include "tlog.h"
#include "taosdef.h"
#include "types.h"
#define GREEN "\033[1;32m"
#define NC "\033[0m"
@ -49,6 +51,7 @@ typedef struct {
// char autoCommit[8]; // true, false
// char autoOffsetRest[16]; // none, earliest, latest
TdFilePtr pConsumeRowsFile;
int32_t ifCheckData;
int64_t expectMsgCnt;
@ -129,7 +132,6 @@ void initLogFile() {
char tmpString[128];
sprintf(filename, "%s/../log/tmqlog_%s.txt", configDir, getCurrentTimeString(tmpString));
// sprintf(filename, "%s/../log/tmqlog.txt", configDir);
#ifdef WINDOWS
for (int i = 2; i < sizeof(filename); i++) {
if (filename[i] == ':') filename[i] = '-';
@ -296,35 +298,165 @@ int32_t saveConsumeContentToTbl(SThreadInfo* pInfo, char* buf) {
return 0;
}
static char *shellFormatTimestamp(char *buf, int64_t val, int32_t precision) {
//if (shell.args.is_raw_time) {
// sprintf(buf, "%" PRId64, val);
// return buf;
//}
time_t tt;
int32_t ms = 0;
if (precision == TSDB_TIME_PRECISION_NANO) {
tt = (time_t)(val / 1000000000);
ms = val % 1000000000;
} else if (precision == TSDB_TIME_PRECISION_MICRO) {
tt = (time_t)(val / 1000000);
ms = val % 1000000;
} else {
tt = (time_t)(val / 1000);
ms = val % 1000;
}
/*
comment out as it make testcases like select_with_tags.sim fail.
but in windows, this may cause the call to localtime crash if tt < 0,
need to find a better solution.
if (tt < 0) {
tt = 0;
}
*/
#ifdef WINDOWS
if (tt < 0) tt = 0;
#endif
if (tt <= 0 && ms < 0) {
tt--;
if (precision == TSDB_TIME_PRECISION_NANO) {
ms += 1000000000;
} else if (precision == TSDB_TIME_PRECISION_MICRO) {
ms += 1000000;
} else {
ms += 1000;
}
}
struct tm *ptm = taosLocalTime(&tt, NULL);
size_t pos = strftime(buf, 35, "%Y-%m-%d %H:%M:%S", ptm);
if (precision == TSDB_TIME_PRECISION_NANO) {
sprintf(buf + pos, ".%09d", ms);
} else if (precision == TSDB_TIME_PRECISION_MICRO) {
sprintf(buf + pos, ".%06d", ms);
} else {
sprintf(buf + pos, ".%03d", ms);
}
return buf;
}
static void shellDumpFieldToFile(TdFilePtr pFile, const char *val, TAOS_FIELD *field, int32_t length, int32_t precision) {
if (val == NULL) {
taosFprintfFile(pFile, "%s", TSDB_DATA_NULL_STR);
return;
}
int n;
char buf[TSDB_MAX_BYTES_PER_ROW];
switch (field->type) {
case TSDB_DATA_TYPE_BOOL:
taosFprintfFile(pFile, "%d", ((((int32_t)(*((char *)val))) == 1) ? 1 : 0));
break;
case TSDB_DATA_TYPE_TINYINT:
taosFprintfFile(pFile, "%d", *((int8_t *)val));
break;
case TSDB_DATA_TYPE_UTINYINT:
taosFprintfFile(pFile, "%u", *((uint8_t *)val));
break;
case TSDB_DATA_TYPE_SMALLINT:
taosFprintfFile(pFile, "%d", *((int16_t *)val));
break;
case TSDB_DATA_TYPE_USMALLINT:
taosFprintfFile(pFile, "%u", *((uint16_t *)val));
break;
case TSDB_DATA_TYPE_INT:
taosFprintfFile(pFile, "%d", *((int32_t *)val));
break;
case TSDB_DATA_TYPE_UINT:
taosFprintfFile(pFile, "%u", *((uint32_t *)val));
break;
case TSDB_DATA_TYPE_BIGINT:
taosFprintfFile(pFile, "%" PRId64, *((int64_t *)val));
break;
case TSDB_DATA_TYPE_UBIGINT:
taosFprintfFile(pFile, "%" PRIu64, *((uint64_t *)val));
break;
case TSDB_DATA_TYPE_FLOAT:
taosFprintfFile(pFile, "%.5f", GET_FLOAT_VAL(val));
break;
case TSDB_DATA_TYPE_DOUBLE:
n = snprintf(buf, TSDB_MAX_BYTES_PER_ROW, "%*.9f", length, GET_DOUBLE_VAL(val));
if (n > TMAX(25, length)) {
taosFprintfFile(pFile, "%*.15e", length, GET_DOUBLE_VAL(val));
} else {
taosFprintfFile(pFile, "%s", buf);
}
break;
case TSDB_DATA_TYPE_BINARY:
case TSDB_DATA_TYPE_NCHAR:
case TSDB_DATA_TYPE_JSON:
memcpy(buf, val, length);
buf[length] = 0;
taosFprintfFile(pFile, "\'%s\'", buf);
break;
case TSDB_DATA_TYPE_TIMESTAMP:
shellFormatTimestamp(buf, *(int64_t *)val, precision);
taosFprintfFile(pFile, "'%s'", buf);
break;
default:
break;
}
}
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++) {
if (i > 0) {
taosFprintfFile(pFile, "\n");
}
shellDumpFieldToFile(pFile, (const char *)row[i], fields + i, length[i], precision);
}
taosFprintfFile(pFile, "\n");
}
static int32_t msg_process(TAOS_RES* msg, SThreadInfo* pInfo, int32_t msgIndex) {
char buf[1024];
int32_t totalRows = 0;
// printf("topic: %s\n", tmq_get_topic_name(msg));
int32_t vgroupId = tmq_get_vgroup_id(msg);
const char* dbName = tmq_get_db_name(msg);
taosFprintfFile(g_fp, "msg index:%" PRId64 ", consumerId: %d\n", msgIndex, pInfo->consumerId);
// taosFprintfFile(g_fp, "topic: %s, vgroupId: %d, tableName: %s\n", tmq_get_topic_name(msg), vgroupId,
// tmq_get_table_name(msg));
taosFprintfFile(g_fp, "topic: %s, vgroupId: %d\n", tmq_get_topic_name(msg), vgroupId);
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);
while (1) {
TAOS_ROW row = taos_fetch_row(msg);
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* length = taos_fetch_lengths(msg);
int32_t precision = taos_result_precision(msg);
const char* tbName = tmq_get_table_name(msg);
dumpToFileForCheck(pInfo->pConsumeRowsFile, row, fields, length, numOfFields, precision);
taos_print_row(buf, row, fields, numOfFields);
const char* tbName = tmq_get_table_name(msg);
if (0 != g_stConfInfo.showRowFlag) {
taosFprintfFile(g_fp, "tbname:%s, rows[%d]: %s\n", (tbName != NULL ? tbName : "null table"), totalRows, buf);
if (0 != g_stConfInfo.saveRowFlag) {
saveConsumeContentToTbl(pInfo, buf);
}
//if (0 != g_stConfInfo.saveRowFlag) {
// saveConsumeContentToTbl(pInfo, buf);
//}
}
totalRows++;
@ -463,6 +595,18 @@ void loop_consume(SThreadInfo* pInfo) {
pInfo->consumerId);
pInfo->ts = taosGetTimestampMs();
if (pInfo->ifCheckData) {
char filename[256] = {0};
char tmpString[128];
//sprintf(filename, "%s/../log/consumerid_%d_%s.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);
if (pInfo->pConsumeRowsFile == NULL) {
taosFprintfFile(g_fp, "%s create file fail for save rows content\n", getCurrentTimeString(tmpString));
return;
}
}
while (running) {
TAOS_RES* tmqMsg = tmq_consumer_poll(pInfo->tmq, g_stConfInfo.consumeDelay * 1000);

62
tools/shell/src/shellCommand.c
View File

@ -21,6 +21,7 @@
#define UP 3
#define DOWN 4
#define PSIZE shell.info.promptSize
#define SHELL_INPUT_MAX_COMMAND_SIZE 10000
typedef struct {
char *buffer;
@ -227,6 +228,7 @@ void shellPrintChar(char c, int32_t times) {
}
void shellPositionCursor(int32_t step, int32_t direction) {
#ifndef WINDOWS
if (step > 0) {
if (direction == LEFT) {
fprintf(stdout, "\033[%dD", step);
@ -239,6 +241,7 @@ void shellPositionCursor(int32_t step, int32_t direction) {
}
fflush(stdout);
}
#endif
}
void shellUpdateBuffer(SShellCmd *cmd) {
@ -330,10 +333,14 @@ void shellClearScreen(int32_t ecmd_pos, int32_t cursor_pos) {
int32_t command_x = ecmd_pos / ws_col;
shellPositionCursor(cursor_y, LEFT);
shellPositionCursor(command_x - cursor_x, DOWN);
#ifndef WINDOWS
fprintf(stdout, "\033[2K");
#endif
for (int32_t i = 0; i < command_x; i++) {
shellPositionCursor(1, UP);
#ifndef WINDOWS
fprintf(stdout, "\033[2K");
#endif
}
fflush(stdout);
}
@ -394,6 +401,41 @@ void shellShowOnScreen(SShellCmd *cmd) {
fflush(stdout);
}
char taosGetConsoleChar() {
#ifdef WINDOWS
static void *console = NULL;
if (console == NULL) {
console = GetStdHandle(STD_INPUT_HANDLE);
}
static TdWchar buf[SHELL_INPUT_MAX_COMMAND_SIZE];
static char mbStr[5];
static unsigned long bufLen = 0;
static uint16_t bufIndex = 0, mbStrIndex = 0, mbStrLen = 0;
if (bufLen == 0) {
ReadConsoleW(console, buf, SHELL_INPUT_MAX_COMMAND_SIZE, &bufLen, NULL);
bufIndex = 0;
}
if (mbStrLen == 0){
if (buf[bufIndex] == '\r') {
bufIndex++;
}
mbStrLen = WideCharToMultiByte(CP_UTF8, 0, &buf[bufIndex], 1, mbStr, sizeof(mbStr), NULL, NULL);
mbStrIndex = 0;
bufIndex++;
}
mbStrIndex++;
if (mbStrIndex == mbStrLen) {
mbStrLen = 0;
if (bufIndex == bufLen) {
bufLen = 0;
}
}
return mbStr[mbStrIndex-1];
#else
return (char)getchar(); // getchar() return an 'int32_t' value
#endif
}
int32_t shellReadCommand(char *command) {
SShellHistory *pHistory = &shell.history;
SShellCmd cmd = {0};
@ -407,7 +449,7 @@ int32_t shellReadCommand(char *command) {
// Read input.
char c;
while (1) {
c = (char)getchar(); // getchar() return an 'int32_t' value
c = taosGetConsoleChar();
if (c == EOF) {
return c;
@ -417,7 +459,7 @@ int32_t shellReadCommand(char *command) {
int32_t count = shellCountPrefixOnes(c);
utf8_array[0] = c;
for (int32_t k = 1; k < count; k++) {
c = (char)getchar();
c = taosGetConsoleChar();
utf8_array[k] = c;
}
shellInsertChar(&cmd, utf8_array, count);
@ -472,10 +514,10 @@ int32_t shellReadCommand(char *command) {
break;
}
} else if (c == '\033') {
c = (char)getchar();
c = taosGetConsoleChar();
switch (c) {
case '[':
c = (char)getchar();
c = taosGetConsoleChar();
switch (c) {
case 'A': // Up arrow
if (hist_counter != pHistory->hstart) {
@ -502,35 +544,35 @@ int32_t shellReadCommand(char *command) {
shellMoveCursorLeft(&cmd);
break;
case '1':
if ((c = (char)getchar()) == '~') {
if ((c = taosGetConsoleChar()) == '~') {
// Home key
shellPositionCursorHome(&cmd);
}
break;
case '2':
if ((c = (char)getchar()) == '~') {
if ((c = taosGetConsoleChar()) == '~') {
// Insert key
}
break;
case '3':
if ((c = (char)getchar()) == '~') {
if ((c = taosGetConsoleChar()) == '~') {
// Delete key
shellDeleteChar(&cmd);
}
break;
case '4':
if ((c = (char)getchar()) == '~') {
if ((c = taosGetConsoleChar()) == '~') {
// End key
shellPositionCursorEnd(&cmd);
}
break;
case '5':
if ((c = (char)getchar()) == '~') {
if ((c = taosGetConsoleChar()) == '~') {
// Page up key
}
break;
case '6':
if ((c = (char)getchar()) == '~') {
if ((c = taosGetConsoleChar()) == '~') {
// Page down key
}
break;

4
tools/shell/src/shellEngine.c
View File

@ -393,15 +393,11 @@ void shellPrintNChar(const char *str, int32_t length, int32_t width) {
break;
}
int w = 0;
#ifdef WINDOWS
w = bytes;
#else
if(*(str + pos) == '\t' || *(str + pos) == '\n' || *(str + pos) == '\r'){
w = bytes;
}else{
w = taosWcharWidth(wc);
}
#endif
pos += bytes;
if (w <= 0) {