enh: enh sys scan in vtable query.

Author: Simon9997

include/common/tmsg.h

@@ -3414,6 +3414,12 @@ typedef struct STagScanOperatorParam {
   tb_uid_t       vcUid;
 } STagScanOperatorParam;
 
+typedef struct SSysScanOperatorParam {
+  bool           isVstb;
+  tb_uid_t       uid;
+  int64_t        version;
+} SSysScanOperatorParam;
+
 typedef struct SVTableScanOperatorParam {
   uint64_t        uid;
   SOperatorParam* pTagScanOp;

include/libs/executor/storageapi.h

@@ -302,6 +302,9 @@ typedef struct SStoreMeta {
   int32_t (*metaGetCachedRefDbs)(void* pVnode, tb_uid_t suid, SArray* pList);
   int32_t (*metaPutRefDbsToCache)(void* pVnode, tb_uid_t suid, SArray* pList);
 
+  int32_t (*metaGetVirtualSupertableVersion)(void* pVnode, tb_uid_t suid, int32_t* version);
+  int32_t (*metaGetVirtualNormalChildtableVersion)(void* pVnode, tb_uid_t uid, int32_t* version);
+
   void* (*storeGetIndexInfo)(void* pVnode);
   void* (*getInvertIndex)(void* pVnode);
   // support filter and non-filter cases. [vnodeGetCtbIdList & vnodeGetCtbIdListByFilter]

include/libs/nodes/plannodes.h

@@ -264,6 +264,7 @@ typedef struct SDynQueryCtrlVtbScan {
   uint64_t      uid;
   int32_t       rversion;
   SNodeList*    pOrgVgIds;
+  SNodeList*    pSysScanVgIds;
   SVgroupsInfo* pVgroupList;
 } SDynQueryCtrlVtbScan;
 
@@ -675,6 +676,7 @@ typedef struct SVtbScanDynCtrlBasic {
   SEpSet     mgmtEpSet;
   SNodeList *pScanCols;
   SNodeList *pOrgVgIds;
+  SNodeList *pSysScanVgIds;
 } SVtbScanDynCtrlBasic;
 
 typedef struct SDynQueryCtrlPhysiNode {

source/common/src/msg/tmsg.c

@@ -10996,6 +10996,13 @@ int32_t tSerializeSOperatorParam(SEncoder *pEncoder, SOperatorParam *pOpParam) {
   TAOS_CHECK_RETURN(tEncodeI32(pEncoder, pOpParam->opType));
   TAOS_CHECK_RETURN(tEncodeI32(pEncoder, pOpParam->downstreamIdx));
   switch (pOpParam->opType) {
+    case QUERY_NODE_PHYSICAL_PLAN_SYSTABLE_SCAN: {
+      SSysScanOperatorParam* pSysScan = (SSysScanOperatorParam *)pOpParam->value;
+      TAOS_CHECK_RETURN(tEncodeBool(pEncoder, pSysScan->isVstb));
+      TAOS_CHECK_RETURN(tEncodeI64(pEncoder, pSysScan->version));
+      TAOS_CHECK_RETURN(tEncodeI64(pEncoder, pSysScan->uid));
+      break;
+    }
     case QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN: {
       STagScanOperatorParam *pTagScan = (STagScanOperatorParam *)pOpParam->value;
       TAOS_CHECK_RETURN(tEncodeI64(pEncoder, pTagScan->vcUid));
@@ -11047,6 +11054,17 @@ int32_t tDeserializeSOperatorParam(SDecoder *pDecoder, SOperatorParam *pOpParam)
   TAOS_CHECK_RETURN(tDecodeI32(pDecoder, &pOpParam->opType));
   TAOS_CHECK_RETURN(tDecodeI32(pDecoder, &pOpParam->downstreamIdx));
   switch (pOpParam->opType) {
+    case QUERY_NODE_PHYSICAL_PLAN_SYSTABLE_SCAN: {
+      pOpParam->value = taosMemoryMalloc(sizeof(SSysScanOperatorParam));
+      if (NULL == pOpParam->value) {
+        TAOS_CHECK_RETURN(terrno);
+      }
+      SSysScanOperatorParam* pSysScan = pOpParam->value;
+      TAOS_CHECK_RETURN(tDecodeBool(pDecoder, &pSysScan->isVstb));
+      TAOS_CHECK_RETURN(tDecodeI64(pDecoder, &pSysScan->version));
+      TAOS_CHECK_RETURN(tDecodeI64(pDecoder, &pSysScan->uid));
+      break;
+    }
     case QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN: {
       pOpParam->value = taosMemoryMalloc(sizeof(STagScanOperatorParam));
       if (NULL == pOpParam->value) {

source/common/src/systable.c

@@ -262,6 +262,7 @@ static const SSysDbTableSchema userVctbColsSchema[] = {
     {.name = "col_source", .bytes = TSDB_COL_FNAME_LEN - 1 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = false},
     {.name = "vgroup_id", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = false},
     {.name = "ref_version", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = false},
+    {.name = "vstb_version", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = false},
 };
 
 static const SSysDbTableSchema userTblDistSchema[] = {

source/dnode/vnode/inc/vnode.h

@@ -155,6 +155,9 @@ int32_t metaGetStbStats(void *pVnode, int64_t uid, int64_t *numOfTables, int32_t
 int32_t metaGetCachedRefDbs(void *pVnode, tb_uid_t suid, SArray *pList);
 int32_t metaPutRefDbsToCache(void *pVnode, tb_uid_t suid, SArray *pList);
 
+int32_t metaGetVirtualSupertableVersion(void *pVnode, tb_uid_t suid, int32_t *version);
+int32_t metaGetVirtualNormalChildtableVersion(void *pVnode, tb_uid_t uid, int32_t *version);
+
 // tsdb
 typedef struct STsdbReader STsdbReader;
 

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

@@ -540,8 +540,9 @@ struct SVnode {
   SVMonitorObj  monitor;
   uint32_t      applyQueueErrorCount;
 
-  // Notification Handles
-  SStreamNotifyHandleMap* pNotifyHandleMap;
+  // version for virtual super table
+  TdThreadRwlock versionLock;
+  SHashObj*      pVtableVersion;
 };
 
 #define TD_VID(PVNODE) ((PVNODE)->config.vgId)

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

@@ -995,6 +995,39 @@ int32_t metaRefDbsCacheClear(SMeta* pMeta, uint64_t suid) {
   return code;
 }
 
+int32_t metaGetVirtualSupertableVersion(void *pVnode, tb_uid_t suid, int32_t *version) {
+  int32_t        code = TSDB_CODE_SUCCESS;
+  SVnode*        vnode = (SVnode*)pVnode;
+
+  (void)taosThreadRwlockRdlock(&vnode->versionLock);
+  int32_t* ver = taosHashGet(vnode->pVtableVersion, &suid, sizeof(tb_uid_t));
+  if (ver != NULL) {
+    *version = (*ver);
+  } else {
+    *version = 0;
+  }
+  (void)taosThreadRwlockUnlock(&vnode->versionLock);
+  return code;
+}
+
+int32_t metaGetVirtualNormalChildtableVersion(void *pVnode, tb_uid_t uid, int32_t *version) {
+  int         code = TSDB_CODE_SUCCESS;
+  int32_t     lino = 0;
+  SMetaReader mr = {0};
+  metaReaderDoInit(&mr, ((SVnode *)pVnode)->pMeta, META_READER_LOCK);
+  code = metaReaderGetTableEntryByUid(&mr, uid);
+  TSDB_CHECK_CODE(code, lino, _return);
+
+  *version = mr.me.colRef.version;
+
+_return:
+  if (code != TSDB_CODE_SUCCESS) {
+    metaError("%s failed since %s, line %d", __func__, tstrerror(code), lino);
+  }
+  metaReaderClear(&mr);
+  return code;
+}
+
 int32_t metaGetCachedRefDbs(void* pVnode, tb_uid_t suid, SArray* pList) {
   int32_t        code = TSDB_CODE_SUCCESS;
   int32_t        line = 0;

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

@@ -2540,6 +2540,24 @@ int32_t metaHandleEntry2(SMeta *pMeta, const SMetaEntry *pEntry) {
     pMeta->changed = true;
     metaDebug("vgId:%d, index:%" PRId64 ", handle meta entry success, type:%d tb:%s uid:%" PRId64, vgId,
               pEntry->version, pEntry->type, pEntry->type > 0 ? pEntry->name : "", pEntry->uid);
+    if (type == TSDB_VIRTUAL_CHILD_TABLE) {
+      int64_t suid = pEntry->ctbEntry.suid;
+      (void)taosThreadRwlockWrlock(&pMeta->pVnode->versionLock);
+      int32_t  *ver = taosHashGet(pMeta->pVnode->pVtableVersion, &suid, sizeof(tb_uid_t));
+      if (ver != NULL) {
+        int32_t newVersion = (*ver) + 1;
+        metaDebug("vgId:%d, vtable uid:%" PRId64 " version updated to %d" , TD_VID(pMeta->pVnode), suid, newVersion);
+        code = taosHashRemove(pMeta->pVnode->pVtableVersion, &suid, sizeof(tb_uid_t));
+        if (TSDB_CODE_SUCCESS == code) {
+          code = taosHashPut(pMeta->pVnode->pVtableVersion, &suid, sizeof(tb_uid_t), &newVersion, sizeof(newVersion));
+        }
+      } else {
+        int32_t newVersion = 1;
+        metaDebug("vgId:%d, vtable uid:%" PRId64 " version set to %c", TD_VID(pMeta->pVnode), suid, newVersion);
+        code = taosHashPut(pMeta->pVnode->pVtableVersion, &suid, sizeof(tb_uid_t), &newVersion, sizeof(newVersion));
+      }
+      (void)taosThreadRwlockUnlock(&pMeta->pVnode->versionLock);
+    }
   } else {
     metaErr(vgId, code);
   }

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

@@ -115,6 +115,9 @@ void initMetadataAPI(SStoreMeta* pMeta) {
 
   pMeta->metaGetCachedRefDbs = metaGetCachedRefDbs;
   pMeta->metaPutRefDbsToCache = metaPutRefDbsToCache;
+
+  pMeta->metaGetVirtualSupertableVersion = metaGetVirtualSupertableVersion;
+  pMeta->metaGetVirtualNormalChildtableVersion = metaGetVirtualNormalChildtableVersion;
 }
 
 void initTqAPI(SStoreTqReader* pTq) {