Merge branch 'develop' into feature/add-parser-test-case-to-2.0
This commit is contained in:
commit
b101c02988
|
@ -31,11 +31,13 @@ extern "C" {
|
|||
#include "tscSecondaryMerge.h"
|
||||
#include "tsclient.h"
|
||||
|
||||
#define UTIL_TABLE_IS_SUPERTABLE(metaInfo) \
|
||||
#define UTIL_TABLE_IS_SUPERTABLE(metaInfo) \
|
||||
(((metaInfo)->pTableMeta != NULL) && ((metaInfo)->pTableMeta->tableType == TSDB_SUPER_TABLE))
|
||||
#define UTIL_TABLE_IS_NOMRAL_TABLE(metaInfo) (!(UTIL_TABLE_IS_SUPERTABLE(metaInfo)))
|
||||
#define UTIL_TABLE_IS_CHILD_TABLE(metaInfo) \
|
||||
(((metaInfo)->pTableMeta != NULL) && ((metaInfo)->pTableMeta->tableType == TSDB_CHILD_TABLE))
|
||||
|
||||
#define UTIL_TABLE_IS_NOMRAL_TABLE(metaInfo)\
|
||||
(!(UTIL_TABLE_IS_SUPERTABLE(metaInfo) || UTIL_TABLE_IS_CHILD_TABLE(metaInfo)))
|
||||
|
||||
#define TSDB_COL_IS_TAG(f) (((f)&TSDB_COL_TAG) != 0)
|
||||
|
||||
|
|
|
@ -305,10 +305,8 @@ int32_t tsParseOneColumnData(SSchema *pSchema, SSQLToken *pToken, char *payload,
|
|||
case TSDB_DATA_TYPE_BINARY:
|
||||
// binary data cannot be null-terminated char string, otherwise the last char of the string is lost
|
||||
if (pToken->type == TK_NULL) {
|
||||
*(int16_t*) payload = sizeof(int8_t);
|
||||
payload += VARSTR_HEADER_SIZE;
|
||||
|
||||
*payload = TSDB_DATA_BINARY_NULL;
|
||||
varDataSetLen(payload, sizeof(int8_t));
|
||||
*(uint8_t*) varDataVal(payload) = TSDB_DATA_BINARY_NULL;
|
||||
} else { // too long values will return invalid sql, not be truncated automatically
|
||||
if (pToken->n + VARSTR_HEADER_SIZE > pSchema->bytes) { //todo refactor
|
||||
return tscInvalidSQLErrMsg(msg, "string data overflow", pToken->z);
|
||||
|
@ -321,22 +319,18 @@ int32_t tsParseOneColumnData(SSchema *pSchema, SSQLToken *pToken, char *payload,
|
|||
|
||||
case TSDB_DATA_TYPE_NCHAR:
|
||||
if (pToken->type == TK_NULL) {
|
||||
*(int16_t*) payload = sizeof(int32_t);
|
||||
payload += VARSTR_HEADER_SIZE;
|
||||
|
||||
*(uint32_t*) payload = TSDB_DATA_NCHAR_NULL;
|
||||
varDataSetLen(payload, sizeof(int32_t));
|
||||
*(uint32_t*) varDataVal(payload) = TSDB_DATA_NCHAR_NULL;
|
||||
} else {
|
||||
// if the converted output len is over than pColumnModel->bytes, return error: 'Argument list too long'
|
||||
size_t wcharLength = 0;
|
||||
if (!taosMbsToUcs4(pToken->z, pToken->n, payload + VARSTR_HEADER_SIZE, pSchema->bytes - VARSTR_HEADER_SIZE,
|
||||
&wcharLength)) {
|
||||
|
||||
size_t output = 0;
|
||||
if (!taosMbsToUcs4(pToken->z, pToken->n, varDataVal(payload), pSchema->bytes - VARSTR_HEADER_SIZE, &output)) {
|
||||
char buf[512] = {0};
|
||||
snprintf(buf, tListLen(buf), "%s", strerror(errno));
|
||||
return tscInvalidSQLErrMsg(msg, buf, pToken->z);
|
||||
}
|
||||
|
||||
*(uint16_t*) payload = (uint16_t) (wcharLength);
|
||||
varDataSetLen(payload, output);
|
||||
}
|
||||
break;
|
||||
|
||||
|
@ -480,10 +474,19 @@ int tsParseOneRowData(char **str, STableDataBlocks *pDataBlocks, SSchema schema[
|
|||
char *ptr = payload;
|
||||
|
||||
for (int32_t i = 0; i < spd->numOfCols; ++i) {
|
||||
|
||||
if (!spd->hasVal[i]) { // current column do not have any value to insert, set it to null
|
||||
setNull(ptr, schema[i].type, schema[i].bytes);
|
||||
if (schema[i].type == TSDB_DATA_TYPE_BINARY) {
|
||||
varDataSetLen(ptr, sizeof(int8_t));
|
||||
*(uint8_t*) varDataVal(ptr) = TSDB_DATA_BINARY_NULL;
|
||||
} else if (schema[i].type == TSDB_DATA_TYPE_NCHAR) {
|
||||
varDataSetLen(ptr, sizeof(int32_t));
|
||||
*(uint32_t*) varDataVal(ptr) = TSDB_DATA_NCHAR_NULL;
|
||||
} else {
|
||||
setNull(ptr, schema[i].type, schema[i].bytes);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
ptr += schema[i].bytes;
|
||||
}
|
||||
|
||||
|
@ -1288,6 +1291,7 @@ int tsParseInsertSql(SSqlObj *pSql) {
|
|||
|
||||
pCmd->count = 0;
|
||||
pCmd->command = TSDB_SQL_INSERT;
|
||||
pSql->res.numOfRows = 0;
|
||||
|
||||
SQueryInfo *pQueryInfo = NULL;
|
||||
tscGetQueryInfoDetailSafely(pCmd, pCmd->clauseIndex, &pQueryInfo);
|
||||
|
@ -1300,7 +1304,6 @@ int tsParseInsertSql(SSqlObj *pSql) {
|
|||
return tscInvalidSQLErrMsg(pCmd->payload, "keyword INTO is expected", sToken.z);
|
||||
}
|
||||
|
||||
pSql->res.numOfRows = 0;
|
||||
return doParseInsertSql(pSql, pSql->sqlstr + index);
|
||||
}
|
||||
|
||||
|
|
|
@ -1411,7 +1411,7 @@ int32_t addProjectionExprAndResultField(SQueryInfo* pQueryInfo, tSQLExprItem* pI
|
|||
STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, index.tableIndex);
|
||||
STableMeta* pTableMeta = pTableMetaInfo->pTableMeta;
|
||||
|
||||
if (index.columnIndex >= tscGetNumOfColumns(pTableMeta) && !UTIL_TABLE_IS_CHILD_TABLE(pTableMetaInfo)) {
|
||||
if (index.columnIndex >= tscGetNumOfColumns(pTableMeta) && UTIL_TABLE_IS_NOMRAL_TABLE(pTableMetaInfo)) {
|
||||
return invalidSqlErrMsg(pQueryInfo->msg, msg1);
|
||||
}
|
||||
|
||||
|
|
|
@ -571,14 +571,14 @@ static char *doSerializeTableInfo(SQueryTableMsg* pQueryMsg, SSqlObj *pSql, char
|
|||
if (UTIL_TABLE_IS_NOMRAL_TABLE(pTableMetaInfo) || pTableMetaInfo->pVgroupTables == NULL) {
|
||||
|
||||
SCMVgroupInfo* pVgroupInfo = NULL;
|
||||
if (UTIL_TABLE_IS_NOMRAL_TABLE(pTableMetaInfo)) {
|
||||
pVgroupInfo = &pTableMeta->vgroupInfo;
|
||||
} else {
|
||||
if (UTIL_TABLE_IS_SUPERTABLE(pTableMetaInfo)) {
|
||||
int32_t index = pTableMetaInfo->vgroupIndex;
|
||||
assert(index >= 0);
|
||||
|
||||
|
||||
pVgroupInfo = &pTableMetaInfo->vgroupList->vgroups[index];
|
||||
tscTrace("%p query on stable, vgIndex:%d, numOfVgroups:%d", pSql, index, pTableMetaInfo->vgroupList->numOfVgroups);
|
||||
} else {
|
||||
pVgroupInfo = &pTableMeta->vgroupInfo;
|
||||
}
|
||||
|
||||
tscSetDnodeIpList(pSql, pVgroupInfo);
|
||||
|
@ -2410,7 +2410,7 @@ static int32_t getTableMetaFromMgmt(SSqlObj *pSql, STableMetaInfo *pTableMetaInf
|
|||
|
||||
strncpy(pNewMeterMetaInfo->name, pTableMetaInfo->name, tListLen(pNewMeterMetaInfo->name));
|
||||
memcpy(pNew->cmd.payload, pSql->cmd.payload, TSDB_DEFAULT_PAYLOAD_SIZE); // tag information if table does not exists.
|
||||
tscTrace("%p new pSqlObj:%p to get tableMeta", pSql, pNew);
|
||||
tscTrace("%p new pSqlObj:%p to get tableMeta, auto create:%d", pSql, pNew, pNew->cmd.autoCreated);
|
||||
|
||||
pNew->fp = tscTableMetaCallBack;
|
||||
pNew->param = pSql;
|
||||
|
|
|
@ -592,42 +592,18 @@ void taos_free_result_imp(TAOS_RES *res, int keepCmd) {
|
|||
|
||||
tscTrace("%p code:%d, numOfRows:%d, command:%d", pSql, pRes->code, pRes->numOfRows, pCmd->command);
|
||||
|
||||
void *fp = pSql->fp;
|
||||
if (fp != NULL) {
|
||||
pSql->freed = 1;
|
||||
}
|
||||
|
||||
pSql->freed = 1;
|
||||
tscProcessSql(pSql);
|
||||
|
||||
/*
|
||||
* If release connection msg is sent to vnode, the corresponding SqlObj for async query can not be freed instantly,
|
||||
* since its free operation is delegated to callback function, which is tscProcessMsgFromServer.
|
||||
*/
|
||||
if (fp == NULL) {
|
||||
/*
|
||||
* fp may be released here, so we cannot use the pSql->fp
|
||||
*
|
||||
* In case of handle sync model query, the main SqlObj cannot be freed.
|
||||
* So, we only free part attributes, including allocated resources and references on metermeta/metricmeta
|
||||
* data in cache.
|
||||
*
|
||||
* Then this object will be reused and no free operation is required.
|
||||
*/
|
||||
if (keepCmd) {
|
||||
tscFreeSqlResult(pSql);
|
||||
tscTrace("%p sql result is freed by app while sql command is kept", pSql);
|
||||
} else {
|
||||
tscPartiallyFreeSqlObj(pSql);
|
||||
tscTrace("%p sql result is freed by app", pSql);
|
||||
}
|
||||
} else { // for async release, remove its link
|
||||
STscObj* pObj = pSql->pTscObj;
|
||||
if (pObj->pSql == pSql) {
|
||||
pObj->pSql = NULL;
|
||||
}
|
||||
STscObj* pObj = pSql->pTscObj;
|
||||
if (pObj->pSql == pSql) {
|
||||
pObj->pSql = NULL;
|
||||
}
|
||||
} else {
|
||||
// if no free resource msg is sent to vnode, we free this object immediately.
|
||||
} else { // if no free resource msg is sent to vnode, we free this object immediately.
|
||||
STscObj* pTscObj = pSql->pTscObj;
|
||||
|
||||
if (pTscObj->pSql != pSql) {
|
||||
|
|
|
@ -2024,7 +2024,6 @@ void **doSetResultRowData(SSqlObj *pSql, bool finalResult) {
|
|||
tExprTreeCalcTraverse(pRes->pArithSup->pArithExpr->pExpr, 1, pRes->buffer[i], pRes->pArithSup,
|
||||
TSDB_ORDER_ASC, getArithemicInputSrc);
|
||||
pRes->tsrow[i] = pRes->buffer[i];
|
||||
// free(sas); //todo optimization
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
@ -357,6 +357,7 @@ void tscResetSqlCmdObj(SSqlCmd* pCmd) {
|
|||
pCmd->curSql = NULL;
|
||||
pCmd->msgType = 0;
|
||||
pCmd->parseFinished = 0;
|
||||
pCmd->autoCreated = 0;
|
||||
|
||||
taosHashCleanup(pCmd->pTableList);
|
||||
pCmd->pTableList = NULL;
|
||||
|
|
|
@ -28,16 +28,16 @@ extern "C" {
|
|||
|
||||
#define STR_TO_VARSTR(x, str) do {VarDataLenT __len = strlen(str); \
|
||||
*(VarDataLenT*)(x) = __len; \
|
||||
strncpy((char*)(x) + VARSTR_HEADER_SIZE, (str), __len);} while(0);
|
||||
strncpy(varDataVal(x), (str), __len);} while(0);
|
||||
|
||||
#define STR_WITH_MAXSIZE_TO_VARSTR(x, str, _maxs) do {\
|
||||
char* _e = stpncpy((char*)(x) + VARSTR_HEADER_SIZE, (str), (_maxs));\
|
||||
*(VarDataLenT*)(x) = (_e - (x) - VARSTR_HEADER_SIZE);\
|
||||
char* _e = stpncpy(varDataVal(x), (str), (_maxs));\
|
||||
varDataSetLen(x, (_e - (x) - VARSTR_HEADER_SIZE));\
|
||||
} while(0)
|
||||
|
||||
#define STR_WITH_SIZE_TO_VARSTR(x, str, _size) do {\
|
||||
*(VarDataLenT*)(x) = (_size); \
|
||||
strncpy((char*)(x) + VARSTR_HEADER_SIZE, (str), (_size));\
|
||||
strncpy(varDataVal(x), (str), (_size));\
|
||||
} while(0);
|
||||
|
||||
// ----------------- TSDB COLUMN DEFINITION
|
||||
|
|
|
@ -92,9 +92,9 @@ bool isNull(const char *val, int32_t type) {
|
|||
case TSDB_DATA_TYPE_DOUBLE:
|
||||
return *(uint64_t *)val == TSDB_DATA_DOUBLE_NULL;
|
||||
case TSDB_DATA_TYPE_NCHAR:
|
||||
return *(uint32_t *)val == TSDB_DATA_NCHAR_NULL;
|
||||
return *(uint32_t*) varDataVal(val) == TSDB_DATA_NCHAR_NULL;
|
||||
case TSDB_DATA_TYPE_BINARY:
|
||||
return *(uint8_t *)val == TSDB_DATA_BINARY_NULL;
|
||||
return *(uint8_t *) varDataVal(val) == TSDB_DATA_BINARY_NULL;
|
||||
default:
|
||||
return false;
|
||||
};
|
||||
|
|
|
@ -40,9 +40,10 @@ typedef int16_t VarDataLenT;
|
|||
|
||||
#define varDataLen(v) ((VarDataLenT *)(v))[0]
|
||||
#define varDataTLen(v) (sizeof(VarDataLenT) + varDataLen(v))
|
||||
#define varDataVal(v) ((void *)((char *)v + sizeof(VarDataLenT)))
|
||||
#define varDataVal(v) ((void *)((char *)v + VARSTR_HEADER_SIZE))
|
||||
#define varDataCopy(dst, v) memcpy((dst), (void*) (v), varDataTLen(v))
|
||||
#define varDataLenByData(v) (*(VarDataLenT *)(((char*)(v)) - VARSTR_HEADER_SIZE))
|
||||
#define varDataSetLen(v, _len) (((VarDataLenT *)(v))[0] = (VarDataLenT) (_len))
|
||||
|
||||
// this data type is internally used only in 'in' query to hold the values
|
||||
#define TSDB_DATA_TYPE_ARRAY (TSDB_DATA_TYPE_NCHAR + 1)
|
||||
|
@ -240,7 +241,7 @@ void tsDataSwap(void *pLeft, void *pRight, int32_t type, int32_t size);
|
|||
#define TSDB_MULTI_METERMETA_MAX_NUM 100000 // maximum batch size allowed to load metermeta
|
||||
|
||||
#define TSDB_MIN_CACHE_BLOCK_SIZE 1
|
||||
#define TSDB_MAX_CACHE_BLOCK_SIZE 1000000
|
||||
#define TSDB_MAX_CACHE_BLOCK_SIZE 10240 // 10GB for each vnode
|
||||
#define TSDB_DEFAULT_CACHE_BLOCK_SIZE 16
|
||||
|
||||
#define TSDB_MIN_TOTAL_BLOCKS 2
|
||||
|
|
|
@ -48,16 +48,16 @@ typedef struct tQueryInfo {
|
|||
int32_t colIndex; // index of column in schema
|
||||
uint8_t optr; // expression operator
|
||||
SSchema sch; // schema of tags
|
||||
// tVariant q; // query condition value on the specific schema, filter expression
|
||||
char* q;
|
||||
__compar_fn_t compare; // filter function
|
||||
void* param; // STSchema,
|
||||
} tQueryInfo;
|
||||
|
||||
typedef struct SBinaryFilterSupp {
|
||||
typedef struct SExprTraverseSupp {
|
||||
__result_filter_fn_t fp;
|
||||
__do_filter_suppl_fn_t setupInfoFn;
|
||||
void * pExtInfo;
|
||||
} SBinaryFilterSupp;
|
||||
} SExprTraverseSupp;
|
||||
|
||||
typedef struct tExprNode {
|
||||
uint8_t nodeType;
|
||||
|
@ -81,7 +81,7 @@ void tSQLBinaryExprToString(tExprNode *pExpr, char *dst, int32_t *len);
|
|||
|
||||
void tExprTreeDestroy(tExprNode **pExprs, void (*fp)(void*));
|
||||
|
||||
void tExprTreeTraverse(tExprNode *pExpr, SSkipList *pSkipList, SArray *result, SBinaryFilterSupp *param);
|
||||
void tExprTreeTraverse(tExprNode *pExpr, SSkipList *pSkipList, SArray *result, SExprTraverseSupp *param);
|
||||
|
||||
void tExprTreeCalcTraverse(tExprNode *pExprs, int32_t numOfRows, char *pOutput, void *param, int32_t order,
|
||||
char *(*cb)(void *, const char*, int32_t));
|
||||
|
|
|
@ -544,13 +544,11 @@ static void tQueryIndexColumn(SSkipList* pSkipList, tQueryInfo* pQueryInfo, SArr
|
|||
setQueryCond(pQueryInfo, &cond);
|
||||
|
||||
if (cond.start != NULL) {
|
||||
iter = tSkipListCreateIterFromVal(pSkipList, (char*) &cond.start->v, pSkipList->keyInfo.type, TSDB_ORDER_ASC);
|
||||
iter = tSkipListCreateIterFromVal(pSkipList, (char*) cond.start->v, pSkipList->keyInfo.type, TSDB_ORDER_ASC);
|
||||
} else {
|
||||
iter = tSkipListCreateIterFromVal(pSkipList, (char*) &cond.end->v, pSkipList->keyInfo.type, TSDB_ORDER_DESC);
|
||||
iter = tSkipListCreateIterFromVal(pSkipList, (char*) cond.end->v, pSkipList->keyInfo.type, TSDB_ORDER_DESC);
|
||||
}
|
||||
|
||||
__compar_fn_t func = getKeyComparFunc(pSkipList->keyInfo.type);
|
||||
|
||||
if (cond.start != NULL) {
|
||||
int32_t optr = cond.start->optr;
|
||||
|
||||
|
@ -558,7 +556,7 @@ static void tQueryIndexColumn(SSkipList* pSkipList, tQueryInfo* pQueryInfo, SArr
|
|||
while(tSkipListIterNext(iter)) {
|
||||
SSkipListNode* pNode = tSkipListIterGet(iter);
|
||||
|
||||
int32_t ret = func(SL_GET_NODE_KEY(pSkipList, pNode), cond.start->v);
|
||||
int32_t ret = pQueryInfo->compare(SL_GET_NODE_KEY(pSkipList, pNode), cond.start->v);
|
||||
if (ret == 0) {
|
||||
taosArrayPush(result, SL_GET_NODE_DATA(pNode));
|
||||
} else {
|
||||
|
@ -573,7 +571,7 @@ static void tQueryIndexColumn(SSkipList* pSkipList, tQueryInfo* pQueryInfo, SArr
|
|||
SSkipListNode* pNode = tSkipListIterGet(iter);
|
||||
|
||||
if (comp) {
|
||||
ret = func(SL_GET_NODE_KEY(pSkipList, pNode), cond.start->v);
|
||||
ret = pQueryInfo->compare(SL_GET_NODE_KEY(pSkipList, pNode), cond.start->v);
|
||||
assert(ret >= 0);
|
||||
}
|
||||
|
||||
|
@ -600,7 +598,7 @@ static void tQueryIndexColumn(SSkipList* pSkipList, tQueryInfo* pQueryInfo, SArr
|
|||
SSkipListNode* pNode = tSkipListIterGet(iter);
|
||||
|
||||
if (comp) {
|
||||
ret = func(SL_GET_NODE_KEY(pSkipList, pNode), cond.end->v);
|
||||
ret = pQueryInfo->compare(SL_GET_NODE_KEY(pSkipList, pNode), cond.end->v);
|
||||
assert(ret <= 0);
|
||||
}
|
||||
|
||||
|
@ -708,7 +706,7 @@ static void tArrayTraverse(tExprNode *pExpr, __result_filter_fn_t fp, SArray *pR
|
|||
}
|
||||
}
|
||||
|
||||
static bool filterItem(tExprNode *pExpr, const void *pItem, SBinaryFilterSupp *param) {
|
||||
static bool filterItem(tExprNode *pExpr, const void *pItem, SExprTraverseSupp *param) {
|
||||
tExprNode *pLeft = pExpr->_node.pLeft;
|
||||
tExprNode *pRight = pExpr->_node.pRight;
|
||||
|
||||
|
@ -747,7 +745,7 @@ static bool filterItem(tExprNode *pExpr, const void *pItem, SBinaryFilterSupp *p
|
|||
* @param pSchema tag schemas
|
||||
* @param fp filter callback function
|
||||
*/
|
||||
static void exprTreeTraverseImpl(tExprNode *pExpr, SArray *pResult, SBinaryFilterSupp *param) {
|
||||
static void exprTreeTraverseImpl(tExprNode *pExpr, SArray *pResult, SExprTraverseSupp *param) {
|
||||
size_t size = taosArrayGetSize(pResult);
|
||||
|
||||
SArray* array = taosArrayInit(size, POINTER_BYTES);
|
||||
|
@ -763,7 +761,7 @@ static void exprTreeTraverseImpl(tExprNode *pExpr, SArray *pResult, SBinaryFilte
|
|||
}
|
||||
|
||||
|
||||
static void tSQLBinaryTraverseOnSkipList(tExprNode *pExpr, SArray *pResult, SSkipList *pSkipList, SBinaryFilterSupp *param ) {
|
||||
static void tSQLBinaryTraverseOnSkipList(tExprNode *pExpr, SArray *pResult, SSkipList *pSkipList, SExprTraverseSupp *param ) {
|
||||
SSkipListIterator* iter = tSkipListCreateIter(pSkipList);
|
||||
|
||||
while (tSkipListIterNext(iter)) {
|
||||
|
@ -813,7 +811,7 @@ static void tQueryIndexlessColumn(SSkipList* pSkipList, tQueryInfo* pQueryInfo,
|
|||
|
||||
|
||||
// post-root order traverse syntax tree
|
||||
void tExprTreeTraverse(tExprNode *pExpr, SSkipList *pSkipList, SArray *result, SBinaryFilterSupp *param) {
|
||||
void tExprTreeTraverse(tExprNode *pExpr, SSkipList *pSkipList, SArray *result, SExprTraverseSupp *param) {
|
||||
if (pExpr == NULL) {
|
||||
return;
|
||||
}
|
||||
|
|
|
@ -266,11 +266,15 @@ static pthread_once_t keywordsHashTableInit = PTHREAD_ONCE_INIT;
|
|||
|
||||
int tSQLKeywordCode(const char* z, int n) {
|
||||
pthread_once(&keywordsHashTableInit, doInitKeywordsTable);
|
||||
|
||||
char key[128] = {0};
|
||||
|
||||
char key[512] = {0};
|
||||
if (n > tListLen(key)) { // too long token, can not be any other token type
|
||||
return TK_ID;
|
||||
}
|
||||
|
||||
for (int32_t j = 0; j < n; ++j) {
|
||||
if (z[j] >= 'a' && z[j] <= 'z') {
|
||||
key[j] = (char)(z[j] & 0xDF); // touppercase and set the null-terminated
|
||||
key[j] = (char)(z[j] & 0xDF); // to uppercase and set the null-terminated
|
||||
} else {
|
||||
key[j] = z[j];
|
||||
}
|
||||
|
|
|
@ -502,13 +502,19 @@ static int tsdbAddTableIntoIndex(STsdbMeta *pMeta, STable *pTable) {
|
|||
memcpy(SL_GET_NODE_DATA(pNode), &pTable, POINTER_BYTES);
|
||||
|
||||
tSkipListPut(list, pNode);
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
static int tsdbRemoveTableFromIndex(STsdbMeta *pMeta, STable *pTable) {
|
||||
assert(pTable->type == TSDB_CHILD_TABLE);
|
||||
// TODO
|
||||
assert(pTable->type == TSDB_CHILD_TABLE && pTable != NULL);
|
||||
|
||||
STable* pSTable = tsdbGetTableByUid(pMeta, pTable->superUid);
|
||||
assert(pSTable != NULL);
|
||||
|
||||
char* key = dataRowTuple(pTable->tagVal); // key
|
||||
bool ret = tSkipListRemove(pSTable->pIndex, key);
|
||||
|
||||
assert(ret);
|
||||
return 0;
|
||||
}
|
||||
|
||||
|
|
|
@ -335,11 +335,7 @@ static int32_t getFileCompInfo(STsdbQueryHandle* pQueryHandle, int32_t* numOfBlo
|
|||
pCheckInfo->compSize = compIndex->len;
|
||||
}
|
||||
|
||||
// tsdbLoadCompBlocks(fileGroup, compIndex, pCheckInfo->pCompInfo);
|
||||
STable* pTable = tsdbGetTableByUid(tsdbGetMeta(pQueryHandle->pTsdb), pCheckInfo->tableId.uid);
|
||||
assert(pTable != NULL);
|
||||
|
||||
tsdbSetHelperTable(&pQueryHandle->rhelper, pTable, pQueryHandle->pTsdb);
|
||||
tsdbSetHelperTable(&pQueryHandle->rhelper, pCheckInfo->pTableObj, pQueryHandle->pTsdb);
|
||||
|
||||
tsdbLoadCompInfo(&(pQueryHandle->rhelper), (void *)(pCheckInfo->pCompInfo));
|
||||
SCompInfo* pCompInfo = pCheckInfo->pCompInfo;
|
||||
|
@ -472,6 +468,7 @@ static bool loadFileDataBlock(STsdbQueryHandle* pQueryHandle, SCompBlock* pBlock
|
|||
filterDataInDataBlock(pQueryHandle, pCheckInfo, pBlock, sa);
|
||||
} else { // the whole block is loaded in to buffer
|
||||
pQueryHandle->realNumOfRows = pBlock->numOfPoints;
|
||||
cur->pos = 0;
|
||||
}
|
||||
} else {
|
||||
// query ended in current block
|
||||
|
@ -491,6 +488,7 @@ static bool loadFileDataBlock(STsdbQueryHandle* pQueryHandle, SCompBlock* pBlock
|
|||
filterDataInDataBlock(pQueryHandle, pCheckInfo, pBlock, sa);
|
||||
} else {
|
||||
pQueryHandle->realNumOfRows = pBlock->numOfPoints;
|
||||
cur->pos = pBlock->numOfPoints - 1;
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -568,7 +566,7 @@ static void filterDataInDataBlock(STsdbQueryHandle* pQueryHandle, STableCheckInf
|
|||
SDataBlockInfo blockInfo = getTrueDataBlockInfo(pCheckInfo, pBlock);
|
||||
|
||||
SDataCols* pCols = pQueryHandle->rhelper.pDataCols[0];
|
||||
|
||||
|
||||
int32_t endPos = cur->pos;
|
||||
if (ASCENDING_ORDER_TRAVERSE(pQueryHandle->order) && pQueryHandle->window.ekey > blockInfo.window.ekey) {
|
||||
endPos = blockInfo.rows - 1;
|
||||
|
@ -612,7 +610,6 @@ static void filterDataInDataBlock(STsdbQueryHandle* pQueryHandle, STableCheckInf
|
|||
int32_t reqCols = taosArrayGetSize(pQueryHandle->pColumns);
|
||||
|
||||
for (int32_t i = 0; i < reqCols; ++i) {
|
||||
// int16_t colId = *(int16_t*)taosArrayGet(sa, i);
|
||||
SColumnInfoData* pCol = taosArrayGet(pQueryHandle->pColumns, i);
|
||||
int32_t bytes = pCol->info.bytes;
|
||||
|
||||
|
@ -1236,12 +1233,6 @@ static int32_t getAllTableIdList(STable* pSuperTable, SArray* list) {
|
|||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
typedef struct SExprTreeSupporter {
|
||||
SSchema* pTagSchema;
|
||||
int32_t numOfTags;
|
||||
int32_t optr;
|
||||
} SExprTreeSupporter;
|
||||
|
||||
/**
|
||||
* convert the result pointer to table id instead of table object pointer
|
||||
* @param pRes
|
||||
|
@ -1252,7 +1243,7 @@ static void convertQueryResult(SArray* pRes, SArray* pTableList) {
|
|||
}
|
||||
|
||||
size_t size = taosArrayGetSize(pTableList);
|
||||
for (int32_t i = 0; i < size; ++i) {
|
||||
for (int32_t i = 0; i < size; ++i) { // todo speedup by using reserve space.
|
||||
STable* pTable = taosArrayGetP(pTableList, i);
|
||||
taosArrayPush(pRes, &pTable->tableId);
|
||||
}
|
||||
|
@ -1273,16 +1264,15 @@ static void destroyHelper(void* param) {
|
|||
free(param);
|
||||
}
|
||||
|
||||
static int32_t getTagColumnInfo(SExprTreeSupporter* pSupporter, SSchema* pSchema) {
|
||||
static int32_t getTagColumnIndex(STSchema* pTSchema, SSchema* pSchema) {
|
||||
// filter on table name(TBNAME)
|
||||
if (strcasecmp(pSchema->name, TSQL_TBNAME_L) == 0) {
|
||||
return TSDB_TBNAME_COLUMN_INDEX;
|
||||
}
|
||||
|
||||
for(int32_t i = 0; i < pSupporter->numOfTags; ++i) {
|
||||
if (pSupporter->pTagSchema[i].bytes == pSchema->bytes &&
|
||||
pSupporter->pTagSchema[i].type == pSchema->type &&
|
||||
pSupporter->pTagSchema[i].colId == pSchema->colId) {
|
||||
|
||||
for(int32_t i = 0; i < schemaNCols(pTSchema); ++i) {
|
||||
STColumn* pColumn = &pTSchema->columns[i];
|
||||
if (pColumn->bytes == pSchema->bytes && pColumn->type == pSchema->type && pColumn->colId == pSchema->colId) {
|
||||
return i;
|
||||
}
|
||||
}
|
||||
|
@ -1298,21 +1288,22 @@ void filterPrepare(void* expr, void* param) {
|
|||
|
||||
int32_t i = 0;
|
||||
pExpr->_node.info = calloc(1, sizeof(tQueryInfo));
|
||||
|
||||
SExprTreeSupporter* pSupporter = (SExprTreeSupporter*)param;
|
||||
|
||||
STSchema* pTSSchema = (STSchema*) param;
|
||||
|
||||
tQueryInfo* pInfo = pExpr->_node.info;
|
||||
tVariant* pCond = pExpr->_node.pRight->pVal;
|
||||
SSchema* pSchema = pExpr->_node.pLeft->pSchema;
|
||||
|
||||
// todo : if current super table does not change schema yet, this function may failed, add test case
|
||||
int32_t index = getTagColumnInfo(pSupporter, pSchema);
|
||||
int32_t index = getTagColumnIndex(pTSSchema, pSchema);
|
||||
assert((index >= 0 && i < TSDB_MAX_TAGS) || (index == TSDB_TBNAME_COLUMN_INDEX));
|
||||
|
||||
pInfo->sch = *pSchema;
|
||||
pInfo->colIndex = index;
|
||||
pInfo->optr = pExpr->_node.optr;
|
||||
pInfo->compare = getComparFunc(pSchema->type, pInfo->optr);
|
||||
pInfo->param = pTSSchema;
|
||||
|
||||
if (pInfo->optr == TSDB_RELATION_IN) {
|
||||
pInfo->q = (char*) pCond->arr;
|
||||
|
@ -1436,7 +1427,7 @@ SArray* createTableGroup(SArray* pTableList, STSchema* pTagSchema, SColIndex* pC
|
|||
}
|
||||
|
||||
bool tSkipListNodeFilterCallback(const void* pNode, void* param) {
|
||||
tQueryInfo* pInfo = (tQueryInfo*)param;
|
||||
tQueryInfo* pInfo = (tQueryInfo*) param;
|
||||
|
||||
STable* pTable = *(STable**)(SL_GET_NODE_DATA((SSkipListNode*)pNode));
|
||||
|
||||
|
@ -1447,7 +1438,14 @@ bool tSkipListNodeFilterCallback(const void* pNode, void* param) {
|
|||
val = pTable->name;
|
||||
type = TSDB_DATA_TYPE_BINARY;
|
||||
} else {
|
||||
val = dataRowTuple(pTable->tagVal); // todo not only the first column
|
||||
STSchema* pTSchema = (STSchema*) pInfo->param; // todo table schema is identical to stable schema??
|
||||
|
||||
int32_t offset = pTSchema->columns[pInfo->colIndex].offset;
|
||||
if (pInfo->sch.type == TSDB_DATA_TYPE_BINARY || pInfo->sch.type == TSDB_DATA_TYPE_NCHAR) {
|
||||
val = tdGetRowDataOfCol(pTable->tagVal, pInfo->sch.type, TD_DATA_ROW_HEAD_SIZE + offset);
|
||||
} else {
|
||||
val = dataRowTuple(pTable->tagVal) + offset;
|
||||
}
|
||||
}
|
||||
|
||||
int32_t ret = 0;
|
||||
|
@ -1497,19 +1495,11 @@ bool tSkipListNodeFilterCallback(const void* pNode, void* param) {
|
|||
}
|
||||
|
||||
static int32_t doQueryTableList(STable* pSTable, SArray* pRes, tExprNode* pExpr) {
|
||||
// query according to the binary expression
|
||||
STSchema* pSchema = pSTable->tagSchema;
|
||||
SSchema* schema = calloc(schemaNCols(pSchema), sizeof(SSchema));
|
||||
for (int32_t i = 0; i < schemaNCols(pSchema); ++i) {
|
||||
schema[i].colId = schemaColAt(pSchema, i)->colId;
|
||||
schema[i].type = schemaColAt(pSchema, i)->type;
|
||||
schema[i].bytes = schemaColAt(pSchema, i)->bytes;
|
||||
}
|
||||
|
||||
SExprTreeSupporter s = {.pTagSchema = schema, .numOfTags = schemaNCols(pSTable->tagSchema)};
|
||||
|
||||
SBinaryFilterSupp supp = {
|
||||
.fp = (__result_filter_fn_t)tSkipListNodeFilterCallback, .setupInfoFn = filterPrepare, .pExtInfo = &s,
|
||||
// query according to the expression tree
|
||||
SExprTraverseSupp supp = {
|
||||
.fp = (__result_filter_fn_t) tSkipListNodeFilterCallback,
|
||||
.setupInfoFn = filterPrepare,
|
||||
.pExtInfo = pSTable->tagSchema,
|
||||
};
|
||||
|
||||
SArray* pTableList = taosArrayInit(8, POINTER_BYTES);
|
||||
|
@ -1519,7 +1509,6 @@ static int32_t doQueryTableList(STable* pSTable, SArray* pRes, tExprNode* pExpr)
|
|||
|
||||
convertQueryResult(pRes, pTableList);
|
||||
taosArrayDestroy(pTableList);
|
||||
free(schema);
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
|
@ -1527,10 +1516,17 @@ int32_t tsdbQuerySTableByTagCond(TsdbRepoT *tsdb, int64_t uid, const char *pTagC
|
|||
const char* tbnameCond, STableGroupInfo *pGroupInfo, SColIndex *pColIndex, int32_t numOfCols) {
|
||||
STable* pTable = tsdbGetTableByUid(tsdbGetMeta(tsdb), uid);
|
||||
if (pTable == NULL) {
|
||||
uError("failed to get stable, uid:%, %p" PRIu64, uid);
|
||||
uError("%p failed to get stable, uid:%" PRIu64, tsdb, uid);
|
||||
return TSDB_CODE_INVALID_TABLE_ID;
|
||||
}
|
||||
|
||||
if (pTable->type != TSDB_SUPER_TABLE) {
|
||||
uError("%p query normal tag not allowed, uid:%, tid:%d, name:%s" PRIu64,
|
||||
tsdb, uid, pTable->tableId.tid, pTable->name);
|
||||
|
||||
return TSDB_CODE_OPS_NOT_SUPPORT; //basically, this error is caused by invalid sql issued by client
|
||||
}
|
||||
|
||||
SArray* res = taosArrayInit(8, sizeof(STableId));
|
||||
STSchema* pTagSchema = tsdbGetTableTagSchema(tsdbGetMeta(tsdb), pTable);
|
||||
|
||||
|
|
|
@ -178,10 +178,9 @@ SSkipListNode *tSkipListPut(SSkipList *pSkipList, SSkipListNode *pNode);
|
|||
*
|
||||
* @param pSkipList
|
||||
* @param pKey
|
||||
* @param keyType
|
||||
* @return
|
||||
*/
|
||||
SArray *tSkipListGet(SSkipList *pSkipList, SSkipListKey pKey, int16_t keyType);
|
||||
SArray *tSkipListGet(SSkipList *pSkipList, SSkipListKey pKey);
|
||||
|
||||
/**
|
||||
* get the size of skip list
|
||||
|
@ -242,7 +241,7 @@ void *tSkipListDestroyIter(SSkipListIterator *iter);
|
|||
* true: one node has been removed
|
||||
* false: no node has been removed
|
||||
*/
|
||||
bool tSkipListRemove(SSkipList *pSkipList, SSkipListKey *pKey);
|
||||
bool tSkipListRemove(SSkipList *pSkipList, SSkipListKey key);
|
||||
|
||||
/*
|
||||
* remove the specified node in parameters
|
||||
|
|
|
@ -231,7 +231,6 @@ static UNUSED_FUNC int32_t compareWStrPatternComp(const void* pLeft, const void*
|
|||
return (ret == TSDB_PATTERN_MATCH) ? 0 : 1;
|
||||
}
|
||||
|
||||
// todo promote the type definition before the comparsion
|
||||
__compar_fn_t getComparFunc(int32_t type, int32_t optr) {
|
||||
__compar_fn_t comparFn = NULL;
|
||||
|
||||
|
|
|
@ -74,6 +74,7 @@ static void tSkipListDoInsert(SSkipList *pSkipList, SSkipListNode **forward, SSk
|
|||
static SSkipListNode* tSkipListPushBack(SSkipList *pSkipList, SSkipListNode *pNode);
|
||||
static SSkipListNode* tSkipListPushFront(SSkipList* pSkipList, SSkipListNode *pNode);
|
||||
static SSkipListIterator* doCreateSkipListIterator(SSkipList *pSkipList, int32_t order);
|
||||
static SSkipListNode* tSkipListDoGet(SSkipList *pSkipList, SSkipListKey key);
|
||||
|
||||
static bool initForwardBackwardPtr(SSkipList* pSkipList) {
|
||||
uint32_t maxLevel = pSkipList->maxLevel;
|
||||
|
@ -97,6 +98,7 @@ static bool initForwardBackwardPtr(SSkipList* pSkipList) {
|
|||
|
||||
return true;
|
||||
}
|
||||
|
||||
SSkipList *tSkipListCreate(uint8_t maxLevel, uint8_t keyType, uint8_t keyLen, uint8_t dupKey, uint8_t lock,
|
||||
uint8_t freeNode, __sl_key_fn_t fn) {
|
||||
SSkipList *pSkipList = (SSkipList *)calloc(1, sizeof(SSkipList));
|
||||
|
@ -139,25 +141,6 @@ SSkipList *tSkipListCreate(uint8_t maxLevel, uint8_t keyType, uint8_t keyLen, ui
|
|||
return pSkipList;
|
||||
}
|
||||
|
||||
// static void doRemove(SSkipList *pSkipList, SSkipListNode *pNode, SSkipListNode *forward[]) {
|
||||
// int32_t level = pNode->level;
|
||||
// for (int32_t j = level - 1; j >= 0; --j) {
|
||||
// if ((forward[j]->pForward[j] != NULL) && (forward[j]->pForward[j]->pForward[j])) {
|
||||
// forward[j]->pForward[j]->pForward[j]->pBackward[j] = forward[j];
|
||||
// }
|
||||
//
|
||||
// if (forward[j]->pForward[j] != NULL) {
|
||||
// forward[j]->pForward[j] = forward[j]->pForward[j]->pForward[j];
|
||||
// }
|
||||
// }
|
||||
//
|
||||
// pSkipList->state.nTotalMemSize -= (sizeof(SSkipListNode) + POINTER_BYTES * pNode->level * 2);
|
||||
// removeNodeEachLevel(pSkipList, pNode->level);
|
||||
//
|
||||
// tfree(pNode);
|
||||
// --pSkipList->size;
|
||||
//}
|
||||
|
||||
void *tSkipListDestroy(SSkipList *pSkipList) {
|
||||
if (pSkipList == NULL) {
|
||||
return NULL;
|
||||
|
@ -257,105 +240,10 @@ SSkipListNode *tSkipListPut(SSkipList *pSkipList, SSkipListNode *pNode) {
|
|||
return pNode;
|
||||
}
|
||||
|
||||
void tSkipListDoInsert(SSkipList *pSkipList, SSkipListNode **forward, SSkipListNode *pNode) {
|
||||
DO_MEMSET_PTR_AREA(pNode);
|
||||
|
||||
for (int32_t i = 0; i < pNode->level; ++i) {
|
||||
SSkipListNode *x = forward[i];
|
||||
SL_GET_BACKWARD_POINTER(pNode, i) = x;
|
||||
|
||||
SSkipListNode *next = SL_GET_FORWARD_POINTER(x, i);
|
||||
SL_GET_BACKWARD_POINTER(next, i) = pNode;
|
||||
|
||||
SL_GET_FORWARD_POINTER(pNode, i) = next;
|
||||
SL_GET_FORWARD_POINTER(x, i) = pNode;
|
||||
}
|
||||
|
||||
atomic_add_fetch_32(&pSkipList->size, 1);
|
||||
if (pSkipList->lock) {
|
||||
pthread_rwlock_unlock(pSkipList->lock);
|
||||
}
|
||||
}
|
||||
|
||||
SSkipListNode* tSkipListPushFront(SSkipList* pSkipList, SSkipListNode *pNode) {
|
||||
SSkipListNode* forward[MAX_SKIP_LIST_LEVEL] = {0};
|
||||
for(int32_t i = 0; i < pSkipList->level; ++i) {
|
||||
forward[i] = pSkipList->pHead;
|
||||
}
|
||||
|
||||
tSkipListDoInsert(pSkipList, forward, pNode);
|
||||
return pNode;
|
||||
}
|
||||
|
||||
SSkipListNode* tSkipListPushBack(SSkipList *pSkipList, SSkipListNode *pNode) {
|
||||
// do clear pointer area
|
||||
DO_MEMSET_PTR_AREA(pNode);
|
||||
|
||||
for(int32_t i = 0; i < pNode->level; ++i) {
|
||||
SSkipListNode* prev = SL_GET_BACKWARD_POINTER(pSkipList->pTail, i);
|
||||
SL_GET_FORWARD_POINTER(prev, i) = pNode;
|
||||
SL_GET_FORWARD_POINTER(pNode, i) = pSkipList->pTail;
|
||||
|
||||
SL_GET_BACKWARD_POINTER(pNode, i) = prev;
|
||||
SL_GET_BACKWARD_POINTER(pSkipList->pTail, i) = pNode;
|
||||
}
|
||||
|
||||
pSkipList->lastKey = SL_GET_NODE_KEY(pSkipList, pNode);
|
||||
|
||||
atomic_add_fetch_32(&pSkipList->size, 1);
|
||||
if (pSkipList->lock) {
|
||||
pthread_rwlock_unlock(pSkipList->lock);
|
||||
}
|
||||
|
||||
return pNode;
|
||||
}
|
||||
|
||||
SArray* tSkipListGet(SSkipList *pSkipList, SSkipListKey pKey, int16_t keyType) {
|
||||
int32_t sLevel = pSkipList->level - 1;
|
||||
|
||||
// result list
|
||||
SArray* tSkipListGet(SSkipList *pSkipList, SSkipListKey key) {
|
||||
SArray* sa = taosArrayInit(1, POINTER_BYTES);
|
||||
SSkipListNode *pNode = pSkipList->pHead;
|
||||
|
||||
if (pSkipList->lock) {
|
||||
pthread_rwlock_rdlock(pSkipList->lock);
|
||||
}
|
||||
|
||||
#if SKIP_LIST_RECORD_PERFORMANCE
|
||||
pSkipList->state.queryCount++;
|
||||
#endif
|
||||
|
||||
__compar_fn_t filterComparFn = getComparFunc(pSkipList->keyInfo.type, 0);
|
||||
int32_t ret = -1;
|
||||
for (int32_t i = sLevel; i >= 0; --i) {
|
||||
SSkipListNode *p = SL_GET_FORWARD_POINTER(pNode, i);
|
||||
while (p != pSkipList->pTail) {
|
||||
char *key = SL_GET_NODE_KEY(pSkipList, p);
|
||||
|
||||
if ((ret = filterComparFn(key, pKey)) < 0) {
|
||||
pNode = p;
|
||||
p = SL_GET_FORWARD_POINTER(p, i);
|
||||
} else {
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
// find the qualified key
|
||||
if (ret == 0) {
|
||||
SSkipListNode* pResult = SL_GET_FORWARD_POINTER(pNode, i);
|
||||
taosArrayPush(sa, &pResult);
|
||||
|
||||
// skip list does not allowed duplicated key, abort further retrieve data
|
||||
if (!pSkipList->keyInfo.dupKey) {
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (pSkipList->lock) {
|
||||
pthread_rwlock_unlock(pSkipList->lock);
|
||||
}
|
||||
|
||||
SSkipListNode* pNode = tSkipListDoGet(pSkipList, key);
|
||||
taosArrayPush(sa, &pNode);
|
||||
return sa;
|
||||
}
|
||||
|
||||
|
@ -367,114 +255,6 @@ size_t tSkipListGetSize(const SSkipList* pSkipList) {
|
|||
return pSkipList->size;
|
||||
}
|
||||
|
||||
SSkipListIterator* tSkipListCreateIter(SSkipList *pSkipList) {
|
||||
if (pSkipList == NULL) {
|
||||
return NULL;
|
||||
}
|
||||
|
||||
return doCreateSkipListIterator(pSkipList, TSDB_ORDER_ASC);
|
||||
}
|
||||
|
||||
SSkipListIterator *tSkipListCreateIterFromVal(SSkipList* pSkipList, const char* val, int32_t type, int32_t order) {
|
||||
if (pSkipList == NULL) {
|
||||
return NULL;
|
||||
}
|
||||
|
||||
assert(order == TSDB_ORDER_ASC || order == TSDB_ORDER_DESC);
|
||||
|
||||
if (val == NULL) {
|
||||
return doCreateSkipListIterator(pSkipList, order);
|
||||
} else {
|
||||
|
||||
SSkipListNode *forward[MAX_SKIP_LIST_LEVEL] = {0};
|
||||
|
||||
int32_t ret = -1;
|
||||
__compar_fn_t filterComparFn = getKeyComparFunc(pSkipList->keyInfo.type);
|
||||
SSkipListNode* pNode = pSkipList->pHead;
|
||||
|
||||
for (int32_t i = pSkipList->level - 1; i >= 0; --i) {
|
||||
SSkipListNode *p = SL_GET_FORWARD_POINTER(pNode, i);
|
||||
while (p != pSkipList->pTail) {
|
||||
char *key = SL_GET_NODE_KEY(pSkipList, p);
|
||||
|
||||
if ((ret = filterComparFn(key, val)) < 0) {
|
||||
pNode = p;
|
||||
p = SL_GET_FORWARD_POINTER(p, i);
|
||||
} else {
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
forward[i] = pNode;
|
||||
}
|
||||
|
||||
SSkipListIterator* iter = doCreateSkipListIterator(pSkipList, order);
|
||||
|
||||
// set the initial position
|
||||
if (order == TSDB_ORDER_ASC) {
|
||||
iter->cur = forward[0]; // greater equals than the value
|
||||
} else {
|
||||
iter->cur = SL_GET_FORWARD_POINTER(forward[0], 0);
|
||||
|
||||
if (ret == 0) {
|
||||
assert(iter->cur != pSkipList->pTail);
|
||||
iter->cur = SL_GET_FORWARD_POINTER(iter->cur, 0);
|
||||
}
|
||||
}
|
||||
|
||||
return iter;
|
||||
}
|
||||
}
|
||||
|
||||
bool tSkipListIterNext(SSkipListIterator *iter) {
|
||||
if (iter->pSkipList == NULL) {
|
||||
return false;
|
||||
}
|
||||
|
||||
SSkipList *pSkipList = iter->pSkipList;
|
||||
|
||||
if (pSkipList->lock) {
|
||||
pthread_rwlock_rdlock(pSkipList->lock);
|
||||
}
|
||||
|
||||
if (iter->order == TSDB_ORDER_ASC) { // ascending order iterate
|
||||
if (iter->cur == NULL) {
|
||||
iter->cur = SL_GET_FORWARD_POINTER(pSkipList->pHead, 0);
|
||||
} else {
|
||||
iter->cur = SL_GET_FORWARD_POINTER(iter->cur, 0);
|
||||
}
|
||||
} else { // descending order iterate
|
||||
if (iter->cur == NULL) {
|
||||
iter->cur = SL_GET_BACKWARD_POINTER(pSkipList->pTail, 0);
|
||||
} else {
|
||||
iter->cur = SL_GET_BACKWARD_POINTER(iter->cur, 0);
|
||||
}
|
||||
}
|
||||
|
||||
if (pSkipList->lock) {
|
||||
pthread_rwlock_unlock(pSkipList->lock);
|
||||
}
|
||||
|
||||
return (iter->order == TSDB_ORDER_ASC)? (iter->cur != pSkipList->pTail) : (iter->cur != pSkipList->pHead);
|
||||
}
|
||||
|
||||
SSkipListNode *tSkipListIterGet(SSkipListIterator *iter) {
|
||||
if (iter == NULL || iter->cur == iter->pSkipList->pTail || iter->cur == iter->pSkipList->pHead) {
|
||||
return NULL;
|
||||
} else {
|
||||
return iter->cur;
|
||||
}
|
||||
}
|
||||
|
||||
void* tSkipListDestroyIter(SSkipListIterator* iter) {
|
||||
if (iter == NULL) {
|
||||
return NULL;
|
||||
}
|
||||
|
||||
tfree(iter);
|
||||
return NULL;
|
||||
}
|
||||
|
||||
// static int32_t tSkipListEndParQuery(SSkipList *pSkipList, SSkipListNode *pStartNode, SSkipListKey *pEndKey,
|
||||
// int32_t cond, SSkipListNode ***pRes) {
|
||||
// pthread_rwlock_rdlock(&pSkipList->lock);
|
||||
|
@ -588,25 +368,162 @@ void* tSkipListDestroyIter(SSkipListIterator* iter) {
|
|||
// }
|
||||
//
|
||||
// // compress the minimum level of skip list
|
||||
// while (pSkipList->level > 0 && pSkipList->pHead.pForward[pSkipList->level - 1] == NULL) {
|
||||
// while (pSkipList->level > 0 && SL_GET_FORWARD_POINTER(pSkipList->pHead, pSkipList->level - 1) == NULL) {
|
||||
// pSkipList->level -= 1;
|
||||
// }
|
||||
//
|
||||
// return true;
|
||||
//}
|
||||
//
|
||||
// void tSkipListRemoveNode(SSkipList *pSkipList, SSkipListNode *pNode) {
|
||||
// SSkipListNode *forward[MAX_SKIP_LIST_LEVEL] = {0};
|
||||
//
|
||||
// pthread_rwlock_rdlock(&pSkipList->lock);
|
||||
// for (int32_t i = 0; i < pNode->level; ++i) {
|
||||
// forward[i] = pNode->pBackward[i];
|
||||
// }
|
||||
//
|
||||
// removeSupport(pSkipList, forward, &pNode->key);
|
||||
// pthread_rwlock_unlock(&pSkipList->lock);
|
||||
//}
|
||||
//
|
||||
|
||||
bool tSkipListRemove(SSkipList *pSkipList, SSkipListKey key) {
|
||||
SSkipListNode* pNode = tSkipListDoGet(pSkipList, key);
|
||||
if (pNode != NULL) {
|
||||
tSkipListRemoveNode(pSkipList, pNode);
|
||||
return true;
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
void tSkipListRemoveNode(SSkipList *pSkipList, SSkipListNode *pNode) {
|
||||
int32_t level = pNode->level;
|
||||
|
||||
if (pSkipList->lock) {
|
||||
pthread_rwlock_wrlock(pSkipList->lock);
|
||||
}
|
||||
|
||||
for (int32_t j = level - 1; j >= 0; --j) {
|
||||
SSkipListNode* prev = SL_GET_BACKWARD_POINTER(pNode, j);
|
||||
SSkipListNode* next = SL_GET_FORWARD_POINTER(pNode, j);
|
||||
|
||||
SL_GET_FORWARD_POINTER(prev, j) = next;
|
||||
SL_GET_BACKWARD_POINTER(next, j) = prev;
|
||||
}
|
||||
|
||||
if (pSkipList->keyInfo.freeNode) {
|
||||
tfree(pNode);
|
||||
}
|
||||
|
||||
atomic_sub_fetch_32(&pSkipList->size, 1);
|
||||
|
||||
// compress the minimum level of skip list
|
||||
while (pSkipList->level > 0 && SL_GET_FORWARD_POINTER(pSkipList->pHead, pSkipList->level - 1) == NULL) {
|
||||
pSkipList->level -= 1;
|
||||
}
|
||||
|
||||
if (pSkipList->lock) {
|
||||
pthread_rwlock_unlock(pSkipList->lock);
|
||||
}
|
||||
}
|
||||
|
||||
SSkipListIterator* tSkipListCreateIter(SSkipList *pSkipList) {
|
||||
if (pSkipList == NULL) {
|
||||
return NULL;
|
||||
}
|
||||
|
||||
return doCreateSkipListIterator(pSkipList, TSDB_ORDER_ASC);
|
||||
}
|
||||
|
||||
SSkipListIterator *tSkipListCreateIterFromVal(SSkipList* pSkipList, const char* val, int32_t type, int32_t order) {
|
||||
if (pSkipList == NULL) {
|
||||
return NULL;
|
||||
}
|
||||
|
||||
assert(order == TSDB_ORDER_ASC || order == TSDB_ORDER_DESC);
|
||||
|
||||
if (val == NULL) {
|
||||
return doCreateSkipListIterator(pSkipList, order);
|
||||
} else {
|
||||
|
||||
SSkipListNode *forward[MAX_SKIP_LIST_LEVEL] = {0};
|
||||
|
||||
int32_t ret = -1;
|
||||
__compar_fn_t filterComparFn = getKeyComparFunc(pSkipList->keyInfo.type);
|
||||
SSkipListNode* pNode = pSkipList->pHead;
|
||||
|
||||
for (int32_t i = pSkipList->level - 1; i >= 0; --i) {
|
||||
SSkipListNode *p = SL_GET_FORWARD_POINTER(pNode, i);
|
||||
while (p != pSkipList->pTail) {
|
||||
char *key = SL_GET_NODE_KEY(pSkipList, p);
|
||||
|
||||
if ((ret = filterComparFn(key, val)) < 0) {
|
||||
pNode = p;
|
||||
p = SL_GET_FORWARD_POINTER(p, i);
|
||||
} else {
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
forward[i] = pNode;
|
||||
}
|
||||
|
||||
SSkipListIterator* iter = doCreateSkipListIterator(pSkipList, order);
|
||||
|
||||
// set the initial position
|
||||
if (order == TSDB_ORDER_ASC) {
|
||||
iter->cur = forward[0]; // greater equals than the value
|
||||
} else {
|
||||
iter->cur = SL_GET_FORWARD_POINTER(forward[0], 0);
|
||||
|
||||
if (ret == 0) {
|
||||
assert(iter->cur != pSkipList->pTail);
|
||||
iter->cur = SL_GET_FORWARD_POINTER(iter->cur, 0);
|
||||
}
|
||||
}
|
||||
|
||||
return iter;
|
||||
}
|
||||
}
|
||||
|
||||
bool tSkipListIterNext(SSkipListIterator *iter) {
|
||||
if (iter->pSkipList == NULL) {
|
||||
return false;
|
||||
}
|
||||
|
||||
SSkipList *pSkipList = iter->pSkipList;
|
||||
|
||||
if (pSkipList->lock) {
|
||||
pthread_rwlock_rdlock(pSkipList->lock);
|
||||
}
|
||||
|
||||
if (iter->order == TSDB_ORDER_ASC) { // ascending order iterate
|
||||
if (iter->cur == NULL) {
|
||||
iter->cur = SL_GET_FORWARD_POINTER(pSkipList->pHead, 0);
|
||||
} else {
|
||||
iter->cur = SL_GET_FORWARD_POINTER(iter->cur, 0);
|
||||
}
|
||||
} else { // descending order iterate
|
||||
if (iter->cur == NULL) {
|
||||
iter->cur = SL_GET_BACKWARD_POINTER(pSkipList->pTail, 0);
|
||||
} else {
|
||||
iter->cur = SL_GET_BACKWARD_POINTER(iter->cur, 0);
|
||||
}
|
||||
}
|
||||
|
||||
if (pSkipList->lock) {
|
||||
pthread_rwlock_unlock(pSkipList->lock);
|
||||
}
|
||||
|
||||
return (iter->order == TSDB_ORDER_ASC)? (iter->cur != pSkipList->pTail) : (iter->cur != pSkipList->pHead);
|
||||
}
|
||||
|
||||
SSkipListNode *tSkipListIterGet(SSkipListIterator *iter) {
|
||||
if (iter == NULL || iter->cur == iter->pSkipList->pTail || iter->cur == iter->pSkipList->pHead) {
|
||||
return NULL;
|
||||
} else {
|
||||
return iter->cur;
|
||||
}
|
||||
}
|
||||
|
||||
void* tSkipListDestroyIter(SSkipListIterator* iter) {
|
||||
if (iter == NULL) {
|
||||
return NULL;
|
||||
}
|
||||
|
||||
tfree(iter);
|
||||
return NULL;
|
||||
}
|
||||
|
||||
// bool tSkipListRemove(SSkipList *pSkipList, SSkipListKey *pKey) {
|
||||
// SSkipListNode *forward[MAX_SKIP_LIST_LEVEL] = {0};
|
||||
// __compar_fn_t filterComparFn = getComparFunc(pSkipList, pKey->nType);
|
||||
|
@ -668,6 +585,105 @@ void tSkipListPrint(SSkipList *pSkipList, int16_t nlevel) {
|
|||
}
|
||||
}
|
||||
|
||||
void tSkipListDoInsert(SSkipList *pSkipList, SSkipListNode **forward, SSkipListNode *pNode) {
|
||||
DO_MEMSET_PTR_AREA(pNode);
|
||||
|
||||
for (int32_t i = 0; i < pNode->level; ++i) {
|
||||
SSkipListNode *x = forward[i];
|
||||
SL_GET_BACKWARD_POINTER(pNode, i) = x;
|
||||
|
||||
SSkipListNode *next = SL_GET_FORWARD_POINTER(x, i);
|
||||
SL_GET_BACKWARD_POINTER(next, i) = pNode;
|
||||
|
||||
SL_GET_FORWARD_POINTER(pNode, i) = next;
|
||||
SL_GET_FORWARD_POINTER(x, i) = pNode;
|
||||
}
|
||||
|
||||
atomic_add_fetch_32(&pSkipList->size, 1);
|
||||
if (pSkipList->lock) {
|
||||
pthread_rwlock_unlock(pSkipList->lock);
|
||||
}
|
||||
}
|
||||
|
||||
SSkipListNode* tSkipListPushFront(SSkipList* pSkipList, SSkipListNode *pNode) {
|
||||
SSkipListNode* forward[MAX_SKIP_LIST_LEVEL] = {0};
|
||||
for(int32_t i = 0; i < pSkipList->level; ++i) {
|
||||
forward[i] = pSkipList->pHead;
|
||||
}
|
||||
|
||||
tSkipListDoInsert(pSkipList, forward, pNode);
|
||||
return pNode;
|
||||
}
|
||||
|
||||
SSkipListNode* tSkipListPushBack(SSkipList *pSkipList, SSkipListNode *pNode) {
|
||||
// do clear pointer area
|
||||
DO_MEMSET_PTR_AREA(pNode);
|
||||
|
||||
for(int32_t i = 0; i < pNode->level; ++i) {
|
||||
SSkipListNode* prev = SL_GET_BACKWARD_POINTER(pSkipList->pTail, i);
|
||||
SL_GET_FORWARD_POINTER(prev, i) = pNode;
|
||||
SL_GET_FORWARD_POINTER(pNode, i) = pSkipList->pTail;
|
||||
|
||||
SL_GET_BACKWARD_POINTER(pNode, i) = prev;
|
||||
SL_GET_BACKWARD_POINTER(pSkipList->pTail, i) = pNode;
|
||||
}
|
||||
|
||||
pSkipList->lastKey = SL_GET_NODE_KEY(pSkipList, pNode);
|
||||
|
||||
atomic_add_fetch_32(&pSkipList->size, 1);
|
||||
if (pSkipList->lock) {
|
||||
pthread_rwlock_unlock(pSkipList->lock);
|
||||
}
|
||||
|
||||
return pNode;
|
||||
}
|
||||
|
||||
SSkipListNode* tSkipListDoGet(SSkipList *pSkipList, SSkipListKey skey) {
|
||||
SSkipListNode *pNode = pSkipList->pHead;
|
||||
SSkipListNode *pRes = NULL;
|
||||
|
||||
if (pSkipList->lock) {
|
||||
pthread_rwlock_rdlock(pSkipList->lock);
|
||||
}
|
||||
|
||||
#if SKIP_LIST_RECORD_PERFORMANCE
|
||||
pSkipList->state.queryCount++;
|
||||
#endif
|
||||
|
||||
__compar_fn_t cmparFn = getComparFunc(pSkipList->keyInfo.type, 0);
|
||||
|
||||
int32_t ret = -1;
|
||||
for (int32_t i = pSkipList->level - 1; i >= 0; --i) {
|
||||
SSkipListNode *p = SL_GET_FORWARD_POINTER(pNode, i);
|
||||
while (p != pSkipList->pTail) {
|
||||
char *key = SL_GET_NODE_KEY(pSkipList, p);
|
||||
|
||||
if ((ret = cmparFn(key, skey)) < 0) {
|
||||
pNode = p;
|
||||
p = SL_GET_FORWARD_POINTER(p, i);
|
||||
} else {
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
// find the qualified key
|
||||
if (ret == 0) {
|
||||
pRes = SL_GET_FORWARD_POINTER(pNode, i);
|
||||
break;
|
||||
// skip list does not allowed duplicated key, abort further retrieve data
|
||||
// if (!pSkipList->keyInfo.dupKey) {
|
||||
// break;
|
||||
// }
|
||||
}
|
||||
}
|
||||
|
||||
if (pSkipList->lock) {
|
||||
pthread_rwlock_unlock(pSkipList->lock);
|
||||
}
|
||||
|
||||
return pRes;
|
||||
}
|
||||
|
||||
SSkipListIterator* doCreateSkipListIterator(SSkipList *pSkipList, int32_t order) {
|
||||
SSkipListIterator* iter = calloc(1, sizeof(SSkipListIterator));
|
||||
|
||||
|
@ -675,4 +691,4 @@ SSkipListIterator* doCreateSkipListIterator(SSkipList *pSkipList, int32_t order)
|
|||
iter->order = order;
|
||||
|
||||
return iter;
|
||||
}
|
||||
}
|
|
@ -43,4 +43,7 @@ run general/compute/null.sim
|
|||
# run general/compute/diff2.sim
|
||||
|
||||
run general/parse/testSuite.sim
|
||||
run general/field/testSuite.sim
|
||||
|
||||
|
||||
##################################
|
||||
|
|
|
@ -1,11 +1,14 @@
|
|||
run general/field/2.sim
|
||||
run general/field/3.sim
|
||||
run general/field/4.sim
|
||||
run general/field/5.sim
|
||||
run general/field/6.sim
|
||||
run general/field/bigint.sim
|
||||
run general/field/binary.sim
|
||||
# run general/field/single.sim
|
||||
run general/field/bool.sim
|
||||
run general/field/single.sim
|
||||
run general/field/smallint.sim
|
||||
run general/field/tinyint.sim
|
||||
run general/field/int.sim
|
||||
run general/field/bigint.sim
|
||||
run general/field/float.sim
|
||||
run general/field/double.sim
|
||||
# run general/field/binary.sim
|
||||
# run general/field/2.sim
|
||||
# run general/field/3.sim
|
||||
# run general/field/4.sim
|
||||
# run general/field/5.sim
|
||||
# run general/field/6.sim
|
||||
|
|
|
@ -172,8 +172,8 @@ sql create table $tb using $mt tags (-1)
|
|||
# -x ng_tag_v
|
||||
# return -1
|
||||
#ng_tag_v:
|
||||
sql describe $tb
|
||||
if $data23 != -1 then
|
||||
sql select tg from $tb
|
||||
if $data00 != -1 then
|
||||
return -1
|
||||
endi
|
||||
sql drop table $tb
|
||||
|
@ -182,9 +182,9 @@ sql drop table $tb
|
|||
print create_mt.sim unmatched_tag_types
|
||||
sql reset query cache
|
||||
sql create table $tb using $mt tags ('123')
|
||||
sql describe $tb
|
||||
#print data23 = $data23
|
||||
if $data23 != 123 then
|
||||
sql select tg from $tb
|
||||
print data00 = $data00
|
||||
if $data00 != 123 then
|
||||
return -1
|
||||
endi
|
||||
sql drop table $tb
|
||||
|
@ -194,14 +194,14 @@ sql_error create table $tb using $mt tags ('abc')
|
|||
sql drop table if exists $tb
|
||||
sql reset query cache
|
||||
sql create table $tb using $mt tags (1e1)
|
||||
sql describe $tb
|
||||
if $data23 != 10 then
|
||||
sql select tg from $tb
|
||||
if $data00 != 10 then
|
||||
return -1
|
||||
endi
|
||||
sql drop table $tb
|
||||
sql create table $tb using $mt tags ('1e1')
|
||||
sql describe $tb
|
||||
if $data23 != 10 then
|
||||
sql select tg from $tb
|
||||
if $data00 != 10 then
|
||||
return -1
|
||||
endi
|
||||
sql_error create table $tb using $mt tags (2147483649)
|
||||
|
@ -234,13 +234,13 @@ $mt2 = mt2
|
|||
#if $data20 != $mt2 then
|
||||
# return -1
|
||||
#endi
|
||||
#sql describe $mt1
|
||||
#sql select tg from $mt1
|
||||
##print expected $CN_char
|
||||
##print returned $data10
|
||||
#if $data10 != $CN_char then
|
||||
# return -1
|
||||
#endi
|
||||
#sql describe $mt2
|
||||
#sql select tg from $mt2
|
||||
##print expected: $CN_char
|
||||
##print returned: $data20
|
||||
#if $data20 != $CN_char then
|
||||
|
|
|
@ -172,7 +172,7 @@ print ========== create_tb.sim case7: table_name_length_exceeds_limit
|
|||
$tbname32 = _32_aaaabbbbccccddddaaaabbbbcccc
|
||||
$tbname64 = _64_aaaabbbbccccddddaaaabbbbccccddddaaaabbbbccccddddaaaabbbbcccc
|
||||
$tbname63 = _63_aaaabbbbccccddddaaaabbbbccccddddaaaabbbbccccddddaaaabbbbccc
|
||||
$tbname65 = _65_aaaabbbbccccddddaaaabbbbccccddddaaaabbbbccccddddaaaabbbbcccca
|
||||
$tbname65 = _65_aaaabbbbccccddddaaaabbbbccccddddaaaabbbbccccddddaaaabbbbcccca1111111111111111111111111111111111aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa
|
||||
sql create table $tbname32 (ts timestamp, col int)
|
||||
sql insert into $tbname32 values (now, 1)
|
||||
sql create table $tbname64 (ts timestamp, col int)
|
||||
|
|
|
@ -21,7 +21,9 @@ $stb = $stbPrefix . $i
|
|||
|
||||
sql drop database $db -x step1
|
||||
step1:
|
||||
sql create database $db maxrows 400 cache 4096 maxTables 4
|
||||
|
||||
sql create database $db maxrows 400 cache 1 maxTables 4
|
||||
|
||||
sql use $db
|
||||
sql create table $stb (ts timestamp, c1 int, c2 bigint, c3 float, c4 double, c5 smallint, c6 tinyint, c7 bool, c8 binary(10), c9 nchar(10)) tags(t1 int)
|
||||
|
||||
|
|
|
@ -56,8 +56,8 @@ endi
|
|||
if $data07 != 1 then
|
||||
return -1
|
||||
endi
|
||||
#if $data08 != NULL then
|
||||
if $data08 != BINARY then
|
||||
print expect BINARY, actual: $data08
|
||||
return -1
|
||||
endi
|
||||
#if $data09 != NULL then
|
||||
|
|
|
@ -23,7 +23,7 @@ $stb = $stbPrefix . $i
|
|||
|
||||
sql drop database $db -x step1
|
||||
step1:
|
||||
sql create database $db cache 2048
|
||||
sql create database $db cache 16
|
||||
print ====== create tables
|
||||
sql use $db
|
||||
|
||||
|
|
|
@ -22,7 +22,7 @@ $stb = $stbPrefix . $i
|
|||
|
||||
sql drop database $db -x step1
|
||||
step1:
|
||||
sql create database $db cache 2048
|
||||
sql create database $db cache 16
|
||||
print ====== create tables
|
||||
sql use $db
|
||||
|
||||
|
|
|
@ -22,7 +22,7 @@ $stb = $stbPrefix . $i
|
|||
|
||||
sql drop database $db -x step1
|
||||
step1:
|
||||
sql create database $db cache 16384
|
||||
sql create database $db cache 16
|
||||
print ====== create tables
|
||||
sql use $db
|
||||
|
||||
|
|
Loading…
Reference in New Issue