Merge remote-tracking branch 'origin/main' into fix/main_bugfix_wxy
This commit is contained in:
commit
b9c870b4ce
|
@ -41,6 +41,12 @@ typedef struct SBlockOrderInfo {
|
||||||
BMCharPos(bm_, r_) |= (1u << (7u - BitPos(r_))); \
|
BMCharPos(bm_, r_) |= (1u << (7u - BitPos(r_))); \
|
||||||
} while (0)
|
} while (0)
|
||||||
|
|
||||||
|
#define colDataSetNull_f_s(c_, r_) \
|
||||||
|
do { \
|
||||||
|
colDataSetNull_f((c_)->nullbitmap, r_); \
|
||||||
|
memset(((char*)(c_)->pData) + (c_)->info.bytes * (r_), 0, (c_)->info.bytes); \
|
||||||
|
} while (0)
|
||||||
|
|
||||||
#define colDataClearNull_f(bm_, r_) \
|
#define colDataClearNull_f(bm_, r_) \
|
||||||
do { \
|
do { \
|
||||||
BMCharPos(bm_, r_) &= ((char)(~(1u << (7u - BitPos(r_))))); \
|
BMCharPos(bm_, r_) &= ((char)(~(1u << (7u - BitPos(r_))))); \
|
||||||
|
@ -136,7 +142,7 @@ static FORCE_INLINE void colDataAppendNULL(SColumnInfoData* pColumnInfoData, uin
|
||||||
if (IS_VAR_DATA_TYPE(pColumnInfoData->info.type)) {
|
if (IS_VAR_DATA_TYPE(pColumnInfoData->info.type)) {
|
||||||
colDataSetNull_var(pColumnInfoData, currentRow); // it is a null value of VAR type.
|
colDataSetNull_var(pColumnInfoData, currentRow); // it is a null value of VAR type.
|
||||||
} else {
|
} else {
|
||||||
colDataSetNull_f(pColumnInfoData->nullbitmap, currentRow);
|
colDataSetNull_f_s(pColumnInfoData, currentRow);
|
||||||
}
|
}
|
||||||
|
|
||||||
pColumnInfoData->hasNull = true;
|
pColumnInfoData->hasNull = true;
|
||||||
|
@ -151,6 +157,7 @@ static FORCE_INLINE void colDataAppendNNULL(SColumnInfoData* pColumnInfoData, ui
|
||||||
for (int32_t i = start; i < start + nRows; ++i) {
|
for (int32_t i = start; i < start + nRows; ++i) {
|
||||||
colDataSetNull_f(pColumnInfoData->nullbitmap, i);
|
colDataSetNull_f(pColumnInfoData->nullbitmap, i);
|
||||||
}
|
}
|
||||||
|
memset(pColumnInfoData->pData + start * pColumnInfoData->info.bytes, 0, pColumnInfoData->info.bytes * nRows);
|
||||||
}
|
}
|
||||||
|
|
||||||
pColumnInfoData->hasNull = true;
|
pColumnInfoData->hasNull = true;
|
||||||
|
@ -231,7 +238,6 @@ int32_t blockDataSort_rv(SSDataBlock* pDataBlock, SArray* pOrderInfo, bool nullF
|
||||||
|
|
||||||
int32_t colInfoDataEnsureCapacity(SColumnInfoData* pColumn, uint32_t numOfRows, bool clearPayload);
|
int32_t colInfoDataEnsureCapacity(SColumnInfoData* pColumn, uint32_t numOfRows, bool clearPayload);
|
||||||
int32_t blockDataEnsureCapacity(SSDataBlock* pDataBlock, uint32_t numOfRows);
|
int32_t blockDataEnsureCapacity(SSDataBlock* pDataBlock, uint32_t numOfRows);
|
||||||
int32_t blockDataEnsureCapacityNoClear(SSDataBlock* pDataBlock, uint32_t numOfRows);
|
|
||||||
|
|
||||||
void colInfoDataCleanup(SColumnInfoData* pColumn, uint32_t numOfRows);
|
void colInfoDataCleanup(SColumnInfoData* pColumn, uint32_t numOfRows);
|
||||||
void blockDataCleanup(SSDataBlock* pDataBlock);
|
void blockDataCleanup(SSDataBlock* pDataBlock);
|
||||||
|
|
|
@ -69,7 +69,7 @@ int32_t colDataAppend(SColumnInfoData* pColumnInfoData, uint32_t currentRow, con
|
||||||
if (IS_VAR_DATA_TYPE(pColumnInfoData->info.type)) {
|
if (IS_VAR_DATA_TYPE(pColumnInfoData->info.type)) {
|
||||||
pColumnInfoData->varmeta.offset[currentRow] = -1; // it is a null value of VAR type.
|
pColumnInfoData->varmeta.offset[currentRow] = -1; // it is a null value of VAR type.
|
||||||
} else {
|
} else {
|
||||||
colDataSetNull_f(pColumnInfoData->nullbitmap, currentRow);
|
colDataSetNull_f_s(pColumnInfoData, currentRow);
|
||||||
}
|
}
|
||||||
|
|
||||||
pColumnInfoData->hasNull = true;
|
pColumnInfoData->hasNull = true;
|
||||||
|
@ -825,7 +825,7 @@ static int32_t blockDataAssign(SColumnInfoData* pCols, const SSDataBlock* pDataB
|
||||||
} else {
|
} else {
|
||||||
for (int32_t j = 0; j < pDataBlock->info.rows; ++j) {
|
for (int32_t j = 0; j < pDataBlock->info.rows; ++j) {
|
||||||
if (colDataIsNull_f(pSrc->nullbitmap, index[j])) {
|
if (colDataIsNull_f(pSrc->nullbitmap, index[j])) {
|
||||||
colDataSetNull_f(pDst->nullbitmap, j);
|
colDataSetNull_f_s(pDst, j);
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
memcpy(pDst->pData + j * pDst->info.bytes, pSrc->pData + index[j] * pDst->info.bytes, pDst->info.bytes);
|
memcpy(pDst->pData + j * pDst->info.bytes, pSrc->pData + index[j] * pDst->info.bytes, pDst->info.bytes);
|
||||||
|
@ -1161,15 +1161,16 @@ void blockDataEmpty(SSDataBlock* pDataBlock) {
|
||||||
pInfo->window.skey = 0;
|
pInfo->window.skey = 0;
|
||||||
}
|
}
|
||||||
|
|
||||||
// todo temporarily disable it
|
/*
|
||||||
|
* NOTE: the type of the input column may be TSDB_DATA_TYPE_NULL, which is used to denote
|
||||||
|
* the all NULL value in this column. It is an internal representation of all NULL value column, and no visible to
|
||||||
|
* any users. The length of TSDB_DATA_TYPE_NULL is 0, and it is an special case.
|
||||||
|
*/
|
||||||
static int32_t doEnsureCapacity(SColumnInfoData* pColumn, const SDataBlockInfo* pBlockInfo, uint32_t numOfRows, bool clearPayload) {
|
static int32_t doEnsureCapacity(SColumnInfoData* pColumn, const SDataBlockInfo* pBlockInfo, uint32_t numOfRows, bool clearPayload) {
|
||||||
if (numOfRows <= 0 || numOfRows <= pBlockInfo->capacity) {
|
if (numOfRows <= 0 || numOfRows <= pBlockInfo->capacity) {
|
||||||
return TSDB_CODE_SUCCESS;
|
return TSDB_CODE_SUCCESS;
|
||||||
}
|
}
|
||||||
|
|
||||||
// todo temp disable it
|
|
||||||
// ASSERT(pColumn->info.bytes != 0);
|
|
||||||
|
|
||||||
int32_t existedRows = pBlockInfo->rows;
|
int32_t existedRows = pBlockInfo->rows;
|
||||||
|
|
||||||
if (IS_VAR_DATA_TYPE(pColumn->info.type)) {
|
if (IS_VAR_DATA_TYPE(pColumn->info.type)) {
|
||||||
|
@ -1194,7 +1195,8 @@ static int32_t doEnsureCapacity(SColumnInfoData* pColumn, const SDataBlockInfo*
|
||||||
return TSDB_CODE_FAILED;
|
return TSDB_CODE_FAILED;
|
||||||
}
|
}
|
||||||
|
|
||||||
// make sure the allocated memory is MALLOC_ALIGN_BYTES aligned
|
// here we employ the aligned malloc function, to make sure that the address of allocated memory is aligned
|
||||||
|
// to MALLOC_ALIGN_BYTES
|
||||||
tmp = taosMemoryMallocAlign(MALLOC_ALIGN_BYTES, numOfRows * pColumn->info.bytes);
|
tmp = taosMemoryMallocAlign(MALLOC_ALIGN_BYTES, numOfRows * pColumn->info.bytes);
|
||||||
if (tmp == NULL) {
|
if (tmp == NULL) {
|
||||||
return TSDB_CODE_OUT_OF_MEMORY;
|
return TSDB_CODE_OUT_OF_MEMORY;
|
||||||
|
@ -1208,7 +1210,7 @@ static int32_t doEnsureCapacity(SColumnInfoData* pColumn, const SDataBlockInfo*
|
||||||
|
|
||||||
pColumn->pData = tmp;
|
pColumn->pData = tmp;
|
||||||
|
|
||||||
// todo remove it soon
|
// check if the allocated memory is aligned to the requried bytes.
|
||||||
#if defined LINUX
|
#if defined LINUX
|
||||||
if ((((uint64_t)pColumn->pData) & (MALLOC_ALIGN_BYTES - 1)) != 0x0) {
|
if ((((uint64_t)pColumn->pData) & (MALLOC_ALIGN_BYTES - 1)) != 0x0) {
|
||||||
return TSDB_CODE_FAILED;
|
return TSDB_CODE_FAILED;
|
||||||
|
@ -1249,25 +1251,6 @@ int32_t blockDataEnsureCapacity(SSDataBlock* pDataBlock, uint32_t numOfRows) {
|
||||||
return TSDB_CODE_SUCCESS;
|
return TSDB_CODE_SUCCESS;
|
||||||
}
|
}
|
||||||
|
|
||||||
size_t numOfCols = taosArrayGetSize(pDataBlock->pDataBlock);
|
|
||||||
for (int32_t i = 0; i < numOfCols; ++i) {
|
|
||||||
SColumnInfoData* p = taosArrayGet(pDataBlock->pDataBlock, i);
|
|
||||||
code = doEnsureCapacity(p, &pDataBlock->info, numOfRows, true);
|
|
||||||
if (code) {
|
|
||||||
return code;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
pDataBlock->info.capacity = numOfRows;
|
|
||||||
return TSDB_CODE_SUCCESS;
|
|
||||||
}
|
|
||||||
|
|
||||||
int32_t blockDataEnsureCapacityNoClear(SSDataBlock* pDataBlock, uint32_t numOfRows) {
|
|
||||||
int32_t code = 0;
|
|
||||||
if (numOfRows == 0 || numOfRows <= pDataBlock->info.capacity) {
|
|
||||||
return TSDB_CODE_SUCCESS;
|
|
||||||
}
|
|
||||||
|
|
||||||
size_t numOfCols = taosArrayGetSize(pDataBlock->pDataBlock);
|
size_t numOfCols = taosArrayGetSize(pDataBlock->pDataBlock);
|
||||||
for (int32_t i = 0; i < numOfCols; ++i) {
|
for (int32_t i = 0; i < numOfCols; ++i) {
|
||||||
SColumnInfoData* p = taosArrayGet(pDataBlock->pDataBlock, i);
|
SColumnInfoData* p = taosArrayGet(pDataBlock->pDataBlock, i);
|
||||||
|
|
|
@ -14,8 +14,8 @@
|
||||||
*/
|
*/
|
||||||
|
|
||||||
#define _DEFAULT_SOURCE
|
#define _DEFAULT_SOURCE
|
||||||
#include "vmInt.h"
|
|
||||||
#include "tjson.h"
|
#include "tjson.h"
|
||||||
|
#include "vmInt.h"
|
||||||
|
|
||||||
#define MAX_CONTENT_LEN 2 * 1024 * 1024
|
#define MAX_CONTENT_LEN 2 * 1024 * 1024
|
||||||
|
|
||||||
|
@ -46,102 +46,108 @@ SVnodeObj **vmGetVnodeListFromHash(SVnodeMgmt *pMgmt, int32_t *numOfVnodes) {
|
||||||
return pVnodes;
|
return pVnodes;
|
||||||
}
|
}
|
||||||
|
|
||||||
int32_t vmGetVnodeListFromFile(SVnodeMgmt *pMgmt, SWrapperCfg **ppCfgs, int32_t *numOfVnodes) {
|
static int32_t vmDecodeVnodeList(SJson *pJson, SVnodeMgmt *pMgmt, SWrapperCfg **ppCfgs, int32_t *numOfVnodes) {
|
||||||
int32_t code = TSDB_CODE_INVALID_JSON_FORMAT;
|
int32_t code = -1;
|
||||||
int32_t len = 0;
|
|
||||||
int32_t maxLen = MAX_CONTENT_LEN;
|
|
||||||
char *content = taosMemoryCalloc(1, maxLen + 1);
|
|
||||||
cJSON *root = NULL;
|
|
||||||
FILE *fp = NULL;
|
|
||||||
char file[PATH_MAX] = {0};
|
|
||||||
SWrapperCfg *pCfgs = NULL;
|
SWrapperCfg *pCfgs = NULL;
|
||||||
TdFilePtr pFile = NULL;
|
*ppCfgs = NULL;
|
||||||
|
|
||||||
snprintf(file, sizeof(file), "%s%svnodes.json", pMgmt->path, TD_DIRSEP);
|
SJson *vnodes = tjsonGetObjectItem(pJson, "vnodes");
|
||||||
|
if (vnodes == NULL) return -1;
|
||||||
pFile = taosOpenFile(file, TD_FILE_READ);
|
|
||||||
if (pFile == NULL) {
|
|
||||||
dInfo("file %s not exist", file);
|
|
||||||
code = 0;
|
|
||||||
goto _OVER;
|
|
||||||
}
|
|
||||||
|
|
||||||
if (content == NULL) {
|
|
||||||
terrno = TSDB_CODE_OUT_OF_MEMORY;
|
|
||||||
return -1;
|
|
||||||
}
|
|
||||||
|
|
||||||
len = (int32_t)taosReadFile(pFile, content, maxLen);
|
|
||||||
if (len <= 0) {
|
|
||||||
dError("failed to read %s since content is null", file);
|
|
||||||
goto _OVER;
|
|
||||||
}
|
|
||||||
|
|
||||||
content[len] = 0;
|
|
||||||
root = cJSON_Parse(content);
|
|
||||||
if (root == NULL) {
|
|
||||||
dError("failed to read %s since invalid json format", file);
|
|
||||||
goto _OVER;
|
|
||||||
}
|
|
||||||
|
|
||||||
cJSON *vnodes = cJSON_GetObjectItem(root, "vnodes");
|
|
||||||
if (!vnodes || vnodes->type != cJSON_Array) {
|
|
||||||
dError("failed to read %s since vnodes not found", file);
|
|
||||||
goto _OVER;
|
|
||||||
}
|
|
||||||
|
|
||||||
int32_t vnodesNum = cJSON_GetArraySize(vnodes);
|
int32_t vnodesNum = cJSON_GetArraySize(vnodes);
|
||||||
if (vnodesNum > 0) {
|
if (vnodesNum > 0) {
|
||||||
pCfgs = taosMemoryCalloc(vnodesNum, sizeof(SWrapperCfg));
|
pCfgs = taosMemoryCalloc(vnodesNum, sizeof(SWrapperCfg));
|
||||||
if (pCfgs == NULL) {
|
if (pCfgs == NULL) return -1;
|
||||||
dError("failed to read %s since out of memory", file);
|
}
|
||||||
code = TSDB_CODE_OUT_OF_MEMORY;
|
|
||||||
goto _OVER;
|
for (int32_t i = 0; i < vnodesNum; ++i) {
|
||||||
}
|
SJson *vnode = tjsonGetArrayItem(vnodes, i);
|
||||||
|
if (vnode == NULL) goto _OVER;
|
||||||
for (int32_t i = 0; i < vnodesNum; ++i) {
|
|
||||||
cJSON *vnode = cJSON_GetArrayItem(vnodes, i);
|
SWrapperCfg *pCfg = &pCfgs[i];
|
||||||
SWrapperCfg *pCfg = &pCfgs[i];
|
tjsonGetInt32ValueFromDouble(vnode, "vgId", pCfg->vgId, code);
|
||||||
|
if (code < 0) goto _OVER;
|
||||||
cJSON *vgId = cJSON_GetObjectItem(vnode, "vgId");
|
tjsonGetInt32ValueFromDouble(vnode, "dropped", pCfg->dropped, code);
|
||||||
if (!vgId || vgId->type != cJSON_Number) {
|
if (code < 0) goto _OVER;
|
||||||
dError("failed to read %s since vgId not found", file);
|
tjsonGetInt32ValueFromDouble(vnode, "vgVersion", pCfg->vgVersion, code);
|
||||||
taosMemoryFree(pCfgs);
|
if (code < 0) goto _OVER;
|
||||||
goto _OVER;
|
|
||||||
}
|
snprintf(pCfg->path, sizeof(pCfg->path), "%s%svnode%d", pMgmt->path, TD_DIRSEP, pCfg->vgId);
|
||||||
pCfg->vgId = vgId->valueint;
|
|
||||||
snprintf(pCfg->path, sizeof(pCfg->path), "%s%svnode%d", pMgmt->path, TD_DIRSEP, pCfg->vgId);
|
|
||||||
|
|
||||||
cJSON *dropped = cJSON_GetObjectItem(vnode, "dropped");
|
|
||||||
if (!dropped || dropped->type != cJSON_Number) {
|
|
||||||
dError("failed to read %s since dropped not found", file);
|
|
||||||
taosMemoryFree(pCfgs);
|
|
||||||
goto _OVER;
|
|
||||||
}
|
|
||||||
pCfg->dropped = dropped->valueint;
|
|
||||||
|
|
||||||
cJSON *vgVersion = cJSON_GetObjectItem(vnode, "vgVersion");
|
|
||||||
if (!vgVersion || vgVersion->type != cJSON_Number) {
|
|
||||||
dError("failed to read %s since vgVersion not found", file);
|
|
||||||
taosMemoryFree(pCfgs);
|
|
||||||
goto _OVER;
|
|
||||||
}
|
|
||||||
pCfg->vgVersion = vgVersion->valueint;
|
|
||||||
}
|
|
||||||
|
|
||||||
*ppCfgs = pCfgs;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
*numOfVnodes = vnodesNum;
|
|
||||||
code = 0;
|
code = 0;
|
||||||
dInfo("succcessed to read file %s, numOfVnodes:%d", file, vnodesNum);
|
*ppCfgs = pCfgs;
|
||||||
|
*numOfVnodes = vnodesNum;
|
||||||
|
|
||||||
_OVER:
|
_OVER:
|
||||||
if (content != NULL) taosMemoryFree(content);
|
if (*ppCfgs == NULL) taosMemoryFree(pCfgs);
|
||||||
if (root != NULL) cJSON_Delete(root);
|
return code;
|
||||||
|
}
|
||||||
|
|
||||||
|
int32_t vmGetVnodeListFromFile(SVnodeMgmt *pMgmt, SWrapperCfg **ppCfgs, int32_t *numOfVnodes) {
|
||||||
|
int32_t code = -1;
|
||||||
|
TdFilePtr pFile = NULL;
|
||||||
|
char *pData = NULL;
|
||||||
|
SJson *pJson = NULL;
|
||||||
|
char file[PATH_MAX] = {0};
|
||||||
|
SWrapperCfg *pCfgs = NULL;
|
||||||
|
snprintf(file, sizeof(file), "%s%svnodes.json", pMgmt->path, TD_DIRSEP);
|
||||||
|
|
||||||
|
if (taosStatFile(file, NULL, NULL) < 0) {
|
||||||
|
dInfo("vnode file:%s not exist", file);
|
||||||
|
return 0;
|
||||||
|
}
|
||||||
|
|
||||||
|
pFile = taosOpenFile(file, TD_FILE_READ);
|
||||||
|
if (pFile == NULL) {
|
||||||
|
terrno = TAOS_SYSTEM_ERROR(errno);
|
||||||
|
dError("failed to open vnode file:%s since %s", file, terrstr());
|
||||||
|
goto _OVER;
|
||||||
|
}
|
||||||
|
|
||||||
|
int64_t size = 0;
|
||||||
|
if (taosFStatFile(pFile, &size, NULL) < 0) {
|
||||||
|
terrno = TAOS_SYSTEM_ERROR(errno);
|
||||||
|
dError("failed to fstat mnode file:%s since %s", file, terrstr());
|
||||||
|
goto _OVER;
|
||||||
|
}
|
||||||
|
|
||||||
|
pData = taosMemoryMalloc(size + 1);
|
||||||
|
if (pData == NULL) {
|
||||||
|
terrno = TSDB_CODE_OUT_OF_MEMORY;
|
||||||
|
goto _OVER;
|
||||||
|
}
|
||||||
|
|
||||||
|
if (taosReadFile(pFile, pData, size) != size) {
|
||||||
|
terrno = TAOS_SYSTEM_ERROR(errno);
|
||||||
|
dError("failed to read vnode file:%s since %s", file, terrstr());
|
||||||
|
goto _OVER;
|
||||||
|
}
|
||||||
|
|
||||||
|
pData[size] = '\0';
|
||||||
|
|
||||||
|
pJson = tjsonParse(pData);
|
||||||
|
if (pJson == NULL) {
|
||||||
|
terrno = TSDB_CODE_INVALID_JSON_FORMAT;
|
||||||
|
goto _OVER;
|
||||||
|
}
|
||||||
|
|
||||||
|
if (vmDecodeVnodeList(pJson, pMgmt, ppCfgs, numOfVnodes) < 0) {
|
||||||
|
terrno = TSDB_CODE_INVALID_JSON_FORMAT;
|
||||||
|
goto _OVER;
|
||||||
|
}
|
||||||
|
|
||||||
|
code = 0;
|
||||||
|
dInfo("succceed to read vnode file %s", file);
|
||||||
|
|
||||||
|
_OVER:
|
||||||
|
if (pData != NULL) taosMemoryFree(pData);
|
||||||
|
if (pJson != NULL) cJSON_Delete(pJson);
|
||||||
if (pFile != NULL) taosCloseFile(&pFile);
|
if (pFile != NULL) taosCloseFile(&pFile);
|
||||||
|
|
||||||
terrno = code;
|
if (code != 0) {
|
||||||
|
dError("failed to read vnode file:%s since %s", file, terrstr());
|
||||||
|
}
|
||||||
return code;
|
return code;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -41,14 +41,49 @@ static void dmGetDnodeEp(SDnodeData *pData, int32_t dnodeId, char *pEp, char *pF
|
||||||
taosThreadRwlockUnlock(&pData->lock);
|
taosThreadRwlockUnlock(&pData->lock);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
static int32_t dmDecodeEps(SJson *pJson, SDnodeData *pData) {
|
||||||
|
int32_t code = 0;
|
||||||
|
|
||||||
|
tjsonGetInt32ValueFromDouble(pJson, "dnodeId", pData->dnodeId, code);
|
||||||
|
if (code < 0) return -1;
|
||||||
|
tjsonGetNumberValue(pJson, "dnodeVer", pData->dnodeVer, code);
|
||||||
|
if (code < 0) return -1;
|
||||||
|
tjsonGetNumberValue(pJson, "clusterId", pData->clusterId, code);
|
||||||
|
if (code < 0) return -1;
|
||||||
|
tjsonGetInt32ValueFromDouble(pJson, "dropped", pData->dropped, code);
|
||||||
|
if (code < 0) return -1;
|
||||||
|
|
||||||
|
SJson *dnodes = tjsonGetObjectItem(pJson, "dnodes");
|
||||||
|
if (dnodes == NULL) return 0;
|
||||||
|
int32_t numOfDnodes = tjsonGetArraySize(dnodes);
|
||||||
|
|
||||||
|
for (int32_t i = 0; i < numOfDnodes; ++i) {
|
||||||
|
SJson *dnode = tjsonGetArrayItem(dnodes, i);
|
||||||
|
if (dnode == NULL) return -1;
|
||||||
|
|
||||||
|
SDnodeEp dnodeEp = {0};
|
||||||
|
tjsonGetInt32ValueFromDouble(dnode, "id", dnodeEp.id, code);
|
||||||
|
if (code < 0) return -1;
|
||||||
|
code = tjsonGetStringValue(dnode, "fqdn", dnodeEp.ep.fqdn);
|
||||||
|
if (code < 0) return -1;
|
||||||
|
tjsonGetUInt16ValueFromDouble(dnode, "port", dnodeEp.ep.port, code);
|
||||||
|
if (code < 0) return -1;
|
||||||
|
tjsonGetInt8ValueFromDouble(dnode, "isMnode", dnodeEp.isMnode, code);
|
||||||
|
if (code < 0) return -1;
|
||||||
|
|
||||||
|
if (taosArrayPush(pData->dnodeEps, &dnodeEp) == NULL) return -1;
|
||||||
|
}
|
||||||
|
|
||||||
|
return 0;
|
||||||
|
}
|
||||||
|
|
||||||
int32_t dmReadEps(SDnodeData *pData) {
|
int32_t dmReadEps(SDnodeData *pData) {
|
||||||
int32_t code = TSDB_CODE_INVALID_JSON_FORMAT;
|
int32_t code = -1;
|
||||||
int32_t len = 0;
|
|
||||||
int32_t maxLen = 256 * 1024;
|
|
||||||
char *content = taosMemoryCalloc(1, maxLen + 1);
|
|
||||||
cJSON *root = NULL;
|
|
||||||
char file[PATH_MAX] = {0};
|
|
||||||
TdFilePtr pFile = NULL;
|
TdFilePtr pFile = NULL;
|
||||||
|
char *content = NULL;
|
||||||
|
SJson *pJson = NULL;
|
||||||
|
char file[PATH_MAX] = {0};
|
||||||
|
snprintf(file, sizeof(file), "%s%sdnode%sdnode.json", tsDataDir, TD_DIRSEP, TD_DIRSEP);
|
||||||
|
|
||||||
pData->dnodeEps = taosArrayInit(1, sizeof(SDnodeEp));
|
pData->dnodeEps = taosArrayInit(1, sizeof(SDnodeEp));
|
||||||
if (pData->dnodeEps == NULL) {
|
if (pData->dnodeEps == NULL) {
|
||||||
|
@ -56,113 +91,63 @@ int32_t dmReadEps(SDnodeData *pData) {
|
||||||
goto _OVER;
|
goto _OVER;
|
||||||
}
|
}
|
||||||
|
|
||||||
snprintf(file, sizeof(file), "%s%sdnode%sdnode.json", tsDataDir, TD_DIRSEP, TD_DIRSEP);
|
if (taosStatFile(file, NULL, NULL) < 0) {
|
||||||
pFile = taosOpenFile(file, TD_FILE_READ);
|
dInfo("dnode file:%s not exist", file);
|
||||||
if (pFile == NULL) {
|
|
||||||
code = 0;
|
code = 0;
|
||||||
goto _OVER;
|
goto _OVER;
|
||||||
}
|
}
|
||||||
|
|
||||||
len = (int32_t)taosReadFile(pFile, content, maxLen);
|
pFile = taosOpenFile(file, TD_FILE_READ);
|
||||||
if (len <= 0) {
|
if (pFile == NULL) {
|
||||||
dError("failed to read %s since content is null", file);
|
terrno = TAOS_SYSTEM_ERROR(errno);
|
||||||
|
dError("failed to open dnode file:%s since %s", file, terrstr());
|
||||||
goto _OVER;
|
goto _OVER;
|
||||||
}
|
}
|
||||||
|
|
||||||
content[len] = 0;
|
int64_t size = 0;
|
||||||
root = cJSON_Parse(content);
|
if (taosFStatFile(pFile, &size, NULL) < 0) {
|
||||||
if (root == NULL) {
|
terrno = TAOS_SYSTEM_ERROR(errno);
|
||||||
dError("failed to read %s since invalid json format", file);
|
dError("failed to fstat dnode file:%s since %s", file, terrstr());
|
||||||
goto _OVER;
|
goto _OVER;
|
||||||
}
|
}
|
||||||
|
|
||||||
cJSON *dnodeId = cJSON_GetObjectItem(root, "dnodeId");
|
content = taosMemoryMalloc(size + 1);
|
||||||
if (!dnodeId || dnodeId->type != cJSON_Number) {
|
if (content == NULL) {
|
||||||
dError("failed to read %s since dnodeId not found", file);
|
terrno = TSDB_CODE_OUT_OF_MEMORY;
|
||||||
goto _OVER;
|
|
||||||
}
|
|
||||||
pData->dnodeId = dnodeId->valueint;
|
|
||||||
|
|
||||||
cJSON *dnodeVer = cJSON_GetObjectItem(root, "dnodeVer");
|
|
||||||
if (!dnodeVer || dnodeVer->type != cJSON_String) {
|
|
||||||
dError("failed to read %s since dnodeVer not found", file);
|
|
||||||
goto _OVER;
|
|
||||||
}
|
|
||||||
pData->dnodeVer = atoll(dnodeVer->valuestring);
|
|
||||||
|
|
||||||
cJSON *clusterId = cJSON_GetObjectItem(root, "clusterId");
|
|
||||||
if (!clusterId || clusterId->type != cJSON_String) {
|
|
||||||
dError("failed to read %s since clusterId not found", file);
|
|
||||||
goto _OVER;
|
|
||||||
}
|
|
||||||
pData->clusterId = atoll(clusterId->valuestring);
|
|
||||||
|
|
||||||
cJSON *dropped = cJSON_GetObjectItem(root, "dropped");
|
|
||||||
if (!dropped || dropped->type != cJSON_Number) {
|
|
||||||
dError("failed to read %s since dropped not found", file);
|
|
||||||
goto _OVER;
|
|
||||||
}
|
|
||||||
pData->dropped = dropped->valueint;
|
|
||||||
|
|
||||||
cJSON *dnodes = cJSON_GetObjectItem(root, "dnodes");
|
|
||||||
if (!dnodes || dnodes->type != cJSON_Array) {
|
|
||||||
dError("failed to read %s since dnodes not found", file);
|
|
||||||
goto _OVER;
|
goto _OVER;
|
||||||
}
|
}
|
||||||
|
|
||||||
int32_t numOfDnodes = cJSON_GetArraySize(dnodes);
|
if (taosReadFile(pFile, content, size) != size) {
|
||||||
if (numOfDnodes <= 0) {
|
terrno = TAOS_SYSTEM_ERROR(errno);
|
||||||
dError("failed to read %s since numOfDnodes:%d invalid", file, numOfDnodes);
|
dError("failed to read dnode file:%s since %s", file, terrstr());
|
||||||
goto _OVER;
|
goto _OVER;
|
||||||
}
|
}
|
||||||
|
|
||||||
for (int32_t i = 0; i < numOfDnodes; ++i) {
|
content[size] = '\0';
|
||||||
cJSON *node = cJSON_GetArrayItem(dnodes, i);
|
|
||||||
if (node == NULL) break;
|
|
||||||
|
|
||||||
SDnodeEp dnodeEp = {0};
|
pJson = tjsonParse(content);
|
||||||
|
if (pJson == NULL) {
|
||||||
|
terrno = TSDB_CODE_INVALID_JSON_FORMAT;
|
||||||
|
goto _OVER;
|
||||||
|
}
|
||||||
|
|
||||||
cJSON *did = cJSON_GetObjectItem(node, "id");
|
if (dmDecodeEps(pJson, pData) < 0) {
|
||||||
if (!did || did->type != cJSON_Number) {
|
terrno = TSDB_CODE_INVALID_JSON_FORMAT;
|
||||||
dError("failed to read %s since dnodeId not found", file);
|
goto _OVER;
|
||||||
goto _OVER;
|
|
||||||
}
|
|
||||||
|
|
||||||
dnodeEp.id = did->valueint;
|
|
||||||
|
|
||||||
cJSON *dnodeFqdn = cJSON_GetObjectItem(node, "fqdn");
|
|
||||||
if (!dnodeFqdn || dnodeFqdn->type != cJSON_String || dnodeFqdn->valuestring == NULL) {
|
|
||||||
dError("failed to read %s since dnodeFqdn not found", file);
|
|
||||||
goto _OVER;
|
|
||||||
}
|
|
||||||
tstrncpy(dnodeEp.ep.fqdn, dnodeFqdn->valuestring, TSDB_FQDN_LEN);
|
|
||||||
|
|
||||||
cJSON *dnodePort = cJSON_GetObjectItem(node, "port");
|
|
||||||
if (!dnodePort || dnodePort->type != cJSON_Number) {
|
|
||||||
dError("failed to read %s since dnodePort not found", file);
|
|
||||||
goto _OVER;
|
|
||||||
}
|
|
||||||
|
|
||||||
dnodeEp.ep.port = dnodePort->valueint;
|
|
||||||
|
|
||||||
cJSON *isMnode = cJSON_GetObjectItem(node, "isMnode");
|
|
||||||
if (!isMnode || isMnode->type != cJSON_Number) {
|
|
||||||
dError("failed to read %s since isMnode not found", file);
|
|
||||||
goto _OVER;
|
|
||||||
}
|
|
||||||
dnodeEp.isMnode = isMnode->valueint;
|
|
||||||
|
|
||||||
taosArrayPush(pData->dnodeEps, &dnodeEp);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
code = 0;
|
code = 0;
|
||||||
dDebug("succcessed to read file %s", file);
|
dInfo("succceed to read mnode file %s", file);
|
||||||
|
|
||||||
_OVER:
|
_OVER:
|
||||||
if (content != NULL) taosMemoryFree(content);
|
if (content != NULL) taosMemoryFree(content);
|
||||||
if (root != NULL) cJSON_Delete(root);
|
if (pJson != NULL) cJSON_Delete(pJson);
|
||||||
if (pFile != NULL) taosCloseFile(&pFile);
|
if (pFile != NULL) taosCloseFile(&pFile);
|
||||||
|
|
||||||
|
if (code != 0) {
|
||||||
|
dError("failed to read dnode file:%s since %s", file, terrstr());
|
||||||
|
}
|
||||||
|
|
||||||
if (taosArrayGetSize(pData->dnodeEps) == 0) {
|
if (taosArrayGetSize(pData->dnodeEps) == 0) {
|
||||||
SDnodeEp dnodeEp = {0};
|
SDnodeEp dnodeEp = {0};
|
||||||
dnodeEp.isMnode = 1;
|
dnodeEp.isMnode = 1;
|
||||||
|
@ -178,7 +163,6 @@ _OVER:
|
||||||
return -1;
|
return -1;
|
||||||
}
|
}
|
||||||
|
|
||||||
terrno = code;
|
|
||||||
return code;
|
return code;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -247,7 +231,7 @@ _OVER:
|
||||||
|
|
||||||
if (code != 0) {
|
if (code != 0) {
|
||||||
if (terrno == 0) terrno = TAOS_SYSTEM_ERROR(errno);
|
if (terrno == 0) terrno = TAOS_SYSTEM_ERROR(errno);
|
||||||
dInfo("succeed to write dnode file:%s since %s, dnodeVer:%" PRId64, realfile, terrstr(), pData->dnodeVer);
|
dError("failed to write dnode file:%s since %s, dnodeVer:%" PRId64, realfile, terrstr(), pData->dnodeVer);
|
||||||
}
|
}
|
||||||
return code;
|
return code;
|
||||||
}
|
}
|
||||||
|
|
|
@ -19,48 +19,81 @@
|
||||||
|
|
||||||
#define MAXLEN 1024
|
#define MAXLEN 1024
|
||||||
|
|
||||||
int32_t dmReadFile(const char *path, const char *name, bool *pDeployed) {
|
static int32_t dmDecodeFile(SJson *pJson, bool *deployed) {
|
||||||
int32_t code = TSDB_CODE_INVALID_JSON_FORMAT;
|
int32_t code = 0;
|
||||||
int64_t len = 0;
|
int32_t value = 0;
|
||||||
char content[MAXLEN + 1] = {0};
|
|
||||||
cJSON *root = NULL;
|
|
||||||
char file[PATH_MAX] = {0};
|
|
||||||
TdFilePtr pFile = NULL;
|
|
||||||
|
|
||||||
|
tjsonGetInt32ValueFromDouble(pJson, "deployed", value, code);
|
||||||
|
if (code < 0) return -1;
|
||||||
|
|
||||||
|
*deployed = (value != 0);
|
||||||
|
return code;
|
||||||
|
}
|
||||||
|
|
||||||
|
int32_t dmReadFile(const char *path, const char *name, bool *pDeployed) {
|
||||||
|
int32_t code = -1;
|
||||||
|
TdFilePtr pFile = NULL;
|
||||||
|
char *content = NULL;
|
||||||
|
SJson *pJson = NULL;
|
||||||
|
char file[PATH_MAX] = {0};
|
||||||
snprintf(file, sizeof(file), "%s%s%s.json", path, TD_DIRSEP, name);
|
snprintf(file, sizeof(file), "%s%s%s.json", path, TD_DIRSEP, name);
|
||||||
pFile = taosOpenFile(file, TD_FILE_READ);
|
|
||||||
if (pFile == NULL) {
|
if (taosStatFile(file, NULL, NULL) < 0) {
|
||||||
|
dInfo("file:%s not exist", file);
|
||||||
code = 0;
|
code = 0;
|
||||||
goto _OVER;
|
goto _OVER;
|
||||||
}
|
}
|
||||||
|
|
||||||
len = taosReadFile(pFile, content, MAXLEN);
|
pFile = taosOpenFile(file, TD_FILE_READ);
|
||||||
if (len <= 0) {
|
if (pFile == NULL) {
|
||||||
dError("failed to read %s since content is null", file);
|
terrno = TAOS_SYSTEM_ERROR(errno);
|
||||||
|
dError("failed to open file:%s since %s", file, terrstr());
|
||||||
goto _OVER;
|
goto _OVER;
|
||||||
}
|
}
|
||||||
|
|
||||||
root = cJSON_Parse(content);
|
int64_t size = 0;
|
||||||
if (root == NULL) {
|
if (taosFStatFile(pFile, &size, NULL) < 0) {
|
||||||
dError("failed to read %s since invalid json format", file);
|
terrno = TAOS_SYSTEM_ERROR(errno);
|
||||||
|
dError("failed to fstat file:%s since %s", file, terrstr());
|
||||||
goto _OVER;
|
goto _OVER;
|
||||||
}
|
}
|
||||||
|
|
||||||
cJSON *deployed = cJSON_GetObjectItem(root, "deployed");
|
content = taosMemoryMalloc(size + 1);
|
||||||
if (!deployed || deployed->type != cJSON_Number) {
|
if (content == NULL) {
|
||||||
dError("failed to read %s since deployed not found", file);
|
terrno = TSDB_CODE_OUT_OF_MEMORY;
|
||||||
|
goto _OVER;
|
||||||
|
}
|
||||||
|
|
||||||
|
if (taosReadFile(pFile, content, size) != size) {
|
||||||
|
terrno = TAOS_SYSTEM_ERROR(errno);
|
||||||
|
dError("failed to read file:%s since %s", file, terrstr());
|
||||||
|
goto _OVER;
|
||||||
|
}
|
||||||
|
|
||||||
|
content[size] = '\0';
|
||||||
|
|
||||||
|
pJson = tjsonParse(content);
|
||||||
|
if (pJson == NULL) {
|
||||||
|
terrno = TSDB_CODE_INVALID_JSON_FORMAT;
|
||||||
|
goto _OVER;
|
||||||
|
}
|
||||||
|
|
||||||
|
if (dmDecodeFile(pJson, pDeployed) < 0) {
|
||||||
|
terrno = TSDB_CODE_INVALID_JSON_FORMAT;
|
||||||
goto _OVER;
|
goto _OVER;
|
||||||
}
|
}
|
||||||
*pDeployed = deployed->valueint != 0;
|
|
||||||
|
|
||||||
dDebug("succcessed to read file %s, deployed:%d", file, *pDeployed);
|
|
||||||
code = 0;
|
code = 0;
|
||||||
|
dInfo("succceed to read mnode file %s", file);
|
||||||
|
|
||||||
_OVER:
|
_OVER:
|
||||||
if (root != NULL) cJSON_Delete(root);
|
if (content != NULL) taosMemoryFree(content);
|
||||||
|
if (pJson != NULL) cJSON_Delete(pJson);
|
||||||
if (pFile != NULL) taosCloseFile(&pFile);
|
if (pFile != NULL) taosCloseFile(&pFile);
|
||||||
|
|
||||||
terrno = code;
|
if (code != 0) {
|
||||||
|
dError("failed to read dnode file:%s since %s", file, terrstr());
|
||||||
|
}
|
||||||
return code;
|
return code;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -279,7 +279,6 @@ SSDataBlock* doProjectOperation(SOperatorInfo* pOperator) {
|
||||||
// for stream interval
|
// for stream interval
|
||||||
if (pBlock->info.type == STREAM_RETRIEVE || pBlock->info.type == STREAM_DELETE_RESULT ||
|
if (pBlock->info.type == STREAM_RETRIEVE || pBlock->info.type == STREAM_DELETE_RESULT ||
|
||||||
pBlock->info.type == STREAM_DELETE_DATA) {
|
pBlock->info.type == STREAM_DELETE_DATA) {
|
||||||
// printDataBlock1(pBlock, "project1");
|
|
||||||
return pBlock;
|
return pBlock;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -1918,6 +1918,13 @@ static SSDataBlock* doBlockInfoScan(SOperatorInfo* pOperator) {
|
||||||
colDataAppend(pColInfo, 0, p, false);
|
colDataAppend(pColInfo, 0, p, false);
|
||||||
taosMemoryFree(p);
|
taosMemoryFree(p);
|
||||||
|
|
||||||
|
// make the valgrind happy that all memory buffer has been initialized already.
|
||||||
|
if (slotId != 0) {
|
||||||
|
SColumnInfoData* p1 = taosArrayGet(pBlock->pDataBlock, 0);
|
||||||
|
int64_t v = 0;
|
||||||
|
colDataAppendInt64(p1, 0, &v);
|
||||||
|
}
|
||||||
|
|
||||||
pBlock->info.rows = 1;
|
pBlock->info.rows = 1;
|
||||||
pOperator->status = OP_EXEC_DONE;
|
pOperator->status = OP_EXEC_DONE;
|
||||||
return pBlock;
|
return pBlock;
|
||||||
|
|
|
@ -1662,7 +1662,9 @@ int32_t percentileFunction(SqlFunctionCtx* pCtx) {
|
||||||
|
|
||||||
int32_t percentileFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
|
int32_t percentileFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
|
||||||
SVariant* pVal = &pCtx->param[1].param;
|
SVariant* pVal = &pCtx->param[1].param;
|
||||||
|
int32_t code = 0;
|
||||||
double v = 0;
|
double v = 0;
|
||||||
|
|
||||||
GET_TYPED_DATA(v, double, pVal->nType, &pVal->i);
|
GET_TYPED_DATA(v, double, pVal->nType, &pVal->i);
|
||||||
|
|
||||||
SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
|
SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
|
||||||
|
@ -1670,14 +1672,14 @@ int32_t percentileFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
|
||||||
|
|
||||||
tMemBucket* pMemBucket = ppInfo->pMemBucket;
|
tMemBucket* pMemBucket = ppInfo->pMemBucket;
|
||||||
if (pMemBucket != NULL && pMemBucket->total > 0) { // check for null
|
if (pMemBucket != NULL && pMemBucket->total > 0) { // check for null
|
||||||
int32_t code = getPercentile(pMemBucket, v, &ppInfo->result);
|
code = getPercentile(pMemBucket, v, &ppInfo->result);
|
||||||
if (code != TSDB_CODE_SUCCESS) {
|
|
||||||
tMemBucketDestroy(pMemBucket);
|
|
||||||
return code;
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
tMemBucketDestroy(pMemBucket);
|
tMemBucketDestroy(pMemBucket);
|
||||||
|
if (code != TSDB_CODE_SUCCESS) {
|
||||||
|
return code;
|
||||||
|
}
|
||||||
|
|
||||||
return functionFinalize(pCtx, pBlock);
|
return functionFinalize(pCtx, pBlock);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -2644,7 +2646,7 @@ static int32_t doHandleDiff(SDiffInfo* pDiffInfo, int32_t type, const char* pv,
|
||||||
int32_t v = *(int32_t*)pv;
|
int32_t v = *(int32_t*)pv;
|
||||||
int64_t delta = factor * (v - pDiffInfo->prev.i64); // direct previous may be null
|
int64_t delta = factor * (v - pDiffInfo->prev.i64); // direct previous may be null
|
||||||
if (delta < 0 && pDiffInfo->ignoreNegative) {
|
if (delta < 0 && pDiffInfo->ignoreNegative) {
|
||||||
colDataSetNull_f(pOutput->nullbitmap, pos);
|
colDataSetNull_f_s(pOutput, pos);
|
||||||
} else {
|
} else {
|
||||||
colDataAppendInt64(pOutput, pos, &delta);
|
colDataAppendInt64(pOutput, pos, &delta);
|
||||||
}
|
}
|
||||||
|
@ -2657,7 +2659,7 @@ static int32_t doHandleDiff(SDiffInfo* pDiffInfo, int32_t type, const char* pv,
|
||||||
int8_t v = *(int8_t*)pv;
|
int8_t v = *(int8_t*)pv;
|
||||||
int64_t delta = factor * (v - pDiffInfo->prev.i64); // direct previous may be null
|
int64_t delta = factor * (v - pDiffInfo->prev.i64); // direct previous may be null
|
||||||
if (delta < 0 && pDiffInfo->ignoreNegative) {
|
if (delta < 0 && pDiffInfo->ignoreNegative) {
|
||||||
colDataSetNull_f(pOutput->nullbitmap, pos);
|
colDataSetNull_f_s(pOutput, pos);
|
||||||
} else {
|
} else {
|
||||||
colDataAppendInt64(pOutput, pos, &delta);
|
colDataAppendInt64(pOutput, pos, &delta);
|
||||||
}
|
}
|
||||||
|
@ -2668,7 +2670,7 @@ static int32_t doHandleDiff(SDiffInfo* pDiffInfo, int32_t type, const char* pv,
|
||||||
int16_t v = *(int16_t*)pv;
|
int16_t v = *(int16_t*)pv;
|
||||||
int64_t delta = factor * (v - pDiffInfo->prev.i64); // direct previous may be null
|
int64_t delta = factor * (v - pDiffInfo->prev.i64); // direct previous may be null
|
||||||
if (delta < 0 && pDiffInfo->ignoreNegative) {
|
if (delta < 0 && pDiffInfo->ignoreNegative) {
|
||||||
colDataSetNull_f(pOutput->nullbitmap, pos);
|
colDataSetNull_f_s(pOutput, pos);
|
||||||
} else {
|
} else {
|
||||||
colDataAppendInt64(pOutput, pos, &delta);
|
colDataAppendInt64(pOutput, pos, &delta);
|
||||||
}
|
}
|
||||||
|
@ -2680,7 +2682,7 @@ static int32_t doHandleDiff(SDiffInfo* pDiffInfo, int32_t type, const char* pv,
|
||||||
int64_t v = *(int64_t*)pv;
|
int64_t v = *(int64_t*)pv;
|
||||||
int64_t delta = factor * (v - pDiffInfo->prev.i64); // direct previous may be null
|
int64_t delta = factor * (v - pDiffInfo->prev.i64); // direct previous may be null
|
||||||
if (delta < 0 && pDiffInfo->ignoreNegative) {
|
if (delta < 0 && pDiffInfo->ignoreNegative) {
|
||||||
colDataSetNull_f(pOutput->nullbitmap, pos);
|
colDataSetNull_f_s(pOutput, pos);
|
||||||
} else {
|
} else {
|
||||||
colDataAppendInt64(pOutput, pos, &delta);
|
colDataAppendInt64(pOutput, pos, &delta);
|
||||||
}
|
}
|
||||||
|
@ -2691,7 +2693,7 @@ static int32_t doHandleDiff(SDiffInfo* pDiffInfo, int32_t type, const char* pv,
|
||||||
float v = *(float*)pv;
|
float v = *(float*)pv;
|
||||||
double delta = factor * (v - pDiffInfo->prev.d64); // direct previous may be null
|
double delta = factor * (v - pDiffInfo->prev.d64); // direct previous may be null
|
||||||
if ((delta < 0 && pDiffInfo->ignoreNegative) || isinf(delta) || isnan(delta)) { // check for overflow
|
if ((delta < 0 && pDiffInfo->ignoreNegative) || isinf(delta) || isnan(delta)) { // check for overflow
|
||||||
colDataSetNull_f(pOutput->nullbitmap, pos);
|
colDataSetNull_f_s(pOutput, pos);
|
||||||
} else {
|
} else {
|
||||||
colDataAppendDouble(pOutput, pos, &delta);
|
colDataAppendDouble(pOutput, pos, &delta);
|
||||||
}
|
}
|
||||||
|
@ -2702,7 +2704,7 @@ static int32_t doHandleDiff(SDiffInfo* pDiffInfo, int32_t type, const char* pv,
|
||||||
double v = *(double*)pv;
|
double v = *(double*)pv;
|
||||||
double delta = factor * (v - pDiffInfo->prev.d64); // direct previous may be null
|
double delta = factor * (v - pDiffInfo->prev.d64); // direct previous may be null
|
||||||
if ((delta < 0 && pDiffInfo->ignoreNegative) || isinf(delta) || isnan(delta)) { // check for overflow
|
if ((delta < 0 && pDiffInfo->ignoreNegative) || isinf(delta) || isnan(delta)) { // check for overflow
|
||||||
colDataSetNull_f(pOutput->nullbitmap, pos);
|
colDataSetNull_f_s(pOutput, pos);
|
||||||
} else {
|
} else {
|
||||||
colDataAppendDouble(pOutput, pos, &delta);
|
colDataAppendDouble(pOutput, pos, &delta);
|
||||||
}
|
}
|
||||||
|
@ -2737,7 +2739,7 @@ int32_t diffFunction(SqlFunctionCtx* pCtx) {
|
||||||
|
|
||||||
if (colDataIsNull_f(pInputCol->nullbitmap, i)) {
|
if (colDataIsNull_f(pInputCol->nullbitmap, i)) {
|
||||||
if (pDiffInfo->includeNull) {
|
if (pDiffInfo->includeNull) {
|
||||||
colDataSetNull_f(pOutput->nullbitmap, pos);
|
colDataSetNull_f_s(pOutput, pos);
|
||||||
|
|
||||||
numOfElems += 1;
|
numOfElems += 1;
|
||||||
}
|
}
|
||||||
|
@ -2775,8 +2777,7 @@ int32_t diffFunction(SqlFunctionCtx* pCtx) {
|
||||||
|
|
||||||
if (colDataIsNull_f(pInputCol->nullbitmap, i)) {
|
if (colDataIsNull_f(pInputCol->nullbitmap, i)) {
|
||||||
if (pDiffInfo->includeNull) {
|
if (pDiffInfo->includeNull) {
|
||||||
colDataSetNull_f(pOutput->nullbitmap, pos);
|
colDataSetNull_f_s(pOutput, pos);
|
||||||
|
|
||||||
numOfElems += 1;
|
numOfElems += 1;
|
||||||
}
|
}
|
||||||
continue;
|
continue;
|
||||||
|
|
|
@ -92,6 +92,7 @@ static void resetPosInfo(SSlotInfo *pInfo) {
|
||||||
|
|
||||||
int32_t findOnlyResult(tMemBucket *pMemBucket, double *result) {
|
int32_t findOnlyResult(tMemBucket *pMemBucket, double *result) {
|
||||||
ASSERT(pMemBucket->total == 1);
|
ASSERT(pMemBucket->total == 1);
|
||||||
|
terrno = 0;
|
||||||
|
|
||||||
for (int32_t i = 0; i < pMemBucket->numOfSlots; ++i) {
|
for (int32_t i = 0; i < pMemBucket->numOfSlots; ++i) {
|
||||||
tMemBucketSlot *pSlot = &pMemBucket->pSlots[i];
|
tMemBucketSlot *pSlot = &pMemBucket->pSlots[i];
|
||||||
|
|
|
@ -3181,6 +3181,7 @@ bool filterExecuteImplRange(void *pinfo, int32_t numOfRows, SColumnInfoData *pRe
|
||||||
void *colData = colDataGetData(pData, i);
|
void *colData = colDataGetData(pData, i);
|
||||||
if (colData == NULL || colDataIsNull_s(pData, i)) {
|
if (colData == NULL || colDataIsNull_s(pData, i)) {
|
||||||
all = false;
|
all = false;
|
||||||
|
p[i] = 0;
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -1063,11 +1063,11 @@ static int tdbBtreeEncodePayload(SPage *pPage, SCell *pCell, int nHeader, const
|
||||||
} else {
|
} else {
|
||||||
int nLeftKey = kLen;
|
int nLeftKey = kLen;
|
||||||
// pack partial key and nextPgno
|
// pack partial key and nextPgno
|
||||||
memcpy(pCell + nHeader, pKey, nLocal - 4);
|
memcpy(pCell + nHeader, pKey, nLocal - nHeader - sizeof(pgno));
|
||||||
nLeft -= nLocal - 4;
|
nLeft -= nLocal - nHeader - sizeof(pgno);
|
||||||
nLeftKey -= nLocal - 4;
|
nLeftKey -= nLocal - nHeader - sizeof(pgno);
|
||||||
|
|
||||||
memcpy(pCell + nHeader + nLocal - 4, &pgno, sizeof(pgno));
|
memcpy(pCell + nLocal - sizeof(pgno), &pgno, sizeof(pgno));
|
||||||
|
|
||||||
int lastKeyPageSpace = 0;
|
int lastKeyPageSpace = 0;
|
||||||
// pack left key & val to ovpages
|
// pack left key & val to ovpages
|
||||||
|
@ -1087,9 +1087,12 @@ static int tdbBtreeEncodePayload(SPage *pPage, SCell *pCell, int nHeader, const
|
||||||
|
|
||||||
if (lastKeyPage) {
|
if (lastKeyPage) {
|
||||||
if (lastKeyPageSpace >= vLen) {
|
if (lastKeyPageSpace >= vLen) {
|
||||||
memcpy(pBuf + kLen - nLeftKey, pVal, vLen);
|
if (vLen > 0) {
|
||||||
|
memcpy(pBuf + kLen - nLeftKey, pVal, vLen);
|
||||||
|
|
||||||
|
nLeft -= vLen;
|
||||||
|
}
|
||||||
|
|
||||||
nLeft -= vLen;
|
|
||||||
pgno = 0;
|
pgno = 0;
|
||||||
} else {
|
} else {
|
||||||
memcpy(pBuf + kLen - nLeftKey, pVal, lastKeyPageSpace);
|
memcpy(pBuf + kLen - nLeftKey, pVal, lastKeyPageSpace);
|
||||||
|
@ -1111,7 +1114,7 @@ static int tdbBtreeEncodePayload(SPage *pPage, SCell *pCell, int nHeader, const
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
memcpy(pBuf + kLen - nLeft, &pgno, sizeof(pgno));
|
memcpy(pBuf + bytes, &pgno, sizeof(pgno));
|
||||||
|
|
||||||
ret = tdbPageInsertCell(ofp, 0, pBuf, bytes + sizeof(pgno), 0);
|
ret = tdbPageInsertCell(ofp, 0, pBuf, bytes + sizeof(pgno), 0);
|
||||||
if (ret < 0) {
|
if (ret < 0) {
|
||||||
|
|
|
@ -445,6 +445,7 @@
|
||||||
,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/database_pre_suf.py
|
,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/database_pre_suf.py
|
||||||
,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/InsertFuturets.py
|
,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/InsertFuturets.py
|
||||||
,,y,system-test,./pytest.sh python3 ./test.py -f 0-others/show.py
|
,,y,system-test,./pytest.sh python3 ./test.py -f 0-others/show.py
|
||||||
|
,,y,system-test,./pytest.sh python3 ./test.py -f 0-others/information_schema.py
|
||||||
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/abs.py
|
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/abs.py
|
||||||
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/abs.py -R
|
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/abs.py -R
|
||||||
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/and_or_for_byte.py
|
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/and_or_for_byte.py
|
||||||
|
|
|
@ -88,6 +88,7 @@ sql insert into car1 values (now, 1, 1,1 ) (now +1s, 2,2,2) car2 values (now, 1,
|
||||||
|
|
||||||
sql select c1+speed from stb where c1 > 0
|
sql select c1+speed from stb where c1 > 0
|
||||||
if $rows != 3 then
|
if $rows != 3 then
|
||||||
|
print $rows , expect 3
|
||||||
return -1
|
return -1
|
||||||
endi
|
endi
|
||||||
|
|
||||||
|
|
|
@ -0,0 +1,113 @@
|
||||||
|
###################################################################
|
||||||
|
# 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 *
|
||||||
|
from util.cases import *
|
||||||
|
from util.sql import *
|
||||||
|
from util.common import *
|
||||||
|
from util.sqlset import *
|
||||||
|
|
||||||
|
class TDTestCase:
|
||||||
|
def init(self, conn, logSql, replicaVar=1):
|
||||||
|
self.replicaVar = int(replicaVar)
|
||||||
|
tdLog.debug("start to execute %s" % __file__)
|
||||||
|
tdSql.init(conn.cursor())
|
||||||
|
self.setsql = TDSetSql()
|
||||||
|
self.dbname = 'db'
|
||||||
|
self.stbname = 'stb'
|
||||||
|
self.binary_length = 20 # the length of binary for column_dict
|
||||||
|
self.nchar_length = 20 # the length of nchar for column_dict
|
||||||
|
self.ts = 1537146000000
|
||||||
|
self.column_dict = {
|
||||||
|
'ts' : 'timestamp',
|
||||||
|
'col1': 'tinyint',
|
||||||
|
'col2': 'smallint',
|
||||||
|
'col3': 'int',
|
||||||
|
'col4': 'bigint',
|
||||||
|
'col5': 'tinyint unsigned',
|
||||||
|
'col6': 'smallint unsigned',
|
||||||
|
'col7': 'int unsigned',
|
||||||
|
'col8': 'bigint unsigned',
|
||||||
|
'col9': 'float',
|
||||||
|
'col10': 'double',
|
||||||
|
'col11': 'bool',
|
||||||
|
'col12': f'binary({self.binary_length})',
|
||||||
|
'col13': f'nchar({self.nchar_length})'
|
||||||
|
}
|
||||||
|
self.tbnum = 20
|
||||||
|
self.rowNum = 10
|
||||||
|
self.tag_dict = {
|
||||||
|
't0':'int'
|
||||||
|
}
|
||||||
|
self.tag_values = [
|
||||||
|
f'1'
|
||||||
|
]
|
||||||
|
self.binary_str = 'taosdata'
|
||||||
|
self.nchar_str = '涛思数据'
|
||||||
|
self.ins_list = ['ins_dnodes','ins_mnodes','ins_modules','ins_qnodes','ins_snodes','ins_cluster','ins_databases','ins_functions',\
|
||||||
|
'ins_indexes','ins_stables','ins_tables','ins_tags','ins_users','ins_grants','ins_vgroups','ins_configs','ins_dnode_variables',\
|
||||||
|
'ins_topics','ins_subscriptions','ins_streams','ins_stream_tasks','ins_vnodes','ins_user_privileges']
|
||||||
|
self.perf_list = ['perf_connections','perf_queries','perf_consumers','perf_trans','perf_apps']
|
||||||
|
def insert_data(self,column_dict,tbname,row_num):
|
||||||
|
insert_sql = self.setsql.set_insertsql(column_dict,tbname,self.binary_str,self.nchar_str)
|
||||||
|
for i in range(row_num):
|
||||||
|
insert_list = []
|
||||||
|
self.setsql.insert_values(column_dict,i,insert_sql,insert_list,self.ts)
|
||||||
|
def prepare_data(self):
|
||||||
|
tdSql.execute(f"create database if not exists {self.dbname} vgroups 2")
|
||||||
|
tdSql.execute(f'use {self.dbname}')
|
||||||
|
tdSql.execute(self.setsql.set_create_stable_sql(self.stbname,self.column_dict,self.tag_dict))
|
||||||
|
for i in range(self.tbnum):
|
||||||
|
tdSql.execute(f"create table {self.stbname}_{i} using {self.stbname} tags({self.tag_values[0]})")
|
||||||
|
self.insert_data(self.column_dict,f'{self.stbname}_{i}',self.rowNum)
|
||||||
|
def count_check(self):
|
||||||
|
tdSql.query('select count(*) from information_schema.ins_tables')
|
||||||
|
tdSql.checkEqual(tdSql.queryResult[0][0],self.tbnum+len(self.ins_list)+len(self.perf_list))
|
||||||
|
tdSql.query(f'select count(*) from information_schema.ins_tables where db_name = "{self.dbname}"')
|
||||||
|
tdSql.checkEqual(tdSql.queryResult[0][0],self.tbnum)
|
||||||
|
tdSql.query(f'select count(*) from information_schema.ins_tables where db_name = "{self.dbname}" and stable_name = "{self.stbname}"')
|
||||||
|
tdSql.checkEqual(tdSql.queryResult[0][0],self.tbnum)
|
||||||
|
tdSql.execute('create database db1')
|
||||||
|
tdSql.execute('create table stb1 (ts timestamp,c0 int) tags(t0 int)')
|
||||||
|
tdSql.execute('create table tb1 using stb1 tags(1)')
|
||||||
|
tdSql.query(f'select db_name, stable_name, count(*) from information_schema.ins_tables group by db_name, stable_name')
|
||||||
|
for i in tdSql.queryResult:
|
||||||
|
if i[0].lower() == 'information_schema':
|
||||||
|
tdSql.checkEqual(i[2],len(self.ins_list))
|
||||||
|
elif i[0].lower() == self.dbname and i[1] == self.stbname:
|
||||||
|
tdSql.checkEqual(i[2],self.tbnum)
|
||||||
|
elif i[0].lower() == self.dbname and i[1] == 'stb1':
|
||||||
|
tdSql.checkEqual(i[2],1)
|
||||||
|
elif i[0].lower() == 'performance_schema':
|
||||||
|
tdSql.checkEqual(i[2],len(self.perf_list))
|
||||||
|
tdSql.execute('create table db1.ntb (ts timestamp,c0 int)')
|
||||||
|
tdSql.query(f'select db_name, count(*) from information_schema.ins_tables group by db_name')
|
||||||
|
print(tdSql.queryResult)
|
||||||
|
for i in tdSql.queryResult:
|
||||||
|
if i[0].lower() == 'information_schema':
|
||||||
|
tdSql.checkEqual(i[1],len(self.ins_list))
|
||||||
|
elif i[0].lower() == 'performance_schema':
|
||||||
|
tdSql.checkEqual(i[1],len(self.perf_list))
|
||||||
|
elif i[0].lower() == self.dbname:
|
||||||
|
tdSql.checkEqual(i[1],self.tbnum+1)
|
||||||
|
def run(self):
|
||||||
|
self.prepare_data()
|
||||||
|
self.count_check()
|
||||||
|
|
||||||
|
def stop(self):
|
||||||
|
tdSql.close()
|
||||||
|
tdLog.success("%s successfully executed" % __file__)
|
||||||
|
|
||||||
|
tdCases.addWindows(__file__, TDTestCase())
|
||||||
|
tdCases.addLinux(__file__, TDTestCase())
|
Loading…
Reference in New Issue