Merge pull request #17163 from taosdata/enh/optTbnameIN

opt(query): opt tbname_in query

include/libs/index/index.h

@@ -206,12 +206,6 @@ void indexJsonRebuild(SIndexJson* idx, void* iter);
  **/
 bool indexJsonIsRebuild(SIndexJson* idx);
 
-/*
- * init index env
- *
- */
-void indexInit();
-
 /* index filter */
 typedef struct SIndexMetaArg {
   void*    metaEx;
@@ -225,6 +219,12 @@ typedef enum { SFLT_NOT_INDEX, SFLT_COARSE_INDEX, SFLT_ACCURATE_INDEX } SIdxFltS
 SIdxFltStatus idxGetFltStatus(SNode* pFilterNode);
 
 int32_t doFilterTag(SNode* pFilterNode, SIndexMetaArg* metaArg, SArray* result, SIdxFltStatus* status);
+
+/*
+ *  init index env
+ *
+ */
+void indexInit(int32_t threads);
 /*
  * destory index env
  *

source/dnode/mgmt/node_mgmt/src/dmMgmt.c

@@ -193,6 +193,8 @@ int32_t dmInitDnode(SDnode *pDnode, EDndNodeType rtype) {
     goto _OVER;
   }
 
+  indexInit(tsNumOfCommitThreads);
+
   dmReportStartup("dnode-transport", "initialized");
   dDebug("dnode is created, ptr:%p", pDnode);
   code = 0;

source/dnode/vnode/inc/vnode.h

@@ -62,7 +62,9 @@ void    vnodeGetSnapshot(SVnode *pVnode, SSnapshot *pSnapshot);
 void    vnodeGetInfo(SVnode *pVnode, const char **dbname, int32_t *vgId);
 int32_t vnodeProcessCreateTSma(SVnode *pVnode, void *pCont, uint32_t contLen);
 int32_t vnodeGetAllTableList(SVnode *pVnode, uint64_t uid, SArray *list);
+
 int32_t vnodeGetCtbIdList(SVnode *pVnode, int64_t suid, SArray *list);
+int32_t vnodeGetCtbIdListByFilter(SVnode *pVnode, int64_t suid, SArray *list, bool (*filter)(void *arg), void *arg);
 int32_t vnodeGetStbIdList(SVnode *pVnode, int64_t suid, SArray *list);
 void   *vnodeGetIdx(SVnode *pVnode);
 void   *vnodeGetIvtIdx(SVnode *pVnode);
@@ -94,9 +96,12 @@ void        metaReaderClear(SMetaReader *pReader);
 int32_t     metaGetTableEntryByUid(SMetaReader *pReader, tb_uid_t uid);
 int         metaGetTableEntryByName(SMetaReader *pReader, const char *name);
 int32_t     metaGetTableTags(SMeta *pMeta, uint64_t suid, SArray *uidList, SHashObj *tags);
+int32_t     metaGetTableTagsOpt(SMeta *pMeta, uint64_t suid, SArray *uidList, SHashObj *tags);
 int32_t     metaReadNext(SMetaReader *pReader);
 const void *metaGetTableTagVal(void *tag, int16_t type, STagVal *tagVal);
 int         metaGetTableNameByUid(void *meta, uint64_t uid, char *tbName);
+int         metaGetTableUidByName(void *meta, char *tbName, uint64_t *uid);
+int         metaGetTableTypeByName(void *meta, char *tbName, ETableType *tbType);
 bool        metaIsTableExist(SMeta *pMeta, tb_uid_t uid);
 
 typedef struct SMetaFltParam {

source/dnode/vnode/src/inc/vnodeInt.h

@@ -110,22 +110,24 @@ int             metaAlterTable(SMeta* pMeta, int64_t version, SVAlterTbReq* pReq
 SSchemaWrapper* metaGetTableSchema(SMeta* pMeta, tb_uid_t uid, int32_t sver, bool isinline);
 STSchema*       metaGetTbTSchema(SMeta* pMeta, tb_uid_t uid, int32_t sver);
 int32_t         metaGetTbTSchemaEx(SMeta* pMeta, tb_uid_t suid, tb_uid_t uid, int32_t sver, STSchema** ppTSchema);
-tb_uid_t        metaGetTableEntryUidByName(SMeta* pMeta, const char* name);
-int64_t         metaGetTbNum(SMeta* pMeta);
-int64_t         metaGetTimeSeriesNum(SMeta* pMeta);
-SMCtbCursor*    metaOpenCtbCursor(SMeta* pMeta, tb_uid_t uid);
-void            metaCloseCtbCursor(SMCtbCursor* pCtbCur);
-tb_uid_t        metaCtbCursorNext(SMCtbCursor* pCtbCur);
-SMStbCursor*    metaOpenStbCursor(SMeta* pMeta, tb_uid_t uid);
-void            metaCloseStbCursor(SMStbCursor* pStbCur);
-tb_uid_t        metaStbCursorNext(SMStbCursor* pStbCur);
-STSma*          metaGetSmaInfoByIndex(SMeta* pMeta, int64_t indexUid);
-STSmaWrapper*   metaGetSmaInfoByTable(SMeta* pMeta, tb_uid_t uid, bool deepCopy);
-SArray*         metaGetSmaIdsByTable(SMeta* pMeta, tb_uid_t uid);
-SArray*         metaGetSmaTbUids(SMeta* pMeta);
-void*           metaGetIdx(SMeta* pMeta);
-void*           metaGetIvtIdx(SMeta* pMeta);
-int             metaTtlSmaller(SMeta* pMeta, uint64_t time, SArray* uidList);
+int             metaGetTableEntryByName(SMetaReader* pReader, const char* name);
+
+tb_uid_t      metaGetTableEntryUidByName(SMeta* pMeta, const char* name);
+int64_t       metaGetTbNum(SMeta* pMeta);
+int64_t       metaGetTimeSeriesNum(SMeta* pMeta);
+SMCtbCursor*  metaOpenCtbCursor(SMeta* pMeta, tb_uid_t uid);
+void          metaCloseCtbCursor(SMCtbCursor* pCtbCur);
+tb_uid_t      metaCtbCursorNext(SMCtbCursor* pCtbCur);
+SMStbCursor*  metaOpenStbCursor(SMeta* pMeta, tb_uid_t uid);
+void          metaCloseStbCursor(SMStbCursor* pStbCur);
+tb_uid_t      metaStbCursorNext(SMStbCursor* pStbCur);
+STSma*        metaGetSmaInfoByIndex(SMeta* pMeta, int64_t indexUid);
+STSmaWrapper* metaGetSmaInfoByTable(SMeta* pMeta, tb_uid_t uid, bool deepCopy);
+SArray*       metaGetSmaIdsByTable(SMeta* pMeta, tb_uid_t uid);
+SArray*       metaGetSmaTbUids(SMeta* pMeta);
+void*         metaGetIdx(SMeta* pMeta);
+void*         metaGetIvtIdx(SMeta* pMeta);
+int           metaTtlSmaller(SMeta* pMeta, uint64_t time, SArray* uidList);
 
 int32_t metaCreateTSma(SMeta* pMeta, int64_t version, SSmaCfg* pCfg);
 int32_t metaDropTSma(SMeta* pMeta, int64_t indexUid);
@@ -183,8 +185,9 @@ int32_t tqProcessTaskRecoverRsp(STQ* pTq, SRpcMsg* pMsg);
 int32_t tqProcessTaskRetrieveReq(STQ* pTq, SRpcMsg* pMsg);
 int32_t tqProcessTaskRetrieveRsp(STQ* pTq, SRpcMsg* pMsg);
 
-SSubmitReq* tqBlockToSubmit(SVnode* pVnode, const SArray* pBlocks, const STSchema* pSchema, SSchemaWrapper* pTagSchemaWrapper, bool createTb, int64_t suid,
-                            const char* stbFullName, SBatchDeleteReq* pDeleteReq);
+SSubmitReq* tqBlockToSubmit(SVnode* pVnode, const SArray* pBlocks, const STSchema* pSchema,
+                            SSchemaWrapper* pTagSchemaWrapper, bool createTb, int64_t suid, const char* stbFullName,
+                            SBatchDeleteReq* pDeleteReq);
 
 // sma
 int32_t smaInit();

source/dnode/vnode/src/meta/metaOpen.c

@@ -200,36 +200,30 @@ int metaClose(SMeta *pMeta) {
 int32_t metaRLock(SMeta *pMeta) {
   int32_t ret = 0;
 
-  metaTrace("meta rlock %p B", &pMeta->lock);
+  metaTrace("meta rlock %p", &pMeta->lock);
 
   ret = taosThreadRwlockRdlock(&pMeta->lock);
 
-  metaTrace("meta rlock %p E", &pMeta->lock);
-
   return ret;
 }
 
 int32_t metaWLock(SMeta *pMeta) {
   int32_t ret = 0;
 
-  metaTrace("meta wlock %p B", &pMeta->lock);
+  metaTrace("meta wlock %p", &pMeta->lock);
 
   ret = taosThreadRwlockWrlock(&pMeta->lock);
 
-  metaTrace("meta wlock %p E", &pMeta->lock);
-
   return ret;
 }
 
 int32_t metaULock(SMeta *pMeta) {
   int32_t ret = 0;
 
-  metaTrace("meta ulock %p B", &pMeta->lock);
+  metaTrace("meta ulock %p", &pMeta->lock);
 
   ret = taosThreadRwlockUnlock(&pMeta->lock);
 
-  metaTrace("meta ulock %p E", &pMeta->lock);
-
   return ret;
 }
 

source/dnode/vnode/src/meta/metaQuery.c

@@ -202,6 +202,38 @@ int metaGetTableNameByUid(void *meta, uint64_t uid, char *tbName) {
 
   return 0;
 }
+int metaGetTableUidByName(void *meta, char *tbName, uint64_t *uid) {
+  int         code = 0;
+  SMetaReader mr = {0};
+  metaReaderInit(&mr, (SMeta *)meta, 0);
+
+  SMeta       *pMeta = mr.pMeta;
+  SMetaReader *pReader = &mr;
+
+  // query name.idx
+  if (tdbTbGet(pMeta->pNameIdx, tbName, strlen(tbName) + 1, &pReader->pBuf, &pReader->szBuf) < 0) {
+    terrno = TSDB_CODE_PAR_TABLE_NOT_EXIST;
+    metaReaderClear(&mr);
+    return -1;
+  }
+
+  *uid = *(tb_uid_t *)pReader->pBuf;
+  metaReaderClear(&mr);
+
+  return 0;
+}
+
+int metaGetTableTypeByName(void *meta, char *tbName, ETableType *tbType) {
+  int         code = 0;
+  SMetaReader mr = {0};
+  metaReaderInit(&mr, (SMeta *)meta, 0);
+
+  if (metaGetTableEntryByName(&mr, tbName) == 0) {
+    *tbType = mr.me.type;
+  }
+  metaReaderClear(&mr);
+  return 0;
+}
 
 int metaReadNext(SMetaReader *pReader) {
   SMeta *pMeta = pReader->pMeta;
@@ -1102,6 +1134,21 @@ END:
   return ret;
 }
 
+int32_t metaGetTableTagsOpt(SMeta *pMeta, uint64_t suid, SArray *uidList, SHashObj *tags) {
+  int32_t sz = uidList ? taosArrayGetSize(uidList) : 0;
+  for (int i = 0; i < sz; i++) {
+    tb_uid_t  *id = taosArrayGet(uidList, i);
+    SCtbIdxKey ctbIdxKey = {.suid = suid, .uid = *id};
+
+    void   *val = NULL;
+    int32_t len = 0;
+    if (taosHashGet(tags, id, sizeof(tb_uid_t)) == NULL &&
+        0 == tdbTbGet(pMeta->pCtbIdx, &ctbIdxKey, sizeof(SCtbIdxKey), &val, &len)) {
+      taosHashPut(tags, id, sizeof(tb_uid_t), val, len);
+    }
+  }
+  return 0;
+}
 int32_t metaGetTableTags(SMeta *pMeta, uint64_t suid, SArray *uidList, SHashObj *tags) {
   SMCtbCursor *pCur = metaOpenCtbCursor(pMeta, suid);
 

source/dnode/vnode/src/vnd/vnodeQuery.c

@@ -409,6 +409,9 @@ int32_t vnodeGetAllTableList(SVnode *pVnode, uint64_t uid, SArray *list) {
   return TSDB_CODE_SUCCESS;
 }
 
+int32_t vnodeGetCtbIdListByFilter(SVnode *pVnode, int64_t suid, SArray *list, bool (*filter)(void *arg), void *arg) {
+  return 0;
+}
 int32_t vnodeGetCtbIdList(SVnode *pVnode, int64_t suid, SArray *list) {
   SMCtbCursor *pCur = metaOpenCtbCursor(pVnode->pMeta, suid);
 

source/libs/executor/src/executil.c

@@ -26,6 +26,9 @@
 #include "executorimpl.h"
 #include "tcompression.h"
 
+static int32_t optimizeTbnameInCond(void* metaHandle, int64_t suid, SArray* list, SNode* pTagCond);
+static int32_t optimizeTbnameInCondImpl(void* metaHandle, int64_t suid, SArray* list, SNode* pTagCond);
+
 void initResultRowInfo(SResultRowInfo* pResultRowInfo) {
   pResultRowInfo->size = 0;
   pResultRowInfo->cur.pageId = -1;
@@ -407,11 +410,18 @@ static SColumnInfoData* getColInfoResult(void* metaHandle, uint64_t suid, SArray
 
   //  int64_t stt = taosGetTimestampUs();
   tags = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
-  code = metaGetTableTags(metaHandle, suid, uidList, tags);
-  if (code != TSDB_CODE_SUCCESS) {
-    qError("failed to get table tags from meta, reason:%s, suid:%" PRIu64, tstrerror(code), suid);
-    terrno = code;
-    goto end;
+
+  int32_t filter = optimizeTbnameInCond(metaHandle, suid, uidList, pTagCond);
+  if (filter == -1) {
+    code = metaGetTableTags(metaHandle, suid, uidList, tags);
+    if (code != TSDB_CODE_SUCCESS) {
+      qError("failed to get table tags from meta, reason:%s, suid:%" PRIu64, tstrerror(code), suid);
+      terrno = code;
+      goto end;
+    }
+  } else {
+    metaGetTableTagsOpt(metaHandle, suid, uidList, tags);
+    qInfo("succ to get table from meta idx, suid:%" PRIu64, suid);
   }
 
   int32_t rows = taosArrayGetSize(uidList);
@@ -742,6 +752,94 @@ end:
   return code;
 }
 
+static int tableUidCompare(const void* a, const void* b) {
+  uint64_t u1 = *(uint64_t*)a;
+  uint64_t u2 = *(uint64_t*)b;
+  if (u1 == u2) {
+    return 0;
+  }
+  return u1 < u2 ? -1 : 1;
+}
+static int32_t optimizeTbnameInCond(void* metaHandle, int64_t suid, SArray* list, SNode* cond) {
+  if (nodeType(cond) == QUERY_NODE_OPERATOR) {
+    return optimizeTbnameInCondImpl(metaHandle, suid, list, cond);
+  }
+
+  if (nodeType(cond) != QUERY_NODE_LOGIC_CONDITION || ((SLogicConditionNode*)cond)->condType != LOGIC_COND_TYPE_AND) {
+    return -1;
+  }
+
+  bool                 hasTbnameCond = false;
+  SLogicConditionNode* pNode = (SLogicConditionNode*)cond;
+  SNodeList*           pList = (SNodeList*)pNode->pParameterList;
+
+  int32_t len = LIST_LENGTH(pList);
+  if (len <= 0) return -1;
+
+  SListCell* cell = pList->pHead;
+  for (int i = 0; i < len; i++) {
+    if (cell == NULL) break;
+    if (optimizeTbnameInCondImpl(metaHandle, suid, list, cell->pNode) == 0) {
+      hasTbnameCond = true;
+    }
+    cell = cell->pNext;
+  }
+  taosArraySort(list, tableUidCompare);
+  taosArrayRemoveDuplicate(list, tableUidCompare, NULL);
+
+  return hasTbnameCond == true ? 0 : -1;
+}
+static int32_t optimizeTbnameInCondImpl(void* metaHandle, int64_t suid, SArray* list, SNode* pTagCond) {
+  if (nodeType(pTagCond) != QUERY_NODE_OPERATOR) {
+    return -1;
+  }
+  SOperatorNode* pNode = (SOperatorNode*)pTagCond;
+  if (pNode->opType != OP_TYPE_IN) {
+    return -1;
+  }
+  if ((pNode->pLeft != NULL && nodeType(pNode->pLeft) == QUERY_NODE_COLUMN &&
+       ((SColumnNode*)pNode->pLeft)->colType == COLUMN_TYPE_TBNAME) &&
+      (pNode->pRight != NULL && nodeType(pNode->pRight) == QUERY_NODE_NODE_LIST)) {
+    SNodeListNode* pList = (SNodeListNode*)pNode->pRight;
+
+    int32_t len = LIST_LENGTH(pList->pNodeList);
+    if (len <= 0) return -1;
+
+    SListCell* cell = pList->pNodeList->pHead;
+
+    SArray* pTbList = taosArrayInit(len, sizeof(void*));
+    for (int i = 0; i < pList->pNodeList->length; i++) {
+      SValueNode* valueNode = (SValueNode*)cell->pNode;
+      if (!IS_VAR_DATA_TYPE(valueNode->node.resType.type)) {
+        taosArrayDestroy(pTbList);
+        return -1;
+      }
+      char* name = varDataVal(valueNode->datum.p);
+      taosArrayPush(pTbList, &name);
+      cell = cell->pNext;
+    }
+
+    for (int i = 0; i < taosArrayGetSize(pTbList); i++) {
+      char*    name = taosArrayGetP(pTbList, i);
+      uint64_t uid = 0;
+      if (metaGetTableUidByName(metaHandle, name, &uid) == 0) {
+        ETableType tbType = TSDB_TABLE_MAX;
+        if (metaGetTableTypeByName(metaHandle, name, &tbType) == 0 && tbType == TSDB_CHILD_TABLE) {
+          taosArrayPush(list, &uid);
+        } else {
+          taosArrayDestroy(pTbList);
+          return -1;
+        }
+      } else {
+        qWarn("failed to get tableIds from by table name: %s, reason: %s", name, tstrerror(terrno));
+        terrno = 0;
+      }
+    }
+    taosArrayDestroy(pTbList);
+    return 0;
+  }
+  return -1;
+}
 int32_t getTableList(void* metaHandle, void* pVnode, SScanPhysiNode* pScanNode, SNode* pTagCond, SNode* pTagIndexCond,
                      STableListInfo* pListInfo) {
   int32_t code = TSDB_CODE_SUCCESS;
@@ -767,9 +865,6 @@ int32_t getTableList(void* metaHandle, void* pVnode, SScanPhysiNode* pScanNode,
         qError("failed to get tableIds from index, reason:%s, suid:%" PRIu64, tstrerror(code), tableUid);
         code = TDB_CODE_SUCCESS;
       }
-
-      //      int64_t stt1 = taosGetTimestampUs();
-      //      qDebug("generate table list, cost:%ld us", stt1-stt);
     } else if (!pTagCond) {
       vnodeGetCtbIdList(pVnode, pScanNode->suid, res);
     }
@@ -813,7 +908,7 @@ int32_t getTableList(void* metaHandle, void* pVnode, SScanPhysiNode* pScanNode,
   size_t numOfTables = taosArrayGetSize(res);
   for (int i = 0; i < numOfTables; i++) {
     STableKeyInfo info = {.uid = *(uint64_t*)taosArrayGet(res, i), .groupId = 0};
-    void* p = taosArrayPush(pListInfo->pTableList, &info);
+    void*         p = taosArrayPush(pListInfo->pTableList, &info);
     if (p == NULL) {
       taosArrayDestroy(res);
       return TSDB_CODE_OUT_OF_MEMORY;

source/libs/index/src/index.c

@@ -54,11 +54,17 @@
 void*   indexQhandle = NULL;
 int32_t indexRefMgt;
 
+int32_t indexThreads = 5;
+
 static void indexDestroy(void* sIdx);
 
-void indexInit() {
+void indexInit(int32_t threadNum) {
+  indexThreads = threadNum;
+  if (indexThreads <= 1) indexThreads = INDEX_NUM_OF_THREADS;
+}
+void indexEnvInit() {
   // refactor later
-  indexQhandle = taosInitScheduler(INDEX_QUEUE_SIZE, INDEX_NUM_OF_THREADS, "index", NULL);
+  indexQhandle = taosInitScheduler(INDEX_QUEUE_SIZE, indexThreads, "index", NULL);
   indexRefMgt = taosOpenRef(1000, indexDestroy);
 }
 void indexCleanup() {
@@ -99,7 +105,7 @@ static void indexWait(void* idx) {
 
 int indexOpen(SIndexOpts* opts, const char* path, SIndex** index) {
   int ret = TSDB_CODE_SUCCESS;
-  taosThreadOnce(&isInit, indexInit);
+  taosThreadOnce(&isInit, indexEnvInit);
   SIndex* idx = taosMemoryCalloc(1, sizeof(SIndex));
   if (idx == NULL) {
     return TSDB_CODE_OUT_OF_MEMORY;

source/libs/index/src/indexCache.c

@@ -22,7 +22,7 @@
 #define MAX_INDEX_KEY_LEN 256  // test only, change later
 
 #define MEM_TERM_LIMIT     10 * 10000
-#define MEM_THRESHOLD      512 * 1024
+#define MEM_THRESHOLD      8 * 512 * 1024  // 8M
 #define MEM_SIGNAL_QUIT    MEM_THRESHOLD * 20
 #define MEM_ESTIMATE_RADIO 1.5
 

source/libs/index/src/indexTfile.c

@@ -555,6 +555,7 @@ int tfileWriterPut(TFileWriter* tw, void* data, bool order) {
     taosArraySort(v->tableId, idxUidCompare);
     taosArrayRemoveDuplicate(v->tableId, idxUidCompare, NULL);
     int32_t tbsz = taosArrayGetSize(v->tableId);
+    if (tbsz == 0) continue;
     fstOffset += TF_TABLE_TATOAL_SIZE(tbsz);
   }
   tfileWriteFstOffset(tw, fstOffset);
@@ -566,6 +567,7 @@ int tfileWriterPut(TFileWriter* tw, void* data, bool order) {
     TFileValue* v = taosArrayGetP((SArray*)data, i);
 
     int32_t tbsz = taosArrayGetSize(v->tableId);
+    if (tbsz == 0) continue;
     // check buf has enough space or not
     int32_t ttsz = TF_TABLE_TATOAL_SIZE(tbsz);
 
@@ -592,6 +594,10 @@ int tfileWriterPut(TFileWriter* tw, void* data, bool order) {
   for (size_t i = 0; i < sz; i++) {
     // TODO, fst batch write later
     TFileValue* v = taosArrayGetP((SArray*)data, i);
+
+    int32_t tbsz = taosArrayGetSize(v->tableId);
+    if (tbsz == 0) continue;
+
     if (tfileWriteData(tw, v) != 0) {
       indexError("failed to write data: %s, offset: %d len: %d", v->colVal, v->offset,
                  (int)taosArrayGetSize(v->tableId));

source/libs/transport/src/transComm.c

@@ -193,7 +193,7 @@ bool transReadComplete(SConnBuffer* connBuf) {
 
 int transSetConnOption(uv_tcp_t* stream) {
   uv_tcp_nodelay(stream, 1);
-  int ret = uv_tcp_keepalive(stream, 5, 5);
+  int ret = uv_tcp_keepalive(stream, 5, 60);
   return ret;
 }