[td-225] fix bugs for tags filter
This commit is contained in:
parent
68bf22a9fe
commit
f55b05856b
|
@ -33,10 +33,12 @@ extern "C" {
|
|||
|
||||
#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)
|
||||
|
||||
typedef struct SParsedColElem {
|
||||
|
|
|
@ -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);
|
||||
}
|
||||
|
||||
|
|
|
@ -580,14 +580,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);
|
||||
|
|
|
@ -601,42 +601,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;
|
||||
}
|
||||
|
||||
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;
|
||||
}
|
||||
}
|
||||
} 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) {
|
||||
|
|
|
@ -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);
|
||||
}
|
||||
|
||||
|
@ -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;
|
||||
}
|
||||
|
|
|
@ -1236,12 +1236,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 +1246,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 +1267,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;
|
||||
}
|
||||
}
|
||||
|
@ -1299,20 +1292,21 @@ 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;
|
||||
|
@ -1447,7 +1441,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 +1498,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 +1512,6 @@ static int32_t doQueryTableList(STable* pSTable, SArray* pRes, tExprNode* pExpr)
|
|||
|
||||
convertQueryResult(pRes, pTableList);
|
||||
taosArrayDestroy(pTableList);
|
||||
free(schema);
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
|
@ -1527,10 +1519,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);
|
||||
|
||||
|
|
|
@ -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;
|
||||
|
||||
|
|
Loading…
Reference in New Issue