Skip to content

T
TDengine

taosdata / TDengine
1 年多 前同步成功

通知 1185
Star 22016
Fork 4786
T
TDengine
提交 c15b1e35 编写于 作者: Y yihaoDeng
浏览文件
操作

opt compile

上级 f520de0d
隐藏空白更改
内联 并排
Showing with 71 addition and 59 deletion
include/libs/index/index.h
浏览文件 @ c15b1e35
......@@ -212,6 +212,7 @@ typedef struct SIndexMetaArg {
void* idx;
void* ivtIdx;
uint64_t suid;
int (*metaFilterFunc)(void* metaEx, void* param, SArray* result);
} SIndexMetaArg;
typedef enum { SFLT_NOT_INDEX, SFLT_COARSE_INDEX, SFLT_ACCURATE_INDEX } SIdxFltStatus;
......
include/libs/scalar/filter.h
浏览文件 @ c15b1e35
......@@ -55,6 +55,16 @@ extern bool filterRangeExecute(SFilterInfo *info, SColumnDataAgg **pColsAgg,
int32_t filterPartitionCond(SNode **pCondition, SNode **pPrimaryKeyCond, SNode **pTagIndexCond, SNode **pTagCond,
SNode **pOtherCond);
typedef struct SMetaFltParam {
tb_uid_t suid;
int16_t cid;
int16_t type;
void *val;
bool reverse;
bool equal;
int (*filterFunc)(void *a, void *b, int16_t type);
} SMetaFltParam;
#ifdef __cplusplus
}
#endif
......
source/dnode/vnode/inc/vnode.h
浏览文件 @ c15b1e35
......@@ -26,6 +26,7 @@
#include "tfs.h"
#include "wal.h"
#include "filter.h"
#include "tcommon.h"
#include "tfs.h"
#include "tgrant.h"
......@@ -125,35 +126,24 @@ int32_t metaUidFilterCachePut(SMeta *pMeta, uint64_t suid, const void *pKey, in
int32_t payloadLen, double selectivityRatio);
int32_t metaUidCacheClear(SMeta *pMeta, uint64_t suid);
tb_uid_t metaGetTableEntryUidByName(SMeta *pMeta, const char *name);
int32_t metaTbGroupCacheClear(SMeta* pMeta, uint64_t suid);
int32_t metaGetCachedTbGroup(SMeta* pMeta, tb_uid_t suid, const uint8_t* pKey, int32_t keyLen, SArray** pList);
int32_t metaPutTbGroupToCache(SMeta* pMeta, uint64_t suid, const void* pKey, int32_t keyLen, void* pPayload,
int32_t payloadLen);
int32_t metaTbGroupCacheClear(SMeta *pMeta, uint64_t suid);
int32_t metaGetCachedTbGroup(SMeta *pMeta, tb_uid_t suid, const uint8_t *pKey, int32_t keyLen, SArray **pList);
int32_t metaPutTbGroupToCache(SMeta *pMeta, uint64_t suid, const void *pKey, int32_t keyLen, void *pPayload,
int32_t payloadLen);
int64_t metaGetTbNum(SMeta *pMeta);
int64_t metaGetNtbNum(SMeta *pMeta);
int64_t metaGetTbNum(SMeta *pMeta);
int64_t metaGetNtbNum(SMeta *pMeta);
typedef struct {
int64_t uid;
int64_t ctbNum;
} SMetaStbStats;
int32_t metaGetStbStats(SMeta *pMeta, int64_t uid, SMetaStbStats *pInfo);
typedef struct SMetaFltParam {
tb_uid_t suid;
int16_t cid;
int16_t type;
void *val;
bool reverse;
bool equal;
int (*filterFunc)(void *a, void *b, int16_t type);
} SMetaFltParam;
// TODO, refactor later
int32_t metaFilterTableIds(SMeta *pMeta, SMetaFltParam *param, SArray *results);
int32_t metaFilterCreateTime(SMeta *pMeta, SMetaFltParam *parm, SArray *pUids);
int32_t metaFilterTableName(SMeta *pMeta, SMetaFltParam *param, SArray *pUids);
int32_t metaFilterTtl(SMeta *pMeta, SMetaFltParam *param, SArray *pUids);
int32_t metaFilterTableIds(void *pMeta, void *param, SArray *results);
int32_t metaFilterCreateTime(void *pMeta, void *parm, SArray *pUids);
int32_t metaFilterTableName(void *pMeta, void *param, SArray *pUids);
int32_t metaFilterTtl(void *pMeta, void *param, SArray *pUids);
#if 1 // refact APIs below (TODO)
typedef SVCreateTbReq STbCfg;
......@@ -183,7 +173,8 @@ typedef struct STsdbReader STsdbReader;
#define CACHESCAN_RETRIEVE_LAST 0x8
int32_t tsdbReaderOpen(SVnode *pVnode, SQueryTableDataCond *pCond, void *pTableList, int32_t numOfTables,
SSDataBlock *pResBlock, STsdbReader **ppReader, const char *idstr, bool countOnly, SHashObj** pIgnoreTables);
SSDataBlock *pResBlock, STsdbReader **ppReader, const char *idstr, bool countOnly,
SHashObj **pIgnoreTables);
int32_t tsdbSetTableList(STsdbReader *pReader, const void *pTableList, int32_t num);
void tsdbReaderSetId(STsdbReader *pReader, const char *idstr);
void tsdbReaderClose(STsdbReader *pReader);
......@@ -198,9 +189,9 @@ void *tsdbGetIdx(SMeta *pMeta);
void *tsdbGetIvtIdx(SMeta *pMeta);
uint64_t tsdbGetReaderMaxVersion(STsdbReader *pReader);
void tsdbReaderSetCloseFlag(STsdbReader *pReader);
int64_t tsdbGetLastTimestamp(SVnode* pVnode, void* pTableList, int32_t numOfTables, const char* pIdStr);
int64_t tsdbGetLastTimestamp(SVnode *pVnode, void *pTableList, int32_t numOfTables, const char *pIdStr);
int32_t tsdbReuseCacherowsReader(void* pReader, void* pTableIdList, int32_t numOfTables);
int32_t tsdbReuseCacherowsReader(void *pReader, void *pTableIdList, int32_t numOfTables);
int32_t tsdbCacherowsReaderOpen(void *pVnode, int32_t type, void *pTableIdList, int32_t numOfTables, int32_t numOfCols,
SArray *pCidList, int32_t *pSlotIds, uint64_t suid, void **pReader, const char *idstr);
int32_t tsdbRetrieveCacheRows(void *pReader, SSDataBlock *pResBlock, const int32_t *slotIds, const int32_t *dstSlotIds,
......@@ -265,13 +256,13 @@ int32_t tqReaderAddTbUidList(STqReader *pReader, const SArray *pTableUidList);
int32_t tqReaderRemoveTbUidList(STqReader *pReader, const SArray *tbUidList);
int32_t tqSeekVer(STqReader *pReader, int64_t ver, const char *id);
bool tqNextBlockInWal(STqReader* pReader, const char* idstr);
bool tqNextBlockImpl(STqReader *pReader, const char* idstr);
bool tqNextBlockInWal(STqReader *pReader, const char *idstr);
bool tqNextBlockImpl(STqReader *pReader, const char *idstr);
int32_t extractMsgFromWal(SWalReader* pReader, void** pItem, const char* id);
int32_t extractMsgFromWal(SWalReader *pReader, void **pItem, const char *id);
int32_t tqReaderSetSubmitMsg(STqReader *pReader, void *msgStr, int32_t msgLen, int64_t ver);
bool tqNextDataBlockFilterOut(STqReader *pReader, SHashObj *filterOutUids);
int32_t tqRetrieveDataBlock(STqReader *pReader, const char* idstr);
int32_t tqRetrieveDataBlock(STqReader *pReader, const char *idstr);
int32_t tqRetrieveTaosxBlock(STqReader *pReader, SArray *blocks, SArray *schemas, SSubmitTbData **pSubmitTbDataRet);
int32_t vnodeEnqueueStreamMsg(SVnode *pVnode, SRpcMsg *pMsg);
......
source/dnode/vnode/src/meta/metaQuery.c
浏览文件 @ c15b1e35
......@@ -1083,8 +1083,10 @@ typedef struct {
int32_t vLen;
} SIdxCursor;
int32_t metaFilterCreateTime(SMeta *pMeta, SMetaFltParam *param, SArray *pUids) {
int32_t ret = 0;
int32_t metaFilterCreateTime(void *meta, void *arg, SArray *pUids) {
SMeta *pMeta = meta;
SMetaFltParam *param = arg;
int32_t ret = 0;
SIdxCursor *pCursor = NULL;
pCursor = (SIdxCursor *)taosMemoryCalloc(1, sizeof(SIdxCursor));
......@@ -1141,9 +1143,11 @@ END:
return ret;
}
int32_t metaFilterTableName(SMeta *pMeta, SMetaFltParam *param, SArray *pUids) {
int32_t ret = 0;
char *buf = NULL;
int32_t metaFilterTableName(void *meta, void *arg, SArray *pUids) {
SMeta *pMeta = meta;
SMetaFltParam *param = arg;
int32_t ret = 0;
char *buf = NULL;
STagIdxKey *pKey = NULL;
int32_t nKey = 0;
......@@ -1206,9 +1210,11 @@ END:
return ret;
}
int32_t metaFilterTtl(SMeta *pMeta, SMetaFltParam *param, SArray *pUids) {
int32_t ret = 0;
char *buf = NULL;
int32_t metaFilterTtl(void *meta, void *arg, SArray *pUids) {
SMeta *pMeta = meta;
SMetaFltParam *param = arg;
int32_t ret = 0;
char *buf = NULL;
STtlIdxKey *pKey = NULL;
int32_t nKey = 0;
......@@ -1235,7 +1241,10 @@ END:
// impl later
return 0;
}
int32_t metaFilterTableIds(SMeta *pMeta, SMetaFltParam *param, SArray *pUids) {
int32_t metaFilterTableIds(void *meta, void *arg, SArray *pUids) {
SMeta *pMeta = meta;
SMetaFltParam *param = arg;
SMetaEntry oStbEntry = {0};
int32_t ret = -1;
char *buf = NULL;
......
source/libs/executor/src/executil.c
浏览文件 @ c15b1e35
......@@ -435,7 +435,6 @@ static void genTagFilterDigest(const SNode* pTagCond, T_MD5_CTX* pContext) {
taosMemoryFree(payload);
}
static void genTbGroupDigest(const SNode* pGroup, uint8_t* filterDigest, T_MD5_CTX* pContext) {
char* payload = NULL;
int32_t len = 0;
......@@ -453,8 +452,8 @@ static void genTbGroupDigest(const SNode* pGroup, uint8_t* filterDigest, T_MD5_C
taosMemoryFree(payload);
}
int32_t getColInfoResultForGroupby(void* metaHandle, SNodeList* group, STableListInfo* pTableListInfo, uint8_t *digest) {
int32_t getColInfoResultForGroupby(void* metaHandle, SNodeList* group, STableListInfo* pTableListInfo,
uint8_t* digest) {
int32_t code = TSDB_CODE_SUCCESS;
SArray* pBlockList = NULL;
SSDataBlock* pResBlock = NULL;
......@@ -492,18 +491,19 @@ int32_t getColInfoResultForGroupby(void* metaHandle, SNodeList* group, STableLis
if (tsTagFilterCache) {
SNodeListNode* listNode = (SNodeListNode*)nodesMakeNode(QUERY_NODE_NODE_LIST);
listNode->pNodeList = group;
genTbGroupDigest((SNode *)listNode, digest, &context);
genTbGroupDigest((SNode*)listNode, digest, &context);
nodesFree(listNode);
metaGetCachedTbGroup(metaHandle, pTableListInfo->idInfo.suid, context.digest, tListLen(context.digest), &tableList);
if (tableList) {
taosArrayDestroy(pTableListInfo->pTableList);
pTableListInfo->pTableList = tableList;
qDebug("retrieve tb group list from cache, numOfTables:%d", (int32_t)taosArrayGetSize(pTableListInfo->pTableList));
qDebug("retrieve tb group list from cache, numOfTables:%d",
(int32_t)taosArrayGetSize(pTableListInfo->pTableList));
goto end;
}
}
pUidTagList = taosArrayInit(8, sizeof(STUidTagInfo));
for (int32_t i = 0; i < rows; ++i) {
STableKeyInfo* pkeyInfo = taosArrayGet(pTableListInfo->pTableList, i);
......@@ -632,9 +632,10 @@ int32_t getColInfoResultForGroupby(void* metaHandle, SNodeList* group, STableLis
if (tsTagFilterCache) {
tableList = taosArrayDup(pTableListInfo->pTableList, NULL);
metaPutTbGroupToCache(metaHandle, pTableListInfo->idInfo.suid, context.digest, tListLen(context.digest), tableList, taosArrayGetSize(tableList) * sizeof(STableKeyInfo));
metaPutTbGroupToCache(metaHandle, pTableListInfo->idInfo.suid, context.digest, tListLen(context.digest), tableList,
taosArrayGetSize(tableList) * sizeof(STableKeyInfo));
}
// int64_t st2 = taosGetTimestampUs();
// qDebug("calculate tag block rows:%d, cost:%ld us", rows, st2-st1);
......@@ -839,7 +840,6 @@ static int32_t optimizeTbnameInCondImpl(void* metaHandle, SArray* pExistedUidLis
return -1;
}
static SSDataBlock* createTagValBlockForFilter(SArray* pColList, int32_t numOfTables, SArray* pUidTagList,
void* metaHandle) {
SSDataBlock* pResBlock = createDataBlock();
......@@ -1096,8 +1096,11 @@ int32_t getTableList(void* metaHandle, void* pVnode, SScanPhysiNode* pScanNode,
// failed to find the result in the cache, let try to calculate the results
if (pTagIndexCond) {
void* pIndex = tsdbGetIvtIdx(metaHandle);
SIndexMetaArg metaArg = {
.metaEx = metaHandle, .idx = tsdbGetIdx(metaHandle), .ivtIdx = pIndex, .suid = pScanNode->uid};
SIndexMetaArg metaArg = {.metaEx = metaHandle,
.idx = tsdbGetIdx(metaHandle),
.ivtIdx = pIndex,
.suid = pScanNode->uid,
.metaFilterFunc = metaFilterTableIds};
status = SFLT_NOT_INDEX;
code = doFilterTag(pTagIndexCond, &metaArg, pUidList, &status);
......@@ -1514,7 +1517,7 @@ static int32_t setSelectValueColumnInfo(SqlFunctionCtx* pCtx, int32_t numOfOutpu
return TSDB_CODE_OUT_OF_MEMORY;
}
SHashObj *pSelectFuncs = taosHashInit(8, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_ENTRY_LOCK);
SHashObj* pSelectFuncs = taosHashInit(8, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_ENTRY_LOCK);
for (int32_t i = 0; i < numOfOutput; ++i) {
const char* pName = pCtx[i].pExpr->pExpr->_function.functionName;
if ((strcmp(pName, "_select_value") == 0) || (strcmp(pName, "_group_key") == 0)) {
......@@ -1841,9 +1844,7 @@ uint64_t tableListGetSize(const STableListInfo* pTableList) {
return taosArrayGetSize(pTableList->pTableList);
}
uint64_t tableListGetSuid(const STableListInfo* pTableList) {
return pTableList->idInfo.suid;
}
uint64_t tableListGetSuid(const STableListInfo* pTableList) { return pTableList->idInfo.suid; }
STableKeyInfo* tableListGetInfo(const STableListInfo* pTableList, int32_t index) {
if (taosArrayGetSize(pTableList->pTableList) == 0) {
......@@ -2033,8 +2034,8 @@ static int32_t sortTableGroup(STableListInfo* pTableListInfo) {
return TDB_CODE_SUCCESS;
}
int32_t buildGroupIdMapForAllTables(STableListInfo* pTableListInfo, SReadHandle* pHandle, SScanPhysiNode* pScanNode, SNodeList* group,
bool groupSort, uint8_t *digest) {
int32_t buildGroupIdMapForAllTables(STableListInfo* pTableListInfo, SReadHandle* pHandle, SScanPhysiNode* pScanNode,
SNodeList* group, bool groupSort, uint8_t* digest) {
int32_t code = TSDB_CODE_SUCCESS;
bool groupByTbname = groupbyTbname(group);
......@@ -2086,7 +2087,8 @@ int32_t createScanTableListInfo(SScanPhysiNode* pScanNode, SNodeList* pGroupTags
}
uint8_t digest[17] = {0};
int32_t code = getTableList(pHandle->meta, pHandle->vnode, pScanNode, pTagCond, pTagIndexCond, pTableListInfo, digest, idStr);
int32_t code =
getTableList(pHandle->meta, pHandle->vnode, pScanNode, pTagCond, pTagIndexCond, pTableListInfo, digest, idStr);
if (code != TSDB_CODE_SUCCESS) {
qError("failed to getTableList, code: %s", tstrerror(code));
return code;
......
source/libs/index/CMakeLists.txt
浏览文件 @ c15b1e35
......@@ -12,7 +12,6 @@ target_link_libraries(
PUBLIC os
PUBLIC util
PUBLIC common
PUBLIC vnode
PUBLIC nodes
PUBLIC scalar
PUBLIC function
......
source/libs/index/src/indexFilter.c
浏览文件 @ c15b1e35
......@@ -13,6 +13,7 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "filter.h"
#include "index.h"
#include "indexComm.h"
#include "indexInt.h"
......@@ -20,7 +21,6 @@
#include "querynodes.h"
#include "scalar.h"
#include "tdatablock.h"
#include "vnode.h"
// clang-format off
#define SIF_ERR_RET(c) do { int32_t _code = c; if (_code != TSDB_CODE_SUCCESS) { terrno = _code; return _code; } } while (0)
......@@ -659,7 +659,7 @@ static int32_t sifDoIndex(SIFParam *left, SIFParam *right, int8_t operType, SIFP
} else {
if (sifSetFltParam(left, right, &typedata, &param) != 0) return -1;
}
ret = metaFilterTableIds(arg->metaEx, &param, output->result);
ret = arg->metaFilterFunc(arg->metaEx, &param, output->result);
}
return ret;
}
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或