diff --git a/src/client/inc/tsclient.h b/src/client/inc/tsclient.h index 5d326931cb32067b0e6b8b2240abb993907d30c8..c91943e232864ccba69150325bb12c08eb5dee5d 100644 --- a/src/client/inc/tsclient.h +++ b/src/client/inc/tsclient.h @@ -328,6 +328,7 @@ typedef struct { char ** buffer; // Buffer used to put multibytes encoded using unicode (wchar_t) SColumnIndex* pColumnIndex; + TAOS_FIELD* final; SArithmeticSupport *pArithSup; // support the arithmetic expression calculation on agg functions struct SLocalMerger *pLocalMerger; } SSqlRes; diff --git a/src/client/src/tscParseInsert.c b/src/client/src/tscParseInsert.c index 926ee44b70c59fa791719f11795a7be1cd176217..920937928fae775305e0b4e695092bc6d458ff6d 100644 --- a/src/client/src/tscParseInsert.c +++ b/src/client/src/tscParseInsert.c @@ -937,6 +937,10 @@ static int32_t tscCheckIfCreateTable(char **sqlstr, SSqlObj *pSql, char** boundC return ret; } + if (sql == NULL) { + return TSDB_CODE_TSC_INVALID_SQL; + } + code = tscGetTableMetaEx(pSql, pTableMetaInfo, true); if (TSDB_CODE_TSC_ACTION_IN_PROGRESS == code) { return code; @@ -945,6 +949,10 @@ static int32_t tscCheckIfCreateTable(char **sqlstr, SSqlObj *pSql, char** boundC } else { sql = sToken.z; + if (sql == NULL) { + return TSDB_CODE_TSC_INVALID_SQL; + } + code = tscGetTableMetaEx(pSql, pTableMetaInfo, false); if (pCmd->curSql == NULL) { assert(code == TSDB_CODE_TSC_ACTION_IN_PROGRESS); @@ -952,10 +960,6 @@ static int32_t tscCheckIfCreateTable(char **sqlstr, SSqlObj *pSql, char** boundC } *sqlstr = sql; - - if (*sqlstr == NULL) { - code = TSDB_CODE_TSC_INVALID_SQL; - } return code; } diff --git a/src/client/src/tscSchemaUtil.c b/src/client/src/tscSchemaUtil.c index 67352ca71cab03bb9cd2f6b920b97abebbe1420a..2ea382132b8de9a97bd75d1e80b03a6eb6131756 100644 --- a/src/client/src/tscSchemaUtil.c +++ b/src/client/src/tscSchemaUtil.c @@ -144,8 +144,9 @@ SNewVgroupInfo createNewVgroupInfo(SVgroupMsg *pVgroupMsg) { SNewVgroupInfo info = {0}; info.numOfEps = pVgroupMsg->numOfEps; info.vgId = pVgroupMsg->vgId; - info.inUse = 0; + info.inUse = 0; // 0 is the default value of inUse in case of multiple replica + assert(info.numOfEps >= 1 && info.vgId >= 1); for(int32_t i = 0; i < pVgroupMsg->numOfEps; ++i) { tstrncpy(info.ep[i].fqdn, pVgroupMsg->epAddr[i].fqdn, TSDB_FQDN_LEN); info.ep[i].port = pVgroupMsg->epAddr[i].port; diff --git a/src/client/src/tscServer.c b/src/client/src/tscServer.c index f1c1d2da719d63973953382734489aeca1e58258..8889e25177fe23668c67bbba363e0efa06a017d2 100644 --- a/src/client/src/tscServer.c +++ b/src/client/src/tscServer.c @@ -34,6 +34,7 @@ int tscKeepConn[TSDB_SQL_MAX] = {0}; TSKEY tscGetSubscriptionProgress(void* sub, int64_t uid, TSKEY dflt); void tscUpdateSubscriptionProgress(void* sub, int64_t uid, TSKEY ts); void tscSaveSubscriptionProgress(void* sub); +static int32_t extractSTableQueryVgroupId(STableMetaInfo* pTableMetaInfo); static int32_t minMsgSize() { return tsRpcHeadSize + 100; } static int32_t getWaitingTimeInterval(int32_t count) { @@ -78,7 +79,8 @@ static void tscEpSetHtons(SRpcEpSet *s) { for (int32_t i = 0; i < s->numOfEps; i++) { s->port[i] = htons(s->port[i]); } -} +} + bool tscEpSetIsEqual(SRpcEpSet *s1, SRpcEpSet *s2) { if (s1->numOfEps != s2->numOfEps || s1->inUse != s2->inUse) { return false; @@ -111,19 +113,22 @@ static void tscDumpEpSetFromVgroupInfo(SRpcEpSet *pEpSet, SNewVgroupInfo *pVgrou } } -static void tscUpdateVgroupInfo(SSqlObj *pObj, SRpcEpSet *pEpSet) { - SSqlCmd *pCmd = &pObj->cmd; +static void tscUpdateVgroupInfo(SSqlObj *pSql, SRpcEpSet *pEpSet) { + SSqlCmd *pCmd = &pSql->cmd; STableMetaInfo *pTableMetaInfo = tscGetTableMetaInfoFromCmd(pCmd, pCmd->clauseIndex, 0); if (pTableMetaInfo == NULL || pTableMetaInfo->pTableMeta == NULL) { return; } - int32_t vgId = pTableMetaInfo->pTableMeta->vgId; + int32_t vgId = -1; if (pTableMetaInfo->pTableMeta->tableType == TSDB_SUPER_TABLE) { - assert(vgId == 0); - return; + vgId = extractSTableQueryVgroupId(pTableMetaInfo); + } else { + vgId = pTableMetaInfo->pTableMeta->vgId; } + assert(vgId > 0); + SNewVgroupInfo vgroupInfo = {.vgId = -1}; taosHashGetClone(tscVgroupMap, &vgId, sizeof(vgId), NULL, &vgroupInfo, sizeof(SNewVgroupInfo)); assert(vgroupInfo.numOfEps > 0 && vgroupInfo.vgId > 0); @@ -138,6 +143,33 @@ static void tscUpdateVgroupInfo(SSqlObj *pObj, SRpcEpSet *pEpSet) { tscDebug("after: EndPoint in use:%d, numOfEps:%d", vgroupInfo.inUse, vgroupInfo.numOfEps); taosHashPut(tscVgroupMap, &vgId, sizeof(vgId), &vgroupInfo, sizeof(SNewVgroupInfo)); + + // Update the local cached epSet info cached by SqlObj + int32_t inUse = pSql->epSet.inUse; + tscDumpEpSetFromVgroupInfo(&pSql->epSet, &vgroupInfo); + tscDebug("%p update the epSet in SqlObj, in use before:%d, after:%d", pSql, inUse, pSql->epSet.inUse); + +} + +int32_t extractSTableQueryVgroupId(STableMetaInfo* pTableMetaInfo) { + assert(pTableMetaInfo != NULL); + + int32_t vgIndex = pTableMetaInfo->vgroupIndex; + int32_t vgId = -1; + + if (pTableMetaInfo->pVgroupTables == NULL) { + SVgroupsInfo *pVgroupInfo = pTableMetaInfo->vgroupList; + assert(pVgroupInfo->vgroups[vgIndex].vgId > 0 && vgIndex < pTableMetaInfo->vgroupList->numOfVgroups); + vgId = pVgroupInfo->vgroups[vgIndex].vgId; + } else { + int32_t numOfVgroups = (int32_t)taosArrayGetSize(pTableMetaInfo->pVgroupTables); + assert(vgIndex >= 0 && vgIndex < numOfVgroups); + + SVgroupTableInfo *pTableIdList = taosArrayGet(pTableMetaInfo->pVgroupTables, vgIndex); + vgId = pTableIdList->vgInfo.vgId; + } + + return vgId; } void tscProcessHeartBeatRsp(void *param, TAOS_RES *tres, int code) { @@ -515,21 +547,22 @@ int tscBuildFetchMsg(SSqlObj *pSql, SSqlInfo *pInfo) { if (UTIL_TABLE_IS_SUPER_TABLE(pTableMetaInfo)) { int32_t vgIndex = pTableMetaInfo->vgroupIndex; + int32_t vgId = -1; + if (pTableMetaInfo->pVgroupTables == NULL) { SVgroupsInfo *pVgroupInfo = pTableMetaInfo->vgroupList; assert(pVgroupInfo->vgroups[vgIndex].vgId > 0 && vgIndex < pTableMetaInfo->vgroupList->numOfVgroups); - - pRetrieveMsg->header.vgId = htonl(pVgroupInfo->vgroups[vgIndex].vgId); - tscDebug("%p build fetch msg from vgId:%d, vgIndex:%d, qId:%" PRIu64, pSql, pVgroupInfo->vgroups[vgIndex].vgId, vgIndex, pSql->res.qId); + vgId = pVgroupInfo->vgroups[vgIndex].vgId; } else { int32_t numOfVgroups = (int32_t)taosArrayGetSize(pTableMetaInfo->pVgroupTables); assert(vgIndex >= 0 && vgIndex < numOfVgroups); SVgroupTableInfo* pTableIdList = taosArrayGet(pTableMetaInfo->pVgroupTables, vgIndex); - - pRetrieveMsg->header.vgId = htonl(pTableIdList->vgInfo.vgId); - tscDebug("%p build fetch msg from vgId:%d, vgIndex:%d, qId:%" PRIu64, pSql, pTableIdList->vgInfo.vgId, vgIndex, pSql->res.qId); + vgId = pTableIdList->vgInfo.vgId; } + + pRetrieveMsg->header.vgId = htonl(vgId); + tscDebug("%p build fetch msg from vgId:%d, vgIndex:%d, qId:%" PRIu64, pSql, vgId, vgIndex, pSql->res.qId); } else { STableMeta* pTableMeta = pTableMetaInfo->pTableMeta; pRetrieveMsg->header.vgId = htonl(pTableMeta->vgId); @@ -1980,7 +2013,7 @@ int tscProcessTableMetaRsp(SSqlObj *pSql) { (vgroupInfo.inUse < 0)) { // vgroup info exists, compare with it vgroupInfo = createNewVgroupInfo(&pMetaMsg->vgroup); taosHashPut(tscVgroupMap, &vgId, sizeof(vgId), &vgroupInfo, sizeof(vgroupInfo)); - tscDebug("add new VgroupInfo, vgId:%d, total:%d", vgId, (int32_t) taosHashGetSize(tscVgroupMap)); + tscDebug("add new VgroupInfo, vgId:%d, total cached:%d", vgId, (int32_t) taosHashGetSize(tscVgroupMap)); } } @@ -2132,18 +2165,33 @@ int tscProcessSTableVgroupRsp(SSqlObj *pSql) { tscError("%p empty vgroup info", pSql); } else { for (int32_t j = 0; j < pInfo->vgroupList->numOfVgroups; ++j) { - //just init, no need to lock - SVgroupInfo *pVgroups = &pInfo->vgroupList->vgroups[j]; + // just init, no need to lock + SVgroupInfo *pVgroup = &pInfo->vgroupList->vgroups[j]; SVgroupMsg *vmsg = &pVgroupMsg->vgroups[j]; - pVgroups->vgId = htonl(vmsg->vgId); - pVgroups->numOfEps = vmsg->numOfEps; + vmsg->vgId = htonl(vmsg->vgId); + vmsg->numOfEps = vmsg->numOfEps; + for (int32_t k = 0; k < vmsg->numOfEps; ++k) { + vmsg->epAddr[k].port = htons(vmsg->epAddr[k].port); + } + + SNewVgroupInfo newVi = createNewVgroupInfo(vmsg); + pVgroup->numOfEps = newVi.numOfEps; + pVgroup->vgId = newVi.vgId; + for (int32_t k = 0; k < vmsg->numOfEps; ++k) { + pVgroup->epAddr[k].port = newVi.ep[k].port; + pVgroup->epAddr[k].fqdn = strndup(newVi.ep[k].fqdn, TSDB_FQDN_LEN); + } - assert(pVgroups->numOfEps >= 1 && pVgroups->vgId >= 1); + // check if current buffer contains the vgroup info. + // If not, add it + SNewVgroupInfo existVgroupInfo = {.inUse = -1}; + taosHashGetClone(tscVgroupMap, &newVi.vgId, sizeof(newVi.vgId), NULL, &existVgroupInfo, sizeof(SNewVgroupInfo)); - for (int32_t k = 0; k < pVgroups->numOfEps; ++k) { - pVgroups->epAddr[k].port = htons(vmsg->epAddr[k].port); - pVgroups->epAddr[k].fqdn = strndup(vmsg->epAddr[k].fqdn, tListLen(vmsg->epAddr[k].fqdn)); + if (((existVgroupInfo.inUse >= 0) && !vgroupInfoIdentical(&existVgroupInfo, vmsg)) || + (existVgroupInfo.inUse < 0)) { // vgroup info exists, compare with it + taosHashPut(tscVgroupMap, &newVi.vgId, sizeof(newVi.vgId), &newVi, sizeof(newVi)); + tscDebug("add new VgroupInfo, vgId:%d, total cached:%d", newVi.vgId, (int32_t) taosHashGetSize(tscVgroupMap)); } } } diff --git a/src/client/src/tscSql.c b/src/client/src/tscSql.c index 93d0e9fd092b7e91fc8028e30f73a13bdb02627d..13c8f025eaf7b3fb0cdf5e0a7fbc6328e8bd12e6 100644 --- a/src/client/src/tscSql.c +++ b/src/client/src/tscSql.c @@ -405,6 +405,7 @@ int taos_affected_rows(TAOS_RES *tres) { TAOS_FIELD *taos_fetch_fields(TAOS_RES *res) { SSqlObj *pSql = (SSqlObj *)res; + SSqlRes *pRes = &pSql->res; if (pSql == NULL || pSql->signature != pSql) return 0; SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, 0); @@ -419,7 +420,7 @@ TAOS_FIELD *taos_fetch_fields(TAOS_RES *res) { SFieldInfo *pFieldInfo = &pQueryInfo->fieldsInfo; - if (pFieldInfo->final == NULL) { + if (pRes->final == NULL) { TAOS_FIELD* f = calloc(pFieldInfo->numOfOutput, sizeof(TAOS_FIELD)); int32_t j = 0; @@ -439,10 +440,10 @@ TAOS_FIELD *taos_fetch_fields(TAOS_RES *res) { } } - pFieldInfo->final = f; + pRes->final = f; } - return pFieldInfo->final; + return pRes->final; } static bool needToFetchNewBlock(SSqlObj* pSql) { diff --git a/src/client/src/tscUtil.c b/src/client/src/tscUtil.c index 3cb84ad72ac4d62c30d85561ed798b5b4e6fe3b3..4e8bfdf06481d5026bc935e3e1445a419694b3cd 100644 --- a/src/client/src/tscUtil.c +++ b/src/client/src/tscUtil.c @@ -429,6 +429,8 @@ static void tscDestroyResPointerInfo(SSqlRes* pRes) { tfree(pRes->pArithSup->data); tfree(pRes->pArithSup); } + + tfree(pRes->final); pRes->data = NULL; // pRes->data points to the buffer of pRsp, no need to free } @@ -1176,7 +1178,6 @@ void tscFieldInfoClear(SFieldInfo* pFieldInfo) { } taosArrayDestroy(pFieldInfo->internalField); - tfree(pFieldInfo->final); memset(pFieldInfo, 0, sizeof(SFieldInfo)); } diff --git a/src/common/inc/tglobal.h b/src/common/inc/tglobal.h index 3f96466cc00fc3250860e79bc3dd85e6fd0368d7..26475834d5aa91a87e6f8e14e86b211d5b68d6d2 100644 --- a/src/common/inc/tglobal.h +++ b/src/common/inc/tglobal.h @@ -39,6 +39,7 @@ extern int8_t tsEnableTelemetryReporting; extern char tsEmail[]; extern char tsArbitrator[]; extern int8_t tsArbOnline; +extern int32_t tsDnodeId; // common extern int tsRpcTimer; diff --git a/src/common/src/tglobal.c b/src/common/src/tglobal.c index 5f4ce046ed5359bab8c0f175ff0c08d6da939700..69b01e6c089c0af814ebfb49724458990d5320fc 100644 --- a/src/common/src/tglobal.c +++ b/src/common/src/tglobal.c @@ -43,6 +43,7 @@ int8_t tsEnableVnodeBak = 1; int8_t tsEnableTelemetryReporting = 1; int8_t tsArbOnline = 0; char tsEmail[TSDB_FQDN_LEN] = {0}; +int32_t tsDnodeId = 0; // common int32_t tsRpcTimer = 1000; @@ -212,7 +213,7 @@ float tsAvailTmpDirectorySpace = 0; float tsAvailDataDirGB = 0; float tsUsedDataDirGB = 0; float tsReservedTmpDirectorySpace = 1.0f; -float tsMinimalDataDirGB = 1.0f; +float tsMinimalDataDirGB = 2.0f; int32_t tsTotalMemoryMB = 0; uint32_t tsVersion = 0; diff --git a/src/cq/src/cqMain.c b/src/cq/src/cqMain.c index f620eb4dd55754146ddd164d372bb1233cf94aba..d4d202267c37365adc2647df3874185bc669b4a2 100644 --- a/src/cq/src/cqMain.c +++ b/src/cq/src/cqMain.c @@ -73,6 +73,7 @@ static void cqProcessStreamRes(void *param, TAOS_RES *tres, TAOS_ROW row); static void cqCreateStream(SCqContext *pContext, SCqObj *pObj); int32_t cqObjRef = -1; +int32_t cqVnodeNum = 0; void cqRmFromList(SCqObj *pObj) { //LOCK in caller @@ -166,6 +167,8 @@ void *cqOpen(void *ahandle, const SCqCfg *pCfg) { return NULL; } + atomic_add_fetch_32(&cqVnodeNum, 1); + cqCreateRef(); pContext->tmrCtrl = taosTmrInit(0, 0, 0, "CQ"); @@ -240,6 +243,13 @@ void cqClose(void *handle) { if (hasCq == 0) { freeSCqContext(pContext); } + + int32_t remainn = atomic_sub_fetch_32(&cqVnodeNum, 1); + if (remainn <= 0) { + int32_t ref = cqObjRef; + cqObjRef = -1; + taosCloseRef(ref); + } } void cqStart(void *handle) { diff --git a/src/dnode/src/dnodeCfg.c b/src/dnode/src/dnodeCfg.c index fd5956b37f28603893b43ca91f8cfd1177dccad2..c573d709f5ba250177aac41cefad74132d7797c8 100644 --- a/src/dnode/src/dnodeCfg.c +++ b/src/dnode/src/dnodeCfg.c @@ -17,6 +17,7 @@ #include "os.h" #include "cJSON.h" #include "dnodeCfg.h" +#include "tglobal.h" static SDnodeCfg tsCfg = {0}; static pthread_mutex_t tsCfgMutex; @@ -70,6 +71,7 @@ static void dnodeResetCfg(SDnodeCfg *cfg) { pthread_mutex_lock(&tsCfgMutex); tsCfg.dnodeId = cfg->dnodeId; + tsDnodeId = cfg->dnodeId; tstrncpy(tsCfg.clusterId, cfg->clusterId, TSDB_CLUSTER_ID_LEN); dnodePrintCfg(cfg); dnodeWriteCfg(); diff --git a/src/inc/query.h b/src/inc/query.h index 461e8723e7215bc8cbc8fff95f5442ab278f232f..38078cf21fcededd96dad833bbf1d22de55bb8ac 100644 --- a/src/inc/query.h +++ b/src/inc/query.h @@ -92,6 +92,7 @@ void** qRegisterQInfo(void* pMgmt, uint64_t qId, void *qInfo); void** qAcquireQInfo(void* pMgmt, uint64_t key); void** qReleaseQInfo(void* pMgmt, void* pQInfo, bool freeHandle); bool checkQIdEqual(void *qHandle, uint64_t qId); +int64_t genQueryId(void); #ifdef __cplusplus } diff --git a/src/inc/taoserror.h b/src/inc/taoserror.h index 8daccee1f255e5956bb8308bac5d8d05d1088642..eff4eecbc1baaea7f96a6560757bd2000eeab1bf 100644 --- a/src/inc/taoserror.h +++ b/src/inc/taoserror.h @@ -163,6 +163,7 @@ int32_t* taosGetErrno(); #define TSDB_CODE_MND_INVALID_TABLE_NAME TAOS_DEF_ERROR_CODE(0, 0x0362) //"Table does not exist") #define TSDB_CODE_MND_INVALID_TABLE_TYPE TAOS_DEF_ERROR_CODE(0, 0x0363) //"Invalid table type in tsdb") #define TSDB_CODE_MND_TOO_MANY_TAGS TAOS_DEF_ERROR_CODE(0, 0x0364) //"Too many tags") +#define TSDB_CODE_MND_TOO_MANY_COLUMNS TAOS_DEF_ERROR_CODE(0, 0x0365) //"Too many columns") #define TSDB_CODE_MND_TOO_MANY_TIMESERIES TAOS_DEF_ERROR_CODE(0, 0x0366) //"Too many time series") #define TSDB_CODE_MND_NOT_SUPER_TABLE TAOS_DEF_ERROR_CODE(0, 0x0367) //"Not super table") // operation only available for super table #define TSDB_CODE_MND_COL_NAME_TOO_LONG TAOS_DEF_ERROR_CODE(0, 0x0368) //"Tag name too long") diff --git a/src/kit/taosdemo/taosdemo.c b/src/kit/taosdemo/taosdemo.c index c6d5933dc770ee1b36e09a8435b14edf98930b39..c34b00fc8c159d83f3e574ba204e3d9b40569961 100644 --- a/src/kit/taosdemo/taosdemo.c +++ b/src/kit/taosdemo/taosdemo.c @@ -2949,9 +2949,6 @@ static int readSampleFromCsvFileToMem( continue; } - verbosePrint("readLen=%ld stb->lenOfOneRow=%d getRows=%d\n", (long)readLen, - superTblInfo->lenOfOneRow, getRows); - memcpy(superTblInfo->sampleDataBuf + getRows * superTblInfo->lenOfOneRow, line, readLen); getRows++; @@ -3527,6 +3524,7 @@ static bool getMetaFromInsertJsonFile(cJSON* root) { goto PARSE_OVER; } + /* cJSON* batchCreateTbl = cJSON_GetObjectItem(stbInfo, "batch_create_tbl_num"); if (batchCreateTbl && batchCreateTbl->type == cJSON_Number) { g_Dbs.db[i].superTbls[j].batchCreateTableNum = batchCreateTbl->valueint; @@ -3536,6 +3534,7 @@ static bool getMetaFromInsertJsonFile(cJSON* root) { printf("ERROR: failed to read json, batch_create_tbl_num not found\n"); goto PARSE_OVER; } + */ cJSON *childTblExists = cJSON_GetObjectItem(stbInfo, "child_table_exists"); // yes, no if (childTblExists @@ -4619,7 +4618,8 @@ static int generateDataTail(char *tableName, int32_t tableSeq, } static int generateSQLHead(char *tableName, int32_t tableSeq, - threadInfo* pThreadInfo, SSuperTable* superTblInfo, char *buffer) + threadInfo* pThreadInfo, SSuperTable* superTblInfo, + char *buffer, int remainderBufLen) { int len; if (superTblInfo) { @@ -4671,7 +4671,62 @@ static int generateSQLHead(char *tableName, int32_t tableSeq, return len; } -static int generateProgressiveDataBuffer(char *pTblName, +static int generateInterlaceDataBuffer( + char *tableName, int batchPerTbl, int i, int batchPerTblTimes, + int32_t tableSeq, + threadInfo *pThreadInfo, char *buffer, + int64_t insertRows, + int64_t startTime, + int *pRemainderBufLen) +{ + char *pstr = buffer; + SSuperTable* superTblInfo = pThreadInfo->superTblInfo; + + int headLen = generateSQLHead(tableName, tableSeq, pThreadInfo, + superTblInfo, pstr, *pRemainderBufLen); + + if (headLen <= 0) { + return 0; + } + // generate data buffer + verbosePrint("[%d] %s() LN%d i=%d buffer:\n%s\n", + pThreadInfo->threadID, __func__, __LINE__, i, buffer); + + pstr += headLen; + *pRemainderBufLen -= headLen; + + int dataLen = 0; + + verbosePrint("[%d] %s() LN%d i=%d batchPerTblTimes=%d batchPerTbl = %d\n", + pThreadInfo->threadID, __func__, __LINE__, + i, batchPerTblTimes, batchPerTbl); + + if (superTblInfo) { + if (0 == strncasecmp(superTblInfo->startTimestamp, "now", 3)) { + startTime = taosGetTimestamp(pThreadInfo->time_precision); + } + } else { + startTime = 1500000000000; + } + int k = generateDataTail( + tableName, tableSeq, pThreadInfo, superTblInfo, + batchPerTbl, pstr, *pRemainderBufLen, insertRows, 0, + startTime, + &(pThreadInfo->samplePos), &dataLen); + + if (k > 0) { + pstr += dataLen; + *pRemainderBufLen -= dataLen; + } else { + pstr -= headLen; + pstr[0] = '\0'; + } + + return k; +} + +static int generateProgressiveDataBuffer( + char *tableName, int32_t tableSeq, threadInfo *pThreadInfo, char *buffer, int64_t insertRows, @@ -4691,6 +4746,7 @@ static int generateProgressiveDataBuffer(char *pTblName, assert(buffer != NULL); + int k = 0; int maxSqlLen = superTblInfo?superTblInfo->maxSqlLen:g_args.max_sql_len; int remainderBufLen = maxSqlLen; @@ -4698,14 +4754,17 @@ static int generateProgressiveDataBuffer(char *pTblName, char *pstr = buffer; - int headLen = generateSQLHead(pTblName, tableSeq, pThreadInfo, superTblInfo, - buffer); + int headLen = generateSQLHead(tableName, tableSeq, pThreadInfo, superTblInfo, + buffer, remainderBufLen); + + if (headLen <= 0) { + return 0; + } pstr += headLen; remainderBufLen -= headLen; - int k; int dataLen; - k = generateDataTail(pTblName, tableSeq, pThreadInfo, superTblInfo, + k = generateDataTail(tableName, tableSeq, pThreadInfo, superTblInfo, g_args.num_of_RPR, pstr, remainderBufLen, insertRows, startFrom, startTime, pSamplePos, &dataLen); @@ -4811,50 +4870,23 @@ static void* syncWriteInterlace(threadInfo *pThreadInfo) { return NULL; } - int headLen; - if (i == 0) { - headLen = generateSQLHead(tableName, tableSeq, pThreadInfo, - superTblInfo, pstr); - } else { - headLen = snprintf(pstr, TSDB_TABLE_NAME_LEN, "%s.%s values", - pThreadInfo->db_name, - tableName); - } - - // generate data buffer - verbosePrint("[%d] %s() LN%d i=%d buffer:\n%s\n", - pThreadInfo->threadID, __func__, __LINE__, i, buffer); - - pstr += headLen; - remainderBufLen -= headLen; - - int dataLen = 0; - - verbosePrint("[%d] %s() LN%d i=%d batchPerTblTimes=%d batchPerTbl = %d\n", - pThreadInfo->threadID, __func__, __LINE__, - i, batchPerTblTimes, batchPerTbl); - - if (superTblInfo) { - if (0 == strncasecmp(superTblInfo->startTimestamp, "now", 3)) { - startTime = taosGetTimestamp(pThreadInfo->time_precision); - } - } else { - startTime = 1500000000000; - } - int generated = generateDataTail( - tableName, tableSeq, pThreadInfo, superTblInfo, - batchPerTbl, pstr, remainderBufLen, insertRows, 0, + int generated = generateInterlaceDataBuffer( + tableName, batchPerTbl, i, batchPerTblTimes, + tableSeq, + pThreadInfo, pstr, + insertRows, startTime, - &(pThreadInfo->samplePos), &dataLen); + &remainderBufLen); if (generated < 0) { debugPrint("[%d] %s() LN%d, generated data is %d\n", pThreadInfo->threadID, __func__, __LINE__, generated); goto free_and_statistics_interlace; + } else if (generated == 0) { + break; } - pstr += dataLen; - remainderBufLen -= dataLen; + tableSeq ++; recOfBatch += batchPerTbl; // startTime += batchPerTbl * superTblInfo->timeStampStep; pThreadInfo->totalInsertRows += batchPerTbl; @@ -4862,7 +4894,6 @@ static void* syncWriteInterlace(threadInfo *pThreadInfo) { pThreadInfo->threadID, __func__, __LINE__, batchPerTbl, recOfBatch); - tableSeq ++; if (insertMode == INTERLACE_INSERT_MODE) { if (tableSeq == pThreadInfo->start_table_from + pThreadInfo->ntables) { // turn to first table diff --git a/src/mnode/src/mnodeDnode.c b/src/mnode/src/mnodeDnode.c index 80473ba5ae2ad2ea82a6686510d21467b3fb6e90..85d9f94b884475cbe3cd2efb0c0d662e86f86f15 100644 --- a/src/mnode/src/mnodeDnode.c +++ b/src/mnode/src/mnodeDnode.c @@ -437,14 +437,14 @@ static int32_t mnodeCheckClusterCfgPara(const SClusterCfg *clusterCfg) { return TAOS_DN_OFF_TIME_ZONE_NOT_MATCH; } - if (0 != strncasecmp(clusterCfg->locale, tsLocale, strlen(tsLocale))) { - mError("\"locale\"[%s - %s] cfg parameters inconsistent", clusterCfg->locale, tsLocale); - return TAOS_DN_OFF_LOCALE_NOT_MATCH; - } - if (0 != strncasecmp(clusterCfg->charset, tsCharset, strlen(tsCharset))) { - mError("\"charset\"[%s - %s] cfg parameters inconsistent.", clusterCfg->charset, tsCharset); - return TAOS_DN_OFF_CHARSET_NOT_MATCH; - } + // if (0 != strncasecmp(clusterCfg->locale, tsLocale, strlen(tsLocale))) { + // mError("\"locale\"[%s - %s] cfg parameters inconsistent", clusterCfg->locale, tsLocale); + // return TAOS_DN_OFF_LOCALE_NOT_MATCH; + // } + // if (0 != strncasecmp(clusterCfg->charset, tsCharset, strlen(tsCharset))) { + // mError("\"charset\"[%s - %s] cfg parameters inconsistent.", clusterCfg->charset, tsCharset); + // return TAOS_DN_OFF_CHARSET_NOT_MATCH; + // } if (clusterCfg->enableBalance != tsEnableBalance) { mError("\"balance\"[%d - %d] cfg parameters inconsistent", clusterCfg->enableBalance, tsEnableBalance); diff --git a/src/mnode/src/mnodeTable.c b/src/mnode/src/mnodeTable.c index 39eca8819d8ace50b6fa82fadb1d4464247568d6..2a8e941fcbca384d019fa00bb6a16f68960508f0 100644 --- a/src/mnode/src/mnodeTable.c +++ b/src/mnode/src/mnodeTable.c @@ -1037,6 +1037,19 @@ static int32_t mnodeProcessCreateSuperTableMsg(SMnodeMsg *pMsg) { SCreateTableMsg* pCreate = (SCreateTableMsg*)((char*)pCreate1 + sizeof(SCMCreateTableMsg)); + int16_t numOfTags = htons(pCreate->numOfTags); + if (numOfTags > TSDB_MAX_TAGS) { + mError("msg:%p, app:%p table:%s, failed to create, too many tags", pMsg, pMsg->rpcMsg.ahandle, pCreate->tableName); + return TSDB_CODE_MND_TOO_MANY_TAGS; + } + + int16_t numOfColumns = htons(pCreate->numOfColumns); + int32_t numOfCols = numOfColumns + numOfTags; + if (numOfCols > TSDB_MAX_COLUMNS) { + mError("msg:%p, app:%p table:%s, failed to create, too many columns", pMsg, pMsg->rpcMsg.ahandle, pCreate->tableName); + return TSDB_CODE_MND_TOO_MANY_COLUMNS; + } + SSTableObj * pStable = calloc(1, sizeof(SSTableObj)); if (pStable == NULL) { mError("msg:%p, app:%p table:%s, failed to create, no enough memory", pMsg, pMsg->rpcMsg.ahandle, pCreate->tableName); @@ -1050,10 +1063,9 @@ static int32_t mnodeProcessCreateSuperTableMsg(SMnodeMsg *pMsg) { pStable->uid = (us << 24) + ((sdbGetVersion() & ((1ul << 16) - 1ul)) << 8) + (taosRand() & ((1ul << 8) - 1ul)); pStable->sversion = 0; pStable->tversion = 0; - pStable->numOfColumns = htons(pCreate->numOfColumns); - pStable->numOfTags = htons(pCreate->numOfTags); + pStable->numOfColumns = numOfColumns; + pStable->numOfTags = numOfTags; - int32_t numOfCols = pStable->numOfColumns + pStable->numOfTags; int32_t schemaSize = numOfCols * sizeof(SSchema); pStable->schema = (SSchema *)calloc(1, schemaSize); if (pStable->schema == NULL) { @@ -1064,11 +1076,6 @@ static int32_t mnodeProcessCreateSuperTableMsg(SMnodeMsg *pMsg) { memcpy(pStable->schema, pCreate->schema, numOfCols * sizeof(SSchema)); - if (pStable->numOfColumns > TSDB_MAX_COLUMNS || pStable->numOfTags > TSDB_MAX_TAGS) { - mError("msg:%p, app:%p table:%s, failed to create, too many columns", pMsg, pMsg->rpcMsg.ahandle, pCreate->tableName); - return TSDB_CODE_MND_INVALID_TABLE_NAME; - } - pStable->nextColId = 0; for (int32_t col = 0; col < numOfCols; col++) { @@ -1340,6 +1347,11 @@ static int32_t mnodeAddSuperTableColumn(SMnodeMsg *pMsg, SSchema schema[], int32 return TSDB_CODE_MND_APP_ERROR; } + if (pStable->numOfColumns + ncols + pStable->numOfTags > TSDB_MAX_COLUMNS) { + mError("msg:%p, app:%p stable:%s, add column, too many columns", pMsg, pMsg->rpcMsg.ahandle, pStable->info.tableId); + return TSDB_CODE_MND_TOO_MANY_COLUMNS; + } + for (int32_t i = 0; i < ncols; i++) { if (mnodeFindSuperTableColumnIndex(pStable, schema[i].name) > 0) { mError("msg:%p, app:%p stable:%s, add column, column:%s already exist", pMsg, pMsg->rpcMsg.ahandle, diff --git a/src/mnode/src/mnodeVgroup.c b/src/mnode/src/mnodeVgroup.c index 7eb3122d83032ed9759236bf999210008cdb908c..7222c8d1a0f711d12f3d3d575b1f4c909d123746 100644 --- a/src/mnode/src/mnodeVgroup.c +++ b/src/mnode/src/mnodeVgroup.c @@ -994,6 +994,7 @@ void mnodeSendSyncVgroupMsg(SVgObj *pVgroup) { mDebug("vgId:%d, send sync all vnodes msg, numOfVnodes:%d db:%s", pVgroup->vgId, pVgroup->numOfVnodes, pVgroup->dbName); for (int32_t i = 0; i < pVgroup->numOfVnodes; ++i) { + if (pVgroup->vnodeGid[i].role != TAOS_SYNC_ROLE_SLAVE) continue; SRpcEpSet epSet = mnodeGetEpSetFromIp(pVgroup->vnodeGid[i].pDnode->dnodeEp); mDebug("vgId:%d, index:%d, send sync vnode msg to dnode %s", pVgroup->vgId, i, pVgroup->vnodeGid[i].pDnode->dnodeEp); diff --git a/src/query/src/qExecutor.c b/src/query/src/qExecutor.c index d2c68f5ee816691074ead78fa664674807882e35..dc524d95a957f3de13a45315c819b9ae0fe9a4d6 100644 --- a/src/query/src/qExecutor.c +++ b/src/query/src/qExecutor.c @@ -105,6 +105,30 @@ int32_t getMaximumIdleDurationSec() { return tsShellActivityTimer * 2; } + +int64_t genQueryId(void) { + int64_t uid = 0; + int64_t did = tsDnodeId; + + uid = did << 54; + + int64_t pid = ((int64_t)taosGetPId()) & 0x3FF; + + uid |= pid << 44; + + int64_t ts = taosGetTimestampMs() & 0x1FFFFFFFF; + + uid |= ts << 11; + + int64_t sid = atomic_add_fetch_64(&queryHandleId, 1) & 0x7FF; + + uid |= sid; + + return uid; +} + + + static void getNextTimeWindow(SQuery* pQuery, STimeWindow* tw) { int32_t factor = GET_FORWARD_DIRECTION_FACTOR(pQuery->order.order); if (pQuery->interval.intervalUnit != 'n' && pQuery->interval.intervalUnit != 'y') { @@ -6445,6 +6469,8 @@ SQInfo* createQInfoImpl(SQueryTableMsg* pQueryMsg, SSqlGroupbyExpr* pGroupbyExpr goto _cleanup_qinfo; } + pQInfo->qId = *qId; + // to make sure third party won't overwrite this structure pQInfo->signature = pQInfo; SQuery* pQuery = &pQInfo->query; @@ -6581,8 +6607,6 @@ SQInfo* createQInfoImpl(SQueryTableMsg* pQueryMsg, SSqlGroupbyExpr* pGroupbyExpr // todo refactor pQInfo->query.queryBlockDist = (numOfOutput == 1 && pExprs[0].base.colInfo.colId == TSDB_BLOCK_DIST_COLUMN_INDEX); - pQInfo->qId = atomic_add_fetch_64(&queryHandleId, 1); - *qId = pQInfo->qId; qDebug("qmsg:%p QInfo:%" PRIu64 "-%p created", pQueryMsg, pQInfo->qId, pQInfo); return pQInfo; diff --git a/src/query/src/qFilterfunc.c b/src/query/src/qFilterfunc.c index 884f7e653f0ef05b3f7eb0318ac1e6342b9a539e..dabce88423f762029a4b27dc3fb4731fd28e6406 100644 --- a/src/query/src/qFilterfunc.c +++ b/src/query/src/qFilterfunc.c @@ -124,7 +124,7 @@ bool greaterEqualOperator(SColumnFilterElem *pFilter, const char *minval, const bool equalOperator(SColumnFilterElem *pFilter, const char *minval, const char *maxval, int16_t type) { SColumnFilterInfo *pFilterInfo = &pFilter->filterInfo; - if (IS_SIGNED_NUMERIC_TYPE(type) || type == TSDB_DATA_TYPE_BOOL) { + if (IS_SIGNED_NUMERIC_TYPE(type) || type == TSDB_DATA_TYPE_BOOL || type == TSDB_DATA_TYPE_TIMESTAMP) { int64_t minv = -1, maxv = -1; GET_TYPED_DATA(minv, int64_t, type, minval); GET_TYPED_DATA(maxv, int64_t, type, maxval); @@ -202,7 +202,7 @@ bool likeOperator(SColumnFilterElem *pFilter, const char *minval, const char *ma bool notEqualOperator(SColumnFilterElem *pFilter, const char *minval, const char *maxval, int16_t type) { SColumnFilterInfo *pFilterInfo = &pFilter->filterInfo; - if (IS_SIGNED_NUMERIC_TYPE(type) || type == TSDB_DATA_TYPE_BOOL) { + if (IS_SIGNED_NUMERIC_TYPE(type) || type == TSDB_DATA_TYPE_BOOL || type == TSDB_DATA_TYPE_TIMESTAMP) { int64_t minv = -1, maxv = -1; GET_TYPED_DATA(minv, int64_t, type, minval); GET_TYPED_DATA(maxv, int64_t, type, maxval); diff --git a/src/query/src/qResultbuf.c b/src/query/src/qResultbuf.c index c5dd6b3cac66cfa58f74e4ff0897761bd1704ce7..f83caf2d8f006a0b3932d09c217523f8d35cb826 100644 --- a/src/query/src/qResultbuf.c +++ b/src/query/src/qResultbuf.c @@ -287,6 +287,10 @@ static void lruListMoveToFront(SList *pList, SPageInfo* pi) { tdListPrependNode(pList, pi->pn); } +static FORCE_INLINE size_t getAllocPageSize(int32_t pageSize) { + return pageSize + POINTER_BYTES + 2 + sizeof(tFilePage); +} + tFilePage* getNewDataBuf(SDiskbasedResultBuf* pResultBuf, int32_t groupId, int32_t* pageId) { pResultBuf->statis.getPages += 1; @@ -311,7 +315,7 @@ tFilePage* getNewDataBuf(SDiskbasedResultBuf* pResultBuf, int32_t groupId, int32 // allocate buf if (availablePage == NULL) { - pi->pData = calloc(1, pResultBuf->pageSize + POINTER_BYTES + 2); // add extract bytes in case of zipped buffer increased. + pi->pData = calloc(1, getAllocPageSize(pResultBuf->pageSize)); // add extract bytes in case of zipped buffer increased. } else { pi->pData = availablePage; } @@ -355,7 +359,7 @@ tFilePage* getResBufPage(SDiskbasedResultBuf* pResultBuf, int32_t id) { } if (availablePage == NULL) { - (*pi)->pData = calloc(1, pResultBuf->pageSize + POINTER_BYTES); + (*pi)->pData = calloc(1, getAllocPageSize(pResultBuf->pageSize)); } else { (*pi)->pData = availablePage; } diff --git a/src/query/src/queryMain.c b/src/query/src/queryMain.c index 06a7ee210f15d54086fde18bedf5f9bb393d5d91..84cd7fe93b07c2d20b8d6f2aabd0097d983b402e 100644 --- a/src/query/src/queryMain.c +++ b/src/query/src/queryMain.c @@ -197,6 +197,7 @@ int32_t qCreateQueryInfo(void* tsdb, int32_t vgId, SQueryTableMsg* pQueryMsg, qi return code; } + bool qTableQuery(qinfo_t qinfo, uint64_t *qId) { SQInfo *pQInfo = (SQInfo *)qinfo; assert(pQInfo && pQInfo->signature == pQInfo); diff --git a/src/rpc/src/rpcMain.c b/src/rpc/src/rpcMain.c index 3e8c1d9180c1a18ff7b6a8a07f69f53e90fe8ef5..db3c72c2fc8a1a1c326c37974bfde0741f93004b 100644 --- a/src/rpc/src/rpcMain.c +++ b/src/rpc/src/rpcMain.c @@ -1367,8 +1367,7 @@ static void rpcProcessConnError(void *param, void *id) { tDebug("%s %p, connection error happens", pRpc->label, pContext->ahandle); - if (pContext->numOfTry >= pContext->epSet.numOfEps - || pContext->msgType == TSDB_MSG_TYPE_FETCH) { + if (pContext->numOfTry >= pContext->epSet.numOfEps || pContext->msgType == TSDB_MSG_TYPE_FETCH) { rpcMsg.msgType = pContext->msgType+1; rpcMsg.ahandle = pContext->ahandle; rpcMsg.code = pContext->code; diff --git a/src/sync/src/syncMain.c b/src/sync/src/syncMain.c index 2f1576940bc1effe250e3804e03bd4cc9e4e5b90..e5f2d94c4a9696500e0aef7b9593c6cc0daf2792 100644 --- a/src/sync/src/syncMain.c +++ b/src/sync/src/syncMain.c @@ -409,23 +409,22 @@ void syncConfirmForward(int64_t rid, uint64_t version, int32_t code, bool force) syncReleaseNode(pNode); } -#if 1 void syncRecover(int64_t rid) { SSyncPeer *pPeer; SSyncNode *pNode = syncAcquireNode(rid); if (pNode == NULL) return; - // to do: add a few lines to check if recover is OK - // if take this node to unsync state, the whole system may not work - nodeRole = TAOS_SYNC_ROLE_UNSYNCED; (*pNode->notifyRoleFp)(pNode->vgId, nodeRole); - nodeVersion = 0; pthread_mutex_lock(&pNode->mutex); + nodeVersion = 0; + for (int32_t i = 0; i < pNode->replica; ++i) { + if (i == pNode->selfIndex) continue; + pPeer = pNode->peerInfo[i]; if (pPeer->peerFd >= 0) { syncRestartConnection(pPeer); @@ -436,7 +435,6 @@ void syncRecover(int64_t rid) { syncReleaseNode(pNode); } -#endif int32_t syncGetNodesRole(int64_t rid, SNodesRole *pNodesRole) { SSyncNode *pNode = syncAcquireNode(rid); @@ -1000,17 +998,24 @@ static void syncProcessForwardFromPeer(char *cont, SSyncPeer *pPeer) { sTrace("%s, forward is received, hver:%" PRIu64 ", len:%d", pPeer->id, pHead->version, pHead->len); + int32_t code = 0; if (nodeRole == TAOS_SYNC_ROLE_SLAVE) { // nodeVersion = pHead->version; - (*pNode->writeToCacheFp)(pNode->vgId, pHead, TAOS_QTYPE_FWD, NULL); + code = (*pNode->writeToCacheFp)(pNode->vgId, pHead, TAOS_QTYPE_FWD, NULL); } else { if (nodeSStatus != TAOS_SYNC_STATUS_INIT) { - syncSaveIntoBuffer(pPeer, pHead); + code = syncSaveIntoBuffer(pPeer, pHead); } else { sError("%s, forward discarded since sstatus:%s, hver:%" PRIu64, pPeer->id, syncStatus[nodeSStatus], pHead->version); + code = -1; } } + + if (code != 0) { + sError("%s, failed to process fwd msg, hver:%" PRIu64 ", len:%d", pPeer->id, pHead->version, pHead->len); + syncRestartConnection(pPeer); + } } static void syncProcessPeersStatusMsg(SPeersStatus *pPeersStatus, SSyncPeer *pPeer) { diff --git a/src/tsdb/src/tsdbRead.c b/src/tsdb/src/tsdbRead.c index ea72760568293043b0cb415ee1beb6c0866ab5f9..cd97b2a9d6d5b8a29bd9670a5061c4f0af57d7d4 100644 --- a/src/tsdb/src/tsdbRead.c +++ b/src/tsdb/src/tsdbRead.c @@ -2861,12 +2861,6 @@ int32_t tsdbRetrieveDataBlockStatisInfo(TsdbQueryHandleT* pQueryHandle, SDataSta if (pHandle->statis[i].numOfNull == -1) { // set the column data are all NULL pHandle->statis[i].numOfNull = pBlockInfo->compBlock->numOfRows; } - - SColumnInfo* pColInfo = taosArrayGet(pHandle->pColumns, i); - if (pColInfo->type == TSDB_DATA_TYPE_TIMESTAMP) { - pHandle->statis[i].min = pBlockInfo->compBlock->keyFirst; - pHandle->statis[i].max = pBlockInfo->compBlock->keyLast; - } } int64_t elapsed = taosGetTimestampUs() - stime; diff --git a/src/util/src/terror.c b/src/util/src/terror.c index 918ccc493543c785313df67c7f2a8eced6ba8fb7..586a886f47ab4ea71d30689f88c2a386a24a0fde 100644 --- a/src/util/src/terror.c +++ b/src/util/src/terror.c @@ -175,6 +175,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_TABLE_ID, "Table name too long") TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_TABLE_NAME, "Table does not exist") TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_TABLE_TYPE, "Invalid table type in tsdb") TAOS_DEFINE_ERROR(TSDB_CODE_MND_TOO_MANY_TAGS, "Too many tags") +TAOS_DEFINE_ERROR(TSDB_CODE_MND_TOO_MANY_COLUMNS, "Too many columns") TAOS_DEFINE_ERROR(TSDB_CODE_MND_TOO_MANY_TIMESERIES, "Too many time series") TAOS_DEFINE_ERROR(TSDB_CODE_MND_NOT_SUPER_TABLE, "Not super table") // operation only available for super table TAOS_DEFINE_ERROR(TSDB_CODE_MND_COL_NAME_TOO_LONG, "Tag name too long") diff --git a/src/vnode/inc/vnodeInt.h b/src/vnode/inc/vnodeInt.h index 4aa07196a79513c476072b58d62787bd7a5a966d..d770a38e371c9920c438d810f699ab399be15833 100644 --- a/src/vnode/inc/vnodeInt.h +++ b/src/vnode/inc/vnodeInt.h @@ -37,6 +37,7 @@ extern int32_t vDebugFlag; typedef struct { int32_t vgId; // global vnode group ID int32_t refCount; // reference count + int64_t queuedWMsgSize; int32_t queuedWMsg; int32_t queuedRMsg; int32_t flowctrlLevel; diff --git a/src/vnode/src/vnodeMain.c b/src/vnode/src/vnodeMain.c index 69d94aff87388a96664b40b7db1d1c1e906ca8b2..ee8ed9e0fa3507363f04841060011de85422efec 100644 --- a/src/vnode/src/vnodeMain.c +++ b/src/vnode/src/vnodeMain.c @@ -99,8 +99,13 @@ int32_t vnodeSync(int32_t vgId) { return TSDB_CODE_VND_INVALID_VGROUP_ID; } - if (pVnode->role != TAOS_SYNC_ROLE_MASTER) { + if (pVnode->role == TAOS_SYNC_ROLE_SLAVE) { vInfo("vgId:%d, vnode will sync, refCount:%d pVnode:%p", pVnode->vgId, pVnode->refCount, pVnode); + + pVnode->version = 0; + pVnode->fversion = 0; + walResetVersion(pVnode->wal, pVnode->fversion); + syncRecover(pVnode->sync); } diff --git a/src/vnode/src/vnodeRead.c b/src/vnode/src/vnodeRead.c index a3dd8e1354e3ecf8a5372e9bdedb996b7bee8a30..2448fada502177a0bcb4cf3deb194cce68e7854e 100644 --- a/src/vnode/src/vnodeRead.c +++ b/src/vnode/src/vnodeRead.c @@ -208,6 +208,7 @@ static void vnodeBuildNoResultQueryRsp(SRspRet *pRet) { pRsp->completed = true; } + static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SVReadMsg *pRead) { void * pCont = pRead->pCont; int32_t contLen = pRead->contLen; @@ -226,7 +227,7 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SVReadMsg *pRead) { if (contLen != 0) { qinfo_t pQInfo = NULL; - uint64_t qId = 0; + uint64_t qId = genQueryId(); code = qCreateQueryInfo(pVnode->tsdb, pVnode->vgId, pQueryTableMsg, &pQInfo, &qId); SQueryTableRsp *pRsp = (SQueryTableRsp *)rpcMallocCont(sizeof(SQueryTableRsp)); diff --git a/src/vnode/src/vnodeWrite.c b/src/vnode/src/vnodeWrite.c index a0be52db7afead12094e6440c24314796c3f1ddd..aab685e67843fe63b9c3860254988a9ec5ec8d98 100644 --- a/src/vnode/src/vnodeWrite.c +++ b/src/vnode/src/vnodeWrite.c @@ -25,6 +25,7 @@ #include "vnodeStatus.h" #define MAX_QUEUED_MSG_NUM 100000 +#define MAX_QUEUED_MSG_SIZE 1024*1024*1024 //1GB extern void * tsDnodeTmr; static int32_t (*vnodeProcessWriteMsgFp[TSDB_MSG_TYPE_MAX])(SVnodeObj *, void *pCont, SRspRet *); @@ -269,6 +270,13 @@ static int32_t vnodeWriteToWQueueImp(SVWriteMsg *pWrite) { } } + if (tsAvailDataDirGB <= tsMinimalDataDirGB) { + vError("vgId:%d, failed to write into vwqueue since no diskspace, avail:%fGB", pVnode->vgId, tsAvailDataDirGB); + taosFreeQitem(pWrite); + vnodeRelease(pVnode); + return TSDB_CODE_VND_NO_DISKSPACE; + } + if (!vnodeInReadyOrUpdatingStatus(pVnode)) { vError("vgId:%d, failed to write into vwqueue, vstatus is %s, refCount:%d pVnode:%p", pVnode->vgId, vnodeStatus[pVnode->status], pVnode->refCount, pVnode); @@ -278,14 +286,17 @@ static int32_t vnodeWriteToWQueueImp(SVWriteMsg *pWrite) { } int32_t queued = atomic_add_fetch_32(&pVnode->queuedWMsg, 1); - if (queued > MAX_QUEUED_MSG_NUM) { + int64_t queuedSize = atomic_add_fetch_64(&pVnode->queuedWMsgSize, pWrite->pHead.len); + + if (queued > MAX_QUEUED_MSG_NUM || queuedSize > MAX_QUEUED_MSG_SIZE) { int32_t ms = (queued / MAX_QUEUED_MSG_NUM) * 10 + 3; if (ms > 100) ms = 100; vDebug("vgId:%d, too many msg:%d in vwqueue, flow control %dms", pVnode->vgId, queued, ms); taosMsleep(ms); } - vTrace("vgId:%d, write into vwqueue, refCount:%d queued:%d", pVnode->vgId, pVnode->refCount, pVnode->queuedWMsg); + vTrace("vgId:%d, write into vwqueue, refCount:%d queued:%d size:%" PRId64, pVnode->vgId, pVnode->refCount, + pVnode->queuedWMsg, pVnode->queuedWMsgSize); taosWriteQitem(pVnode->wqueue, pWrite->qtype, pWrite); return TSDB_CODE_SUCCESS; @@ -308,7 +319,10 @@ void vnodeFreeFromWQueue(void *vparam, SVWriteMsg *pWrite) { SVnodeObj *pVnode = vparam; int32_t queued = atomic_sub_fetch_32(&pVnode->queuedWMsg, 1); - vTrace("vgId:%d, msg:%p, app:%p, free from vwqueue, queued:%d", pVnode->vgId, pWrite, pWrite->rpcMsg.ahandle, queued); + int64_t queuedSize = atomic_sub_fetch_64(&pVnode->queuedWMsgSize, pWrite->pHead.len); + + vTrace("vgId:%d, msg:%p, app:%p, free from vwqueue, queued:%d size:%" PRId64, pVnode->vgId, pWrite, + pWrite->rpcMsg.ahandle, queued, queuedSize); taosFreeQitem(pWrite); vnodeRelease(pVnode); @@ -344,7 +358,9 @@ static void vnodeFlowCtrlMsgToWQueue(void *param, void *tmrId) { static int32_t vnodePerformFlowCtrl(SVWriteMsg *pWrite) { SVnodeObj *pVnode = pWrite->pVnode; if (pWrite->qtype != TAOS_QTYPE_RPC) return 0; - if (pVnode->queuedWMsg < MAX_QUEUED_MSG_NUM && pVnode->flowctrlLevel <= 0) return 0; + if (pVnode->queuedWMsg < MAX_QUEUED_MSG_NUM && pVnode->queuedWMsgSize < MAX_QUEUED_MSG_SIZE && + pVnode->flowctrlLevel <= 0) + return 0; if (tsEnableFlowCtrl == 0) { int32_t ms = (int32_t)pow(2, pVnode->flowctrlLevel + 2); diff --git a/tests/examples/c/asyncdemo.c b/tests/examples/c/asyncdemo.c index d711ce22c1e5a22b643bbbcb33468cd91713c861..f2a96dd8256782960f9ad114229cd47714c9d1d9 100644 --- a/tests/examples/c/asyncdemo.c +++ b/tests/examples/c/asyncdemo.c @@ -28,7 +28,8 @@ int points = 5; int numOfTables = 3; -int tablesProcessed = 0; +int tablesInsertProcessed = 0; +int tablesSelectProcessed = 0; int64_t st, et; typedef struct { @@ -134,6 +135,9 @@ int main(int argc, char *argv[]) gettimeofday(&systemTime, NULL); st = systemTime.tv_sec * 1000000 + systemTime.tv_usec; + tablesInsertProcessed = 0; + tablesSelectProcessed = 0; + for (i = 0; iname); - tablesProcessed++; - if (tablesProcessed >= numOfTables) { + tablesInsertProcessed++; + if (tablesInsertProcessed >= numOfTables) { gettimeofday(&systemTime, NULL); et = systemTime.tv_sec * 1000000 + systemTime.tv_usec; printf("%lld mseconds to insert %d data points\n", (et - st) / 1000, points*numOfTables); @@ -251,15 +263,17 @@ void taos_retrieve_call_back(void *param, TAOS_RES *tres, int numOfRows) //taos_free_result(tres); printf("%d rows data retrieved from %s\n", pTable->rowsRetrieved, pTable->name); - tablesProcessed++; - if (tablesProcessed >= numOfTables) { + tablesSelectProcessed++; + if (tablesSelectProcessed >= numOfTables) { gettimeofday(&systemTime, NULL); et = systemTime.tv_sec * 1000000 + systemTime.tv_usec; printf("%lld mseconds to query %d data rows\n", (et - st) / 1000, points * numOfTables); } + + taos_free_result(tres); } - taos_free_result(tres); + } void taos_select_call_back(void *param, TAOS_RES *tres, int code) @@ -276,6 +290,4 @@ void taos_select_call_back(void *param, TAOS_RES *tres, int code) taos_cleanup(); exit(1); } - - taos_free_result(tres); } diff --git a/tests/pytest/fulltest.sh b/tests/pytest/fulltest.sh index b64fb2be11c240ff17572eccd54bc453566c6f16..e7e05866369666eb40ccbc6186a7a60ea3595dbf 100755 --- a/tests/pytest/fulltest.sh +++ b/tests/pytest/fulltest.sh @@ -258,6 +258,8 @@ python3 test.py -f subscribe/singlemeter.py #python3 test.py -f subscribe/stability.py python3 test.py -f subscribe/supertable.py +# topic +python3 ./test.py -f topic/topicQuery.py #======================p3-end=============== #======================p4-start=============== diff --git a/tests/pytest/insert/basic.py b/tests/pytest/insert/basic.py index dcb5834d559968a964c4057c61d9b6a40cf8ab15..f23f38651aa57590a14824077e003812ea10d8bd 100644 --- a/tests/pytest/insert/basic.py +++ b/tests/pytest/insert/basic.py @@ -43,6 +43,9 @@ class TDTestCase: tdSql.query("select * from tb") tdSql.checkRows(insertRows + 4) + # test case for https://jira.taosdata.com:18080/browse/TD-3716: + tdSql.error("insert into tb(now, 1)") + def stop(self): tdSql.close() tdLog.success("%s successfully executed" % __file__) diff --git a/tests/pytest/topic/topicQuery.py b/tests/pytest/topic/topicQuery.py new file mode 100644 index 0000000000000000000000000000000000000000..1ee3c3a4274d67a76fdae442b2d26a4de29392a2 --- /dev/null +++ b/tests/pytest/topic/topicQuery.py @@ -0,0 +1,91 @@ +################################################################### +# Copyright (c) 2016 by TAOS Technologies, Inc. +# All rights reserved. +# +# This file is proprietary and confidential to TAOS Technologies. +# No part of this file may be reproduced, stored, transmitted, +# disclosed or used in any form or by any means other than as +# expressly provided by the written permission from Jianhui Tao +# +################################################################### + +# -*- coding: utf-8 -*- + +from util.log import tdLog +from util.cases import tdCases +from util.sql import tdSql + +class TDTestCase: + def init(self, conn, logSql): + tdLog.debug("start to execute %s" % __file__) + tdSql.init(conn.cursor(), logSql) + + self.ts = 1538548685000 + + def run(self): + tdSql.prepare() + + # test case for https://jira.taosdata.com:18080/browse/TD-3679 + print("==============step1") + tdSql.execute( + "create topic tq_test partitions 10") + tdSql.execute( + "insert into tq_test.p1(off, ts, content) values(0, %d, 'aaaa')" % self.ts) + tdSql.execute( + "insert into tq_test.p1(off, ts, content) values(1, %d, 'aaaa')" % (self.ts + 1)) + tdSql.execute( + "insert into tq_test.p1(off, ts, content) values(2, %d, 'aaaa')" % (self.ts + 2)) + tdSql.execute( + "insert into tq_test.p1(off, ts, content) values(3, %d, 'aaaa')" % (self.ts + 3)) + + print("==============step2") + + tdSql.query("select * from tq_test.p1") + tdSql.checkRows(4) + + tdSql.query("select * from tq_test.p1 where ts >= %d" % self.ts) + tdSql.checkRows(4) + + tdSql.query("select * from tq_test.p1 where ts > %d" % self.ts) + tdSql.checkRows(3) + + tdSql.query("select * from tq_test.p1 where ts = %d" % self.ts) + tdSql.checkRows(1) + + + tdSql.execute("use db") + tdSql.execute("create table test(ts timestamp, start timestamp, value int)") + tdSql.execute("insert into test values(%d, %d, 1)" % (self.ts, self.ts)) + tdSql.execute("insert into test values(%d, %d, 1)" % (self.ts + 1, self.ts + 1)) + tdSql.execute("insert into test values(%d, %d, 1)" % (self.ts + 2, self.ts + 2)) + tdSql.execute("insert into test values(%d, %d, 1)" % (self.ts + 3, self.ts + 3)) + + tdSql.query("select * from test") + tdSql.checkRows(4) + + tdSql.query("select * from test where ts >= %d" % self.ts) + tdSql.checkRows(4) + + tdSql.query("select * from test where ts > %d" % self.ts) + tdSql.checkRows(3) + + tdSql.query("select * from test where ts = %d" % self.ts) + tdSql.checkRows(1) + + tdSql.query("select * from test where start >= %d" % self.ts) + tdSql.checkRows(4) + + tdSql.query("select * from test where start > %d" % self.ts) + tdSql.checkRows(3) + + tdSql.query("select * from test where start = %d" % self.ts) + tdSql.checkRows(1) + + + def stop(self): + tdSql.close() + tdLog.success("%s successfully executed" % __file__) + + +tdCases.addWindows(__file__, TDTestCase()) +tdCases.addLinux(__file__, TDTestCase()) \ No newline at end of file