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

Merge pull request #13418 from taosdata/enh/idxFilter 

enh: refactor tag filter code
This commit is contained in:
Yihao Deng 2022-06-02 19:11:54 +08:00 committed by GitHub
parent 2a8b999ca8 69f1960e2e
commit 3eb1d48815
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 149 additions and 114 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
include/common/ttypes.h
View File

@ -49,7 +49,8 @@ typedef struct {
#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))
#define IS_VAR_DATA_TYPE(t) (((t) == TSDB_DATA_TYPE_VARCHAR) || ((t) == TSDB_DATA_TYPE_NCHAR) || ((t) == TSDB_DATA_TYPE_JSON))
#define IS_VAR_DATA_TYPE(t) \
(((t) == TSDB_DATA_TYPE_VARCHAR) || ((t) == TSDB_DATA_TYPE_NCHAR) || ((t) == TSDB_DATA_TYPE_JSON))
#define IS_STR_DATA_TYPE(t) (((t) == TSDB_DATA_TYPE_VARCHAR) || ((t) == TSDB_DATA_TYPE_NCHAR))
#define varDataNetLen(v) (htons(((VarDataLenT *)(v))[0]))
@ -179,6 +180,7 @@ typedef struct {
} \
} while (0)
//TODO: use varchar(0) to represent NULL type
#define IS_VAR_NULL_TYPE(_t, _b) ((_t) == TSDB_DATA_TYPE_VARCHAR && (_b) == 0)
#define IS_NULL_TYPE(_t) ((_t) == TSDB_DATA_TYPE_NULL)
@ -189,7 +191,8 @@ typedef struct {
#define IS_INTEGER_TYPE(_t) ((IS_SIGNED_NUMERIC_TYPE(_t)) || (IS_UNSIGNED_NUMERIC_TYPE(_t)))
#define IS_NUMERIC_TYPE(_t) ((IS_SIGNED_NUMERIC_TYPE(_t)) || (IS_UNSIGNED_NUMERIC_TYPE(_t)) || (IS_FLOAT_TYPE(_t)))
#define IS_MATHABLE_TYPE(_t) (IS_NUMERIC_TYPE(_t) || (_t) == (TSDB_DATA_TYPE_BOOL) || (_t) == (TSDB_DATA_TYPE_TIMESTAMP))
#define IS_MATHABLE_TYPE(_t) \
(IS_NUMERIC_TYPE(_t) || (_t) == (TSDB_DATA_TYPE_BOOL) || (_t) == (TSDB_DATA_TYPE_TIMESTAMP))
#define IS_VALID_TINYINT(_t) ((_t) >= INT8_MIN && (_t) <= INT8_MAX)
#define IS_VALID_SMALLINT(_t) ((_t) >= INT16_MIN && (_t) <= INT16_MAX)
@ -246,7 +249,7 @@ typedef struct tDataTypeDescriptor {
int16_t type;
int16_t nameLen;
int32_t bytes;
char *name;
char * name;
int64_t minValue;
int64_t maxValue;
int32_t (*compFunc)(const char *const input, int32_t inputSize, const int32_t nelements, char *const output,

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

@ -576,15 +576,15 @@ SArray *metaGetSmaTbUids(SMeta *pMeta) {
const void *metaGetTableTagVal(SMetaEntry *pEntry, int16_t type, STagVal *val) {
ASSERT(pEntry->type == TSDB_CHILD_TABLE);
STag *tag = (STag *)pEntry->ctbEntry.pTags;
if (type == TSDB_DATA_TYPE_JSON){
if(tag->nTag == 0){
if (type == TSDB_DATA_TYPE_JSON) {
if (tag->nTag == 0) {
return NULL;
}
return tag;
}
bool find = tTagGet(tag, val);
if(!find){
if (!find) {
return NULL;
}
return val;
@ -605,8 +605,6 @@ typedef struct {
int32_t metaFilteTableIds(SMeta *pMeta, SMetaFltParam *param, SArray *pUids) {
SIdxCursor *pCursor = NULL;
char *tagData = param->val;
int32_t ret = 0, valid = 0;
pCursor = (SIdxCursor *)taosMemoryCalloc(1, sizeof(SIdxCursor));
pCursor->pMeta = pMeta;
@ -623,12 +621,16 @@ int32_t metaFilteTableIds(SMeta *pMeta, SMetaFltParam *param, SArray *pUids) {
int32_t nKey = 0;
int32_t nTagData = 0;
if(IS_VAR_DATA_TYPE(param->type)){
nTagData = strlen(param->val);
}else{
void * tagData = NULL;
if (IS_VAR_DATA_TYPE(param->type)) {
tagData = varDataVal(param->val);
nTagData = varDataLen(param->val);
} else {
tagData = param->val;
nTagData = tDataTypes[param->type].bytes;
}
ret = metaCreateTagIdxKey(pCursor->suid, pCursor->cid, param->val, nTagData, pCursor->type,
ret = metaCreateTagIdxKey(pCursor->suid, pCursor->cid, tagData, nTagData, pCursor->type,
param->reverse ? INT64_MAX : INT64_MIN, &pKey, &nKey);
if (ret != 0) {
goto END;
@ -637,6 +639,7 @@ int32_t metaFilteTableIds(SMeta *pMeta, SMetaFltParam *param, SArray *pUids) {
if (tdbTbcMoveTo(pCursor->pCur, pKey, nKey, &cmp) < 0) {
goto END;
}
void * entryKey = NULL, *entryVal = NULL;
int32_t nEntryKey, nEntryVal;
while (1) {
@ -649,7 +652,12 @@ int32_t metaFilteTableIds(SMeta *pMeta, SMetaFltParam *param, SArray *pUids) {
int32_t cmp = (*param->filterFunc)(p->data, pKey->data, pKey->type);
if (cmp == 0) {
// match
tb_uid_t tuid = *(tb_uid_t *)(p->data + tDataTypes[pCursor->type].bytes);
tb_uid_t tuid = 0;
if (IS_VAR_DATA_TYPE(pKey->type)) {
tuid = *(tb_uid_t *)(p->data + varDataTLen(p->data));
} else {
tuid = *(tb_uid_t *)(p->data + tDataTypes[pCursor->type].bytes);
}
taosArrayPush(pUids, &tuid);
} else if (cmp == 1) {
// not match but should continue to iter

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

@ -49,9 +49,9 @@ int metaCreateSTable(SMeta *pMeta, int64_t version, SVCreateStbReq *pReq) {
int vLen = 0;
const void *pKey = NULL;
const void *pVal = NULL;
void *pBuf = NULL;
void * pBuf = NULL;
int32_t szBuf = 0;
void *p = NULL;
void * p = NULL;
SMetaReader mr = {0};
// validate req
@ -105,7 +105,7 @@ int metaDropSTable(SMeta *pMeta, int64_t verison, SVDropStbReq *pReq) {
}
// drop all child tables
TBC *pCtbIdxc = NULL;
TBC * pCtbIdxc = NULL;
SArray *pArray = taosArrayInit(8, sizeof(tb_uid_t));
tdbTbcOpen(pMeta->pCtbIdx, &pCtbIdxc, &pMeta->txn);
@ -160,8 +160,8 @@ _exit:
int metaAlterSTable(SMeta *pMeta, int64_t version, SVCreateStbReq *pReq) {
SMetaEntry oStbEntry = {0};
SMetaEntry nStbEntry = {0};
TBC *pUidIdxc = NULL;
TBC *pTbDbc = NULL;
TBC * pUidIdxc = NULL;
TBC * pTbDbc = NULL;
const void *pData;
int nData;
int64_t oversion;
@ -280,7 +280,7 @@ _err:
}
int metaDropTable(SMeta *pMeta, int64_t version, SVDropTbReq *pReq, SArray *tbUids) {
void *pData = NULL;
void * pData = NULL;
int nData = 0;
int rc = 0;
tb_uid_t uid;
@ -306,7 +306,7 @@ int metaDropTable(SMeta *pMeta, int64_t version, SVDropTbReq *pReq, SArray *tbUi
}
static int metaDropTableByUid(SMeta *pMeta, tb_uid_t uid, int *type) {
void *pData = NULL;
void * pData = NULL;
int nData = 0;
int rc = 0;
int64_t version;
@ -341,15 +341,16 @@ static int metaDropTableByUid(SMeta *pMeta, tb_uid_t uid, int *type) {
return 0;
}
static int metaAlterTableColumn(SMeta *pMeta, int64_t version, SVAlterTbReq *pAlterTbReq, STableMetaRsp *pMetaRsp) {
void * pVal = NULL;
int nVal = 0;
const void *pData = NULL;
const void * pData = NULL;
int nData = 0;
int ret = 0;
tb_uid_t uid;
int64_t oversion;
SSchema *pColumn = NULL;
SSchema * pColumn = NULL;
SMetaEntry entry = {0};
SSchemaWrapper *pSchema;
int c;
@ -499,7 +500,7 @@ _err:
static int metaUpdateTableTagVal(SMeta *pMeta, int64_t version, SVAlterTbReq *pAlterTbReq) {
SMetaEntry ctbEntry = {0};
SMetaEntry stbEntry = {0};
void *pVal = NULL;
void * pVal = NULL;
int nVal = 0;
int ret;
int c;
@ -530,7 +531,7 @@ static int metaUpdateTableTagVal(SMeta *pMeta, int64_t version, SVAlterTbReq *pA
oversion = *(int64_t *)pData;
// search table.db
TBC *pTbDbc = NULL;
TBC * pTbDbc = NULL;
SDecoder dc1 = {0};
SDecoder dc2 = {0};
@ -554,7 +555,7 @@ static int metaUpdateTableTagVal(SMeta *pMeta, int64_t version, SVAlterTbReq *pA
metaDecodeEntry(&dc2, &stbEntry);
SSchemaWrapper *pTagSchema = &stbEntry.stbEntry.schemaTag;
SSchema *pColumn = NULL;
SSchema * pColumn = NULL;
int32_t iCol = 0;
for (;;) {
pColumn = NULL;
@ -584,8 +585,8 @@ static int metaUpdateTableTagVal(SMeta *pMeta, int64_t version, SVAlterTbReq *pA
memcpy((void *)ctbEntry.ctbEntry.pTags, pAlterTbReq->pTagVal, pAlterTbReq->nTagVal);
} else {
const STag *pOldTag = (const STag *)ctbEntry.ctbEntry.pTags;
STag *pNewTag = NULL;
SArray *pTagArray = taosArrayInit(pTagSchema->nCols, sizeof(STagVal));
STag * pNewTag = NULL;
SArray * pTagArray = taosArrayInit(pTagSchema->nCols, sizeof(STagVal));
if (!pTagArray) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
@ -669,8 +670,8 @@ int metaAlterTable(SMeta *pMeta, int64_t version, SVAlterTbReq *pReq, STableMeta
static int metaSaveToTbDb(SMeta *pMeta, const SMetaEntry *pME) {
STbDbKey tbDbKey;
void *pKey = NULL;
void *pVal = NULL;
void * pKey = NULL;
void * pVal = NULL;
int kLen = 0;
int vLen = 0;
SEncoder coder = {0};
@ -753,16 +754,11 @@ static int metaUpdateCtbIdx(SMeta *pMeta, const SMetaEntry *pME) {
int metaCreateTagIdxKey(tb_uid_t suid, int32_t cid, const void *pTagData, int32_t nTagData, int8_t type, tb_uid_t uid,
STagIdxKey **ppTagIdxKey, int32_t *nTagIdxKey) {
// int32_t nTagData = 0;
// if (pTagData) {
// if (IS_VAR_DATA_TYPE(type)) {
// nTagData = varDataTLen(pTagData);
// } else {
// nTagData = tDataTypes[type].bytes;
// }
// }
if (IS_VAR_DATA_TYPE(type)) {
*nTagIdxKey = sizeof(STagIdxKey) + nTagData + VARSTR_HEADER_SIZE + sizeof(tb_uid_t);
} else {
*nTagIdxKey = sizeof(STagIdxKey) + nTagData + sizeof(tb_uid_t);
}
*ppTagIdxKey = (STagIdxKey *)taosMemoryMalloc(*nTagIdxKey);
if (*ppTagIdxKey == NULL) {
@ -774,8 +770,16 @@ int metaCreateTagIdxKey(tb_uid_t suid, int32_t cid, const void *pTagData, int32_
(*ppTagIdxKey)->cid = cid;
(*ppTagIdxKey)->isNull = (pTagData == NULL) ? 1 : 0;
(*ppTagIdxKey)->type = type;
if (nTagData) memcpy((*ppTagIdxKey)->data, pTagData, nTagData);
// refactor
if (IS_VAR_DATA_TYPE(type)) {
memcpy((*ppTagIdxKey)->data, (uint16_t *)&nTagData, VARSTR_HEADER_SIZE);
memcpy((*ppTagIdxKey)->data + VARSTR_HEADER_SIZE, pTagData, nTagData);
*(tb_uid_t *)((*ppTagIdxKey)->data + VARSTR_HEADER_SIZE + nTagData) = uid;
} else {
memcpy((*ppTagIdxKey)->data, pTagData, nTagData);
*(tb_uid_t *)((*ppTagIdxKey)->data + nTagData) = uid;
}
return 0;
}
@ -785,14 +789,14 @@ static void metaDestroyTagIdxKey(STagIdxKey *pTagIdxKey) {
}
static int metaUpdateTagIdx(SMeta *pMeta, const SMetaEntry *pCtbEntry) {
void *pData = NULL;
void * pData = NULL;
int nData = 0;
STbDbKey tbDbKey = {0};
SMetaEntry stbEntry = {0};
STagIdxKey *pTagIdxKey = NULL;
STagIdxKey * pTagIdxKey = NULL;
int32_t nTagIdxKey;
const SSchema *pTagColumn; // = &stbEntry.stbEntry.schema.pSchema[0];
const void *pTagData = NULL; //
const void * pTagData = NULL; //
int32_t nTagData = 0;
SDecoder dc = {0};
@ -849,7 +853,7 @@ static int metaUpdateTagIdx(SMeta *pMeta, const SMetaEntry *pCtbEntry) {
static int metaSaveToSkmDb(SMeta *pMeta, const SMetaEntry *pME) {
SEncoder coder = {0};
void *pVal = NULL;
void * pVal = NULL;
int vLen = 0;
int rcode = 0;
SSkmDbKey skmDbKey = {0};

79
source/libs/index/src/indexFilter.c
View File

@ -265,23 +265,24 @@ static int32_t sifExecFunction(SFunctionNode *node, SIFCtx *ctx, SIFParam *outpu
typedef int (*Filter)(void *a, void *b, int16_t dtype);
int sifGreaterThan(void *a, void *b, int16_t dtype) {
__compar_fn_t func = indexGetCompar(dtype);
__compar_fn_t func = getComparFunc(dtype, 0);
return tDoCompare(func, QUERY_GREATER_THAN, a, b);
}
int sifGreaterEqual(void *a, void *b, int16_t dtype) {
__compar_fn_t func = indexGetCompar(dtype);
__compar_fn_t func = getComparFunc(dtype, 0);
return tDoCompare(func, QUERY_GREATER_EQUAL, a, b);
}
int sifLessEqual(void *a, void *b, int16_t dtype) {
__compar_fn_t func = indexGetCompar(dtype);
__compar_fn_t func = getComparFunc(dtype, 0);
return tDoCompare(func, QUERY_LESS_EQUAL, a, b);
}
int sifLessThan(void *a, void *b, int16_t dtype) {
__compar_fn_t func = indexGetCompar(dtype);
__compar_fn_t func = getComparFunc(dtype, 0);
return (int)tDoCompare(func, QUERY_LESS_THAN, a, b);
}
int sifEqual(void *a, void *b, int16_t dtype) {
__compar_fn_t func = indexGetCompar(dtype);
__compar_fn_t func = getComparFunc(dtype, 0);
//__compar_fn_t func = indexGetCompar(dtype);
return (int)tDoCompare(func, QUERY_TERM, a, b);
}
static Filter sifGetFilterFunc(EIndexQueryType type, bool *reverse) {
@ -397,38 +398,62 @@ static int32_t sifDefaultFunc(SIFParam *left, SIFParam *right, SIFParam *output)
return TSDB_CODE_QRY_INVALID_INPUT;
}
static sif_func_t sifGetOperFn(int32_t funcId) {
static int32_t sifGetOperFn(int32_t funcId, sif_func_t *func, SIdxFltStatus *status) {
// impl later
*status = SFLT_ACCURATE_INDEX;
switch (funcId) {
case OP_TYPE_GREATER_THAN:
return sifGreaterThanFunc;
*func = sifGreaterThanFunc;
return 0;
case OP_TYPE_GREATER_EQUAL:
return sifGreaterEqualFunc;
*func = sifGreaterEqualFunc;
return 0;
case OP_TYPE_LOWER_THAN:
return sifLessThanFunc;
*func = sifLessThanFunc;
return 0;
case OP_TYPE_LOWER_EQUAL:
return sifLessEqualFunc;
*func = sifLessEqualFunc;
return 0;
case OP_TYPE_EQUAL:
return sifEqualFunc;
*func = sifEqualFunc;
return 0;
case OP_TYPE_NOT_EQUAL:
return sifNotEqualFunc;
*status = SFLT_NOT_INDEX;
*func = sifNotEqualFunc;
return 0;
case OP_TYPE_IN:
return sifInFunc;
*status = SFLT_NOT_INDEX;
*func = sifInFunc;
return 0;
case OP_TYPE_NOT_IN:
return sifNotInFunc;
*status = SFLT_NOT_INDEX;
*func = sifNotInFunc;
return 0;
case OP_TYPE_LIKE:
return sifLikeFunc;
*status = SFLT_NOT_INDEX;
*func = sifLikeFunc;
return 0;
case OP_TYPE_NOT_LIKE:
return sifNotLikeFunc;
*status = SFLT_NOT_INDEX;
*func = sifNotLikeFunc;
return 0;
case OP_TYPE_MATCH:
return sifMatchFunc;
*status = SFLT_NOT_INDEX;
*func = sifMatchFunc;
return 0;
case OP_TYPE_NMATCH:
return sifNotMatchFunc;
*status = SFLT_NOT_INDEX;
*func = sifNotMatchFunc;
return 0;
default:
return sifNullFunc;
*status = SFLT_NOT_INDEX;
*func = sifNullFunc;
return 0;
}
return sifNullFunc;
return 0;
}
// typedef struct filterFuncDict {
static int32_t sifExecOper(SOperatorNode *node, SIFCtx *ctx, SIFParam *output) {
int32_t code = 0;
int32_t nParam = sifGetOperParamNum(node->opType);
@ -441,18 +466,14 @@ static int32_t sifExecOper(SOperatorNode *node, SIFCtx *ctx, SIFParam *output) {
// ugly code, refactor later
output->arg = ctx->arg;
sif_func_t operFn = sifGetOperFn(node->opType);
if (ctx->noExec && operFn == NULL) {
output->status = SFLT_NOT_INDEX;
} else {
output->status = SFLT_ACCURATE_INDEX;
}
sif_func_t operFn = sifNullFunc;
code = sifGetOperFn(node->opType, &operFn, &output->status);
if (ctx->noExec) {
SIF_RET(code);
}
} else {
return operFn(&params[0], nParam > 1 ? &params[1] : NULL, output);
}
_return:
taosMemoryFree(params);
SIF_RET(code);
@ -477,7 +498,7 @@ static int32_t sifExecLogic(SLogicConditionNode *node, SIFCtx *ctx, SIFParam *ou
} else if (node->condType == LOGIC_COND_TYPE_OR) {
taosArrayAddAll(output->result, params[m].result);
} else if (node->condType == LOGIC_COND_TYPE_NOT) {
taosArrayAddAll(output->result, params[m].result);
// taosArrayAddAll(output->result, params[m].result);
}
}
} else {

1
source/libs/transport/src/transCli.c
View File

@ -1151,7 +1151,6 @@ void transSetDefaultAddr(void* ahandle, const char* ip, const char* fqdn) {
SCliThrdObj* thrd = ((SCliObj*)pTransInst->tcphandle)->pThreadObj[i];
tDebug("update epset at thread:%d, threadID:%" PRId64 "", i, thrd->thread);
tsem_t* pSem = pCtx->pSem;
transSendAsync(thrd->asyncPool, &(cliMsg->q));
}
}