other: merge with original branch.

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

@@ -142,7 +142,7 @@ typedef struct SMetaInfo {
   int64_t version;
   int32_t skmVer;
 } SMetaInfo;
-int32_t metaGetInfo(SMeta* pMeta, int64_t uid, SMetaInfo* pInfo);
+int32_t metaGetInfo(SMeta* pMeta, int64_t uid, SMetaInfo* pInfo, SMetaReader* pReader);
 
 typedef struct {
   int64_t uid;

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

@@ -165,8 +165,9 @@ int metaGetTableEntryByUid(SMetaReader *pReader, tb_uid_t uid) {
   version1 = ((SUidIdxVal *)pReader->pBuf)[0].version;
   return metaGetTableEntryByVersion(pReader, version1, uid);
   */
+
   SMetaInfo info;
-  if (metaGetInfo(pMeta, uid, &info) == TSDB_CODE_NOT_FOUND) {
+  if (metaGetInfo(pMeta, uid, &info, pReader) == TSDB_CODE_NOT_FOUND) {
     terrno = TSDB_CODE_PAR_TABLE_NOT_EXIST;
     return -1;
   }
@@ -623,7 +624,7 @@ int32_t metaGetTbTSchemaEx(SMeta *pMeta, tb_uid_t suid, tb_uid_t uid, int32_t sv
   SSkmDbKey skmDbKey;
   if (sver <= 0) {
     SMetaInfo info;
-    if (metaGetInfo(pMeta, suid ? suid : uid, &info) == 0) {
+    if (metaGetInfo(pMeta, suid ? suid : uid, &info, NULL) == 0) {
       sver = info.skmVer;
     } else {
       TBC *pSkmDbC = NULL;
@@ -1388,10 +1389,11 @@ int32_t metaGetTableTags(SMeta *pMeta, uint64_t suid, SArray *uidList, SHashObj
 
 int32_t metaCacheGet(SMeta *pMeta, int64_t uid, SMetaInfo *pInfo);
 
-int32_t metaGetInfo(SMeta *pMeta, int64_t uid, SMetaInfo *pInfo) {
+int32_t metaGetInfo(SMeta *pMeta, int64_t uid, SMetaInfo *pInfo, SMetaReader *pReader) {
   int32_t code = 0;
   void   *pData = NULL;
   int     nData = 0;
+  int     lock = 0;
 
   metaRLock(pMeta);
 
@@ -1416,11 +1418,22 @@ int32_t metaGetInfo(SMeta *pMeta, int64_t uid, SMetaInfo *pInfo) {
   pInfo->version = ((SUidIdxVal *)pData)->version;
   pInfo->skmVer = ((SUidIdxVal *)pData)->skmVer;
 
+  if (pReader != NULL) {
+    lock = !(pReader->flags & META_READER_NOLOCK);
+    if (lock) {
+      metaULock(pReader->pMeta);
+      // metaReaderReleaseLock(pReader);
+    }
+  }
   // upsert the cache
   metaWLock(pMeta);
   metaCacheUpsert(pMeta, pInfo);
   metaULock(pMeta);
 
+  if (lock) {
+    metaRLock(pReader->pMeta);
+  }
+
 _exit:
   tdbFree(pData);
   return code;

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

@@ -207,7 +207,7 @@ int metaCreateSTable(SMeta *pMeta, int64_t version, SVCreateStbReq *pReq) {
     tb_uid_t uid = *(tb_uid_t *)pData;
     tdbFree(pData);
     SMetaInfo info;
-    metaGetInfo(pMeta, uid, &info);
+    metaGetInfo(pMeta, uid, &info, NULL);
     if (info.uid == info.suid) {
       return 0;
     } else {

source/dnode/vnode/src/tsdb/tsdbMemTable.c

@@ -104,7 +104,7 @@ int32_t tsdbInsertTableData(STsdb *pTsdb, int64_t version, SSubmitMsgIter *pMsgI
   tb_uid_t   uid = pMsgIter->uid;
 
   SMetaInfo info;
-  code = metaGetInfo(pTsdb->pVnode->pMeta, uid, &info);
+  code = metaGetInfo(pTsdb->pVnode->pMeta, uid, &info, NULL);
   if (code) {
     code = TSDB_CODE_TDB_TABLE_NOT_EXIST;
     goto _err;
@@ -114,7 +114,7 @@ int32_t tsdbInsertTableData(STsdb *pTsdb, int64_t version, SSubmitMsgIter *pMsgI
     goto _err;
   }
   if (info.suid) {
-    metaGetInfo(pTsdb->pVnode->pMeta, info.suid, &info);
+    metaGetInfo(pTsdb->pVnode->pMeta, info.suid, &info, NULL);
   }
   if (pMsgIter->sversion != info.skmVer) {
     tsdbError("vgId:%d, req sver:%d, skmVer:%d suid:%" PRId64 " uid:%" PRId64, TD_VID(pTsdb->pVnode),
@@ -153,7 +153,7 @@ int32_t tsdbDeleteTableData(STsdb *pTsdb, int64_t version, tb_uid_t suid, tb_uid
 
   // check if table exists
   SMetaInfo info;
-  code = metaGetInfo(pTsdb->pVnode->pMeta, uid, &info);
+  code = metaGetInfo(pTsdb->pVnode->pMeta, uid, &info, NULL);
   if (code) {
     code = TSDB_CODE_TDB_TABLE_NOT_EXIST;
     goto _err;