From 2e507dc8b74bd2ea0c0a09005b54c136c3ab9f95 Mon Sep 17 00:00:00 2001 From: Xiaoyu Wang Date: Fri, 17 Dec 2021 23:01:27 -0500 Subject: [PATCH 1/4] TD-12193 physical plan serialization --- include/libs/planner/planner.h | 4 +- source/libs/planner/src/physicalPlanJson.c | 191 +++++++++++++++++++++ 2 files changed, 194 insertions(+), 1 deletion(-) diff --git a/include/libs/planner/planner.h b/include/libs/planner/planner.h index f86ce77803..16271310e5 100644 --- a/include/libs/planner/planner.h +++ b/include/libs/planner/planner.h @@ -108,7 +108,7 @@ typedef struct SProjectPhyNode { typedef struct SExchangePhyNode { SPhyNode node; uint64_t srcTemplateId; // template id of datasource suplans - SArray *pSourceEpSet; // SEpSet, scheduler fill by calling qSetSuplanExecutionNode + SArray *pSrcEndPoints; // SEpAddrMsg, scheduler fill by calling qSetSuplanExecutionNode } SExchangePhyNode; typedef struct SSubplanId { @@ -129,6 +129,7 @@ typedef struct SSubplan { typedef struct SQueryDag { uint64_t queryId; + int32_t numOfSubplans; SArray *pSubplans; // Element is SArray*, and nested element is SSubplan. The execution level of subplan, starting from 0. } SQueryDag; @@ -137,6 +138,7 @@ typedef struct SQueryDag { */ int32_t qCreateQueryDag(const struct SQueryStmtInfo* pQueryInfo, struct SEpSet* pQnode, struct SQueryDag** pDag); +// @eps is an array of SEpAddr structures int32_t qSetSubplanExecutionNode(SSubplan* subplan, SArray* eps); int32_t qExplainQuery(const struct SQueryStmtInfo* pQueryInfo, struct SEpSet* pQnode, char** str); diff --git a/source/libs/planner/src/physicalPlanJson.c b/source/libs/planner/src/physicalPlanJson.c index 7c0dc25627..2510797158 100644 --- a/source/libs/planner/src/physicalPlanJson.c +++ b/source/libs/planner/src/physicalPlanJson.c @@ -463,6 +463,191 @@ static bool exprInfoFromJson(const cJSON* json, void* obj) { return res; } +static const char* jkTimeWindowStartKey = "StartKey"; +static const char* jkTimeWindowEndKey = "EndKey"; + +static bool timeWindowToJson(const void* obj, cJSON* json) { + const STimeWindow* win = (const STimeWindow*)obj; + bool res = cJSON_AddNumberToObject(json, jkTimeWindowStartKey, win->skey); + if (res) { + res = cJSON_AddNumberToObject(json, jkTimeWindowEndKey, win->ekey); + } + return res; +} + +static bool timeWindowFromJson(const cJSON* json, void* obj) { + STimeWindow* win = (STimeWindow*)obj; + win->skey = getNumber(json, jkTimeWindowStartKey); + win->ekey = getNumber(json, jkTimeWindowEndKey); + return true; +} + +static const char* jkScanNodeTableId = "TableId"; +static const char* jkScanNodeTableType = "TableType"; + +static bool scanNodeToJson(const void* obj, cJSON* json) { + const SScanPhyNode* scan = (const SScanPhyNode*)obj; + bool res = cJSON_AddNumberToObject(json, jkScanNodeTableId, scan->uid); + if (res) { + res = cJSON_AddNumberToObject(json, jkScanNodeTableType, scan->tableType); + } + return res; +} + +static bool scanNodeFromJson(const cJSON* json, void* obj) { + SScanPhyNode* scan = (SScanPhyNode*)obj; + scan->uid = getNumber(json, jkScanNodeTableId); + scan->tableType = getNumber(json, jkScanNodeTableType); + return true; +} + +static const char* jkTableScanNodeFlag = "Flag"; +static const char* jkTableScanNodeWindow = "Window"; +static const char* jkTableScanNodeTagsConditions = "TagsConditions"; + +static bool tableScanNodeToJson(const void* obj, cJSON* json) { + const STableScanPhyNode* scan = (const STableScanPhyNode*)obj; + bool res = scanNodeToJson(obj, json); + if (res) { + res = cJSON_AddNumberToObject(json, jkTableScanNodeFlag, scan->scanFlag); + } + if (res) { + res = addObject(json, jkTableScanNodeWindow, timeWindowToJson, &scan->window); + } + if (res) { + res = addArray(json, jkTableScanNodeTagsConditions, exprInfoToJson, scan->pTagsConditions); + } + return res; +} + +static bool tableScanNodeFromJson(const cJSON* json, void* obj) { + STableScanPhyNode* scan = (STableScanPhyNode*)obj; + bool res = scanNodeFromJson(json, obj); + if (res) { + scan->scanFlag = getNumber(json, jkTableScanNodeFlag); + } + if (res) { + res = fromObject(json, jkTableScanNodeWindow, timeWindowFromJson, &scan->window, true); + } + if (res) { + res = fromArray(json, jkTableScanNodeTagsConditions, exprInfoFromJson, &scan->pTagsConditions, sizeof(SExprInfo)); + } + return res; +} + +static const char* jkEpAddrFqdn = "Fqdn"; +static const char* jkEpAddrPort = "Port"; + +static bool epAddrToJson(const void* obj, cJSON* json) { + const SEpAddrMsg* ep = (const SEpAddrMsg*)obj; + bool res = cJSON_AddStringToObject(json, jkEpAddrFqdn, ep->fqdn); + if (res) { + res = cJSON_AddNumberToObject(json, jkEpAddrPort, ep->port); + } + return res; +} + +static bool epAddrFromJson(const cJSON* json, void* obj) { + SEpAddrMsg* ep = (SEpAddrMsg*)obj; + copyString(json, jkEpAddrFqdn, ep->fqdn); + ep->port = getNumber(json, jkEpAddrPort); + return true; +} + +static const char* jkExchangeNodeSrcTemplateId = "SrcTemplateId"; +static const char* jkExchangeNodeSrcEndPoints = "SrcEndPoints"; + +static bool exchangeNodeToJson(const void* obj, cJSON* json) { + const SExchangePhyNode* exchange = (const SExchangePhyNode*)obj; + bool res = cJSON_AddNumberToObject(json, jkExchangeNodeSrcTemplateId, exchange->srcTemplateId); + if (res) { + res = addArray(json, jkExchangeNodeSrcEndPoints, epAddrToJson, exchange->pSrcEndPoints); + } + return res; +} + +static bool exchangeNodeFromJson(const cJSON* json, void* obj) { + SExchangePhyNode* exchange = (SExchangePhyNode*)obj; + exchange->srcTemplateId = getNumber(json, jkExchangeNodeSrcTemplateId); + return fromArray(json, jkExchangeNodeSrcEndPoints, epAddrFromJson, &exchange->pSrcEndPoints, sizeof(SEpAddrMsg)); +} + +static bool specificPhyNodeToJson(const void* obj, cJSON* json) { + const SPhyNode* phyNode = (const SPhyNode*)obj; + switch (phyNode->info.type) { + case OP_TableScan: + case OP_DataBlocksOptScan: + case OP_TableSeqScan: + return tableScanNodeToJson(obj, json); + case OP_TagScan: + case OP_SystemTableScan: + return scanNodeToJson(obj, json); + case OP_Aggregate: + break; // todo + case OP_Project: + return true; + case OP_Groupby: + case OP_Limit: + case OP_SLimit: + case OP_TimeWindow: + case OP_SessionWindow: + case OP_StateWindow: + case OP_Fill: + case OP_MultiTableAggregate: + case OP_MultiTableTimeInterval: + case OP_Filter: + case OP_Distinct: + case OP_Join: + case OP_AllTimeWindow: + case OP_AllMultiTableTimeInterval: + case OP_Order: + break; // todo + case OP_Exchange: + return exchangeNodeToJson(obj, json); + default: + break; + } + return false; +} + +static bool specificPhyNodeFromJson(const cJSON* json, void* obj) { + SPhyNode* phyNode = (SPhyNode*)obj; + switch (phyNode->info.type) { + case OP_TableScan: + case OP_DataBlocksOptScan: + case OP_TableSeqScan: + return tableScanNodeFromJson(json, obj); + case OP_TagScan: + case OP_SystemTableScan: + return scanNodeFromJson(json, obj); + case OP_Aggregate: + break; // todo + case OP_Project: + return true; + case OP_Groupby: + case OP_Limit: + case OP_SLimit: + case OP_TimeWindow: + case OP_SessionWindow: + case OP_StateWindow: + case OP_Fill: + case OP_MultiTableAggregate: + case OP_MultiTableTimeInterval: + case OP_Filter: + case OP_Distinct: + case OP_Join: + case OP_AllTimeWindow: + case OP_AllMultiTableTimeInterval: + case OP_Order: + break; // todo + case OP_Exchange: + return exchangeNodeFromJson(json, obj); + default: + break; + } + return false; +} + static const char* jkPnodeName = "Name"; static const char* jkPnodeTargets = "Targets"; static const char* jkPnodeConditions = "Conditions"; @@ -484,6 +669,9 @@ static bool phyNodeToJson(const void* obj, cJSON* jNode) { if (res) { res = addArray(jNode, jkPnodeChildren, phyNodeToJson, phyNode->pChildren); } + if (res) { + res = addObject(jNode, phyNode->info.name, specificPhyNodeToJson, phyNode); + } return res; } @@ -501,6 +689,9 @@ static bool phyNodeFromJson(const cJSON* json, void* obj) { if (res) { res = fromArray(json, jkPnodeChildren, phyNodeFromJson, &node->pChildren, sizeof(SSlotSchema)); } + if (res) { + res = fromObject(json, node->info.name, specificPhyNodeFromJson, node, true); + } return res; } From 1afa3de4c1b7aeb4262aad62a3780c5d356ab652 Mon Sep 17 00:00:00 2001 From: Xiaoyu Wang Date: Fri, 17 Dec 2021 23:34:02 -0500 Subject: [PATCH 2/4] TD-12194 Improve the planner interface. --- include/libs/planner/planner.h | 7 ++++-- source/libs/planner/inc/plannerInt.h | 3 ++- source/libs/planner/src/physicalPlan.c | 30 +++++++++++++++----------- source/libs/planner/src/planner.c | 4 ++++ 4 files changed, 28 insertions(+), 16 deletions(-) diff --git a/include/libs/planner/planner.h b/include/libs/planner/planner.h index 16271310e5..3f4d21a746 100644 --- a/include/libs/planner/planner.h +++ b/include/libs/planner/planner.h @@ -138,8 +138,11 @@ typedef struct SQueryDag { */ int32_t qCreateQueryDag(const struct SQueryStmtInfo* pQueryInfo, struct SEpSet* pQnode, struct SQueryDag** pDag); -// @eps is an array of SEpAddr structures -int32_t qSetSubplanExecutionNode(SSubplan* subplan, SArray* eps); +// Set datasource of this subplan, multiple calls may be made to a subplan. +// @subplan subplan to be schedule +// @templateId templateId of a group of datasource subplans of this @subplan +// @eps Execution location of this group of datasource subplans, is an array of SEpAddr structures +int32_t qSetSubplanExecutionNode(SSubplan* subplan, uint64_t templateId, SArray* eps); int32_t qExplainQuery(const struct SQueryStmtInfo* pQueryInfo, struct SEpSet* pQnode, char** str); diff --git a/source/libs/planner/inc/plannerInt.h b/source/libs/planner/inc/plannerInt.h index 68a9518680..19563a8a0c 100644 --- a/source/libs/planner/inc/plannerInt.h +++ b/source/libs/planner/inc/plannerInt.h @@ -100,8 +100,9 @@ int32_t queryPlanToString(struct SQueryPlanNode* pQueryNode, char** str); int32_t queryPlanToSql(struct SQueryPlanNode* pQueryNode, char** sql); int32_t createDag(SQueryPlanNode* pQueryNode, struct SCatalog* pCatalog, SQueryDag** pDag); - +int32_t setSubplanExecutionNode(SSubplan* subplan, uint64_t templateId, SArray* eps); int32_t subPlanToString(const SSubplan *pPhyNode, char** str); +int32_t stringToSubplan(const char* str, SSubplan** subplan); /** * Destroy the query plan object. diff --git a/source/libs/planner/src/physicalPlan.c b/source/libs/planner/src/physicalPlan.c index 5a891d4b28..f187ec0ec9 100644 --- a/source/libs/planner/src/physicalPlan.c +++ b/source/libs/planner/src/physicalPlan.c @@ -19,6 +19,13 @@ #define STORE_CURRENT_SUBPLAN(cxt) SSubplan* _ = cxt->pCurrentSubplan #define RECOVERY_CURRENT_SUBPLAN(cxt) cxt->pCurrentSubplan = _ +typedef struct SPlanContext { + struct SCatalog* pCatalog; + struct SQueryDag* pDag; + SSubplan* pCurrentSubplan; + SSubplanId nextId; +} SPlanContext; + static const char* gOpName[] = { "Unknown", #define INCLUDE_AS_NAME @@ -26,12 +33,14 @@ static const char* gOpName[] = { #undef INCLUDE_AS_NAME }; -typedef struct SPlanContext { - struct SCatalog* pCatalog; - struct SQueryDag* pDag; - SSubplan* pCurrentSubplan; - SSubplanId nextId; -} SPlanContext; +int32_t opNameToOpType(const char* name) { + for (int32_t i = 1; i < sizeof(gOpName) / sizeof(gOpName[0]); ++i) { + if (strcmp(name, gOpName[i])) { + return i; + } + } + return OP_Unknown; +} static void toDataBlockSchema(SQueryPlanNode* pPlanNode, SDataBlockSchema* dataBlockSchema) { SWAP(dataBlockSchema->pSchema, pPlanNode->pSchema, SSchema*); @@ -216,11 +225,6 @@ int32_t createDag(SQueryPlanNode* pQueryNode, struct SCatalog* pCatalog, SQueryD return TSDB_CODE_SUCCESS; } -int32_t opNameToOpType(const char* name) { - for (int32_t i = 1; i < sizeof(gOpName) / sizeof(gOpName[0]); ++i) { - if (strcmp(name, gOpName[i])) { - return i; - } - } - return OP_Unknown; +int32_t setSubplanExecutionNode(SSubplan* subplan, uint64_t templateId, SArray* eps) { + //todo } diff --git a/source/libs/planner/src/planner.c b/source/libs/planner/src/planner.c index 3d2bb895d5..3a90acb5fd 100644 --- a/source/libs/planner/src/planner.c +++ b/source/libs/planner/src/planner.c @@ -46,6 +46,10 @@ int32_t qCreateQueryDag(const struct SQueryStmtInfo* pQueryInfo, struct SEpSet* return TSDB_CODE_SUCCESS; } +int32_t qSetSubplanExecutionNode(SSubplan* subplan, uint64_t templateId, SArray* eps) { + return setSubplanExecutionNode(subplan, templateId, eps); +} + int32_t qSubPlanToString(const SSubplan *subplan, char** str) { return subPlanToString(subplan, str); } From c51780622f04119eeb213883c625986b6130af7d Mon Sep 17 00:00:00 2001 From: yihaoDeng Date: Sat, 18 Dec 2021 18:16:07 +0800 Subject: [PATCH 3/4] adjust index interface --- cmake/cmake.options | 7 +++ include/libs/index/index.h | 20 ++++++-- source/libs/index/inc/indexInt.h | 19 ++++--- source/libs/index/inc/index_cache.h | 8 +-- source/libs/index/src/index.c | 77 ++++++++++++++++++----------- source/libs/index/src/index_cache.c | 32 ++++++------ 6 files changed, 100 insertions(+), 63 deletions(-) diff --git a/cmake/cmake.options b/cmake/cmake.options index edaab3bd45..e44a38b3f5 100644 --- a/cmake/cmake.options +++ b/cmake/cmake.options @@ -37,6 +37,7 @@ option( off ) + option( BUILD_WITH_NURAFT "If build with NuRaft" @@ -54,3 +55,9 @@ option( "If use doxygen build documents" OFF ) + +option( + USE_INVERTEDINDEX + "If use invertedIndex" + ON +) diff --git a/include/libs/index/index.h b/include/libs/index/index.h index 0885ce151e..2cc2aeb6a0 100644 --- a/include/libs/index/index.h +++ b/include/libs/index/index.h @@ -24,6 +24,7 @@ extern "C" { #endif typedef struct SIndex SIndex; +typedef struct SIndexTerm SIndexTerm; typedef struct SIndexOpts SIndexOpts; typedef struct SIndexMultiTermQuery SIndexMultiTermQuery; typedef struct SArray SIndexMultiTerm; @@ -35,7 +36,7 @@ typedef enum { ADD_INDEX, // add index on specify column DROP_INDEX, // drop existed index DROP_SATBLE // drop stable -} SIndexColumnType; +} SIndexOperOnColumn; typedef enum { MUST = 0, SHOULD = 1, NOT = 2 } EIndexOperatorType; typedef enum { QUERY_TERM = 0, QUERY_PREFIX = 1, QUERY_SUFFIX = 2,QUERY_REGEX = 3} EIndexQueryType; @@ -45,7 +46,7 @@ typedef enum { QUERY_TERM = 0, QUERY_PREFIX = 1, QUERY_SUFFIX = 2,QUERY_REGEX = */ SIndexMultiTermQuery *indexMultiTermQueryCreate(EIndexOperatorType oper); void indexMultiTermQueryDestroy(SIndexMultiTermQuery *pQuery); -int indexMultiTermQueryAdd(SIndexMultiTermQuery *pQuery, const char *field, int32_t nFields, const char *value, int32_t nValue, EIndexQueryType type); +int indexMultiTermQueryAdd(SIndexMultiTermQuery *pQuery, SIndexTerm *term, EIndexQueryType type); /* * @param: * @param: @@ -61,8 +62,8 @@ int indexRebuild(SIndex *index, SIndexOpts *opt); * @param */ SIndexMultiTerm *indexMultiTermCreate(); -int indexMultiTermAdd(SIndexMultiTerm *terms, const char *field, int32_t nFields, const char *value, int32_t nValue); -void indexMultiTermDestroy(SIndexMultiTerm *terms); +int indexMultiTermAdd(SIndexMultiTerm *terms, SIndexTerm *term); +void indexMultiTermDestroy(SIndexMultiTerm *terms); /* * @param: * @param: @@ -70,6 +71,17 @@ void indexMultiTermDestroy(SIndexMultiTerm *terms); SIndexOpts *indexOptsCreate(); void indexOptsDestroy(SIndexOpts *opts); + +/* + * @param: + * @param: + */ + +SIndexTerm *indexTermCreate(int64_t suid, SIndexOperOnColumn operType, uint8_t colType, + const char *colName, int32_t nColName, const char *colVal, int32_t nColVal); +void indexTermDestroy(SIndexTerm *p); + + #ifdef __cplusplus } #endif diff --git a/source/libs/index/inc/indexInt.h b/source/libs/index/inc/indexInt.h index cc740826e9..fb5a9e40b5 100644 --- a/source/libs/index/inc/indexInt.h +++ b/source/libs/index/inc/indexInt.h @@ -60,22 +60,21 @@ struct SIndexMultiTermQuery { // field and key; typedef struct SIndexTerm { - uint8_t type; // term data type, str/interger/json - char *key; - int32_t nKey; - char *val; - int32_t nVal; + int64_t suid; + SIndexOperOnColumn operType; // oper type, add/del/update + uint8_t colType; // term data type, str/interger/json + char *colName; + int32_t nColName; + char *colVal; + int32_t nColVal; } SIndexTerm; typedef struct SIndexTermQuery { - SIndexTerm* field_value; - EIndexQueryType type; + SIndexTerm* term; + EIndexQueryType qType; } SIndexTermQuery; -SIndexTerm *indexTermCreate(const char *key, int32_t nKey, const char *val, int32_t nVal); -void indexTermDestroy(SIndexTerm *p); - #define indexFatal(...) do { if (sDebugFlag & DEBUG_FATAL) { taosPrintLog("index FATAL ", 255, __VA_ARGS__); }} while(0) #define indexError(...) do { if (sDebugFlag & DEBUG_ERROR) { taosPrintLog("index ERROR ", 255, __VA_ARGS__); }} while(0) diff --git a/source/libs/index/inc/index_cache.h b/source/libs/index/inc/index_cache.h index 39107a78ac..b952e16a8e 100644 --- a/source/libs/index/inc/index_cache.h +++ b/source/libs/index/inc/index_cache.h @@ -38,13 +38,13 @@ typedef struct IndexCache { // IndexCache *indexCacheCreate(); -void indexCacheDestroy(IndexCache *cache); +void indexCacheDestroy(void *cache); -int indexCachePut(IndexCache *cache, int16_t fieldId, int16_t fieldType, const char *fieldValue, int32_t fvLen, +int indexCachePut(void *cache, int16_t fieldId, int16_t fieldType, const char *fieldValue, int32_t fvLen, uint32_t version, uint64_t uid, int8_t operType); -int indexCacheGet(IndexCache *cache, uint64_t *rst); -int indexCacheSearch(IndexCache *cache, SIndexMultiTermQuery *query, SArray *result); +int indexCacheGet(void *cache, uint64_t *rst); +int indexCacheSearch(void *cache, SIndexMultiTermQuery *query, SArray *result); #ifdef __cplusplus } diff --git a/source/libs/index/src/index.c b/source/libs/index/src/index.c index 6a2697491d..ca6c2062f1 100644 --- a/source/libs/index/src/index.c +++ b/source/libs/index/src/index.c @@ -46,7 +46,7 @@ SIndex *indexOpen(SIndexOpts *opts, const char *path) { index_t *index = index_open(path); sIdx->index = index; #endif - + sIdx->cache = (void*)indexCacheCreate(); sIdx->tindex = NULL; sIdx->fieldObj = taosHashInit(8, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK); @@ -61,9 +61,12 @@ void indexClose(SIndex *sIdx) { index_close(sIdex->index); sIdx->index = NULL; #endif + +#ifdef USE_INVERTEDINDEX indexCacheDestroy(sIdx->cache); taosHashCleanup(sIdx->fieldObj); pthread_mutex_destroy(&sIdx->mtx); +#endif free(sIdx); return; } @@ -86,6 +89,7 @@ int indexPut(SIndex *index, SArray* fVals, int uid) { index_document_destroy(doc); #endif +#ifdef USE_INVERTEDINDEX //TODO(yihao): reduce the lock range pthread_mutex_lock(&index->mtx); for (int i = 0; i < taosArrayGetSize(fVals); i++) { @@ -106,11 +110,16 @@ int indexPut(SIndex *index, SArray* fVals, int uid) { SIdxFieldInfo *fi = taosHashGet(index->fieldObj, p->key, p->nKey); assert(fi != NULL); int32_t fieldId = fi->fieldId; - int32_t colType = fi->type; + int32_t fieldType = fi->type; int32_t version = index->cVersion; - + int res = indexCachePut(index->cache, fieldId, fieldType, p->val, p->nVal, version, uid, p->operType); + if (ret != 0) { + return + } } pthread_mutex_unlock(&index->mtx); +#endif + return 1; } int indexSearch(SIndex *index, SIndexMultiTermQuery *multiQuerys, SArray *result) { @@ -148,16 +157,26 @@ int indexSearch(SIndex *index, SIndexMultiTermQuery *multiQuerys, SArray *result free(fields); free(keys); free(types); +#endif + +#ifdef USE_INVERTEDINDEX + #endif return 1; } int indexDelete(SIndex *index, SIndexMultiTermQuery *query) { +#ifdef USE_INVERTEDINDEX +#endif return 1; } -int indexRebuild(SIndex *index, SIndexOpts *opts); +int indexRebuild(SIndex *index, SIndexOpts *opts) { +#ifdef USE_INVERTEDINDEX +#endif + +} SIndexOpts *indexOptsCreate() { @@ -184,53 +203,55 @@ SIndexMultiTermQuery *indexMultiTermQueryCreate(EIndexOperatorType opera) { void indexMultiTermQueryDestroy(SIndexMultiTermQuery *pQuery) { for (int i = 0; i < taosArrayGetSize(pQuery->query); i++) { SIndexTermQuery *p = (SIndexTermQuery *)taosArrayGet(pQuery->query, i); - indexTermDestroy(p->field_value); + indexTermDestroy(p->term); } taosArrayDestroy(pQuery->query); free(pQuery); }; -int indexMultiTermQueryAdd(SIndexMultiTermQuery *pQuery, const char *field, int32_t nFields, const char *value, int32_t nValue, EIndexQueryType type){ - SIndexTerm *t = indexTermCreate(field, nFields, value, nValue); - if (t == NULL) {return -1;} - SIndexTermQuery q = {.type = type, .field_value = t}; +int indexMultiTermQueryAdd(SIndexMultiTermQuery *pQuery, SIndexTerm *term, EIndexQueryType qType){ + SIndexTermQuery q = {.qType = qType, .term = term}; taosArrayPush(pQuery->query, &q); return 0; } -SIndexTerm *indexTermCreate(const char *key, int32_t nKey, const char *val, int32_t nVal) { - SIndexTerm *t = (SIndexTerm *)malloc(sizeof(SIndexTerm)); - t->key = (char *)calloc(nKey + 1, 1); - memcpy(t->key, key, nKey); - t->nKey = nKey; +SIndexTerm *indexTermCreate(int64_t suid, SIndexOperOnColumn oper, uint8_t colType, const char *colName, int32_t nColName, const char *colVal, int32_t nColVal) { + SIndexTerm *t = (SIndexTerm *)calloc(1, (sizeof(SIndexTerm))); + if (t == NULL) { return NULL; } - t->val = (char *)calloc(nVal + 1, 1); - memcpy(t->val, val, nVal); - t->nVal = nVal; + t->suid = suid; + t->operType= oper; + t->colType = colType; + + t->colName = (char *)calloc(1, nColName + 1); + memcpy(t->colName, colName, nColName); + t->nColName = nColName; + + t->colVal = (char *)calloc(1, nColVal + 1); + memcpy(t->colVal, colVal, nColVal); + t->nColVal = nColVal; return t; } void indexTermDestroy(SIndexTerm *p) { - free(p->key); - free(p->val); + free(p->colName); + free(p->colVal); free(p); } -SArray *indexMultiTermCreate() { +SIndexMultiTerm *indexMultiTermCreate() { return taosArrayInit(4, sizeof(SIndexTerm *)); } -int indexMultiTermAdd(SArray *array, const char *field, int32_t nField, const char *val, int32_t nVal) { - SIndexTerm *term = indexTermCreate(field, nField, val, nVal); - if (term == NULL) { return -1; } - taosArrayPush(array, &term); +int indexMultiTermAdd(SIndexMultiTerm *terms, SIndexTerm *term) { + taosArrayPush(terms, &term); return 0; } -void indexMultiTermDestroy(SArray *array) { - for (int32_t i = 0; i < taosArrayGetSize(array); i++) { - SIndexTerm *p = taosArrayGetP(array, i); +void indexMultiTermDestroy(SIndexMultiTerm *terms) { + for (int32_t i = 0; i < taosArrayGetSize(terms); i++) { + SIndexTerm *p = taosArrayGetP(terms, i); indexTermDestroy(p); } - taosArrayDestroy(array); + taosArrayDestroy(terms); } void indexInit() { diff --git a/source/libs/index/src/index_cache.c b/source/libs/index/src/index_cache.c index acb8e32157..23f7a08823 100644 --- a/source/libs/index/src/index_cache.c +++ b/source/libs/index/src/index_cache.c @@ -16,7 +16,7 @@ #include "index_cache.h" #include "tcompare.h" -#define MAX_INDEX_KEY_LEN 128 // test only, change later +#define MAX_INDEX_KEY_LEN 256// test only, change later static char* getIndexKey(const void *pData) { return NULL; @@ -96,16 +96,19 @@ IndexCache *indexCacheCreate() { } -void indexCacheDestroy(IndexCache *cache) { - if (cache == NULL) { return; } - tSkipListDestroy(cache->skiplist); - free(cache); +void indexCacheDestroy(void *cache) { + IndexCache *pCache = cache; + if (pCache == NULL) { return; } + tSkipListDestroy(pCache->skiplist); + free(pCache); } -int indexCachePut(IndexCache *cache, int16_t fieldId, int16_t fieldType, const char *fieldValue, int32_t fvLen, +int indexCachePut(void *cache, int16_t fieldId, int16_t fieldType, const char *fieldValue, int32_t fvLen, uint32_t version, uint64_t uid, int8_t operType) { if (cache == NULL) { return -1;} + IndexCache *pCache = cache; + // encode data int32_t total = sizeof(int32_t) + sizeof(fieldId) + sizeof(fieldType) + sizeof(fvLen) + fvLen + sizeof(version) + sizeof(uid) + sizeof(operType); @@ -135,20 +138,15 @@ int indexCachePut(IndexCache *cache, int16_t fieldId, int16_t fieldType, const c memcpy(p, &operType, sizeof(operType)); p += sizeof(operType); - tSkipListPut(cache->skiplist, (void *)buf); + tSkipListPut(pCache->skiplist, (void *)buf); // encode end - } -int indexCacheDel(IndexCache *cache, int32_t fieldId, const char *fieldValue, int32_t fvlen, uint64_t uid, int8_t operType) { - +int indexCacheDel(void *cache, int32_t fieldId, const char *fieldValue, int32_t fvlen, uint64_t uid, int8_t operType) { + IndexCache *pCache = cache; + return 0; } -int indexCacheSearch(IndexCache *cache, SIndexMultiTermQuery *query, SArray *result) { - +int indexCacheSearch(void *cache, SIndexMultiTermQuery *query, SArray *result) { + return 0; } - - - - - From f074a4b3a5c99dcbda3f0f8c857bb92f9d19dd02 Mon Sep 17 00:00:00 2001 From: yihaoDeng Date: Sat, 18 Dec 2021 23:06:08 +0800 Subject: [PATCH 4/4] update cache put --- cmake/cmake.options | 2 +- include/libs/index/index.h | 2 +- source/libs/index/CMakeLists.txt | 6 +- source/libs/index/inc/indexInt.h | 4 +- source/libs/index/src/index.c | 54 ++++++------ source/libs/index/test/CMakeLists.txt | 2 +- .../test/{indexTests.cpp => indexTests.cc} | 86 +++++++++++++++++-- 7 files changed, 116 insertions(+), 40 deletions(-) rename source/libs/index/test/{indexTests.cpp => indexTests.cc} (74%) diff --git a/cmake/cmake.options b/cmake/cmake.options index e44a38b3f5..2384a427e4 100644 --- a/cmake/cmake.options +++ b/cmake/cmake.options @@ -57,7 +57,7 @@ option( ) option( - USE_INVERTEDINDEX + BUILD_WITH_INVERTEDINDEX "If use invertedIndex" ON ) diff --git a/include/libs/index/index.h b/include/libs/index/index.h index 2cc2aeb6a0..3ca8d10603 100644 --- a/include/libs/index/index.h +++ b/include/libs/index/index.h @@ -51,7 +51,7 @@ int indexMultiTermQueryAdd(SIndexMultiTermQuery *pQuery, SIndexTerm * @param: * @param: */ -SIndex* indexOpen(SIndexOpts *opt, const char *path); +int indexOpen(SIndexOpts *opt, const char *path, SIndex **index); void indexClose(SIndex *index); int indexPut(SIndex *index, SIndexMultiTerm *terms, int uid); int indexDelete(SIndex *index, SIndexMultiTermQuery *query); diff --git a/source/libs/index/CMakeLists.txt b/source/libs/index/CMakeLists.txt index f68fc5e61e..4805bd3b77 100644 --- a/source/libs/index/CMakeLists.txt +++ b/source/libs/index/CMakeLists.txt @@ -22,9 +22,13 @@ if (${BUILD_WITH_LUCENE}) index PUBLIC lucene++ ) - endif(${BUILD_WITH_LUCENE}) +if (${BUILD_WITH_INVERTEDINDEX}) + add_definitions(-DUSE_INVERTED_INDEX) +endif(${BUILD_WITH_INVERTEDINDEX}) + + if (${BUILD_TEST}) add_subdirectory(test) endif(${BUILD_TEST}) diff --git a/source/libs/index/inc/indexInt.h b/source/libs/index/inc/indexInt.h index fb5a9e40b5..7e017049e8 100644 --- a/source/libs/index/inc/indexInt.h +++ b/source/libs/index/inc/indexInt.h @@ -37,10 +37,10 @@ struct SIndex { #endif void *cache; void *tindex; - SHashObj *fieldObj;// < field name, field id> + SHashObj *colObj;// < field name, field id> int64_t suid; // current super table id, -1 is normal table - int fieldId; // field id allocated to cache + int colId; // field id allocated to cache int32_t cVersion; // current version allocated to cache pthread_mutex_t mtx; }; diff --git a/source/libs/index/src/index.c b/source/libs/index/src/index.c index ca6c2062f1..08c59d8d43 100644 --- a/source/libs/index/src/index.c +++ b/source/libs/index/src/index.c @@ -22,11 +22,10 @@ #endif -typedef struct SIdxFieldInfo { - int fieldId; // generated by index internal +typedef struct SIdxColInfo { + int colId; // generated by index internal int cVersion; - int type; // field type -} SIdxFieldInfo; +} SIdxColInfo; static pthread_once_t isInit = PTHREAD_ONCE_INIT; static void indexInit(); @@ -38,9 +37,10 @@ static int indexMergeCacheIntoTindex(struct SIndex *sIdx) { indexWarn("suid %" PRIu64 " merge cache into tindex", sIdx->suid); return 0; } -SIndex *indexOpen(SIndexOpts *opts, const char *path) { +int indexOpen(SIndexOpts *opts, const char *path, SIndex **index) { pthread_once(&isInit, indexInit); SIndex *sIdx = calloc(1, sizeof(SIndex)); + if (sIdx == NULL) { return -1; } #ifdef USE_LUCENE index_t *index = index_open(path); @@ -49,11 +49,13 @@ SIndex *indexOpen(SIndexOpts *opts, const char *path) { sIdx->cache = (void*)indexCacheCreate(); sIdx->tindex = NULL; - sIdx->fieldObj = taosHashInit(8, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK); - sIdx->fieldId = 1; + sIdx->colObj = taosHashInit(8, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK); + sIdx->colId = 1; sIdx->cVersion = 1; pthread_mutex_init(&sIdx->mtx, NULL); - return sIdx; + + *index = sIdx; + return 0; } void indexClose(SIndex *sIdx) { @@ -62,16 +64,16 @@ void indexClose(SIndex *sIdx) { sIdx->index = NULL; #endif -#ifdef USE_INVERTEDINDEX +#ifdef USE_INVERTED_INDEX indexCacheDestroy(sIdx->cache); - taosHashCleanup(sIdx->fieldObj); + taosHashCleanup(sIdx->colObj); pthread_mutex_destroy(&sIdx->mtx); #endif free(sIdx); return; } -int indexPut(SIndex *index, SArray* fVals, int uid) { +int indexPut(SIndex *index, SIndexMultiTerm * fVals, int uid) { #ifdef USE_LUCENE index_document_t *doc = index_document_create(); @@ -89,38 +91,38 @@ int indexPut(SIndex *index, SArray* fVals, int uid) { index_document_destroy(doc); #endif -#ifdef USE_INVERTEDINDEX +#ifdef USE_INVERTED_INDEX + //TODO(yihao): reduce the lock range pthread_mutex_lock(&index->mtx); for (int i = 0; i < taosArrayGetSize(fVals); i++) { SIndexTerm *p = taosArrayGetP(fVals, i); - SIdxFieldInfo *fi = taosHashGet(index->fieldObj, p->key, p->nKey); + SIdxColInfo *fi = taosHashGet(index->colObj, p->colName, p->nColName); if (fi == NULL) { - SIdxFieldInfo tfi = {.fieldId = index->fieldId, .type = p->type}; + SIdxColInfo tfi = {.colId = index->colId}; index->cVersion++; - index->fieldId++; - taosHashPut(index->fieldObj, p->key, p->nKey, &tfi, sizeof(tfi)); + index->colId++; + taosHashPut(index->colObj, p->colName, p->nColName, &tfi, sizeof(tfi)); } else { //TODO, del } } + pthread_mutex_unlock(&index->mtx); for (int i = 0; i < taosArrayGetSize(fVals); i++) { SIndexTerm *p = taosArrayGetP(fVals, i); - SIdxFieldInfo *fi = taosHashGet(index->fieldObj, p->key, p->nKey); + SIdxColInfo *fi = taosHashGet(index->colObj, p->colName, p->nColName); assert(fi != NULL); - int32_t fieldId = fi->fieldId; - int32_t fieldType = fi->type; + int32_t colId = fi->colId; int32_t version = index->cVersion; - int res = indexCachePut(index->cache, fieldId, fieldType, p->val, p->nVal, version, uid, p->operType); + int ret = indexCachePut(index->cache, colId, p->colType, p->colVal, p->nColVal, version, uid, p->operType); if (ret != 0) { - return + return ret; } } - pthread_mutex_unlock(&index->mtx); #endif - return 1; + return 0; } int indexSearch(SIndex *index, SIndexMultiTermQuery *multiQuerys, SArray *result) { #ifdef USE_LUCENE @@ -159,7 +161,7 @@ int indexSearch(SIndex *index, SIndexMultiTermQuery *multiQuerys, SArray *result free(types); #endif -#ifdef USE_INVERTEDINDEX +#ifdef USE_INVERTED_INDEX #endif return 1; @@ -167,13 +169,13 @@ int indexSearch(SIndex *index, SIndexMultiTermQuery *multiQuerys, SArray *result int indexDelete(SIndex *index, SIndexMultiTermQuery *query) { -#ifdef USE_INVERTEDINDEX +#ifdef USE_INVERTED_INDEX #endif return 1; } int indexRebuild(SIndex *index, SIndexOpts *opts) { -#ifdef USE_INVERTEDINDEX +#ifdef USE_INVERTED_INDEX #endif } diff --git a/source/libs/index/test/CMakeLists.txt b/source/libs/index/test/CMakeLists.txt index f84f874a23..6eb532b41e 100644 --- a/source/libs/index/test/CMakeLists.txt +++ b/source/libs/index/test/CMakeLists.txt @@ -1,7 +1,7 @@ add_executable(indexTest "") target_sources(indexTest PRIVATE - "indexTests.cpp" + "indexTests.cc" ) target_include_directories ( indexTest PUBLIC diff --git a/source/libs/index/test/indexTests.cpp b/source/libs/index/test/indexTests.cc similarity index 74% rename from source/libs/index/test/indexTests.cpp rename to source/libs/index/test/indexTests.cc index f582536817..9dff2e9ea0 100644 --- a/source/libs/index/test/indexTests.cpp +++ b/source/libs/index/test/indexTests.cc @@ -1,3 +1,17 @@ +/* + * Copyright (c) 2019 TAOS Data, Inc. + * + * This program is free software: you can use, redistribute, and/or modify + * it under the terms of the GNU Affero General Public License, version 3 + * or later ("AGPL"), as published by the Free Software Foundation. + * + * This program is distributed in the hope that it will be useful, but WITHOUT + * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or + * FITNESS FOR A PARTICULAR PURPOSE. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ #include #include #include @@ -61,7 +75,7 @@ class FstReadMemory { // add later bool Search(AutomationCtx *ctx, std::vector &result) { FstStreamBuilder *sb = fstSearch(_fst, ctx); - StreamWithState *st = streamBuilderIntoStream(sb); + StreamWithState *st = streamBuilderIntoStream(sb); StreamWithStateResult *rt = NULL; while ((rt = streamWithStateNextWith(st, NULL)) != NULL) { @@ -279,15 +293,71 @@ void validateFst() { delete m; } +class IndexEnv : public ::testing::Test { + protected: + virtual void SetUp() { + taosRemoveDir(path); + opts = indexOptsCreate(); + int ret = indexOpen(opts, path, &index); + assert(ret == 0); + } + virtual void TearDown() { + indexClose(index); + indexOptsDestroy(opts); + } + + const char *path = "/tmp/tindex"; + SIndexOpts *opts; + SIndex *index; +}; -int main(int argc, char** argv) { - checkFstPerf(); - //checkFstPrefixSearch(); - return 1; +TEST_F(IndexEnv, testPut) { + + // single index column + { + + std::string colName("tag1"), colVal("Hello world"); + SIndexTerm *term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), colVal.c_str(), colVal.size()); + SIndexMultiTerm *terms = indexMultiTermCreate(); + indexMultiTermAdd(terms, term); + + for (size_t i = 0; i < 100; i++) { + int tableId = i; + int ret = indexPut(index, terms, tableId); + assert(ret == 0); + } + indexMultiTermDestroy(terms); + } + // multi index column + { + + SIndexMultiTerm *terms = indexMultiTermCreate(); + { + std::string colName("tag1"), colVal("Hello world"); + SIndexTerm *term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), colVal.c_str(), colVal.size()); + indexMultiTermAdd(terms, term); + } + { + std::string colName("tag2"), colVal("Hello world"); + SIndexTerm *term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), colVal.c_str(), colVal.size()); + indexMultiTermAdd(terms, term); + } + + for (int i = 0; i < 100; i++) { + int tableId = i; + int ret = indexPut(index, terms, tableId); + assert(ret == 0); + } + indexMultiTermDestroy(terms); + } + // +} + +TEST_F(IndexEnv, testDel) { + } -//TEST(IndexFstBuilder, IndexFstInput) { -// -//} + +