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

support column filter

This commit is contained in:
dapan1121 2021-06-17 16:20:01 +08:00
parent 74b1797cfa
commit bc65d9eba7
12 changed files with 727 additions and 95 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
src/client/inc/tscUtil.h
View File

@ -338,6 +338,7 @@ char* strdup_throw(const char* str);
bool vgroupInfoIdentical(SNewVgroupInfo *pExisted, SVgroupMsg* src); bool vgroupInfoIdentical(SNewVgroupInfo *pExisted, SVgroupMsg* src);
SNewVgroupInfo createNewVgroupInfo(SVgroupMsg *pVgroupMsg); SNewVgroupInfo createNewVgroupInfo(SVgroupMsg *pVgroupMsg);
SCond* tsGetTableFilter(SArray* filters, uint64_t uid);
#ifdef __cplusplus #ifdef __cplusplus
} }

162
src/client/src/tscSQLParser.c
View File

@ -137,6 +137,7 @@ static int32_t exprTreeFromSqlExpr(SSqlCmd* pCmd, tExprNode **pExpr, const tSqlE
static bool validateDebugFlag(int32_t v); static bool validateDebugFlag(int32_t v);
static int32_t checkQueryRangeForFill(SSqlCmd* pCmd, SQueryInfo* pQueryInfo); static int32_t checkQueryRangeForFill(SSqlCmd* pCmd, SQueryInfo* pQueryInfo);
static int32_t loadAllTableMeta(SSqlObj* pSql, struct SSqlInfo* pInfo); static int32_t loadAllTableMeta(SSqlObj* pSql, struct SSqlInfo* pInfo);
static tSqlExpr* extractExprForSTable(SSqlCmd* pCmd, tSqlExpr** pExpr, SQueryInfo* pQueryInfo, int32_t tableIndex);
static bool isTimeWindowQuery(SQueryInfo* pQueryInfo) { static bool isTimeWindowQuery(SQueryInfo* pQueryInfo) {
return pQueryInfo->interval.interval > 0 || pQueryInfo->sessionWindow.gap > 0; return pQueryInfo->interval.interval > 0 || pQueryInfo->sessionWindow.gap > 0;
@ -3458,10 +3459,10 @@ static int32_t tablenameCondToString(tSqlExpr* pExpr, SStringBuilder* sb) {
} }
enum { enum {
TSQL_EXPR_TS = 0, TSQL_EXPR_TS = 1,
TSQL_EXPR_TAG = 1, TSQL_EXPR_TAG = 2,
TSQL_EXPR_COLUMN = 2, TSQL_EXPR_COLUMN = 4,
TSQL_EXPR_TBNAME = 3, TSQL_EXPR_TBNAME = 8,
}; };
static int32_t extractColumnFilterInfo(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, SColumnIndex* pIndex, tSqlExpr* pExpr, int32_t sqlOptr) { static int32_t extractColumnFilterInfo(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, SColumnIndex* pIndex, tSqlExpr* pExpr, int32_t sqlOptr) {
@ -3568,6 +3569,63 @@ static int32_t getTablenameCond(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, tSqlExpr*
return ret; return ret;
} }
static int32_t getColQueryCondExpr(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, tSqlExpr** pExpr) {
int32_t ret = TSDB_CODE_SUCCESS;
for (int32_t i = 0; i < pQueryInfo->numOfTables; ++i) {
tSqlExpr* p1 = extractExprForSTable(pCmd, pExpr, pQueryInfo, i);
if (p1 == NULL) { // no query condition on this table
continue;
}
tExprNode* p = NULL;
//SFilterInfo colFilter = {0};
SArray* colList = taosArrayInit(10, sizeof(SColIndex));
ret = exprTreeFromSqlExpr(pCmd, &p, p1, pQueryInfo, colList, NULL);
//if (ret == TSDB_CODE_SUCCESS) {
// ret = filterInitFromTree(p, &colFilter, (int32_t)taosArrayGetSize(colList));
//}
SBufferWriter bw = tbufInitWriter(NULL, false);
TRY(0) {
exprTreeToBinary(&bw, p);
} CATCH(code) {
tbufCloseWriter(&bw);
UNUSED(code);
// TODO: more error handling
} END_TRY
// add to required table column list
STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, i);
int64_t uid = pTableMetaInfo->pTableMeta->id.uid;
SCond cond = {
.uid = uid,
.len = (int32_t)(tbufTell(&bw)),
.cond = tbufGetData(&bw, true)
};
if (pQueryInfo->colCond == NULL) {
pQueryInfo->colCond = taosArrayInit(2, sizeof(SCond));
}
taosArrayPush(pQueryInfo->colCond, &cond);
tSqlExprDestroy(p1);
tExprTreeDestroy(p, NULL);
if (ret) {
break;
}
}
return ret;
}
static int32_t getColumnQueryCondInfo(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, tSqlExpr* pExpr, int32_t relOptr) { static int32_t getColumnQueryCondInfo(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, tSqlExpr* pExpr, int32_t relOptr) {
if (pExpr == NULL) { if (pExpr == NULL) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
@ -3967,6 +4025,17 @@ static int32_t setExprToCond(tSqlExpr** parent, tSqlExpr* pExpr, const char* msg
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static int32_t setNormalExprToCond(tSqlExpr** parent, tSqlExpr* pExpr, int32_t parentOptr) {
if (*parent != NULL) {
*parent = tSqlExprCreate((*parent), pExpr, parentOptr);
} else {
*parent = pExpr;
}
return TSDB_CODE_SUCCESS;
}
static int32_t validateNullExpr(tSqlExpr* pExpr, char* msgBuf) { static int32_t validateNullExpr(tSqlExpr* pExpr, char* msgBuf) {
const char* msg = "only support is [not] null"; const char* msg = "only support is [not] null";
@ -4001,7 +4070,7 @@ static int32_t validateLikeExpr(tSqlExpr* pExpr, STableMeta* pTableMeta, int32_t
} }
static int32_t handleExprInQueryCond(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, tSqlExpr** pExpr, SCondExpr* pCondExpr, static int32_t handleExprInQueryCond(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, tSqlExpr** pExpr, SCondExpr* pCondExpr,
int32_t* type, int32_t parentOptr) { int32_t* type, int32_t parentOptr, tSqlExpr** columnExpr) {
const char* msg1 = "table query cannot use tags filter"; const char* msg1 = "table query cannot use tags filter";
const char* msg2 = "illegal column name"; const char* msg2 = "illegal column name";
const char* msg3 = "only one query time range allowed"; const char* msg3 = "only one query time range allowed";
@ -4098,7 +4167,7 @@ static int32_t handleExprInQueryCond(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, tSql
} }
*pExpr = NULL; // remove this expression *pExpr = NULL; // remove this expression
*type = TSQL_EXPR_TS; *type |= TSQL_EXPR_TAG;
} else if (index.columnIndex >= tscGetNumOfColumns(pTableMeta) || index.columnIndex == TSDB_TBNAME_COLUMN_INDEX) { } else if (index.columnIndex >= tscGetNumOfColumns(pTableMeta) || index.columnIndex == TSDB_TBNAME_COLUMN_INDEX) {
// query on tags, check for tag query condition // query on tags, check for tag query condition
if (UTIL_TABLE_IS_NORMAL_TABLE(pTableMetaInfo)) { if (UTIL_TABLE_IS_NORMAL_TABLE(pTableMetaInfo)) {
@ -4123,7 +4192,7 @@ static int32_t handleExprInQueryCond(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, tSql
return invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg6); return invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg6);
} }
*type = TSQL_EXPR_TBNAME; *type |= TSQL_EXPR_TAG;
*pExpr = NULL; *pExpr = NULL;
} else { } else {
if (pRight != NULL && pRight->tokenId == TK_ID) { // join on tag columns for stable query if (pRight != NULL && pRight->tokenId == TK_ID) { // join on tag columns for stable query
@ -4140,17 +4209,17 @@ static int32_t handleExprInQueryCond(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, tSql
// *pExpr, NULL, parentOptr); // *pExpr, NULL, parentOptr);
} }
*type = TSQL_EXPR_TAG; *type |= TSQL_EXPR_TAG;
} }
} else { // query on other columns } else { // query on other columns
*type = TSQL_EXPR_COLUMN; *type |= TSQL_EXPR_COLUMN;
if (pRight->tokenId == TK_ID) { // other column cannot be served as the join column if (pRight->tokenId == TK_ID) { // other column cannot be served as the join column
return invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg5); return invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg5);
} }
ret = setExprToCond(&pCondExpr->pColumnCond, *pExpr, NULL, parentOptr, pQueryInfo->msg); ret = setNormalExprToCond(columnExpr, *pExpr, parentOptr);
*pExpr = NULL; // remove it from expr tree *pExpr = NULL; // remove it from expr tree
} }
@ -4158,12 +4227,16 @@ static int32_t handleExprInQueryCond(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, tSql
} }
int32_t getQueryCondExpr(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, tSqlExpr** pExpr, SCondExpr* pCondExpr, int32_t getQueryCondExpr(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, tSqlExpr** pExpr, SCondExpr* pCondExpr,
int32_t* type, int32_t parentOptr) { int32_t* type, int32_t parentOptr, tSqlExpr** columnExpr) {
if (pExpr == NULL) { if (pExpr == NULL) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
const char* msg1 = "query condition between different columns must use 'AND'"; tSqlExpr *columnLeft = NULL;
tSqlExpr *columnRight = NULL;
int32_t ret = 0;
const char* msg1 = "query condition between columns and tags/timestamp must use 'AND'";
if ((*pExpr)->flags & (1 << EXPR_FLAG_TS_ERROR)) { if ((*pExpr)->flags & (1 << EXPR_FLAG_TS_ERROR)) {
return TSDB_CODE_TSC_INVALID_OPERATION; return TSDB_CODE_TSC_INVALID_OPERATION;
@ -4176,45 +4249,66 @@ int32_t getQueryCondExpr(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, tSqlExpr** pExpr
return TSDB_CODE_TSC_INVALID_OPERATION; return TSDB_CODE_TSC_INVALID_OPERATION;
} }
int32_t leftType = -1; int32_t leftType = 0;
int32_t rightType = -1; int32_t rightType = 0;
if (!tSqlExprIsParentOfLeaf(*pExpr)) { if (!tSqlExprIsParentOfLeaf(*pExpr)) {
int32_t ret = getQueryCondExpr(pCmd, pQueryInfo, &(*pExpr)->pLeft, pCondExpr, &leftType, (*pExpr)->tokenId); ret = getQueryCondExpr(pCmd, pQueryInfo, &(*pExpr)->pLeft, pCondExpr, &leftType, (*pExpr)->tokenId, &columnLeft);
if (ret != TSDB_CODE_SUCCESS) { if (ret != TSDB_CODE_SUCCESS) {
return ret; goto err_ret;
} }
ret = getQueryCondExpr(pCmd, pQueryInfo, &(*pExpr)->pRight, pCondExpr, &rightType, (*pExpr)->tokenId); ret = getQueryCondExpr(pCmd, pQueryInfo, &(*pExpr)->pRight, pCondExpr, &rightType, (*pExpr)->tokenId, &columnRight);
if (ret != TSDB_CODE_SUCCESS) { if (ret != TSDB_CODE_SUCCESS) {
return ret; goto err_ret;
} }
/* /*
* if left child and right child do not belong to the same group, the sub * if left child and right child do not belong to the same group, the sub
* expression is not valid for parent node, it must be TK_AND operator. * expression is not valid for parent node, it must be TK_AND operator.
*/ */
if (leftType != rightType) { if (((leftType != rightType) || (leftType == (TSQL_EXPR_COLUMN|TSQL_EXPR_TAG ))) && (*pExpr)->tokenId == TK_OR) {
if ((*pExpr)->tokenId == TK_OR && (leftType + rightType != TSQL_EXPR_TBNAME + TSQL_EXPR_TAG)) { ret = invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg1);
return invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg1); goto err_ret;
}
} }
*type = rightType; if (columnLeft && columnRight) {
setNormalExprToCond(&columnLeft, columnRight, (*pExpr)->tokenId);
*columnExpr = columnLeft;
} else {
*columnExpr = columnLeft ? columnLeft : columnRight;
}
*type = leftType|rightType;
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
exchangeExpr(*pExpr); exchangeExpr(*pExpr);
if (pLeft->tokenId == TK_ID && pRight->tokenId == TK_TIMESTAMP && (pRight->flags & (1 << EXPR_FLAG_TIMESTAMP_VAR))) { if (pLeft->tokenId == TK_ID && pRight->tokenId == TK_TIMESTAMP && (pRight->flags & (1 << EXPR_FLAG_TIMESTAMP_VAR))) {
return TSDB_CODE_TSC_INVALID_OPERATION; ret = TSDB_CODE_TSC_INVALID_OPERATION;
goto err_ret;
} }
if ((pLeft->flags & (1 << EXPR_FLAG_TS_ERROR)) || (pRight->flags & (1 << EXPR_FLAG_TS_ERROR))) { if ((pLeft->flags & (1 << EXPR_FLAG_TS_ERROR)) || (pRight->flags & (1 << EXPR_FLAG_TS_ERROR))) {
return TSDB_CODE_TSC_INVALID_OPERATION; ret = TSDB_CODE_TSC_INVALID_OPERATION;
goto err_ret;
} }
return handleExprInQueryCond(pCmd, pQueryInfo, pExpr, pCondExpr, type, parentOptr); ret = handleExprInQueryCond(pCmd, pQueryInfo, pExpr, pCondExpr, type, parentOptr, columnExpr);
if (ret) {
goto err_ret;
}
return TSDB_CODE_SUCCESS;
err_ret:
tSqlExprDestroy(columnLeft);
tSqlExprDestroy(columnRight);
return ret;
} }
static void doExtractExprForSTable(SSqlCmd* pCmd, tSqlExpr** pExpr, SQueryInfo* pQueryInfo, tSqlExpr** pOut, int32_t tableIndex) { static void doExtractExprForSTable(SSqlCmd* pCmd, tSqlExpr** pExpr, SQueryInfo* pQueryInfo, tSqlExpr** pOut, int32_t tableIndex) {
@ -4595,9 +4689,9 @@ static int32_t getTagQueryCondExpr(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, SCondE
SArray* colList = taosArrayInit(10, sizeof(SColIndex)); SArray* colList = taosArrayInit(10, sizeof(SColIndex));
ret = exprTreeFromSqlExpr(pCmd, &p, p1, pQueryInfo, colList, NULL); ret = exprTreeFromSqlExpr(pCmd, &p, p1, pQueryInfo, colList, NULL);
if (ret == TSDB_CODE_SUCCESS) { //if (ret == TSDB_CODE_SUCCESS) {
ret = filterInitFromTree(p, &pQueryInfo->colFilter, (int32_t)taosArrayGetSize(colList)); // ret = filterInitFromTree(p, &pQueryInfo->tagFilter, (int32_t)taosArrayGetSize(colList));
} //}
SBufferWriter bw = tbufInitWriter(NULL, false); SBufferWriter bw = tbufInitWriter(NULL, false);
@ -4632,7 +4726,7 @@ static int32_t getTagQueryCondExpr(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, SCondE
} }
tSqlExprDestroy(p1); tSqlExprDestroy(p1);
//tExprTreeDestroy(p, NULL); TODO tExprTreeDestroy(p, NULL); //TODO
taosArrayDestroy(colList); taosArrayDestroy(colList);
if (pQueryInfo->tagCond.pCond != NULL && taosArrayGetSize(pQueryInfo->tagCond.pCond) > 0 && !UTIL_TABLE_IS_SUPER_TABLE(pTableMetaInfo)) { if (pQueryInfo->tagCond.pCond != NULL && taosArrayGetSize(pQueryInfo->tagCond.pCond) > 0 && !UTIL_TABLE_IS_SUPER_TABLE(pTableMetaInfo)) {
@ -4763,11 +4857,12 @@ int32_t validateWhereNode(SQueryInfo* pQueryInfo, tSqlExpr** pExpr, SSqlObj* pSq
} }
int32_t type = 0; int32_t type = 0;
if ((ret = getQueryCondExpr(&pSql->cmd, pQueryInfo, pExpr, &condExpr, &type, (*pExpr)->tokenId)) != TSDB_CODE_SUCCESS) { if ((ret = getQueryCondExpr(&pSql->cmd, pQueryInfo, pExpr, &condExpr, &type, (*pExpr)->tokenId, &condExpr.pColumnCond)) != TSDB_CODE_SUCCESS) {
return ret; return ret;
} }
tSqlExprCompact(pExpr); tSqlExprCompact(pExpr);
tSqlExprCompact(&condExpr.pColumnCond);
// after expression compact, the expression tree is only include tag query condition // after expression compact, the expression tree is only include tag query condition
condExpr.pTagCond = (*pExpr); condExpr.pTagCond = (*pExpr);
@ -4792,6 +4887,11 @@ int32_t validateWhereNode(SQueryInfo* pQueryInfo, tSqlExpr** pExpr, SSqlObj* pSq
goto PARSE_WHERE_EXIT; goto PARSE_WHERE_EXIT;
} }
if ((ret = getColQueryCondExpr(&pSql->cmd, pQueryInfo, &condExpr.pColumnCond)) != TSDB_CODE_SUCCESS) {
goto PARSE_WHERE_EXIT;
}
// 5. other column query condition // 5. other column query condition
if ((ret = getColumnQueryCondInfo(&pSql->cmd, pQueryInfo, condExpr.pColumnCond, TK_AND)) != TSDB_CODE_SUCCESS) { if ((ret = getColumnQueryCondInfo(&pSql->cmd, pQueryInfo, condExpr.pColumnCond, TK_AND)) != TSDB_CODE_SUCCESS) {
goto PARSE_WHERE_EXIT; goto PARSE_WHERE_EXIT;

10
src/client/src/tscServer.c
View File

@ -888,6 +888,16 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
serializeColFilterInfo(pCol->flist.filterInfo, pCol->flist.numOfFilters, &pMsg); serializeColFilterInfo(pCol->flist.filterInfo, pCol->flist.numOfFilters, &pMsg);
} }
if (pQueryInfo->colCond && taosArrayGetSize(pQueryInfo->colCond) > 0) {
SCond *pCond = tsGetTableFilter(pQueryInfo->colCond, pTableMeta->id.uid);
if (pCond != NULL && pCond->cond != NULL) {
pQueryMsg->colCondLen = htons(pCond->len);
memcpy(pMsg, pCond->cond, pCond->len);
pMsg += pCond->len;
}
}
for (int32_t i = 0; i < query.numOfOutput; ++i) { for (int32_t i = 0; i < query.numOfOutput; ++i) {
code = serializeSqlExpr(&query.pExpr1[i].base, pTableMetaInfo, &pMsg, pSql->self, true); code = serializeSqlExpr(&query.pExpr1[i].base, pTableMetaInfo, &pMsg, pSql->self, true);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {

18
src/client/src/tscUtil.c
View File

@ -118,6 +118,24 @@ SCond* tsGetSTableQueryCond(STagCond* pTagCond, uint64_t uid) {
return NULL; return NULL;
} }
SCond* tsGetTableFilter(SArray* filters, uint64_t uid) {
if (filters == NULL) {
return NULL;
}
size_t size = taosArrayGetSize(filters);
for (int32_t i = 0; i < size; ++i) {
SCond* cond = taosArrayGet(filters, i);
if (uid == cond->uid) {
return cond;
}
}
return NULL;
}
void tsSetSTableQueryCond(STagCond* pTagCond, uint64_t uid, SBufferWriter* bw) { void tsSetSTableQueryCond(STagCond* pTagCond, uint64_t uid, SBufferWriter* bw) {
if (tbufTell(bw) == 0) { if (tbufTell(bw) == 0) {
return; return;

1
src/inc/taosmsg.h
View File

@ -484,6 +484,7 @@ typedef struct {
SInterval interval; SInterval interval;
SSessionWindow sw; // session window SSessionWindow sw; // session window
uint16_t tagCondLen; // tag length in current query uint16_t tagCondLen; // tag length in current query
uint16_t colCondLen; // column length in current query
uint32_t tbnameCondLen; // table name filter condition string length uint32_t tbnameCondLen; // table name filter condition string length
int16_t numOfGroupCols; // num of group by columns int16_t numOfGroupCols; // num of group by columns
int16_t orderByIdx; int16_t orderByIdx;

10
src/query/inc/qExecutor.h
View File

@ -225,7 +225,9 @@ typedef struct SQueryAttr {
int32_t numOfFilterCols; int32_t numOfFilterCols;
int64_t* fillVal; int64_t* fillVal;
SOrderedPrjQueryInfo prjInfo; // limit value for each vgroup, only available in global order projection query. SOrderedPrjQueryInfo prjInfo; // limit value for each vgroup, only available in global order projection query.
SSingleColumnFilterInfo* pFilterInfo; SSingleColumnFilterInfo* pFilterInfo;
SFilterInfo *pFilters;
void* tsdb; void* tsdb;
SMemRef memRef; SMemRef memRef;
@ -352,6 +354,7 @@ typedef struct SQInfo {
typedef struct SQueryParam { typedef struct SQueryParam {
char *sql; char *sql;
char *tagCond; char *tagCond;
char *colCond;
char *tbnameCond; char *tbnameCond;
char *prevResult; char *prevResult;
SArray *pTableIdList; SArray *pTableIdList;
@ -360,6 +363,8 @@ typedef struct SQueryParam {
SExprInfo *pExprs; SExprInfo *pExprs;
SExprInfo *pSecExprs; SExprInfo *pSecExprs;
SFilterInfo *pFilters;
SColIndex *pGroupColIndex; SColIndex *pGroupColIndex;
SColumnInfo *pTagColumnInfo; SColumnInfo *pTagColumnInfo;
SGroupbyExpr *pGroupbyExpr; SGroupbyExpr *pGroupbyExpr;
@ -540,6 +545,7 @@ SSDataBlock* doSLimit(void* param, bool* newgroup);
int32_t doCreateFilterInfo(SColumnInfo* pCols, int32_t numOfCols, int32_t numOfFilterCols, SSingleColumnFilterInfo** pFilterInfo, uint64_t qId); int32_t doCreateFilterInfo(SColumnInfo* pCols, int32_t numOfCols, int32_t numOfFilterCols, SSingleColumnFilterInfo** pFilterInfo, uint64_t qId);
void doSetFilterColumnInfo(SSingleColumnFilterInfo* pFilterInfo, int32_t numOfFilterCols, SSDataBlock* pBlock); void doSetFilterColumnInfo(SSingleColumnFilterInfo* pFilterInfo, int32_t numOfFilterCols, SSDataBlock* pBlock);
void doSetFilterColInfo(SFilterInfo *pFilters, SSDataBlock* pBlock);
bool doFilterDataBlock(SSingleColumnFilterInfo* pFilterInfo, int32_t numOfFilterCols, int32_t numOfRows, int8_t* p); bool doFilterDataBlock(SSingleColumnFilterInfo* pFilterInfo, int32_t numOfFilterCols, int32_t numOfRows, int8_t* p);
void doCompactSDataBlock(SSDataBlock* pBlock, int32_t numOfRows, int8_t* p); void doCompactSDataBlock(SSDataBlock* pBlock, int32_t numOfRows, int8_t* p);
@ -559,9 +565,11 @@ int32_t createQueryFunc(SQueriedTableInfo* pTableInfo, int32_t numOfOutput, SExp
int32_t createIndirectQueryFuncExprFromMsg(SQueryTableMsg *pQueryMsg, int32_t numOfOutput, SExprInfo **pExprInfo, int32_t createIndirectQueryFuncExprFromMsg(SQueryTableMsg *pQueryMsg, int32_t numOfOutput, SExprInfo **pExprInfo,
SSqlExpr **pExpr, SExprInfo *prevExpr); SSqlExpr **pExpr, SExprInfo *prevExpr);
int32_t createQueryFilter(char *data, uint16_t len, SFilterInfo** pFilters);
SGroupbyExpr *createGroupbyExprFromMsg(SQueryTableMsg *pQueryMsg, SColIndex *pColIndex, int32_t *code); SGroupbyExpr *createGroupbyExprFromMsg(SQueryTableMsg *pQueryMsg, SColIndex *pColIndex, int32_t *code);
SQInfo *createQInfoImpl(SQueryTableMsg *pQueryMsg, SGroupbyExpr *pGroupbyExpr, SExprInfo *pExprs, SQInfo *createQInfoImpl(SQueryTableMsg *pQueryMsg, SGroupbyExpr *pGroupbyExpr, SExprInfo *pExprs,
SExprInfo *pSecExprs, STableGroupInfo *pTableGroupInfo, SColumnInfo* pTagCols, int32_t vgId, char* sql, uint64_t *qId); SExprInfo *pSecExprs, STableGroupInfo *pTableGroupInfo, SColumnInfo* pTagCols, SFilterInfo* pFilters, int32_t vgId, char* sql, uint64_t *qId);
int32_t initQInfo(STsBufInfo* pTsBufInfo, void* tsdb, void* sourceOptr, SQInfo* pQInfo, SQueryParam* param, char* start, int32_t initQInfo(STsBufInfo* pTsBufInfo, void* tsdb, void* sourceOptr, SQInfo* pQInfo, SQueryParam* param, char* start,
int32_t prevResultLen, void* merger); int32_t prevResultLen, void* merger);

34
src/query/inc/qFilter.h
View File

@ -22,6 +22,10 @@ extern "C" {
#include "texpr.h" #include "texpr.h"
#define FILTER_DEFAULT_UNIT_SIZE 4
#define FILTER_DEFAULT_FIELD_SIZE 4
#define FILTER_DEFAULT_GROUP_UNIT_SIZE 2
enum { enum {
F_FIELD_COLUMN = 0, F_FIELD_COLUMN = 0,
F_FIELD_VALUE, F_FIELD_VALUE,
@ -41,10 +45,16 @@ typedef struct SFilterField {
} SFilterField; } SFilterField;
typedef struct SFilterFields { typedef struct SFilterFields {
uint16_t size;
uint16_t num; uint16_t num;
SFilterField *fields; SFilterField *fields;
} SFilterFields; } SFilterFields;
typedef struct SFilterFieldId {
uint16_t type;
uint16_t idx;
} SFilterFieldId;
typedef struct SFilterGroup { typedef struct SFilterGroup {
uint16_t unitNum; uint16_t unitNum;
uint16_t *unitIdxs; uint16_t *unitIdxs;
@ -58,14 +68,15 @@ typedef struct SFilterCompare {
typedef struct SFilterUnit { typedef struct SFilterUnit {
SFilterCompare compare; SFilterCompare compare;
SFilterField *left; SFilterFieldId left;
SFilterField *right; SFilterFieldId right;
} SFilterUnit; } SFilterUnit;
typedef struct SFilterInfo { typedef struct SFilterInfo {
uint16_t unitSize;
uint16_t unitNum; uint16_t unitNum;
uint16_t groupNum; uint16_t groupNum;
SFilterFields fileds[F_FIELD_MAX]; SFilterFields fields[F_FIELD_MAX];
SFilterGroup *groups; SFilterGroup *groups;
SFilterUnit *units; SFilterUnit *units;
uint8_t *unitRes; // result uint8_t *unitRes; // result
@ -81,11 +92,24 @@ typedef struct SFilterInfo {
#define CHK_LRETV(c,...) do { if (c) { qError(__VA_ARGS__); return; } } while (0) #define CHK_LRETV(c,...) do { if (c) { qError(__VA_ARGS__); return; } } while (0)
#define CHK_LRET(c, r,...) do { if (c) { qError(__VA_ARGS__); return r; } } while (0) #define CHK_LRET(c, r,...) do { if (c) { qError(__VA_ARGS__); return r; } } while (0)
#define FILTER_GET_FIELD(i, id) (&((i)->fields[(id).type].fields[(id).idx]))
#define FILTER_GET_COL_FIELD_DATA(fi, ri) ((fi)->data + ((SSchema *)((fi)->desc))->bytes * (ri))
#define FILTER_GET_VAL_FIELD_DATA(fi) (&((tVariant *)((fi)->desc))->i64)
#define FILTER_UNIT_CLR_F(i) memset((i)->unitFlags, 0, (i)->unitNum * sizeof(*info->unitFlags))
#define FILTER_UNIT_SET_F(i, idx) (i)->unitFlags[idx] = 1
#define FILTER_UNIT_GET_F(i, idx) ((i)->unitFlags[idx])
#define FILTER_UNIT_GET_R(i, idx) ((i)->unitRes[idx])
#define FILTER_UNIT_SET_R(i, idx, v) (i)->unitRes[idx] = (v)
typedef int32_t(*filter_desc_compare_func)(const void *, const void *); typedef int32_t(*filter_desc_compare_func)(const void *, const void *);
extern int32_t filterInitFromTree(tExprNode* tree, SFilterInfo *info, int32_t colSize); extern int32_t filterInitFromTree(tExprNode* tree, SFilterInfo **pinfo);
extern bool filterExecute(SFilterInfo *info, int32_t numOfRows, int8_t* p);
extern int32_t filterSetColData(SFilterInfo *info, int16_t colId, void *data);
#ifdef __cplusplus #ifdef __cplusplus
} }

7
src/query/inc/qTableMeta.h
View File

@ -89,6 +89,11 @@ typedef struct STableMetaInfo {
struct SQInfo; // global merge operator struct SQInfo; // global merge operator
struct SQueryAttr; // query object struct SQueryAttr; // query object
typedef struct STableFilter {
uint64_t uid;
SFilterInfo info;
} STableFilter;
typedef struct SQueryInfo { typedef struct SQueryInfo {
int16_t command; // the command may be different for each subclause, so keep it seperately. int16_t command; // the command may be different for each subclause, so keep it seperately.
uint32_t type; // query/insert type uint32_t type; // query/insert type
@ -106,7 +111,7 @@ typedef struct SQueryInfo {
SLimitVal slimit; SLimitVal slimit;
STagCond tagCond; STagCond tagCond;
SFilterInfo colFilter; SArray * colCond;
SOrderVal order; SOrderVal order;
int16_t fillType; // final result fill type int16_t fillType; // final result fill type

93
src/query/src/qExecutor.c
View File

@ -2557,6 +2557,49 @@ void filterRowsInDataBlock(SQueryRuntimeEnv* pRuntimeEnv, SSingleColumnFilterInf
tfree(p); tfree(p);
} }
void filterColRowsInDataBlock(SQueryRuntimeEnv* pRuntimeEnv, SSDataBlock* pBlock, bool ascQuery) {
int32_t numOfRows = pBlock->info.rows;
int8_t *p = calloc(numOfRows, sizeof(int8_t));
bool all = true;
if (pRuntimeEnv->pTsBuf != NULL) {
SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, 0);
TSKEY* k = (TSKEY*) pColInfoData->pData;
for (int32_t i = 0; i < numOfRows; ++i) {
int32_t offset = ascQuery? i:(numOfRows - i - 1);
int32_t ret = doTSJoinFilter(pRuntimeEnv, k[offset], ascQuery);
if (ret == TS_JOIN_TAG_NOT_EQUALS) {
break;
} else if (ret == TS_JOIN_TS_NOT_EQUALS) {
all = false;
continue;
} else {
assert(ret == TS_JOIN_TS_EQUAL);
p[offset] = true;
}
if (!tsBufNextPos(pRuntimeEnv->pTsBuf)) {
break;
}
}
// save the cursor status
pRuntimeEnv->current->cur = tsBufGetCursor(pRuntimeEnv->pTsBuf);
} else {
all = filterExecute(pRuntimeEnv->pQueryAttr->pFilters, numOfRows, p);
}
if (!all) {
doCompactSDataBlock(pBlock, numOfRows, p);
}
tfree(p);
}
static SColumnInfo* doGetTagColumnInfoById(SColumnInfo* pTagColList, int32_t numOfTags, int16_t colId); static SColumnInfo* doGetTagColumnInfoById(SColumnInfo* pTagColList, int32_t numOfTags, int16_t colId);
static void doSetTagValueInParam(void* pTable, int32_t tagColId, tVariant *tag, int16_t type, int16_t bytes); static void doSetTagValueInParam(void* pTable, int32_t tagColId, tVariant *tag, int16_t type, int16_t bytes);
@ -2597,6 +2640,15 @@ void doSetFilterColumnInfo(SSingleColumnFilterInfo* pFilterInfo, int32_t numOfFi
} }
} }
void doSetFilterColInfo(SFilterInfo * pFilters, SSDataBlock* pBlock) {
for (int32_t j = 0; j < pBlock->info.numOfCols; ++j) {
SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, j);
filterSetColData(pFilters, pColInfo->info.colId, pColInfo->pData);
}
}
int32_t loadDataBlockOnDemand(SQueryRuntimeEnv* pRuntimeEnv, STableScanInfo* pTableScanInfo, SSDataBlock* pBlock, int32_t loadDataBlockOnDemand(SQueryRuntimeEnv* pRuntimeEnv, STableScanInfo* pTableScanInfo, SSDataBlock* pBlock,
uint32_t* status) { uint32_t* status) {
*status = BLK_DATA_NO_NEEDED; *status = BLK_DATA_NO_NEEDED;
@ -2735,9 +2787,9 @@ int32_t loadDataBlockOnDemand(SQueryRuntimeEnv* pRuntimeEnv, STableScanInfo* pTa
return terrno; return terrno;
} }
doSetFilterColumnInfo(pQueryAttr->pFilterInfo, pQueryAttr->numOfFilterCols, pBlock); doSetFilterColInfo(pQueryAttr->pFilters, pBlock);
if (pQueryAttr->numOfFilterCols > 0 || pRuntimeEnv->pTsBuf != NULL) { if (pQueryAttr->pFilters != NULL || pRuntimeEnv->pTsBuf != NULL) {
filterRowsInDataBlock(pRuntimeEnv, pQueryAttr->pFilterInfo, pQueryAttr->numOfFilterCols, pBlock, ascQuery); filterColRowsInDataBlock(pRuntimeEnv, pBlock, ascQuery);
} }
} }
@ -6365,6 +6417,7 @@ int32_t convertQueryMsg(SQueryTableMsg *pQueryMsg, SQueryParam* param) {
pQueryMsg->numOfOutput = htons(pQueryMsg->numOfOutput); pQueryMsg->numOfOutput = htons(pQueryMsg->numOfOutput);
pQueryMsg->numOfGroupCols = htons(pQueryMsg->numOfGroupCols); pQueryMsg->numOfGroupCols = htons(pQueryMsg->numOfGroupCols);
pQueryMsg->tagCondLen = htons(pQueryMsg->tagCondLen); pQueryMsg->tagCondLen = htons(pQueryMsg->tagCondLen);
pQueryMsg->colCondLen = htons(pQueryMsg->colCondLen);
pQueryMsg->tsBuf.tsOffset = htonl(pQueryMsg->tsBuf.tsOffset); pQueryMsg->tsBuf.tsOffset = htonl(pQueryMsg->tsBuf.tsOffset);
pQueryMsg->tsBuf.tsLen = htonl(pQueryMsg->tsBuf.tsLen); pQueryMsg->tsBuf.tsLen = htonl(pQueryMsg->tsBuf.tsLen);
pQueryMsg->tsBuf.tsNumOfBlocks = htonl(pQueryMsg->tsBuf.tsNumOfBlocks); pQueryMsg->tsBuf.tsNumOfBlocks = htonl(pQueryMsg->tsBuf.tsNumOfBlocks);
@ -6415,6 +6468,18 @@ int32_t convertQueryMsg(SQueryTableMsg *pQueryMsg, SQueryParam* param) {
} }
} }
if (pQueryMsg->colCondLen > 0) {
param->colCond = calloc(1, pQueryMsg->colCondLen);
if (param->colCond == NULL) {
code = TSDB_CODE_QRY_OUT_OF_MEMORY;
goto _cleanup;
}
memcpy(param->colCond, pMsg, pQueryMsg->colCondLen);
pMsg += pQueryMsg->colCondLen;
}
param->tableScanOperator = pQueryMsg->tableScanOperator; param->tableScanOperator = pQueryMsg->tableScanOperator;
param->pExpr = calloc(pQueryMsg->numOfOutput, POINTER_BYTES); param->pExpr = calloc(pQueryMsg->numOfOutput, POINTER_BYTES);
if (param->pExpr == NULL) { if (param->pExpr == NULL) {
@ -6831,6 +6896,25 @@ int32_t createQueryFunc(SQueriedTableInfo* pTableInfo, int32_t numOfOutput, SExp
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t createQueryFilter(char *data, uint16_t len, SFilterInfo** pFilters) {
tExprNode* expr = NULL;
TRY(TSDB_MAX_TAG_CONDITIONS) {
expr = exprTreeFromBinary(data, len);
} CATCH( code ) {
CLEANUP_EXECUTE();
return code;
} END_TRY
if (expr == NULL) {
qError("failed to create expr tree");
return TSDB_CODE_QRY_APP_ERROR;
}
return filterInitFromTree(expr, pFilters);
}
// todo refactor // todo refactor
int32_t createIndirectQueryFuncExprFromMsg(SQueryTableMsg* pQueryMsg, int32_t numOfOutput, SExprInfo** pExprInfo, int32_t createIndirectQueryFuncExprFromMsg(SQueryTableMsg* pQueryMsg, int32_t numOfOutput, SExprInfo** pExprInfo,
SSqlExpr** pExpr, SExprInfo* prevExpr) { SSqlExpr** pExpr, SExprInfo* prevExpr) {
@ -7061,7 +7145,7 @@ FORCE_INLINE bool checkQIdEqual(void *qHandle, uint64_t qId) {
} }
SQInfo* createQInfoImpl(SQueryTableMsg* pQueryMsg, SGroupbyExpr* pGroupbyExpr, SExprInfo* pExprs, SQInfo* createQInfoImpl(SQueryTableMsg* pQueryMsg, SGroupbyExpr* pGroupbyExpr, SExprInfo* pExprs,
SExprInfo* pSecExprs, STableGroupInfo* pTableGroupInfo, SColumnInfo* pTagCols, int32_t vgId, SExprInfo* pSecExprs, STableGroupInfo* pTableGroupInfo, SColumnInfo* pTagCols, SFilterInfo* pFilters, int32_t vgId,
char* sql, uint64_t *qId) { char* sql, uint64_t *qId) {
int16_t numOfCols = pQueryMsg->numOfCols; int16_t numOfCols = pQueryMsg->numOfCols;
int16_t numOfOutput = pQueryMsg->numOfOutput; int16_t numOfOutput = pQueryMsg->numOfOutput;
@ -7110,6 +7194,7 @@ SQInfo* createQInfoImpl(SQueryTableMsg* pQueryMsg, SGroupbyExpr* pGroupbyExpr, S
pQueryAttr->needReverseScan = pQueryMsg->needReverseScan; pQueryAttr->needReverseScan = pQueryMsg->needReverseScan;
pQueryAttr->stateWindow = pQueryMsg->stateWindow; pQueryAttr->stateWindow = pQueryMsg->stateWindow;
pQueryAttr->vgId = vgId; pQueryAttr->vgId = vgId;
pQueryAttr->pFilters = pFilters;
pQueryAttr->tableCols = calloc(numOfCols, sizeof(SSingleColumnFilterInfo)); pQueryAttr->tableCols = calloc(numOfCols, sizeof(SSingleColumnFilterInfo));
if (pQueryAttr->tableCols == NULL) { if (pQueryAttr->tableCols == NULL) {

276
src/query/src/qFilter.c
View File

@ -15,6 +15,7 @@
#include "os.h" #include "os.h"
#include "queryLog.h" #include "queryLog.h"
#include "qFilter.h" #include "qFilter.h"
#include "tcompare.h"
OptrStr gOptrStr[] = { OptrStr gOptrStr[] = {
{TSDB_RELATION_INVALID, "invalid"}, {TSDB_RELATION_INVALID, "invalid"},
@ -53,21 +54,10 @@ filter_desc_compare_func gDescCompare [F_FIELD_MAX] = {
filterFieldValDescCompare filterFieldValDescCompare
}; };
int32_t filterMergeGroup(SFilterGroup *gp1, SFilterGroup *gp2, SArray* group) {
int32_t filterMergeGroup(SArray* group, SArray* left, SArray* right) {
int32_t leftSize = (int32_t)taosArrayGetSize(left);
int32_t rightSize = (int32_t)taosArrayGetSize(right);
CHK_LRET(taosArrayGetSize(left) <= 0, TSDB_CODE_QRY_APP_ERROR, "empty group");
CHK_LRET(taosArrayGetSize(right) <= 0, TSDB_CODE_QRY_APP_ERROR, "empty group");
SFilterGroup gp = {0}; SFilterGroup gp = {0};
for (int32_t l = 0; l < leftSize; ++l) { //TODO CHECK DUP
SFilterGroup *gp1 = taosArrayGet(left, l);
for (int32_t r = 0; r < rightSize; ++r) {
SFilterGroup *gp2 = taosArrayGet(right, r);
gp.unitNum = gp1->unitNum + gp2->unitNum; gp.unitNum = gp1->unitNum + gp2->unitNum;
gp.unitIdxs = calloc(gp.unitNum, sizeof(*gp.unitIdxs)); gp.unitIdxs = calloc(gp.unitNum, sizeof(*gp.unitIdxs));
@ -77,6 +67,25 @@ int32_t filterMergeGroup(SArray* group, SArray* left, SArray* right) {
gp.unitFlags = NULL; gp.unitFlags = NULL;
taosArrayPush(group, &gp); taosArrayPush(group, &gp);
return TSDB_CODE_SUCCESS;
}
int32_t filterMergeGroups(SArray* group, SArray* left, SArray* right) {
int32_t leftSize = (int32_t)taosArrayGetSize(left);
int32_t rightSize = (int32_t)taosArrayGetSize(right);
CHK_LRET(taosArrayGetSize(left) <= 0, TSDB_CODE_QRY_APP_ERROR, "empty group");
CHK_LRET(taosArrayGetSize(right) <= 0, TSDB_CODE_QRY_APP_ERROR, "empty group");
for (int32_t l = 0; l < leftSize; ++l) {
SFilterGroup *gp1 = taosArrayGet(left, l);
for (int32_t r = 0; r < rightSize; ++r) {
SFilterGroup *gp2 = taosArrayGet(right, r);
filterMergeGroup(gp1, gp2, group);
} }
} }
@ -95,9 +104,10 @@ int32_t filterGetFiled(SFilterFields* fields, int32_t type, void *v) {
} }
SFilterField* filterAddField(SFilterInfo *info, tExprNode *node) { int32_t filterAddField(SFilterInfo *info, tExprNode *node, SFilterFieldId *fid) {
CHK_LRET(node == NULL, NULL, "empty node"); CHK_LRET(node == NULL, TSDB_CODE_QRY_APP_ERROR, "empty node");
CHK_LRET(node->nodeType != TSQL_NODE_COL && node->nodeType != TSQL_NODE_VALUE, NULL, "invalid nodeType"); CHK_LRET(node->nodeType != TSQL_NODE_COL && node->nodeType != TSQL_NODE_VALUE, TSDB_CODE_QRY_APP_ERROR, "invalid nodeType");
int32_t type, idx = -1; int32_t type, idx = -1;
uint16_t *num; uint16_t *num;
void *v; void *v;
@ -110,26 +120,39 @@ SFilterField* filterAddField(SFilterInfo *info, tExprNode *node) {
v = node->pVal; v = node->pVal;
} }
num = &info->fileds[type].num; num = &info->fields[type].num;
if (num > 0) { if (*num > 0) {
idx = filterGetFiled(&info->fileds[type], type, v); idx = filterGetFiled(&info->fields[type], type, v);
} }
if (idx < 0) { if (idx < 0) {
idx = *num; idx = *num;
info->fileds[type].fields[idx].type = type; if (idx >= info->fields[type].size) {
info->fileds[type].fields[idx].desc = v; info->fields[type].size += FILTER_DEFAULT_FIELD_SIZE;
info->fields[type].fields = realloc(info->fields[type].fields, info->fields[type].size * sizeof(SFilterField));
}
info->fields[type].fields[idx].type = type;
info->fields[type].fields[idx].desc = v;
++(*num); ++(*num);
} }
return &info->fileds[type].fields[idx]; fid->type = type;
fid->idx = idx;
return TSDB_CODE_SUCCESS;
} }
int32_t filterAddUnit(SFilterInfo *info, uint8_t optr, SFilterField *left, SFilterField *right) { int32_t filterAddUnit(SFilterInfo *info, uint8_t optr, SFilterFieldId *left, SFilterFieldId *right) {
if (info->unitNum >= info->unitSize) {
info->unitSize += FILTER_DEFAULT_UNIT_SIZE;
info->units = realloc(info->units, info->unitSize * sizeof(SFilterUnit));
}
info->units[info->unitNum].compare.optr = optr; info->units[info->unitNum].compare.optr = optr;
info->units[info->unitNum].left = left; info->units[info->unitNum].left = *left;
info->units[info->unitNum].right = right; info->units[info->unitNum].right = *right;
++info->unitNum; ++info->unitNum;
@ -138,12 +161,20 @@ int32_t filterAddUnit(SFilterInfo *info, uint8_t optr, SFilterField *left, SFilt
int32_t filterAddGroup(SFilterGroup *group, uint16_t unitIdx) { int32_t filterAddGroup(SFilterGroup *group, uint16_t unitIdx) {
group->unitNum = 1; group->unitNum = 1;
group->unitIdxs= calloc(1, sizeof(*group->unitIdxs)); group->unitIdxs= calloc(group->unitNum, sizeof(*group->unitIdxs));
group->unitIdxs[0] = unitIdx; group->unitIdxs[0] = unitIdx;
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static void filterFreeGroup(void *pItem) {
SFilterGroup* p = (SFilterGroup*) pItem;
if (p) {
tfree(p->unitIdxs);
tfree(p->unitFlags);
}
}
int32_t filterTreeToGroup(tExprNode* tree, SFilterInfo *info, SArray* group) { int32_t filterTreeToGroup(tExprNode* tree, SFilterInfo *info, SArray* group) {
int32_t code = TSDB_CODE_SUCCESS; int32_t code = TSDB_CODE_SUCCESS;
@ -161,7 +192,10 @@ int32_t filterTreeToGroup(tExprNode* tree, SFilterInfo *info, SArray* group) {
ERR_JRET(filterTreeToGroup(tree->_node.pLeft, info, leftGroup)); ERR_JRET(filterTreeToGroup(tree->_node.pLeft, info, leftGroup));
ERR_JRET(filterTreeToGroup(tree->_node.pRight, info, rightGroup)); ERR_JRET(filterTreeToGroup(tree->_node.pRight, info, rightGroup));
ERR_JRET(filterMergeGroup(group, leftGroup, rightGroup)); ERR_JRET(filterMergeGroups(group, leftGroup, rightGroup));
taosArrayDestroyEx(leftGroup, filterFreeGroup);
taosArrayDestroyEx(rightGroup, filterFreeGroup);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -173,10 +207,11 @@ int32_t filterTreeToGroup(tExprNode* tree, SFilterInfo *info, SArray* group) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
SFilterField *left = filterAddField(info, tree->_node.pLeft); SFilterFieldId left, right;
SFilterField *right = filterAddField(info, tree->_node.pRight); filterAddField(info, tree->_node.pLeft, &left);
filterAddField(info, tree->_node.pRight, &right);
filterAddUnit(info, tree->_node.optr, left, right); filterAddUnit(info, tree->_node.optr, &left, &right);
SFilterGroup fgroup = {0}; SFilterGroup fgroup = {0};
filterAddGroup(&fgroup, info->unitNum - 1); filterAddGroup(&fgroup, info->unitNum - 1);
@ -185,32 +220,52 @@ int32_t filterTreeToGroup(tExprNode* tree, SFilterInfo *info, SArray* group) {
_err_return: _err_return:
taosArrayDestroy(leftGroup); taosArrayDestroyEx(leftGroup, filterFreeGroup);
taosArrayDestroy(rightGroup); taosArrayDestroyEx(rightGroup, filterFreeGroup);
return code; return code;
} }
int32_t filterInitUnitFunc(SFilterInfo *info) {
for (uint16_t i = 0; i < info->unitNum; ++i) {
SFilterUnit* unit = &info->units[i];
SFilterField *left = FILTER_GET_FIELD(info, unit->left);
unit->compare.pCompareFunc = getComparFunc(left->type, unit->compare.optr);
}
return TSDB_CODE_SUCCESS;
}
void filterDumpInfoToString(SFilterInfo *info) { void filterDumpInfoToString(SFilterInfo *info) {
CHK_LRETV(info == NULL, "FilterInfo: empty"); CHK_LRETV(info == NULL, "FilterInfo: empty");
qDebug("FilterInfo:"); qDebug("FilterInfo:");
qDebug("Col F Num:%u", info->fileds[F_FIELD_COLUMN].num); qDebug("Field Col Num:%u", info->fields[F_FIELD_COLUMN].num);
for (uint16_t i = 0; i < info->fileds[F_FIELD_COLUMN].num; ++i) { for (uint16_t i = 0; i < info->fields[F_FIELD_COLUMN].num; ++i) {
SFilterField *field = &info->fileds[F_FIELD_COLUMN].fields[i]; SFilterField *field = &info->fields[F_FIELD_COLUMN].fields[i];
SSchema *sch = field->desc; SSchema *sch = field->desc;
qDebug("COL%d => [%d][%s]", i, sch->colId, sch->name); qDebug("COL%d => [%d][%s]", i, sch->colId, sch->name);
} }
qDebug("Field Val Num:%u", info->fields[F_FIELD_VALUE].num);
for (uint16_t i = 0; i < info->fields[F_FIELD_VALUE].num; ++i) {
SFilterField *field = &info->fields[F_FIELD_VALUE].fields[i];
tVariant *var = field->desc;
qDebug("VAL%d => [type:%d][val:%" PRIu64"]", i, var->nType, var->u64); //TODO
}
qDebug("Unit Num:%u", info->unitNum); qDebug("Unit Num:%u", info->unitNum);
for (uint16_t i = 0; i < info->unitNum; ++i) { for (uint16_t i = 0; i < info->unitNum; ++i) {
SFilterUnit *unit = &info->units[i]; SFilterUnit *unit = &info->units[i];
SFilterField *left = unit->left; SFilterField *left = FILTER_GET_FIELD(info, unit->left);
SFilterField *right = unit->right; SFilterField *right = FILTER_GET_FIELD(info, unit->right);
SSchema *sch = left->desc; SSchema *sch = left->desc;
tVariant *var = right->desc; tVariant *var = right->desc;
qDebug("UNIT%d => [%d][%s] %s %" PRId64, i, sch->colId, sch->name, gOptrStr[unit->compare.optr].str, IS_NUMERIC_TYPE(var->nType) ? var->i64 : -1); qDebug("UNIT%d => [%d][%s] %s %" PRId64, i, sch->colId, sch->name, gOptrStr[unit->compare.optr].str, IS_NUMERIC_TYPE(var->nType) ? var->i64 : -1); //TODO
} }
qDebug("Group Num:%u", info->groupNum); qDebug("Group Num:%u", info->groupNum);
@ -224,24 +279,33 @@ void filterDumpInfoToString(SFilterInfo *info) {
} }
} }
int32_t filterInitFromTree(tExprNode* tree, SFilterInfo *info, int32_t colSize) { int32_t filterInitFromTree(tExprNode* tree, SFilterInfo **pinfo) {
int32_t code = TSDB_CODE_SUCCESS; int32_t code = TSDB_CODE_SUCCESS;
SFilterInfo *info = NULL;
CHK_RET(colSize <= 0, code); CHK_LRET(tree == NULL || pinfo == NULL, TSDB_CODE_QRY_APP_ERROR, "invalid param");
CHK_LRET(tree == NULL || info == NULL, TSDB_CODE_QRY_APP_ERROR, "invalid param");
if (*pinfo == NULL) {
*pinfo = calloc(1, sizeof(SFilterInfo));
}
info = *pinfo;
SArray* group = taosArrayInit(4, sizeof(SFilterGroup)); SArray* group = taosArrayInit(4, sizeof(SFilterGroup));
info->units = calloc(colSize, sizeof(SFilterUnit)); info->unitSize = FILTER_DEFAULT_UNIT_SIZE;
info->units = calloc(info->unitSize, sizeof(SFilterUnit));
info->fileds[F_FIELD_COLUMN].num = 0; info->fields[F_FIELD_COLUMN].num = 0;
info->fileds[F_FIELD_COLUMN].fields = calloc(colSize, sizeof(SFilterField)); info->fields[F_FIELD_COLUMN].size = FILTER_DEFAULT_FIELD_SIZE;
info->fileds[F_FIELD_VALUE].num = 0; info->fields[F_FIELD_COLUMN].fields = calloc(info->fields[F_FIELD_COLUMN].size, sizeof(SFilterField));
info->fileds[F_FIELD_VALUE].fields = calloc(colSize, sizeof(SFilterField)); info->fields[F_FIELD_VALUE].num = 0;
info->fields[F_FIELD_VALUE].size = FILTER_DEFAULT_FIELD_SIZE;
info->fields[F_FIELD_VALUE].fields = calloc(info->fields[F_FIELD_VALUE].size, sizeof(SFilterField));
code = filterTreeToGroup(tree, info, group); code = filterTreeToGroup(tree, info, group);
ERR_RET(code); ERR_JRET(code);
size_t groupSize = taosArrayGetSize(group); size_t groupSize = taosArrayGetSize(group);
@ -253,13 +317,20 @@ int32_t filterInitFromTree(tExprNode* tree, SFilterInfo *info, int32_t colSize)
for (size_t i = 0; i < groupSize; ++i) { for (size_t i = 0; i < groupSize; ++i) {
SFilterGroup *pg = taosArrayGet(group, i); SFilterGroup *pg = taosArrayGet(group, i);
info->groups[i].unitNum = pg->unitNum; info->groups[i] = *pg;
info->groups[i].unitIdxs = pg->unitIdxs;
info->groups[i].unitFlags = pg->unitFlags;
} }
ERR_JRET(filterInitUnitFunc(info));
info->unitRes = malloc(info->unitNum * sizeof(*info->unitRes));
info->unitFlags = malloc(info->unitNum * sizeof(*info->unitFlags));
filterDumpInfoToString(info); filterDumpInfoToString(info);
_err_return:
taosArrayDestroy(group);
return code; return code;
} }
@ -269,4 +340,109 @@ void filterFreeInfo(SFilterInfo *info) {
//TODO //TODO
} }
int32_t filterSetColData(SFilterInfo *info, int16_t colId, void *data) {
CHK_LRET(info == NULL, TSDB_CODE_QRY_APP_ERROR, "info NULL");
CHK_LRET(info->fields[F_FIELD_COLUMN].num <= 0, TSDB_CODE_QRY_APP_ERROR, "no column fileds");
for (uint16_t i = 0; i < info->fields[F_FIELD_COLUMN].num; ++i) {
SFilterField* fi = &info->fields[F_FIELD_COLUMN].fields[i];
SSchema* sch = fi->desc;
if (sch->colId == colId) {
fi->data = data;
break;
}
}
return TSDB_CODE_SUCCESS;
}
bool filterDoCompare(SFilterUnit *unit, void *left, void *right) {
int32_t ret = unit->compare.pCompareFunc(left, right);
switch (unit->compare.optr) {
case TSDB_RELATION_EQUAL: {
return ret == 0;
}
case TSDB_RELATION_NOT_EQUAL: {
return ret != 0;
}
case TSDB_RELATION_GREATER_EQUAL: {
return ret >= 0;
}
case TSDB_RELATION_GREATER: {
return ret > 0;
}
case TSDB_RELATION_LESS_EQUAL: {
return ret <= 0;
}
case TSDB_RELATION_LESS: {
return ret < 0;
}
case TSDB_RELATION_LIKE: {
return ret == 0;
}
case TSDB_RELATION_IN: {
return ret == 1;
}
default:
assert(false);
}
return true;
}
bool filterExecute(SFilterInfo *info, int32_t numOfRows, int8_t* p) {
bool all = true;
for (int32_t i = 0; i < numOfRows; ++i) {
FILTER_UNIT_CLR_F(info);
p[i] = 0;
for (uint16_t g = 0; g < info->groupNum; ++g) {
SFilterGroup* group = &info->groups[g];
bool qualified = true;
for (uint16_t u = 0; u < group->unitNum; ++u) {
uint16_t uidx = group->unitIdxs[u];
uint8_t ures = 0;
if (FILTER_UNIT_GET_F(info, uidx)) {
ures = FILTER_UNIT_GET_R(info, uidx);
} else {
SFilterUnit *unit = &info->units[uidx];
SFilterField *left = FILTER_GET_FIELD(info, unit->left);
SFilterField *right = FILTER_GET_FIELD(info, unit->right);
ures = filterDoCompare(unit, FILTER_GET_COL_FIELD_DATA(left, i), FILTER_GET_VAL_FIELD_DATA(right));
FILTER_UNIT_SET_R(info, uidx, ures);
FILTER_UNIT_SET_F(info, uidx);
}
if (!ures) {
qualified = ures;
break;
}
}
if (qualified) {
p[i] = 1;
break;
}
}
if (p[i] != 1) {
all = false;
}
}
return all;
}

8
src/query/src/queryMain.c
View File

@ -103,6 +103,12 @@ int32_t qCreateQueryInfo(void* tsdb, int32_t vgId, SQueryTableMsg* pQueryMsg, qi
} }
} }
if (param.colCond != NULL) {
if ((code = createQueryFilter(param.colCond, pQueryMsg->colCondLen, &param.pFilters)) != TSDB_CODE_SUCCESS) {
goto _over;
}
}
param.pGroupbyExpr = createGroupbyExprFromMsg(pQueryMsg, param.pGroupColIndex, &code); param.pGroupbyExpr = createGroupbyExprFromMsg(pQueryMsg, param.pGroupColIndex, &code);
if ((param.pGroupbyExpr == NULL && pQueryMsg->numOfGroupCols != 0) || code != TSDB_CODE_SUCCESS) { if ((param.pGroupbyExpr == NULL && pQueryMsg->numOfGroupCols != 0) || code != TSDB_CODE_SUCCESS) {
goto _over; goto _over;
@ -162,7 +168,7 @@ int32_t qCreateQueryInfo(void* tsdb, int32_t vgId, SQueryTableMsg* pQueryMsg, qi
assert(pQueryMsg->stableQuery == isSTableQuery); assert(pQueryMsg->stableQuery == isSTableQuery);
(*pQInfo) = createQInfoImpl(pQueryMsg, param.pGroupbyExpr, param.pExprs, param.pSecExprs, &tableGroupInfo, (*pQInfo) = createQInfoImpl(pQueryMsg, param.pGroupbyExpr, param.pExprs, param.pSecExprs, &tableGroupInfo,
param.pTagColumnInfo, vgId, param.sql, qId); param.pTagColumnInfo, param.pFilters, vgId, param.sql, qId);
param.sql = NULL; param.sql = NULL;
param.pExprs = NULL; param.pExprs = NULL;

198
tests/script/general/parser/condition.sim Normal file
View File

@ -0,0 +1,198 @@
system sh/stop_dnodes.sh
system sh/deploy.sh -n dnode1 -i 1
system sh/cfg.sh -n dnode1 -c walLevel -v 1
system sh/cfg.sh -n dnode1 -c maxtablespervnode -v 4
system sh/exec.sh -n dnode1 -s start
sleep 100
sql connect
sql drop database if exists cdb
sql create database if not exists cdb
sql use cdb
sql create table stb1 (ts timestamp, c1 int, c2 float, c3 bigint, c4 smallint, c5 tinyint, c6 double, c7 bool, c8 binary(10), c9 nchar(9)) TAGS(t1 int, t2 binary(10), t3 double)
sql create table tb1 using stb1 tags(1,'1',1.0)
sql create table tb2 using stb1 tags(2,'2',2.0)
sql create table tb3 using stb1 tags(3,'3',3.0)
sql create table tb4 using stb1 tags(4,'4',4.0)
sql create table tb5 using stb1 tags(5,'5',5.0)
sql create table tb6 using stb1 tags(6,'6',6.0)
sql insert into tb1 values ('2021-05-05 18:19:00',1,1.0,1,1,1,1.0,true ,'1','1')
sql insert into tb1 values ('2021-05-05 18:19:01',2,2.0,2,2,2,2.0,true ,'2','2')
sql insert into tb1 values ('2021-05-05 18:19:02',3,3.0,3,3,3,3.0,false,'3','3')
sql insert into tb1 values ('2021-05-05 18:19:03',4,4.0,4,4,4,4.0,false,'4','4')
sql insert into tb1 values ('2021-05-05 18:19:04',11,11.0,11,11,11,11.0,true ,'11','11')
sql insert into tb1 values ('2021-05-05 18:19:05',12,12.0,12,12,12,12.0,true ,'12','12')
sql insert into tb1 values ('2021-05-05 18:19:06',13,13.0,13,13,13,13.0,false,'13','13')
sql insert into tb1 values ('2021-05-05 18:19:07',14,14.0,14,14,14,14.0,false,'14','14')
sql insert into tb2 values ('2021-05-05 18:19:08',21,21.0,21,21,21,21.0,true ,'21','21')
sql insert into tb2 values ('2021-05-05 18:19:09',22,22.0,22,22,22,22.0,true ,'22','22')
sql insert into tb2 values ('2021-05-05 18:19:10',23,23.0,23,23,23,23.0,false,'23','23')
sql insert into tb2 values ('2021-05-05 18:19:11',24,24.0,24,24,24,24.0,false,'24','24')
sql insert into tb3 values ('2021-05-05 18:19:12',31,31.0,31,31,31,31.0,true ,'31','31')
sql insert into tb3 values ('2021-05-05 18:19:13',32,32.0,32,32,32,32.0,true ,'32','32')
sql insert into tb3 values ('2021-05-05 18:19:14',33,33.0,33,33,33,33.0,false,'33','33')
sql insert into tb3 values ('2021-05-05 18:19:15',34,34.0,34,34,34,34.0,false,'34','34')
sql insert into tb4 values ('2021-05-05 18:19:16',41,41.0,41,41,41,41.0,true ,'41','41')
sql insert into tb4 values ('2021-05-05 18:19:17',42,42.0,42,42,42,42.0,true ,'42','42')
sql insert into tb4 values ('2021-05-05 18:19:18',43,43.0,43,43,43,43.0,false,'43','43')
sql insert into tb4 values ('2021-05-05 18:19:19',44,44.0,44,44,44,44.0,false,'44','44')
sql insert into tb5 values ('2021-05-05 18:19:20',51,51.0,51,51,51,51.0,true ,'51','51')
sql insert into tb5 values ('2021-05-05 18:19:21',52,52.0,52,52,52,52.0,true ,'52','52')
sql insert into tb5 values ('2021-05-05 18:19:22',53,53.0,53,53,53,53.0,false,'53','53')
sql insert into tb5 values ('2021-05-05 18:19:23',54,54.0,54,54,54,54.0,false,'54','54')
sql insert into tb6 values ('2021-05-05 18:19:24',61,61.0,61,61,61,61.0,true ,'61','61')
sql insert into tb6 values ('2021-05-05 18:19:25',62,62.0,62,62,62,62.0,true ,'62','62')
sql insert into tb6 values ('2021-05-05 18:19:26',63,63.0,63,63,63,63.0,false,'63','63')
sql insert into tb6 values ('2021-05-05 18:19:27',64,64.0,64,64,64,64.0,false,'64','64')
sleep 100
print "column test"
sql select * from stb1 where c1 > 0
if $rows != 28 then
return -1
endi
#sql select * from stb1 where c1 > 0 and c1 > 3
#sql select * from stb1 where c1 > 0 or c1 > 3
#sql select * from stb1 where c1 > 0 and c1 > 3 and c1 < 2
#sql select * from stb1 where c1 > 0 or c1 > 3 or c1 < 1
#sql select * from stb1 where c1 > 0 and c1 > 3 or c1 < 1
#sql select * from stb1 where c1 > 0 or c1 > 3 and c1 < 1
#sql select * from stb1 where c1 > 0 and c1 > 3 and c1 < 1 and c1 > 4
#sql select * from stb1 where c1 > 0 and c1 > 3 and c1 < 1 or c1 > 4
#sql select * from stb1 where c1 > 0 and c1 > 3 or c1 < 1 and c1 > 4
#sql select * from stb1 where c1 > 0 or c1 > 3 and c1 < 1 and c1 > 4
#sql select * from stb1 where c1 > 0 and c1 > 3 or c1 < 1 or c1 > 4
#sql select * from stb1 where c1 > 0 or c1 > 3 and c1 < 1 or c1 > 4
#sql select * from stb1 where c1 > 0 or c1 > 3 or c1 < 1 and c1 > 4
#sql select * from stb1 where c1 > 0 or c1 > 3 or c1 < 1 or c1 > 4
#
#sql select * from stb1 where (c1 > 0 and c1 > 3) and c1 < 2
#sql select * from stb1 where c1 > 0 and (c1 > 3 and c1 < 2)
#sql select * from stb1 where (c1 > 0 or c1 > 3) or c1 < 1
#sql select * from stb1 where c1 > 0 or (c1 > 3 or c1 < 1)
#sql select * from stb1 where (c1 > 0 and c1 > 3) or c1 < 1
#sql select * from stb1 where c1 > 0 and (c1 > 3 or c1 < 1)
#sql select * from stb1 where (c1 > 0 or c1 > 3) and c1 < 1
#sql select * from stb1 where c1 > 0 or (c1 > 3 and c1 < 1)
#sql select * from stb1 where (c1 > 0 and c1 > 3) and (c1 < 1 and c1 > 4)
#sql select * from stb1 where (c1 > 0 and c1 > 3 and c1 < 1) and c1 > 4
#sql select * from stb1 where c1 > 0 and (c1 > 3 and c1 < 1) and c1 > 4
#sql select * from stb1 where c1 > 0 and (c1 > 3 and c1 < 1 or c1 > 4)
#sql select * from stb1 where (c1 > 0 and c1 > 3) or (c1 < 1 and c1 > 4)
#sql select * from stb1 where c1 > 0 and (c1 > 3 or c1 < 1) and c1 > 4
#sql select * from stb1 where (c1 > 0 and c1 > 3 or c1 < 1) and c1 > 4
#sql select * from stb1 where c1 > 0 and (c1 > 3 or c1 < 1 and c1 > 4)
#sql select * from stb1 where (c1 > 0 or c1 > 3) and (c1 < 1 and c1 > 4)
#sql select * from stb1 where c1 > 0 or (c1 > 3 and c1 < 1 and c1 > 4)
#sql select * from stb1 where (c1 > 0 or c1 > 3 and c1 < 1) and c1 > 4
#sql select * from stb1 where c1 > 0 or (c1 > 3 and c1 < 1) and c1 > 4
#sql select * from stb1 where (c1 > 0 and c1 > 3) or (c1 < 1 or c1 > 4)
#sql select * from stb1 where c1 > 0 and (c1 > 3 or c1 < 1 or c1 > 4)
#sql select * from stb1 where (c1 > 0 and c1 > 3 or c1 < 1) or c1 > 4
#sql select * from stb1 where c1 > 0 and (c1 > 3 or c1 < 1) or c1 > 4
#sql select * from stb1 where (c1 > 0 or c1 > 3) and (c1 < 1 or c1 > 4)
#sql select * from stb1 where c1 > 0 or (c1 > 3 and c1 < 1 or c1 > 4)
#sql select * from stb1 where (c1 > 0 or c1 > 3 and c1 < 1) or c1 > 4
#sql select * from stb1 where c1 > 0 or (c1 > 3 and c1 < 1) or c1 > 4
#sql select * from stb1 where (c1 > 0 or c1 > 3) or (c1 < 1 and c1 > 4)
#sql select * from stb1 where (c1 > 0 or c1 > 3 or c1 < 1) and c1 > 4
#sql select * from stb1 where c1 > 0 or (c1 > 3 or c1 < 1 and c1 > 4)
#sql select * from stb1 where c1 > 0 or (c1 > 3 or c1 < 1) and c1 > 4
#sql select * from stb1 where (c1 > 0 or c1 > 3) or (c1 < 1 or c1 > 4)
#sql select * from stb1 where c1 > 0 or (c1 > 3 or c1 < 1 or c1 > 4)
#sql select * from stb1 where (c1 > 0 or c1 > 3 or c1 < 1) or c1 > 4
#sql select * from stb1 where c1 > 0 or (c1 > 3 or c1 < 1) or c1 > 4
sql select * from stb1 where (c1 > 40 or c1 < 20) and (c2 < 53 or c2 >= 63) and c3 > 1 and c3 < 5
print "ts test"
print "tbname test"
print "tag test"
print "join test"
print "column&ts test"
print "column&tbname test"
print "column&tag test"
#sql_error select * from stb1 where t1 > 0 or c1 > 0
#sql_error select * from stb1 where c1 > 0 or t1 > 0
#sql_error select * from stb1 where t1 > 0 or c1 > 0 or t1 > 1
#sql_error select * from stb1 where c1 > 0 or t1 > 0 or c1 > 1
#sql_error select * from stb1 where t1 > 0 and c1 > 0 or t1 > 1
#sql_error select * from stb1 where c1 > 0 or t1 > 0 and c1 > 1
#sql_error select * from stb1 where c1 > 0 or t1 > 0 and c1 > 1
#sql_error select * from stb1 where t1 > 0 or t1 > 0 and c1 > 1
#sql_error select * from stb1 where (c1 > 0 and t1 > 0 ) or (t1 > 1 and c1 > 3)
#sql_error select * from stb1 where (c1 > 0 and t1 > 0 ) or t1 > 1
#
#
#sql select * from stb1 where c1 > 0 and t1 > 0
#sql select * from stb1 where t1 > 0 and c1 > 0
#sql select * from stb1 where t1 > 0 and t1 > 3 and c1 > 1
#sql select * from stb1 where t1 > 0 and c1 > 0 and t1 > 1
#sql select * from stb1 where c1 > 0 and t1 > 0 and c1 > 1
#sql select * from stb1 where c1 > 0 and (t1 > 0 or t1 > 1)
#sql select * from stb1 where (t1 > 0 or t1 > 2 ) and (c1 > 1 or c1 > 3)
#sql select * from stb1 where c1 > 0 and (t1 > 0 or t1 > 1)
print "column&join test"
print "ts&tbname test"
print "ts&tag test"
print "ts&join test"
print "tbname&tag test"
print "tbname&join test"
print "tag&join test"
print "column&ts&tbname test"
print "column&ts&tag test"
print "column&ts&join test"
print "column&tbname&tag test"
print "column&tbname&join test"
print "column&tag&join test"
print "ts&tbname&tag test"
print "ts&tbname&join test"
print "ts&tag&join test"
print "tbname&tag&join test"
print "column&ts&tbname&tag test"
print "column&ts&tbname&join test"
print "column&ts&tag&join test"
print "column&tbname&tag&join test"
print "ts&tbname&tag&join test"
print "column&ts&tbname&tag&join test"
#system sh/exec.sh -n dnode1 -s stop -x SIGINT