[TD-98] merge develop branch

.codecov.yml

+coverage:
+  precision: 2
+  notify:
+    email:
+      default:
+        to:
+          - sdsang@taosdata.com
+          - &author
+        only_pulls: false
+        layout: reach, diff, flags, files
+        flags: null
+        paths: null

.travis.yml

@@ -146,8 +146,8 @@ matrix:
           fi
 
           cd ${TRAVIS_BUILD_DIR}
-          lcov -d . --capture -o coverage.info
-          lcov -l coverage.info || exit $?
+          lcov -d . --capture --rc lcov_branch_coverage=1 -o coverage.info
+          lcov -l --rc lcov_branch_coverage=1 coverage.info || exit $?
 
           gem install coveralls-lcov
 
@@ -166,7 +166,7 @@ matrix:
             exit $?
           fi
 
-          bash <(curl -s https://codecov.io/bash) -f coverage.info
+          bash <(curl -s https://codecov.io/bash) -y .codecov.yml -f coverage.info
           if [ "$?" -eq "0" ]; then
             echo -e "${GREEN} ## Uploaded to Codecov! ## ${NC} "
           else

src/common/inc/dataformat.h

@@ -111,7 +111,6 @@ typedef struct SDataCol {
   int     len;
   int     offset;
   void *  pData; // Original data
-  void *  pCData; // Compressed data
 } SDataCol;
 
 typedef struct {
@@ -133,9 +132,12 @@ typedef struct {
 SDataCols *tdNewDataCols(int maxRowSize, int maxCols, int maxRows);
 void       tdResetDataCols(SDataCols *pCols);
 void       tdInitDataCols(SDataCols *pCols, STSchema *pSchema);
+SDataCols *tdDupDataCols(SDataCols *pCols, bool keepData);
 void       tdFreeDataCols(SDataCols *pCols);
 void       tdAppendDataRowToDataCol(SDataRow row, SDataCols *pCols);
 void       tdPopDataColsPoints(SDataCols *pCols, int pointsToPop);
+int        tdMergeDataCols(SDataCols *target, SDataCols *src, int rowsToMerge);
+void       tdMergeTwoDataCols(SDataCols *target, SDataCols *src1, int *iter1, SDataCols *src2, int *iter2, int tRows);
 
 #ifdef __cplusplus
 }

src/common/src/dataformat.c

@@ -13,6 +13,7 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 #include "dataformat.h"
+#include "tutil.h"
 
 static int tdFLenFromSchema(STSchema *pSchema);
 
@@ -338,6 +339,28 @@ void tdFreeDataCols(SDataCols *pCols) {
   }
 }
 
+SDataCols *tdDupDataCols(SDataCols *pDataCols, bool keepData) {
+  SDataCols *pRet = tdNewDataCols(pDataCols->maxRowSize, pDataCols->maxCols, pDataCols->maxPoints);
+  if (pRet == NULL) return NULL;
+
+  pRet->numOfCols = pDataCols->numOfCols;
+  pRet->sversion = pDataCols->sversion;
+  if (keepData) pRet->numOfPoints = pDataCols->numOfPoints;
+
+  for (int i = 0; i < pDataCols->numOfCols; i++) {
+    pRet->cols[i].type = pDataCols->cols[i].type;
+    pRet->cols[i].colId = pDataCols->cols[i].colId;
+    pRet->cols[i].bytes = pDataCols->cols[i].bytes;
+    pRet->cols[i].len = pDataCols->cols[i].len;
+    pRet->cols[i].offset = pDataCols->cols[i].offset;
+    pRet->cols[i].pData = (void *)((char *)pRet->buf + ((char *)(pDataCols->cols[i].pData) - (char *)(pDataCols->buf)));
+
+    if (keepData) memcpy(pRet->cols[i].pData, pDataCols->cols[i].pData, pRet->cols[i].bytes * pDataCols->numOfPoints);
+  }
+
+  return pRet;
+}
+
 void tdResetDataCols(SDataCols *pCols) {
   pCols->numOfPoints = 0;
   for (int i = 0; i < pCols->maxCols; i++) {
@@ -382,6 +405,58 @@ static int tdFLenFromSchema(STSchema *pSchema) {
   return ret;
 }
 
-int tdMergeDataCols(SDataCols *target, SDataCols *source) {
+int tdMergeDataCols(SDataCols *target, SDataCols *source, int rowsToMerge) {
+  ASSERT(rowsToMerge > 0 && rowsToMerge <= source->numOfPoints);
+
+  SDataCols *pTarget = tdDupDataCols(target, true);
+  if (pTarget == NULL) goto _err;
+  // tdResetDataCols(target);
+
+  int iter1 = 0;
+  int iter2 = 0;
+  tdMergeTwoDataCols(target,pTarget, &iter1, source, &iter2, pTarget->numOfPoints + rowsToMerge);
+
+  tdFreeDataCols(pTarget);
   return 0;
+
+_err:
+  tdFreeDataCols(pTarget);
+  return -1;
+}
+
+void tdMergeTwoDataCols(SDataCols *target, SDataCols *src1, int *iter1, SDataCols *src2, int *iter2, int tRows) {
+  tdResetDataCols(target);
+
+  while (target->numOfPoints < tRows) {
+    if (*iter1 >= src1->numOfPoints && *iter2 >= src2->numOfPoints) break;
+
+    TSKEY key1 = (*iter1 >= src1->numOfPoints) ? INT64_MAX : ((TSKEY *)(src1->cols[0].pData))[*iter1];
+    TSKEY key2 = (*iter2 >= src2->numOfPoints) ? INT64_MAX : ((TSKEY *)(src2->cols[0].pData))[*iter2];
+
+    if (key1 < key2) {
+      for (int i = 0; i < src1->numOfCols; i++) {
+        ASSERT(target->cols[i].type == src1->cols[i].type);
+        memcpy((void *)((char *)(target->cols[i].pData) + TYPE_BYTES[target->cols[i].type] * target->numOfPoints),
+               (void *)((char *)(src1->cols[i].pData) + TYPE_BYTES[target->cols[i].type] * (*iter1)),
+               TYPE_BYTES[target->cols[i].type]);
+        target->cols[i].len += TYPE_BYTES[target->cols[i].type];
+      }
+
+      target->numOfPoints++;
+      (*iter1)++;
+    } else if (key1 > key2) {
+      for (int i = 0; i < src2->numOfCols; i++) {
+        ASSERT(target->cols[i].type == src2->cols[i].type);
+        memcpy((void *)((char *)(target->cols[i].pData) + TYPE_BYTES[target->cols[i].type] * target->numOfPoints),
+               (void *)((char *)(src2->cols[i].pData) + TYPE_BYTES[src2->cols[i].type] * (*iter2)),
+               TYPE_BYTES[target->cols[i].type]);
+        target->cols[i].len += TYPE_BYTES[target->cols[i].type];
+      }
+
+      target->numOfPoints++;
+      (*iter2)++;
+    } else {
+      ASSERT(false);
+    }
+  }
 }
\ No newline at end of file

src/dnode/CMakeLists.txt

@@ -27,7 +27,7 @@ IF ((TD_LINUX_64) OR (TD_LINUX_32 AND TD_ARM))
   ENDIF ()
 
   IF (TD_SYNC)
-    TARGET_LINK_LIBRARIES(taosd replica sync)
+    TARGET_LINK_LIBRARIES(taosd balance sync)
   ENDIF ()
 
   SET(PREPARE_ENV_CMD "prepare_env_cmd")

src/dnode/src/dnodeMClient.c

@@ -23,12 +23,13 @@
 #include "tsync.h"
 #include "ttime.h"
 #include "ttimer.h"
-#include "treplica.h"
+#include "tbalance.h"
+#include "vnode.h"
+#include "mnode.h"
 #include "dnode.h"
 #include "dnodeMClient.h"
 #include "dnodeModule.h"
 #include "dnodeMgmt.h"
-#include "vnode.h"
 
 #define MPEER_CONTENT_LEN 2000
 
@@ -181,7 +182,7 @@ static void dnodeProcessStatusRsp(SRpcMsg *pMsg) {
              tsMnodeInfos.nodeInfos[i].nodeName);
     }
     dnodeSaveMnodeIpList();
-    replicaNotify();
+    sdbUpdateSync();
   }
 
   taosTmrReset(dnodeSendStatusMsg, tsStatusInterval * 1000, NULL, tsDnodeTmr, &tsStatusTimer);

src/inc/mnode.h

@@ -24,6 +24,7 @@ int32_t mgmtInitSystem();
 int32_t mgmtStartSystem();
 void    mgmtCleanUpSystem();
 void    mgmtStopSystem();
+void    sdbUpdateSync();
 
 #ifdef __cplusplus
 }

src/inc/treplica.h → src/inc/tbalance.h

@@ -13,8 +13,8 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#ifndef TDENGINE_REPLICA_H
-#define TDENGINE_REPLICA_H
+#ifndef TDENGINE_BALANCE_H
+#define TDENGINE_BALANCE_H
 
 #ifdef __cplusplus
 extern "C" {
@@ -23,13 +23,12 @@ extern "C" {
 struct SVgObj;
 struct SDnodeObj;
 
-int32_t replicaInit();
-void    replicaCleanUp();
-void    replicaNotify();
-void    replicaReset();
-int32_t replicaAllocVnodes(struct SVgObj *pVgroup);
-int32_t replicaForwardReqToPeer(void *pHead);
-int32_t replicaDropDnode(struct SDnodeObj *pDnode);
+int32_t balanceInit();
+void    balanceCleanUp();
+void    balanceNotify();
+void    balanceReset();
+int32_t balanceAllocVnodes(struct SVgObj *pVgroup);
+int32_t balanceDropDnode(struct SDnodeObj *pDnode);
 
 #ifdef __cplusplus
 }

src/inc/tsdb.h

@@ -46,6 +46,7 @@ typedef struct {
 // --------- TSDB REPOSITORY CONFIGURATION DEFINITION
 typedef struct {
   int8_t  precision;
+  int8_t  compression;
   int32_t tsdbId;
   int32_t maxTables;            // maximum number of tables this repository can have
   int32_t daysPerFile;          // day per file sharding policy
@@ -60,13 +61,13 @@ STsdbCfg *tsdbCreateDefaultCfg();
 void      tsdbFreeCfg(STsdbCfg *pCfg);
 
 // --------- TSDB REPOSITORY DEFINITION
-typedef void tsdb_repo_t;  // use void to hide implementation details from outside
+typedef void TsdbRepoT;  // use void to hide implementation details from outside
 
-int          tsdbCreateRepo(char *rootDir, STsdbCfg *pCfg, void *limiter);
-int32_t      tsdbDropRepo(tsdb_repo_t *repo);
-tsdb_repo_t *tsdbOpenRepo(char *tsdbDir, STsdbAppH *pAppH);
-int32_t      tsdbCloseRepo(tsdb_repo_t *repo);
-int32_t      tsdbConfigRepo(tsdb_repo_t *repo, STsdbCfg *pCfg);
+int        tsdbCreateRepo(char *rootDir, STsdbCfg *pCfg, void *limiter);
+int32_t    tsdbDropRepo(TsdbRepoT *repo);
+TsdbRepoT *tsdbOpenRepo(char *tsdbDir, STsdbAppH *pAppH);
+int32_t    tsdbCloseRepo(TsdbRepoT *repo);
+int32_t    tsdbConfigRepo(TsdbRepoT *repo, STsdbCfg *pCfg);
 
 // --------- TSDB TABLE DEFINITION
 typedef struct {
@@ -76,15 +77,15 @@ typedef struct {
 
 // --------- TSDB TABLE configuration
 typedef struct {
-  ETableType      type;
-  char *          name;
-  STableId        tableId;
-  int32_t         sversion;
-  char *          sname; // super table name
-  int64_t         superUid;
-  STSchema *      schema;
-  STSchema *      tagSchema;
-  SDataRow        tagValues;
+  ETableType type;
+  char *     name;
+  STableId   tableId;
+  int32_t    sversion;
+  char *     sname;  // super table name
+  int64_t    superUid;
+  STSchema * schema;
+  STSchema * tagSchema;
+  SDataRow   tagValues;
 } STableCfg;
 
 int  tsdbInitTableCfg(STableCfg *config, ETableType type, int64_t uid, int32_t tid);
@@ -96,11 +97,11 @@ int  tsdbTableSetName(STableCfg *config, char *name, bool dup);
 int  tsdbTableSetSName(STableCfg *config, char *sname, bool dup);
 void tsdbClearTableCfg(STableCfg *config);
 
-int32_t tsdbGetTableTagVal(tsdb_repo_t *repo, STableId id, int32_t col, int16_t* type, int16_t* bytes, char** val);
+int32_t tsdbGetTableTagVal(TsdbRepoT *repo, STableId id, int32_t col, int16_t *type, int16_t *bytes, char **val);
 
-int tsdbCreateTable(tsdb_repo_t *repo, STableCfg *pCfg);
-int tsdbDropTable(tsdb_repo_t *pRepo, STableId tableId);
-int tsdbAlterTable(tsdb_repo_t *repo, STableCfg *pCfg);
+int tsdbCreateTable(TsdbRepoT *repo, STableCfg *pCfg);
+int tsdbDropTable(TsdbRepoT *pRepo, STableId tableId);
+int tsdbAlterTable(TsdbRepoT *repo, STableCfg *pCfg);
 
 // the TSDB repository info
 typedef struct STsdbRepoInfo {
@@ -110,7 +111,7 @@ typedef struct STsdbRepoInfo {
   int64_t  tsdbTotalDiskSize;  // the total disk size taken by this TSDB repository
   // TODO: Other informations to add
 } STsdbRepoInfo;
-STsdbRepoInfo *tsdbGetStatus(tsdb_repo_t *pRepo);
+STsdbRepoInfo *tsdbGetStatus(TsdbRepoT *pRepo);
 
 // the meter information report structure
 typedef struct {
@@ -119,7 +120,7 @@ typedef struct {
   int64_t   tableTotalDataSize;  // In bytes
   int64_t   tableTotalDiskSize;  // In bytes
 } STableInfo;
-STableInfo *tsdbGetTableInfo(tsdb_repo_t *pRepo, STableId tid);
+STableInfo *tsdbGetTableInfo(TsdbRepoT *pRepo, STableId tid);
 
 // -- FOR INSERT DATA
 /**
@@ -129,11 +130,11 @@ STableInfo *tsdbGetTableInfo(tsdb_repo_t *pRepo, STableId tid);
  *
  * @return the number of points inserted, -1 for failure and the error number is set
  */
-int32_t tsdbInsertData(tsdb_repo_t *pRepo, SSubmitMsg *pMsg);
+int32_t tsdbInsertData(TsdbRepoT *pRepo, SSubmitMsg *pMsg);
 
 // -- FOR QUERY TIME SERIES DATA
 
-typedef void *tsdb_query_handle_t;  // Use void to hide implementation details
+typedef void *TsdbQueryHandleT;  // Use void to hide implementation details
 
 typedef struct STableGroupList {  // qualified table object list in group
   SArray *pGroupList;
@@ -167,21 +168,21 @@ typedef struct SDataBlockInfo {
 
 typedef struct {
   size_t  numOfTables;
-  SArray* pGroupList;
+  SArray *pGroupList;
 } STableGroupInfo;
 
 typedef struct {
 } SFields;
 
 #define TSDB_TS_GREATER_EQUAL 1
-#define TSDB_TS_LESS_EQUAL    2
+#define TSDB_TS_LESS_EQUAL 2
 
 typedef struct SQueryRowCond {
   int32_t rel;
   TSKEY   ts;
 } SQueryRowCond;
 
-typedef void *tsdbpos_t;
+typedef void *TsdbPosT;
 
 /**
  * Get the data block iterator, starting from position according to the query condition
@@ -189,14 +190,14 @@ typedef void *tsdbpos_t;
  * @param pTableList    table sid list
  * @return
  */
-tsdb_query_handle_t *tsdbQueryTables(tsdb_repo_t* tsdb, STsdbQueryCond *pCond, STableGroupInfo *groupInfo);
+TsdbQueryHandleT *tsdbQueryTables(TsdbRepoT *tsdb, STsdbQueryCond *pCond, STableGroupInfo *groupInfo);
 
 /**
  * move to next block
  * @param pQueryHandle
  * @return
  */
-bool tsdbNextDataBlock(tsdb_query_handle_t *pQueryHandle);
+bool tsdbNextDataBlock(TsdbQueryHandleT *pQueryHandle);
 
 /**
  * Get current data block information
@@ -204,7 +205,7 @@ bool tsdbNextDataBlock(tsdb_query_handle_t *pQueryHandle);
  * @param pQueryHandle
  * @return
  */
-SDataBlockInfo tsdbRetrieveDataBlockInfo(tsdb_query_handle_t *pQueryHandle);
+SDataBlockInfo tsdbRetrieveDataBlockInfo(TsdbQueryHandleT *pQueryHandle);
 
 /**
  *
@@ -216,7 +217,7 @@ SDataBlockInfo tsdbRetrieveDataBlockInfo(tsdb_query_handle_t *pQueryHandle);
  * @pBlockStatis the pre-calculated value for current data blocks. if the block is a cache block, always return 0
  * @return
  */
-int32_t tsdbRetrieveDataBlockStatisInfo(tsdb_query_handle_t *pQueryHandle, SDataStatis **pBlockStatis);
+int32_t tsdbRetrieveDataBlockStatisInfo(TsdbQueryHandleT *pQueryHandle, SDataStatis **pBlockStatis);
 
 /**
  * The query condition with primary timestamp is passed to iterator during its constructor function,
@@ -226,7 +227,7 @@ int32_t tsdbRetrieveDataBlockStatisInfo(tsdb_query_handle_t *pQueryHandle, SData
  * @param pQueryHandle
  * @return
  */
-SArray *tsdbRetrieveDataBlock(tsdb_query_handle_t *pQueryHandle, SArray *pIdList);
+SArray *tsdbRetrieveDataBlock(TsdbQueryHandleT *pQueryHandle, SArray *pIdList);
 
 /**
  *  todo remove the parameter of position, and order type
@@ -238,7 +239,7 @@ SArray *tsdbRetrieveDataBlock(tsdb_query_handle_t *pQueryHandle, SArray *pIdList
  * @param order ascending order or descending order
  * @return
  */
-int32_t tsdbResetQuery(tsdb_query_handle_t *pQueryHandle, STimeWindow *window, tsdbpos_t position, int16_t order);
+int32_t tsdbResetQuery(TsdbQueryHandleT *pQueryHandle, STimeWindow *window, TsdbPosT position, int16_t order);
 
 /**
  * todo remove this function later
@@ -246,7 +247,7 @@ int32_t tsdbResetQuery(tsdb_query_handle_t *pQueryHandle, STimeWindow *window, t
  * @param pIdList
  * @return
  */
-SArray *tsdbRetrieveDataRow(tsdb_query_handle_t *pQueryHandle, SArray *pIdList, SQueryRowCond *pCond);
+SArray *tsdbRetrieveDataRow(TsdbQueryHandleT *pQueryHandle, SArray *pIdList, SQueryRowCond *pCond);
 
 /**
  *  Get iterator for super tables, of which tags values satisfy the tag filter info
@@ -259,7 +260,7 @@ SArray *tsdbRetrieveDataRow(tsdb_query_handle_t *pQueryHandle, SArray *pIdList,
  * @param pTagFilterStr tag filter info
  * @return
  */
-tsdb_query_handle_t *tsdbQueryFromTagConds(STsdbQueryCond *pCond, int16_t stableId, const char *pTagFilterStr);
+TsdbQueryHandleT *tsdbQueryFromTagConds(STsdbQueryCond *pCond, int16_t stableId, const char *pTagFilterStr);
 
 /**
  * Get the qualified tables for (super) table query.
@@ -269,7 +270,7 @@ tsdb_query_handle_t *tsdbQueryFromTagConds(STsdbQueryCond *pCond, int16_t stable
  * @param pQueryHandle
  * @return table sid list. the invoker is responsible for the release of this the sid list.
  */
-SArray *tsdbGetTableList(tsdb_query_handle_t *pQueryHandle);
+SArray *tsdbGetTableList(TsdbQueryHandleT *pQueryHandle);
 
 /**
  * Get the qualified table id for a super table according to the tag query expression.
@@ -277,16 +278,16 @@ SArray *tsdbGetTableList(tsdb_query_handle_t *pQueryHandle);
  * @param pTagCond. tag query condition
  *
  */
-int32_t tsdbQueryByTagsCond(tsdb_repo_t* tsdb, int64_t uid, const char* pTagCond, size_t len, STableGroupInfo* pGroupList,
-                      SColIndex* pColIndex, int32_t numOfCols);
+int32_t tsdbQueryByTagsCond(TsdbRepoT *tsdb, int64_t uid, const char *pTagCond, size_t len, STableGroupInfo *pGroupList,
+                      SColIndex *pColIndex, int32_t numOfCols);
 
-int32_t tsdbGetOneTableGroup(tsdb_repo_t* tsdb, int64_t uid, STableGroupInfo* pGroupInfo);
+int32_t tsdbGetOneTableGroup(TsdbRepoT *tsdb, int64_t uid, STableGroupInfo *pGroupInfo);
 
 /**
  * clean up the query handle
  * @param queryHandle
  */
-void tsdbCleanupQueryHandle(tsdb_query_handle_t queryHandle);
+void tsdbCleanupQueryHandle(TsdbQueryHandleT queryHandle);
 
 #ifdef __cplusplus
 }

src/inc/vnode.h

@@ -47,7 +47,6 @@ void*   vnodeGetVnode(int32_t vgId);      // keep refcount unchanged
 void*   vnodeGetRqueue(void *);
 void*   vnodeGetWqueue(int32_t vgId);
 void*   vnodeGetWal(void *pVnode);
-void*   vnodeGetTsdb(void *pVnode);
 
 int32_t vnodeProcessWrite(void *pVnode, int qtype, void *pHead, void *item);
 void    vnodeBuildStatusMsg(void * param);

src/mnode/inc/mgmtMnode.h

@@ -39,10 +39,9 @@ int32_t mgmtGetMnodesNum();
 void *  mgmtGetNextMnode(void *pNode, struct SMnodeObj **pMnode);
 void    mgmtReleaseMnode(struct SMnodeObj *pMnode);
 
-bool    mgmtIsMaster();
-
+char *  mgmtGetMnodeRoleStr();
 void    mgmtGetMnodeIpList(SRpcIpSet *ipSet, bool usePublicIp);
-void    mgmtGetMnodeList(void *mpeers);
+void    mgmtGetMnodeList(void *mnodes);
 
 #ifdef __cplusplus
 }

src/mnode/inc/mgmtSdb.h

@@ -36,54 +36,47 @@ typedef enum {
   SDB_KEY_STRING, 
   SDB_KEY_INT,
   SDB_KEY_AUTO
-} ESdbKeyType;
+} ESdbKey;
 
 typedef enum {
   SDB_OPER_GLOBAL,
   SDB_OPER_LOCAL
-} ESdbOperType;
+} ESdbOper;
 
 typedef struct {
-  ESdbOperType type;
-  void *  table;
-  void *  pObj;
-  int32_t rowSize;
-  void *  rowData;
-} SSdbOperDesc;
+  ESdbOper type;
+  void *   table;
+  void *   pObj;
+  int32_t  rowSize;
+  void *   rowData;
+} SSdbOper;
 
 typedef struct {
   char   *tableName;
   int32_t hashSessions;
   int32_t maxRowSize;
   int32_t refCountPos;
-  ESdbTable   tableId;
-  ESdbKeyType keyType;
-  int32_t (*insertFp)(SSdbOperDesc *pOper);
-  int32_t (*deleteFp)(SSdbOperDesc *pOper);
-  int32_t (*updateFp)(SSdbOperDesc *pOper);
-  int32_t (*encodeFp)(SSdbOperDesc *pOper);
-  int32_t (*decodeFp)(SSdbOperDesc *pDesc);  
-  int32_t (*destroyFp)(SSdbOperDesc *pDesc);
+  ESdbTable tableId;
+  ESdbKey   keyType;
+  int32_t (*insertFp)(SSdbOper *pOper);
+  int32_t (*deleteFp)(SSdbOper *pOper);
+  int32_t (*updateFp)(SSdbOper *pOper);
+  int32_t (*encodeFp)(SSdbOper *pOper);
+  int32_t (*decodeFp)(SSdbOper *pDesc);  
+  int32_t (*destroyFp)(SSdbOper *pDesc);
   int32_t (*restoredFp)();
 } SSdbTableDesc;
 
-typedef struct {
-  int64_t version;
-  void *  wal;
-  pthread_mutex_t mutex;
-} SSdbObject;
-
 int32_t sdbInit();
 void    sdbCleanUp();
-SSdbObject *sdbGetObj();
-
 void *  sdbOpenTable(SSdbTableDesc *desc);
 void    sdbCloseTable(void *handle);
-int     sdbProcessWrite(void *param, void *data, int type);
+bool    sdbIsMaster();
+void    sdbUpdateMnodeRoles();
 
-int32_t sdbInsertRow(SSdbOperDesc *pOper);
-int32_t sdbDeleteRow(SSdbOperDesc *pOper);
-int32_t sdbUpdateRow(SSdbOperDesc *pOper);
+int32_t sdbInsertRow(SSdbOper *pOper);
+int32_t sdbDeleteRow(SSdbOper *pOper);
+int32_t sdbUpdateRow(SSdbOper *pOper);
 
 void    *sdbGetRow(void *handle, void *key);
 void    *sdbFetchRow(void *handle, void *pNode, void **ppRow);

src/mnode/src/mgmtAcct.c

@@ -30,28 +30,28 @@ void *  tsAcctSdb = NULL;
 int32_t tsAcctUpdateSize;
 static void mgmtCreateRootAcct();
 
-static int32_t mgmtActionAcctDestroy(SSdbOperDesc *pOper) {
+static int32_t mgmtActionAcctDestroy(SSdbOper *pOper) {
   SAcctObj *pAcct = pOper->pObj;
   pthread_mutex_destroy(&pAcct->mutex);
   tfree(pOper->pObj);
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t mgmtAcctActionInsert(SSdbOperDesc *pOper) {
+static int32_t mgmtAcctActionInsert(SSdbOper *pOper) {
   SAcctObj *pAcct = pOper->pObj;
   memset(&pAcct->acctInfo, 0, sizeof(SAcctInfo));
   pthread_mutex_init(&pAcct->mutex, NULL);
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t mgmtActionAcctDelete(SSdbOperDesc *pOper) {
+static int32_t mgmtActionAcctDelete(SSdbOper *pOper) {
   SAcctObj *pAcct = pOper->pObj;
   mgmtDropAllUsers(pAcct);
   mgmtDropAllDbs(pAcct);
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t mgmtActionAcctUpdate(SSdbOperDesc *pOper) {
+static int32_t mgmtActionAcctUpdate(SSdbOper *pOper) {
   SAcctObj *pAcct = pOper->pObj;
   SAcctObj *pSaved = mgmtGetAcct(pAcct->user);
   if (pAcct != pSaved) {
@@ -61,14 +61,14 @@ static int32_t mgmtActionAcctUpdate(SSdbOperDesc *pOper) {
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t mgmtActionActionEncode(SSdbOperDesc *pOper) {
+static int32_t mgmtActionActionEncode(SSdbOper *pOper) {
   SAcctObj *pAcct = pOper->pObj;
   memcpy(pOper->rowData, pAcct, tsAcctUpdateSize);
   pOper->rowSize = tsAcctUpdateSize;
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t mgmtActionAcctDecode(SSdbOperDesc *pOper) {
+static int32_t mgmtActionAcctDecode(SSdbOper *pOper) {
   SAcctObj *pAcct = (SAcctObj *) calloc(1, sizeof(SAcctObj));
   if (pAcct == NULL) return TSDB_CODE_SERV_OUT_OF_MEMORY;
 
@@ -110,7 +110,7 @@ int32_t mgmtInitAccts() {
     return -1;
   }
 
-  mTrace("account table is created");
+  mTrace("table:accounts table is created");
   return acctInit();
 }
 
@@ -179,7 +179,7 @@ static void mgmtCreateRootAcct() {
   pAcct->acctId = sdbGetId(tsAcctSdb);
   pAcct->createdTime = taosGetTimestampMs();
 
-  SSdbOperDesc oper = {
+  SSdbOper oper = {
     .type = SDB_OPER_GLOBAL,
     .table = tsAcctSdb,
     .pObj = pAcct,

src/mnode/src/mgmtReplica.c → src/mnode/src/mgmtBalance.c

@@ -16,7 +16,7 @@
 #define _DEFAULT_SOURCE
 #include "os.h"
 #include "trpc.h"
-#include "treplica.h"
+#include "tbalance.h"
 #include "mgmtDef.h"
 #include "mgmtLog.h"
 #include "mgmtMnode.h"
@@ -25,13 +25,12 @@
 
 #ifndef _SYNC
 
-int32_t replicaInit() { return TSDB_CODE_SUCCESS; }
-void    replicaCleanUp() {}
-void    replicaNotify() {}
-void    replicaReset() {}
-int32_t replicaForwardReqToPeer(void *pHead) { return TSDB_CODE_SUCCESS; }
+int32_t balanceInit() { return TSDB_CODE_SUCCESS; }
+void    balanceCleanUp() {}
+void    balanceNotify() {}
+void    balanceReset() {}
 
-int32_t replicaAllocVnodes(SVgObj *pVgroup) {
+int32_t balanceAllocVnodes(SVgObj *pVgroup) {
   void *     pNode = NULL;
   SDnodeObj *pDnode = NULL;
   SDnodeObj *pSelDnode = NULL;

src/mnode/src/mgmtDServer.c

@@ -21,7 +21,7 @@
 #include "tsystem.h"
 #include "tutil.h"
 #include "tgrant.h"
-#include "treplica.h"
+#include "tbalance.h"
 #include "tglobalcfg.h"
 #include "dnode.h"
 #include "mgmtDef.h"

src/mnode/src/mgmtDb.c

@@ -46,12 +46,12 @@ static void    mgmtProcessCreateDbMsg(SQueuedMsg *pMsg);
 static void    mgmtProcessAlterDbMsg(SQueuedMsg *pMsg);
 static void    mgmtProcessDropDbMsg(SQueuedMsg *pMsg);
 
-static int32_t mgmtDbActionDestroy(SSdbOperDesc *pOper) {
+static int32_t mgmtDbActionDestroy(SSdbOper *pOper) {
   tfree(pOper->pObj);
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t mgmtDbActionInsert(SSdbOperDesc *pOper) {
+static int32_t mgmtDbActionInsert(SSdbOper *pOper) {
   SDbObj *pDb = pOper->pObj;
   SAcctObj *pAcct = mgmtGetAcct(pDb->cfg.acct);
 
@@ -72,7 +72,7 @@ static int32_t mgmtDbActionInsert(SSdbOperDesc *pOper) {
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t mgmtDbActionDelete(SSdbOperDesc *pOper) {
+static int32_t mgmtDbActionDelete(SSdbOper *pOper) {
   SDbObj *pDb = pOper->pObj;
   SAcctObj *pAcct = mgmtGetAcct(pDb->cfg.acct);
 
@@ -84,7 +84,7 @@ static int32_t mgmtDbActionDelete(SSdbOperDesc *pOper) {
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t mgmtDbActionUpdate(SSdbOperDesc *pOper) {
+static int32_t mgmtDbActionUpdate(SSdbOper *pOper) {
   SDbObj *pDb = pOper->pObj;
   SDbObj *pSaved = mgmtGetDb(pDb->name);
   if (pDb != pSaved) {
@@ -94,14 +94,14 @@ static int32_t mgmtDbActionUpdate(SSdbOperDesc *pOper) {
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t mgmtDbActionEncode(SSdbOperDesc *pOper) {
+static int32_t mgmtDbActionEncode(SSdbOper *pOper) {
   SDbObj *pDb = pOper->pObj;
   memcpy(pOper->rowData, pDb, tsDbUpdateSize);
   pOper->rowSize = tsDbUpdateSize;
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t mgmtDbActionDecode(SSdbOperDesc *pOper) {
+static int32_t mgmtDbActionDecode(SSdbOper *pOper) {
   SDbObj *pDb = (SDbObj *) calloc(1, sizeof(SDbObj));
   if (pDb == NULL) return TSDB_CODE_SERV_OUT_OF_MEMORY;
   
@@ -146,7 +146,7 @@ int32_t mgmtInitDbs() {
   mgmtAddShellShowMetaHandle(TSDB_MGMT_TABLE_DB, mgmtGetDbMeta);
   mgmtAddShellShowRetrieveHandle(TSDB_MGMT_TABLE_DB, mgmtRetrieveDbs);
   
-  mTrace("db data is initialized");
+  mTrace("table:dbs table is created");
   return 0;
 }
 
@@ -318,7 +318,7 @@ static int32_t mgmtCreateDb(SAcctObj *pAcct, SCMCreateDbMsg *pCreate) {
   pDb->createdTime = taosGetTimestampMs();
   pDb->cfg = *pCreate;
 
-  SSdbOperDesc oper = {
+  SSdbOper oper = {
     .type = SDB_OPER_GLOBAL,
     .table = tsDbSdb,
     .pObj = pDb,
@@ -671,7 +671,7 @@ static int32_t mgmtSetDbDropping(SDbObj *pDb) {
   if (pDb->status) return TSDB_CODE_SUCCESS;
 
   pDb->status = true;
-  SSdbOperDesc oper = {
+  SSdbOper oper = {
     .type = SDB_OPER_GLOBAL,
     .table = tsDbSdb,
     .pObj = pDb,
@@ -756,7 +756,7 @@ static int32_t mgmtAlterDb(SDbObj *pDb, SCMAlterDbMsg *pAlter) {
 
   if (memcmp(&newCfg, &pDb->cfg, sizeof(SDbCfg)) != 0) {
     pDb->cfg = newCfg;
-    SSdbOperDesc oper = {
+    SSdbOper oper = {
       .type = SDB_OPER_GLOBAL,
       .table = tsDbSdb,
       .pObj = pDb,
@@ -814,7 +814,7 @@ static void mgmtDropDb(SQueuedMsg *pMsg) {
   SDbObj *pDb = pMsg->pDb;
   mPrint("db:%s, drop db from sdb", pDb->name);
 
-  SSdbOperDesc oper = {
+  SSdbOper oper = {
     .type = SDB_OPER_GLOBAL,
     .table = tsDbSdb,
     .pObj = pDb

src/mnode/src/mgmtDnode.c

@@ -16,7 +16,7 @@
 #define _DEFAULT_SOURCE
 #include "os.h"
 #include "tgrant.h"
-#include "treplica.h"
+#include "tbalance.h"
 #include "tglobalcfg.h"
 #include "ttime.h"
 #include "tutil.h"
@@ -52,12 +52,12 @@ static int32_t mgmtRetrieveVnodes(SShowObj *pShow, char *data, int32_t rows, voi
 static int32_t mgmtGetDnodeMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn);
 static int32_t mgmtRetrieveDnodes(SShowObj *pShow, char *data, int32_t rows, void *pConn);
 
-static int32_t mgmtDnodeActionDestroy(SSdbOperDesc *pOper) {
+static int32_t mgmtDnodeActionDestroy(SSdbOper *pOper) {
   tfree(pOper->pObj);
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t mgmtDnodeActionInsert(SSdbOperDesc *pOper) {
+static int32_t mgmtDnodeActionInsert(SSdbOper *pOper) {
   SDnodeObj *pDnode = pOper->pObj;
   if (pDnode->status != TAOS_DN_STATUS_DROPPING) {
     pDnode->status = TAOS_DN_STATUS_OFFLINE;
@@ -72,7 +72,7 @@ static int32_t mgmtDnodeActionInsert(SSdbOperDesc *pOper) {
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t mgmtDnodeActionDelete(SSdbOperDesc *pOper) {
+static int32_t mgmtDnodeActionDelete(SSdbOper *pOper) {
   SDnodeObj *pDnode = pOper->pObj;
   void *     pNode = NULL;
   void *     pLastNode = NULL;
@@ -85,7 +85,7 @@ static int32_t mgmtDnodeActionDelete(SSdbOperDesc *pOper) {
     if (pVgroup == NULL) break;
 
     if (pVgroup->vnodeGid[0].dnodeId == pDnode->dnodeId) {
-      SSdbOperDesc oper = {
+      SSdbOper oper = {
         .type = SDB_OPER_LOCAL,
         .table = tsVgroupSdb,
         .pObj = pVgroup,
@@ -101,7 +101,7 @@ static int32_t mgmtDnodeActionDelete(SSdbOperDesc *pOper) {
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t mgmtDnodeActionUpdate(SSdbOperDesc *pOper) {
+static int32_t mgmtDnodeActionUpdate(SSdbOper *pOper) {
   SDnodeObj *pDnode = pOper->pObj;
   SDnodeObj *pSaved = mgmtGetDnode(pDnode->dnodeId);
   if (pDnode != pSaved) {
@@ -111,14 +111,14 @@ static int32_t mgmtDnodeActionUpdate(SSdbOperDesc *pOper) {
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t mgmtDnodeActionEncode(SSdbOperDesc *pOper) {
+static int32_t mgmtDnodeActionEncode(SSdbOper *pOper) {
   SDnodeObj *pDnode = pOper->pObj;
   memcpy(pOper->rowData, pDnode, tsDnodeUpdateSize);
   pOper->rowSize = tsDnodeUpdateSize;
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t mgmtDnodeActionDecode(SSdbOperDesc *pOper) {
+static int32_t mgmtDnodeActionDecode(SSdbOper *pOper) {
   SDnodeObj *pDnode = (SDnodeObj *) calloc(1, sizeof(SDnodeObj));
   if (pDnode == NULL) return TSDB_CODE_SERV_OUT_OF_MEMORY;
 
@@ -180,7 +180,7 @@ int32_t mgmtInitDnodes() {
   mgmtAddShellShowMetaHandle(TSDB_MGMT_TABLE_DNODE, mgmtGetDnodeMeta);
   mgmtAddShellShowRetrieveHandle(TSDB_MGMT_TABLE_DNODE, mgmtRetrieveDnodes);
  
-  mTrace("dnodes table is created");
+  mTrace("table:dnodes table is created");
   return 0;
 }
 
@@ -221,7 +221,7 @@ void mgmtReleaseDnode(SDnodeObj *pDnode) {
 }
 
 void mgmtUpdateDnode(SDnodeObj *pDnode) {
-  SSdbOperDesc oper = {
+  SSdbOper oper = {
     .type = SDB_OPER_GLOBAL,
     .table = tsDnodeSdb,
     .pObj = pDnode,
@@ -340,7 +340,7 @@ void mgmtProcessDnodeStatusMsg(SRpcMsg *rpcMsg) {
   if (pDnode->status == TAOS_DN_STATUS_OFFLINE) {
     mTrace("dnode:%d, from offline to online", pDnode->dnodeId);
     pDnode->status = TAOS_DN_STATUS_READY;
-    replicaNotify();
+    balanceNotify();
   }
 
   mgmtReleaseDnode(pDnode);
@@ -393,7 +393,7 @@ static int32_t mgmtCreateDnode(uint32_t ip) {
   pDnode->totalVnodes = TSDB_INVALID_VNODE_NUM; 
   sprintf(pDnode->dnodeName, "n%d", sdbGetId(tsDnodeSdb) + 1);
 
-  SSdbOperDesc oper = {
+  SSdbOper oper = {
     .type = SDB_OPER_GLOBAL,
     .table = tsDnodeSdb,
     .pObj = pDnode,
@@ -413,7 +413,7 @@ static int32_t mgmtCreateDnode(uint32_t ip) {
 }
 
 int32_t mgmtDropDnode(SDnodeObj *pDnode) {
-  SSdbOperDesc oper = {
+  SSdbOper oper = {
     .type = SDB_OPER_GLOBAL,
     .table = tsDnodeSdb,
     .pObj = pDnode

src/mnode/src/mgmtMain.c

@@ -17,7 +17,7 @@
 #include "os.h"
 #include "taosdef.h"
 #include "tsched.h"
-#include "treplica.h"
+#include "tbalance.h"
 #include "tgrant.h"
 #include "ttimer.h"
 #include "dnode.h"
@@ -62,7 +62,7 @@ int32_t mgmtStartSystem() {
   }
 
   if (grantInit() < 0) {
-    mError("failed to init grants");
+    mError("failed to init grant");
     return -1;
   }
 
@@ -92,7 +92,7 @@ int32_t mgmtStartSystem() {
   }
 
   if (mgmtInitMnodes() < 0) {
-    mError("failed to init mpeers");
+    mError("failed to init mnodes");
     return -1;
   }
 
@@ -101,8 +101,8 @@ int32_t mgmtStartSystem() {
     return -1;
   }
 
-  if (replicaInit() < 0) {
-    mError("failed to init replica")
+  if (balanceInit() < 0) {
+    mError("failed to init balance")
   }
 
   if (mgmtInitDClient() < 0) {
@@ -144,7 +144,7 @@ void mgmtCleanUpSystem() {
   mPrint("starting to clean up mgmt");
   grantCleanUp();
   mgmtCleanupMnodes();
-  replicaCleanUp();
+  balanceCleanUp();
   mgmtCleanUpShell();
   mgmtCleanupDClient();
   mgmtCleanupDServer();
@@ -161,7 +161,7 @@ void mgmtCleanUpSystem() {
 }
 
 void mgmtStopSystem() {
-  if (mgmtIsMaster()) {
+  if (sdbIsMaster()) {
     mTrace("it is a master mgmt node, it could not be stopped");
     return;
   }

src/mnode/src/mgmtMnode.c

@@ -18,7 +18,7 @@
 #include "taoserror.h"
 #include "trpc.h"
 #include "tsync.h"
-#include "treplica.h"
+#include "tbalance.h"
 #include "tutil.h"
 #include "ttime.h"
 #include "tsocket.h"
@@ -30,18 +30,17 @@
 #include "mgmtShell.h"
 #include "mgmtUser.h"
 
-int32_t tsMnodeIsMaster = true;
 static void *  tsMnodeSdb = NULL;
 static int32_t tsMnodeUpdateSize = 0;
 static int32_t mgmtGetMnodeMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn);
 static int32_t mgmtRetrieveMnodes(SShowObj *pShow, char *data, int32_t rows, void *pConn);
 
-static int32_t mgmtMnodeActionDestroy(SSdbOperDesc *pOper) {
+static int32_t mgmtMnodeActionDestroy(SSdbOper *pOper) {
   tfree(pOper->pObj);
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t mgmtMnodeActionInsert(SSdbOperDesc *pOper) {
+static int32_t mgmtMnodeActionInsert(SSdbOper *pOper) {
   SMnodeObj *pMnode = pOper->pObj;
   SDnodeObj *pDnode = mgmtGetDnode(pMnode->mnodeId);
   if (pDnode == NULL) return TSDB_CODE_DNODE_NOT_EXIST;
@@ -53,7 +52,7 @@ static int32_t mgmtMnodeActionInsert(SSdbOperDesc *pOper) {
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t mgmtMnodeActionDelete(SSdbOperDesc *pOper) {
+static int32_t mgmtMnodeActionDelete(SSdbOper *pOper) {
   SMnodeObj *pMnode = pOper->pObj;
 
   SDnodeObj *pDnode = mgmtGetDnode(pMnode->mnodeId);
@@ -65,7 +64,7 @@ static int32_t mgmtMnodeActionDelete(SSdbOperDesc *pOper) {
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t mgmtMnodeActionUpdate(SSdbOperDesc *pOper) {
+static int32_t mgmtMnodeActionUpdate(SSdbOper *pOper) {
   SMnodeObj *pMnode = pOper->pObj;
   SMnodeObj *pSaved = mgmtGetMnode(pMnode->mnodeId);
   if (pMnode != pSaved) {
@@ -76,14 +75,14 @@ static int32_t mgmtMnodeActionUpdate(SSdbOperDesc *pOper) {
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t mgmtMnodeActionEncode(SSdbOperDesc *pOper) {
+static int32_t mgmtMnodeActionEncode(SSdbOper *pOper) {
   SMnodeObj *pMnode = pOper->pObj;
   memcpy(pOper->rowData, pMnode, tsMnodeUpdateSize);
   pOper->rowSize = tsMnodeUpdateSize;
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t mgmtMnodeActionDecode(SSdbOperDesc *pOper) {
+static int32_t mgmtMnodeActionDecode(SSdbOper *pOper) {
   SMnodeObj *pMnode = calloc(1, sizeof(SMnodeObj));
   if (pMnode == NULL) return TSDB_CODE_SERV_OUT_OF_MEMORY;
 
@@ -133,7 +132,7 @@ int32_t mgmtInitMnodes() {
   mgmtAddShellShowMetaHandle(TSDB_MGMT_TABLE_MNODE, mgmtGetMnodeMeta);
   mgmtAddShellShowRetrieveHandle(TSDB_MGMT_TABLE_MNODE, mgmtRetrieveMnodes);
 
-  mTrace("mnodes table is created");
+  mTrace("table:mnodes table is created");
   return TSDB_CODE_SUCCESS;
 }
 
@@ -157,7 +156,7 @@ void *mgmtGetNextMnode(void *pNode, SMnodeObj **pMnode) {
   return sdbFetchRow(tsMnodeSdb, pNode, (void **)pMnode); 
 }
 
-static char *mgmtGetMnodeRoleStr(int32_t role) {
+char *mgmtGetMnodeRoleStr(int32_t role) {
   switch (role) {
     case TAOS_SYNC_ROLE_OFFLINE:
       return "offline";
@@ -172,8 +171,6 @@ static char *mgmtGetMnodeRoleStr(int32_t role) {
   }
 }
 
-bool mgmtIsMaster() { return tsMnodeIsMaster; }
-
 void mgmtGetMnodeIpList(SRpcIpSet *ipSet, bool usePublicIp) {
   void *pNode = NULL;
   while (1) {
@@ -213,10 +210,8 @@ void mgmtGetMnodeList(void *param) {
     mnodes->nodeInfos[index].nodeIp = htonl(pMnode->pDnode->privateIp);
     mnodes->nodeInfos[index].nodePort = htons(pMnode->pDnode->mnodeDnodePort);
     strcpy(mnodes->nodeInfos[index].nodeName, pMnode->pDnode->dnodeName);
-    mPrint("node:%d role:%s", pMnode->mnodeId, mgmtGetMnodeRoleStr(pMnode->role));
     if (pMnode->role == TAOS_SYNC_ROLE_MASTER) {
       mnodes->inUse = index;
-      mPrint("node:%d inUse:%d", pMnode->mnodeId, mnodes->inUse);
     }
 
     index++;
@@ -231,7 +226,7 @@ int32_t mgmtAddMnode(int32_t dnodeId) {
   pMnode->mnodeId = dnodeId;
   pMnode->createdTime = taosGetTimestampMs();
 
-  SSdbOperDesc oper = {
+  SSdbOper oper = {
     .type = SDB_OPER_GLOBAL,
     .table = tsMnodeSdb,
     .pObj = pMnode,
@@ -252,7 +247,7 @@ int32_t mgmtDropMnode(int32_t dnodeId) {
     return TSDB_CODE_DNODE_NOT_EXIST;
   }
   
-  SSdbOperDesc oper = {
+  SSdbOper oper = {
     .type = SDB_OPER_GLOBAL,
     .table = tsMnodeSdb,
     .pObj = pMnode
@@ -268,6 +263,7 @@ int32_t mgmtDropMnode(int32_t dnodeId) {
 }
 
 static int32_t mgmtGetMnodeMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn) {
+  sdbUpdateMnodeRoles();
   SUserObj *pUser = mgmtGetUserFromConn(pConn, NULL);
   if (pUser == NULL) return 0;
 

src/mnode/src/mgmtShell.c

@@ -144,7 +144,7 @@ static void mgmtProcessMsgFromShell(SRpcMsg *rpcMsg) {
     return;
   }
 
-  if (!mgmtIsMaster()) {
+  if (!sdbIsMaster()) {
     // rpcSendRedirectRsp(rpcMsg->handle, mgmtGetMnodeIpListForRedirect());
     mgmtSendSimpleResp(rpcMsg->handle, TSDB_CODE_NO_MASTER);
     rpcFreeCont(rpcMsg->pCont);

src/mnode/src/mgmtTable.c

@@ -83,12 +83,12 @@ static void mgmtDestroyChildTable(SChildTableObj *pTable) {
   tfree(pTable);
 }
 
-static int32_t mgmtChildTableActionDestroy(SSdbOperDesc *pOper) {
+static int32_t mgmtChildTableActionDestroy(SSdbOper *pOper) {
   mgmtDestroyChildTable(pOper->pObj);
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t mgmtChildTableActionInsert(SSdbOperDesc *pOper) {
+static int32_t mgmtChildTableActionInsert(SSdbOper *pOper) {
   SChildTableObj *pTable = pOper->pObj;
 
   SVgObj *pVgroup = mgmtGetVgroup(pTable->vgId);
@@ -128,7 +128,7 @@ static int32_t mgmtChildTableActionInsert(SSdbOperDesc *pOper) {
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t mgmtChildTableActionDelete(SSdbOperDesc *pOper) {
+static int32_t mgmtChildTableActionDelete(SSdbOper *pOper) {
   SChildTableObj *pTable = pOper->pObj;
   if (pTable->vgId == 0) {
     return TSDB_CODE_INVALID_VGROUP_ID;
@@ -169,7 +169,7 @@ static int32_t mgmtChildTableActionDelete(SSdbOperDesc *pOper) {
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t mgmtChildTableActionUpdate(SSdbOperDesc *pOper) {
+static int32_t mgmtChildTableActionUpdate(SSdbOper *pOper) {
   SChildTableObj *pNew = pOper->pObj;
   SChildTableObj *pTable = mgmtGetChildTable(pNew->info.tableId);
   if (pTable != pNew) {
@@ -186,7 +186,7 @@ static int32_t mgmtChildTableActionUpdate(SSdbOperDesc *pOper) {
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t mgmtChildTableActionEncode(SSdbOperDesc *pOper) {
+static int32_t mgmtChildTableActionEncode(SSdbOper *pOper) {
   const int32_t maxRowSize = sizeof(SChildTableObj) + sizeof(SSchema) * TSDB_MAX_COLUMNS;
   SChildTableObj *pTable = pOper->pObj;
   assert(pTable != NULL && pOper->rowData != NULL);
@@ -208,7 +208,7 @@ static int32_t mgmtChildTableActionEncode(SSdbOperDesc *pOper) {
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t mgmtChildTableActionDecode(SSdbOperDesc *pOper) {
+static int32_t mgmtChildTableActionDecode(SSdbOper *pOper) {
   assert(pOper->rowData != NULL);
   SChildTableObj *pTable = calloc(1, sizeof(SChildTableObj));
   if (pTable == NULL) {
@@ -252,7 +252,7 @@ static int32_t mgmtChildTableActionRestored() {
     SDbObj *pDb = mgmtGetDbByTableId(pTable->info.tableId);
     if (pDb == NULL) {
       mError("ctable:%s, failed to get db, discard it", pTable->info.tableId);
-      SSdbOperDesc desc = {0};
+      SSdbOper desc = {0};
       desc.type = SDB_OPER_LOCAL;
       desc.pObj = pTable;
       desc.table = tsChildTableSdb;
@@ -266,7 +266,7 @@ static int32_t mgmtChildTableActionRestored() {
     if (pVgroup == NULL) {
       mError("ctable:%s, failed to get vgroup:%d sid:%d, discard it", pTable->info.tableId, pTable->vgId, pTable->sid);
       pTable->vgId = 0;
-      SSdbOperDesc desc = {0};
+      SSdbOper desc = {0};
       desc.type = SDB_OPER_LOCAL;
       desc.pObj = pTable;
       desc.table = tsChildTableSdb;
@@ -280,7 +280,7 @@ static int32_t mgmtChildTableActionRestored() {
       mError("ctable:%s, db:%s not match with vgroup:%d db:%s sid:%d, discard it",
              pTable->info.tableId, pDb->name, pTable->vgId, pVgroup->dbName, pTable->sid);
       pTable->vgId = 0;
-      SSdbOperDesc desc = {0};
+      SSdbOper desc = {0};
       desc.type = SDB_OPER_LOCAL;
       desc.pObj = pTable;
       desc.table = tsChildTableSdb;
@@ -292,7 +292,7 @@ static int32_t mgmtChildTableActionRestored() {
     if (pVgroup->tableList == NULL) {
       mError("ctable:%s, vgroup:%d tableList is null", pTable->info.tableId, pTable->vgId);
       pTable->vgId = 0;
-      SSdbOperDesc desc = {0};
+      SSdbOper desc = {0};
       desc.type = SDB_OPER_LOCAL;
       desc.pObj = pTable;
       desc.table = tsChildTableSdb;
@@ -306,7 +306,7 @@ static int32_t mgmtChildTableActionRestored() {
       if (pSuperTable == NULL) {
         mError("ctable:%s, stable:%s not exist", pTable->info.tableId, pTable->superTableId);
         pTable->vgId = 0;
-        SSdbOperDesc desc = {0};
+        SSdbOper desc = {0};
         desc.type = SDB_OPER_LOCAL;
         desc.pObj = pTable;
         desc.table = tsChildTableSdb;
@@ -347,7 +347,7 @@ static int32_t mgmtInitChildTables() {
     return -1;
   }
 
-  mTrace("child table is initialized");
+  mTrace("table:ctables is created");
   return 0;
 }
 
@@ -392,12 +392,12 @@ static void mgmtDestroySuperTable(SSuperTableObj *pStable) {
   tfree(pStable);
 }
 
-static int32_t mgmtSuperTableActionDestroy(SSdbOperDesc *pOper) {
+static int32_t mgmtSuperTableActionDestroy(SSdbOper *pOper) {
   mgmtDestroySuperTable(pOper->pObj);
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t mgmtSuperTableActionInsert(SSdbOperDesc *pOper) {
+static int32_t mgmtSuperTableActionInsert(SSdbOper *pOper) {
   SSuperTableObj *pStable = pOper->pObj;
   SDbObj *pDb = mgmtGetDbByTableId(pStable->info.tableId);
   if (pDb != NULL) {
@@ -408,7 +408,7 @@ static int32_t mgmtSuperTableActionInsert(SSdbOperDesc *pOper) {
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t mgmtSuperTableActionDelete(SSdbOperDesc *pOper) {
+static int32_t mgmtSuperTableActionDelete(SSdbOper *pOper) {
   SSuperTableObj *pStable = pOper->pObj;
   SDbObj *pDb = mgmtGetDbByTableId(pStable->info.tableId);
   if (pDb != NULL) {
@@ -420,7 +420,7 @@ static int32_t mgmtSuperTableActionDelete(SSdbOperDesc *pOper) {
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t mgmtSuperTableActionUpdate(SSdbOperDesc *pOper) {
+static int32_t mgmtSuperTableActionUpdate(SSdbOper *pOper) {
   SSuperTableObj *pNew = pOper->pObj;
   SSuperTableObj *pTable = mgmtGetSuperTable(pNew->info.tableId);
   if (pTable != pNew) {
@@ -435,7 +435,7 @@ static int32_t mgmtSuperTableActionUpdate(SSdbOperDesc *pOper) {
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t mgmtSuperTableActionEncode(SSdbOperDesc *pOper) {
+static int32_t mgmtSuperTableActionEncode(SSdbOper *pOper) {
   const int32_t maxRowSize = sizeof(SChildTableObj) + sizeof(SSchema) * TSDB_MAX_COLUMNS;
 
   SSuperTableObj *pStable = pOper->pObj;
@@ -454,7 +454,7 @@ static int32_t mgmtSuperTableActionEncode(SSdbOperDesc *pOper) {
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t mgmtSuperTableActionDecode(SSdbOperDesc *pOper) {
+static int32_t mgmtSuperTableActionDecode(SSdbOper *pOper) {
   assert(pOper->rowData != NULL);
 
   SSuperTableObj *pStable = (SSuperTableObj *) calloc(1, sizeof(SSuperTableObj));
@@ -505,7 +505,7 @@ static int32_t mgmtInitSuperTables() {
     return -1;
   }
 
-  mTrace("stables is initialized");
+  mTrace("table:stables is created");
   return 0;
 }
 
@@ -731,7 +731,7 @@ static void mgmtProcessCreateSuperTableMsg(SQueuedMsg *pMsg) {
     tschema[col].bytes = htons(tschema[col].bytes);
   }
 
-  SSdbOperDesc oper = {
+  SSdbOper oper = {
     .type = SDB_OPER_GLOBAL,
     .table = tsSuperTableSdb,
     .pObj = pStable,
@@ -755,7 +755,7 @@ static void mgmtProcessDropSuperTableMsg(SQueuedMsg *pMsg) {
     mError("stable:%s, numOfTables:%d not 0", pStable->info.tableId, pStable->numOfTables);
     mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_OTHERS);
   } else {
-    SSdbOperDesc oper = {
+    SSdbOper oper = {
       .type = SDB_OPER_GLOBAL,
       .table = tsSuperTableSdb,
       .pObj = pStable
@@ -806,7 +806,7 @@ static int32_t mgmtAddSuperTableTag(SSuperTableObj *pStable, SSchema schema[], i
   pStable->numOfColumns += ntags;
   pStable->sversion++;
 
-  SSdbOperDesc oper = {
+  SSdbOper oper = {
     .type = SDB_OPER_GLOBAL,
     .table = tsSuperTableSdb,
     .pObj = pStable,
@@ -837,7 +837,7 @@ static int32_t mgmtDropSuperTableTag(SSuperTableObj *pStable, char *tagName) {
   int32_t schemaSize = sizeof(SSchema) * (pStable->numOfTags + pStable->numOfColumns);
   pStable->schema = realloc(pStable->schema, schemaSize);
 
-  SSdbOperDesc oper = {
+  SSdbOper oper = {
     .type = SDB_OPER_GLOBAL,
     .table = tsSuperTableSdb,
     .pObj = pStable,
@@ -872,7 +872,7 @@ static int32_t mgmtModifySuperTableTagName(SSuperTableObj *pStable, char *oldTag
   SSchema *schema = (SSchema *) (pStable->schema + (pStable->numOfColumns + col) * sizeof(SSchema));
   strncpy(schema->name, newTagName, TSDB_COL_NAME_LEN);
 
-  SSdbOperDesc oper = {
+  SSdbOper oper = {
     .type = SDB_OPER_GLOBAL,
     .table = tsSuperTableSdb,
     .pObj = pStable,
@@ -931,7 +931,7 @@ static int32_t mgmtAddSuperTableColumn(SDbObj *pDb, SSuperTableObj *pStable, SSc
     mgmtDecAcctRef(pAcct);
   }
 
-  SSdbOperDesc oper = {
+  SSdbOper oper = {
     .type = SDB_OPER_GLOBAL,
     .table = tsSuperTableSdb,
     .pObj = pStable,
@@ -968,7 +968,7 @@ static int32_t mgmtDropSuperTableColumn(SDbObj *pDb, SSuperTableObj *pStable, ch
     mgmtDecAcctRef(pAcct);
   }
 
-  SSdbOperDesc oper = {
+  SSdbOper oper = {
     .type = SDB_OPER_GLOBAL,
     .table = tsSuperTableSdb,
     .pObj = pStable,
@@ -1116,7 +1116,7 @@ void mgmtDropAllSuperTables(SDbObj *pDropDb) {
     }
 
     if (strncmp(pDropDb->name, pTable->info.tableId, dbNameLen) == 0) {
-      SSdbOperDesc oper = {
+      SSdbOper oper = {
         .type = SDB_OPER_LOCAL,
         .table = tsSuperTableSdb,
         .pObj = pTable,
@@ -1354,7 +1354,7 @@ static SChildTableObj* mgmtDoCreateChildTable(SCMCreateTableMsg *pCreate, SVgObj
     }
   }
   
-  SSdbOperDesc desc = {0};
+  SSdbOper desc = {0};
   desc.type = SDB_OPER_GLOBAL;
   desc.pObj = pTable;
   desc.table = tsChildTableSdb;
@@ -1508,7 +1508,7 @@ static int32_t mgmtAddNormalTableColumn(SDbObj *pDb, SChildTableObj *pTable, SSc
     mgmtDecAcctRef(pAcct);
   }
  
-  SSdbOperDesc oper = {
+  SSdbOper oper = {
     .type = SDB_OPER_GLOBAL,
     .table = tsChildTableSdb,
     .pObj = pTable,
@@ -1542,7 +1542,7 @@ static int32_t mgmtDropNormalTableColumn(SDbObj *pDb, SChildTableObj *pTable, ch
     mgmtDecAcctRef(pAcct);
   }
  
-  SSdbOperDesc oper = {
+  SSdbOper oper = {
     .type = SDB_OPER_GLOBAL,
     .table = tsChildTableSdb,
     .pObj = pTable,
@@ -1687,7 +1687,7 @@ void mgmtDropAllChildTables(SDbObj *pDropDb) {
     }
 
     if (strncmp(pDropDb->name, pTable->info.tableId, dbNameLen) == 0) {
-      SSdbOperDesc oper = {
+      SSdbOper oper = {
         .type = SDB_OPER_LOCAL,
         .table = tsChildTableSdb,
         .pObj = pTable,
@@ -1716,7 +1716,7 @@ static void mgmtDropAllChildTablesInStable(SSuperTableObj *pStable) {
     }
 
     if (pTable->superTable == pStable) {
-      SSdbOperDesc oper = {
+      SSdbOper oper = {
         .type = SDB_OPER_LOCAL,
         .table = tsChildTableSdb,
         .pObj = pTable,
@@ -1805,7 +1805,7 @@ static void mgmtProcessDropChildTableRsp(SRpcMsg *rpcMsg) {
     return;
   }
   
-  SSdbOperDesc oper = {
+  SSdbOper oper = {
     .type = SDB_OPER_GLOBAL,
     .table = tsChildTableSdb,
     .pObj = pTable
@@ -1848,7 +1848,7 @@ static void mgmtProcessCreateChildTableRsp(SRpcMsg *rpcMsg) {
       mError("table:%s, failed to create in dnode, thandle:%p result:%s", pTable->info.tableId,
              queueMsg->thandle, tstrerror(rpcMsg->code));
       
-      SSdbOperDesc oper = {
+      SSdbOper oper = {
         .type = SDB_OPER_GLOBAL,
         .table = tsChildTableSdb,
         .pObj = pTable

src/mnode/src/mgmtUser.c

@@ -36,12 +36,12 @@ static void mgmtProcessCreateUserMsg(SQueuedMsg *pMsg);
 static void mgmtProcessAlterUserMsg(SQueuedMsg *pMsg);
 static void mgmtProcessDropUserMsg(SQueuedMsg *pMsg);
 
-static int32_t mgmtUserActionDestroy(SSdbOperDesc *pOper) {
+static int32_t mgmtUserActionDestroy(SSdbOper *pOper) {
   tfree(pOper->pObj);
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t mgmtUserActionInsert(SSdbOperDesc *pOper) {
+static int32_t mgmtUserActionInsert(SSdbOper *pOper) {
   SUserObj *pUser = pOper->pObj;
   SAcctObj *pAcct = mgmtGetAcct(pUser->acct);
 
@@ -56,7 +56,7 @@ static int32_t mgmtUserActionInsert(SSdbOperDesc *pOper) {
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t mgmtUserActionDelete(SSdbOperDesc *pOper) {
+static int32_t mgmtUserActionDelete(SSdbOper *pOper) {
   SUserObj *pUser = pOper->pObj;
   SAcctObj *pAcct = mgmtGetAcct(pUser->acct);
 
@@ -67,7 +67,7 @@ static int32_t mgmtUserActionDelete(SSdbOperDesc *pOper) {
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t mgmtUserActionUpdate(SSdbOperDesc *pOper) {
+static int32_t mgmtUserActionUpdate(SSdbOper *pOper) {
   SUserObj *pUser = pOper->pObj;
   SUserObj *pSaved = mgmtGetUser(pUser->user);
   if (pUser != pSaved) {
@@ -77,14 +77,14 @@ static int32_t mgmtUserActionUpdate(SSdbOperDesc *pOper) {
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t mgmtUserActionEncode(SSdbOperDesc *pOper) {
+static int32_t mgmtUserActionEncode(SSdbOper *pOper) {
   SUserObj *pUser = pOper->pObj;
   memcpy(pOper->rowData, pUser, tsUserUpdateSize);
   pOper->rowSize = tsUserUpdateSize;
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t mgmtUserActionDecode(SSdbOperDesc *pOper) {
+static int32_t mgmtUserActionDecode(SSdbOper *pOper) {
   SUserObj *pUser = (SUserObj *) calloc(1, sizeof(SUserObj));
   if (pUser == NULL) return TSDB_CODE_SERV_OUT_OF_MEMORY;
 
@@ -137,7 +137,7 @@ int32_t mgmtInitUsers() {
   mgmtAddShellShowMetaHandle(TSDB_MGMT_TABLE_USER, mgmtGetUserMeta);
   mgmtAddShellShowRetrieveHandle(TSDB_MGMT_TABLE_USER, mgmtRetrieveUsers);
   
-  mTrace("user data is initialized");
+  mTrace("table:users table is created");
   return 0;
 }
 
@@ -154,7 +154,7 @@ void mgmtReleaseUser(SUserObj *pUser) {
 }
 
 static int32_t mgmtUpdateUser(SUserObj *pUser) {
-  SSdbOperDesc oper = {
+  SSdbOper oper = {
     .type = SDB_OPER_GLOBAL,
     .table = tsUserSdb,
     .pObj = pUser,
@@ -202,7 +202,7 @@ int32_t mgmtCreateUser(SAcctObj *pAcct, char *name, char *pass) {
     pUser->superAuth = 1;
   }
 
-  SSdbOperDesc oper = {
+  SSdbOper oper = {
     .type = SDB_OPER_GLOBAL,
     .table = tsUserSdb,
     .pObj = pUser,
@@ -219,7 +219,7 @@ int32_t mgmtCreateUser(SAcctObj *pAcct, char *name, char *pass) {
 }
 
 static int32_t mgmtDropUser(SUserObj *pUser) {
-  SSdbOperDesc oper = {
+  SSdbOper oper = {
     .type = SDB_OPER_GLOBAL,
     .table = tsUserSdb,
     .pObj = pUser
@@ -493,7 +493,7 @@ void  mgmtDropAllUsers(SAcctObj *pAcct)  {
     if (pUser == NULL) break;
 
     if (strncmp(pUser->acct, pAcct->user, acctNameLen) == 0) {
-      SSdbOperDesc oper = {
+      SSdbOper oper = {
         .type = SDB_OPER_LOCAL,
         .table = tsUserSdb,
         .pObj = pUser,

src/mnode/src/mgmtVgroup.c

@@ -22,7 +22,7 @@
 #include "tidpool.h"
 #include "tsync.h"
 #include "ttime.h"
-#include "treplica.h"
+#include "tbalance.h"
 #include "mgmtDef.h"
 #include "mgmtLog.h"
 #include "mgmtDb.h"
@@ -48,7 +48,7 @@ static void    mgmtProcessVnodeCfgMsg(SRpcMsg *rpcMsg) ;
 static void mgmtSendDropVgroupMsg(SVgObj *pVgroup, void *ahandle);
 static void mgmtSendCreateVgroupMsg(SVgObj *pVgroup, void *ahandle);
 
-static int32_t mgmtVgroupActionDestroy(SSdbOperDesc *pOper) {
+static int32_t mgmtVgroupActionDestroy(SSdbOper *pOper) {
   SVgObj *pVgroup = pOper->pObj;
   if (pVgroup->idPool) {
     taosIdPoolCleanUp(pVgroup->idPool);
@@ -62,7 +62,7 @@ static int32_t mgmtVgroupActionDestroy(SSdbOperDesc *pOper) {
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t mgmtVgroupActionInsert(SSdbOperDesc *pOper) {
+static int32_t mgmtVgroupActionInsert(SSdbOper *pOper) {
   SVgObj *pVgroup = pOper->pObj;
   SDbObj *pDb = mgmtGetDb(pVgroup->dbName);
   if (pDb == NULL) {
@@ -104,7 +104,7 @@ static int32_t mgmtVgroupActionInsert(SSdbOperDesc *pOper) {
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t mgmtVgroupActionDelete(SSdbOperDesc *pOper) {
+static int32_t mgmtVgroupActionDelete(SSdbOper *pOper) {
   SVgObj *pVgroup = pOper->pObj;
   
   if (pVgroup->pDb != NULL) {
@@ -124,7 +124,7 @@ static int32_t mgmtVgroupActionDelete(SSdbOperDesc *pOper) {
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t mgmtVgroupActionUpdate(SSdbOperDesc *pOper) {
+static int32_t mgmtVgroupActionUpdate(SSdbOper *pOper) {
   SVgObj *pNew = pOper->pObj;
   SVgObj *pVgroup = mgmtGetVgroup(pNew->vgId);
   if (pVgroup != pNew) {
@@ -147,14 +147,14 @@ static int32_t mgmtVgroupActionUpdate(SSdbOperDesc *pOper) {
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t mgmtVgroupActionEncode(SSdbOperDesc *pOper) {
+static int32_t mgmtVgroupActionEncode(SSdbOper *pOper) {
   SVgObj *pVgroup = pOper->pObj;
   memcpy(pOper->rowData, pVgroup, tsVgUpdateSize);
   pOper->rowSize = tsVgUpdateSize;
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t mgmtVgroupActionDecode(SSdbOperDesc *pOper) {
+static int32_t mgmtVgroupActionDecode(SSdbOper *pOper) {
   SVgObj *pVgroup = (SVgObj *) calloc(1, sizeof(SVgObj));
   if (pVgroup == NULL) return TSDB_CODE_SERV_OUT_OF_MEMORY;
 
@@ -199,7 +199,7 @@ int32_t mgmtInitVgroups() {
   mgmtAddDClientRspHandle(TSDB_MSG_TYPE_MD_DROP_VNODE_RSP, mgmtProcessDropVnodeRsp);
   mgmtAddDServerMsgHandle(TSDB_MSG_TYPE_DM_CONFIG_VNODE, mgmtProcessVnodeCfgMsg);
 
-  mTrace("vgroup is initialized");
+  mTrace("table:vgroups is created");
   
   return 0;
 }
@@ -213,7 +213,7 @@ SVgObj *mgmtGetVgroup(int32_t vgId) {
 }
 
 void mgmtUpdateVgroup(SVgObj *pVgroup) {
-  SSdbOperDesc oper = {
+  SSdbOper oper = {
     .type = SDB_OPER_GLOBAL,
     .table = tsVgroupSdb,
     .pObj = pVgroup,
@@ -249,7 +249,7 @@ void mgmtCreateVgroup(SQueuedMsg *pMsg, SDbObj *pDb) {
   strcpy(pVgroup->dbName, pDb->name);
   pVgroup->numOfVnodes = pDb->cfg.replications;
   pVgroup->createdTime = taosGetTimestampMs();
-  if (replicaAllocVnodes(pVgroup) != 0) {
+  if (balanceAllocVnodes(pVgroup) != 0) {
     mError("db:%s, no enough dnode to alloc %d vnodes to vgroup", pDb->name, pVgroup->numOfVnodes);
     free(pVgroup);
     mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_NO_ENOUGH_DNODES);
@@ -257,7 +257,7 @@ void mgmtCreateVgroup(SQueuedMsg *pMsg, SDbObj *pDb) {
     return;
   }
 
-  SSdbOperDesc oper = {
+  SSdbOper oper = {
     .type = SDB_OPER_GLOBAL,
     .table = tsVgroupSdb,
     .pObj = pVgroup,
@@ -289,7 +289,7 @@ void mgmtDropVgroup(SVgObj *pVgroup, void *ahandle) {
   } else {
     mTrace("vgroup:%d, replica:%d is deleting from sdb", pVgroup->vgId, pVgroup->numOfVnodes);
     mgmtSendDropVgroupMsg(pVgroup, NULL);
-    SSdbOperDesc oper = {
+    SSdbOper oper = {
       .type = SDB_OPER_GLOBAL,
       .table = tsVgroupSdb,
       .pObj = pVgroup
@@ -596,7 +596,7 @@ static void mgmtProcessCreateVnodeRsp(SRpcMsg *rpcMsg) {
     SQueuedMsg *newMsg = mgmtCloneQueuedMsg(queueMsg);
     mgmtAddToShellQueue(newMsg);
   } else {
-    SSdbOperDesc oper = {
+    SSdbOper oper = {
       .type = SDB_OPER_GLOBAL,
       .table = tsVgroupSdb,
       .pObj = pVgroup
@@ -659,7 +659,7 @@ static void mgmtProcessDropVnodeRsp(SRpcMsg *rpcMsg) {
 
   if (queueMsg->received != queueMsg->expected) return;
 
-  SSdbOperDesc oper = {
+  SSdbOper oper = {
     .type = SDB_OPER_GLOBAL,
     .table = tsVgroupSdb,
     .pObj = pVgroup
@@ -716,7 +716,7 @@ void mgmtDropAllVgroups(SDbObj *pDropDb) {
     if (pVgroup == NULL) break;
 
     if (strncmp(pDropDb->name, pVgroup->dbName, dbNameLen) == 0) {
-      SSdbOperDesc oper = {
+      SSdbOper oper = {
         .type = SDB_OPER_LOCAL,
         .table = tsVgroupSdb,
         .pObj = pVgroup,

src/query/src/queryExecutor.c

@@ -2479,7 +2479,7 @@ static int64_t doScanAllDataBlocks(SQueryRuntimeEnv *pRuntimeEnv) {
   dTrace("QInfo:%p query start, qrange:%" PRId64 "-%" PRId64 ", lastkey:%" PRId64 ", order:%d",
          GET_QINFO_ADDR(pRuntimeEnv), pQuery->window.skey, pQuery->window.ekey, pQuery->lastKey, pQuery->order.order);
 
-  tsdb_query_handle_t pQueryHandle = pRuntimeEnv->scanFlag == MASTER_SCAN? pRuntimeEnv->pQueryHandle:pRuntimeEnv->pSecQueryHandle;
+  TsdbQueryHandleT pQueryHandle = pRuntimeEnv->scanFlag == MASTER_SCAN? pRuntimeEnv->pQueryHandle:pRuntimeEnv->pSecQueryHandle;
   while (tsdbNextDataBlock(pQueryHandle)) {
     
     if (isQueryKilled(GET_QINFO_ADDR(pRuntimeEnv))) {
@@ -4221,7 +4221,7 @@ static int64_t queryOnDataBlocks(SQInfo *pQInfo) {
   
   int64_t st = taosGetTimestampMs();
   
-  tsdb_query_handle_t *pQueryHandle = pRuntimeEnv->pQueryHandle;
+  TsdbQueryHandleT *pQueryHandle = pRuntimeEnv->pQueryHandle;
   while (tsdbNextDataBlock(pQueryHandle)) {
     if (isQueryKilled(pQInfo)) {
       break;

src/tsdb/inc/tsdbMain.h

@@ -120,7 +120,7 @@ STSchema * tsdbGetTableTagSchema(STsdbMeta *pMeta, STable *pTable);
 #define TSDB_TABLE_OF_ID(pHandle, id) ((pHandle)->pTables)[id]
 #define TSDB_GET_TABLE_OF_NAME(pHandle, name) /* TODO */
 
-STsdbMeta *tsdbGetMeta(tsdb_repo_t *pRepo);
+STsdbMeta *tsdbGetMeta(TsdbRepoT *pRepo);
 
 int32_t tsdbCreateTableImpl(STsdbMeta *pMeta, STableCfg *pCfg);
 int32_t tsdbDropTableImpl(STsdbMeta *pMeta, STableId tableId);
@@ -160,10 +160,10 @@ typedef struct {
   STsdbCacheBlock *curBlock;
   SCacheMem *      mem;
   SCacheMem *      imem;
-  tsdb_repo_t *    pRepo;
+  TsdbRepoT *    pRepo;
 } STsdbCache;
 
-STsdbCache *tsdbInitCache(int maxBytes, int cacheBlockSize, tsdb_repo_t *pRepo);
+STsdbCache *tsdbInitCache(int maxBytes, int cacheBlockSize, TsdbRepoT *pRepo);
 void        tsdbFreeCache(STsdbCache *pCache);
 void *      tsdbAllocFromCache(STsdbCache *pCache, int bytes, TSKEY key);
 
@@ -220,11 +220,12 @@ STsdbFileH *tsdbInitFileH(char *dataDir, int maxFiles);
 void        tsdbCloseFileH(STsdbFileH *pFileH);
 int         tsdbCreateFile(char *dataDir, int fileId, const char *suffix, int maxTables, SFile *pFile, int writeHeader,
                            int toClose);
-int         tsdbCreateFGroup(STsdbFileH *pFileH, char *dataDir, int fid, int maxTables);
+SFileGroup *tsdbCreateFGroup(STsdbFileH *pFileH, char *dataDir, int fid, int maxTables);
 int         tsdbOpenFile(SFile *pFile, int oflag);
 int         tsdbCloseFile(SFile *pFile);
 SFileGroup *tsdbOpenFilesForCommit(STsdbFileH *pFileH, int fid);
 int         tsdbRemoveFileGroup(STsdbFileH *pFile, int fid);
+int         tsdbGetFileName(char *dataDir, int fileId, const char *suffix, char *fname);
 
 #define TSDB_FGROUP_ITER_FORWARD TSDB_ORDER_ASC
 #define TSDB_FGROUP_ITER_BACKWARD TSDB_ORDER_DESC
@@ -270,6 +271,8 @@ typedef struct {
   TSKEY   keyLast;
 } SCompBlock;
 
+// Maximum number of sub-blocks a super-block can have
+#define TSDB_MAX_SUBBLOCKS 8
 #define IS_SUPER_BLOCK(pBlock) ((pBlock)->numOfSubBlocks >= 1)
 #define IS_SUB_BLOCK(pBlock) ((pBlock)->numOfSubBlocks == 0)
 
@@ -307,19 +310,11 @@ typedef struct {
   SCompCol cols[];
 } SCompData;
 
-STsdbFileH *tsdbGetFile(tsdb_repo_t *pRepo);
-
-int tsdbCopyBlockDataInFile(SFile *pOutFile, SFile *pInFile, SCompInfo *pCompInfo, int idx, int isLast,
-                            SDataCols *pCols);
-
-int tsdbLoadCompIdx(SFileGroup *pGroup, void *buf, int maxTables);
-int tsdbLoadCompBlocks(SFileGroup *pGroup, SCompIdx *pIdx, void *buf);
-int tsdbLoadCompCols(SFile *pFile, SCompBlock *pBlock, void *buf);
-int tsdbLoadColData(SFile *pFile, SCompCol *pCol, int64_t blockBaseOffset, void *buf);
-int tsdbLoadDataBlock(SFile *pFile, SCompBlock *pStartBlock, int numOfBlocks, SDataCols *pCols, SCompData *pCompData);
+STsdbFileH *tsdbGetFile(TsdbRepoT *pRepo);
 
+int         tsdbCopyBlockDataInFile(SFile *pOutFile, SFile *pInFile, SCompInfo *pCompInfo, int idx, int isLast,
+                                    SDataCols *pCols);
 SFileGroup *tsdbSearchFGroup(STsdbFileH *pFileH, int fid);
-
 void tsdbGetKeyRangeOfFileId(int32_t daysPerFile, int8_t precision, int32_t fileId, TSKEY *minKey, TSKEY *maxKey);
 
 // TSDB repository definition
@@ -375,9 +370,103 @@ typedef struct {
 int         tsdbInitSubmitMsgIter(SSubmitMsg *pMsg, SSubmitMsgIter *pIter);
 SSubmitBlk *tsdbGetSubmitMsgNext(SSubmitMsgIter *pIter);
 
-int32_t tsdbTriggerCommit(tsdb_repo_t *repo);
-int32_t tsdbLockRepo(tsdb_repo_t *repo);
-int32_t tsdbUnLockRepo(tsdb_repo_t *repo);
+int32_t tsdbTriggerCommit(TsdbRepoT *repo);
+int32_t tsdbLockRepo(TsdbRepoT *repo);
+int32_t tsdbUnLockRepo(TsdbRepoT *repo);
+
+typedef enum { TSDB_WRITE_HELPER, TSDB_READ_HELPER } tsdb_rw_helper_t;
+
+typedef struct {
+  tsdb_rw_helper_t type;  // helper type
+
+  int    maxTables;
+  int    maxRowSize;
+  int    maxRows;
+  int    maxCols;
+  int    minRowsPerFileBlock;
+  int    maxRowsPerFileBlock;
+  int8_t compress;
+} SHelperCfg;
+
+typedef struct {
+  int fid;
+  TSKEY minKey;
+  TSKEY maxKey;
+  // For read/write purpose
+  SFile headF;
+  SFile dataF;
+  SFile lastF;
+  // For write purpose only
+  SFile nHeadF;
+  SFile nLastF;
+} SHelperFile;
+
+typedef struct {
+  int64_t uid;
+  int32_t tid;
+  int32_t sversion;
+} SHelperTable;
+
+typedef struct {
+  // Global configuration
+  SHelperCfg config;
+
+  int8_t state;
+
+  // For file set usage
+  SHelperFile files;
+  SCompIdx *  pCompIdx;
+
+  // For table set usage
+  SHelperTable tableInfo;
+  SCompInfo *  pCompInfo;
+  bool         hasOldLastBlock;
+
+  // For block set usage
+  SCompData *pCompData;
+  SDataCols *pDataCols[2];
+
+} SRWHelper;
+
+// --------- Helper state
+#define TSDB_HELPER_CLEAR_STATE 0x0        // Clear state
+#define TSDB_HELPER_FILE_SET_AND_OPEN 0x1  // File is set
+#define TSDB_HELPER_IDX_LOAD 0x2           // SCompIdx part is loaded
+#define TSDB_HELPER_TABLE_SET 0x4          // Table is set
+#define TSDB_HELPER_INFO_LOAD 0x8          // SCompInfo part is loaded
+#define TSDB_HELPER_FILE_DATA_LOAD 0x10    // SCompData part is loaded
+
+#define TSDB_HELPER_TYPE(h) ((h)->config.type)
+
+#define helperSetState(h, s) (((h)->state) |= (s))
+#define helperClearState(h, s) ((h)->state &= (~(s)))
+#define helperHasState(h, s) ((((h)->state) & (s)) == (s))
+#define blockAtIdx(h, idx) ((h)->pCompInfo->blocks + idx)
+
+int  tsdbInitReadHelper(SRWHelper *pHelper, STsdbRepo *pRepo);
+int  tsdbInitWriteHelper(SRWHelper *pHelper, STsdbRepo *pRepo);
+// int  tsdbInitHelper(SRWHelper *pHelper, SHelperCfg *pCfg);
+void tsdbDestroyHelper(SRWHelper *pHelper);
+void tsdbResetHelper(SRWHelper *pHelper);
+
+// --------- For set operations
+int tsdbSetAndOpenHelperFile(SRWHelper *pHelper, SFileGroup *pGroup);
+// void tsdbSetHelperTable(SRWHelper *pHelper, SHelperTable *pHelperTable, STSchema *pSchema);
+void tsdbSetHelperTable(SRWHelper *pHelper, STable *pTable, STsdbRepo *pRepo);
+int  tsdbCloseHelperFile(SRWHelper *pHelper, bool hasError);
+
+// --------- For read operations
+int tsdbLoadCompIdx(SRWHelper *pHelper, void *target);
+int tsdbLoadCompInfo(SRWHelper *pHelper, void *target);
+int tsdbLoadCompData(SRWHelper *pHelper, SCompBlock *pCompBlock, void *target);
+int tsdbLoadBlockDataCols(SRWHelper *pHelper, SDataCols *pDataCols, int blkIdx, int16_t *colIds, int numOfColIds);
+int tsdbLoadBlockData(SRWHelper *pHelper, SCompBlock *pCompBlock, SDataCols *target);
+
+// --------- For write operations
+int tsdbWriteDataBlock(SRWHelper *pHelper, SDataCols *pDataCols);
+int tsdbMoveLastBlockIfNeccessary(SRWHelper *pHelper);
+int tsdbWriteCompInfo(SRWHelper *pHelper);
+int tsdbWriteCompIdx(SRWHelper *pHelper);
 
 #ifdef __cplusplus
 }

src/tsdb/src/tsdbCache.c

@@ -21,7 +21,7 @@ static int  tsdbAllocBlockFromPool(STsdbCache *pCache);
 static void tsdbFreeBlockList(SList *list);
 static void tsdbFreeCacheMem(SCacheMem *mem);
 
-STsdbCache *tsdbInitCache(int maxBytes, int cacheBlockSize, tsdb_repo_t *pRepo) {
+STsdbCache *tsdbInitCache(int maxBytes, int cacheBlockSize, TsdbRepoT *pRepo) {
   STsdbCache *pCache = (STsdbCache *)calloc(1, sizeof(STsdbCache));
   if (pCache == NULL) return NULL;
 

src/tsdb/src/tsdbFile.c

@@ -21,10 +21,12 @@
 #include <sys/stat.h>
 #include <sys/types.h>
 #include <unistd.h>
+#include <libgen.h>
 
+#include "talgo.h"
+#include "tchecksum.h"
 #include "tsdbMain.h"
 #include "tutil.h"
-#include "talgo.h"
 
 const char *tsdbFileSuffix[] = {
     ".head",  // TSDB_FILE_TYPE_HEAD
@@ -34,7 +36,6 @@ const char *tsdbFileSuffix[] = {
 
 static int compFGroupKey(const void *key, const void *fgroup);
 static int compFGroup(const void *arg1, const void *arg2);
-static int tsdbGetFileName(char *dataDir, int fileId, const char *suffix, char *fname);
 static int tsdbWriteFileHead(SFile *pFile);
 static int tsdbWriteHeadFileIdx(SFile *pFile, int maxTables);
 static int tsdbOpenFGroup(STsdbFileH *pFileH, char *dataDir, int fid);
@@ -93,24 +94,36 @@ static int tsdbOpenFGroup(STsdbFileH *pFileH, char *dataDir, int fid) {
   return 0;
 }
 
-int tsdbCreateFGroup(STsdbFileH *pFileH, char *dataDir, int fid, int maxTables) {
-  if (pFileH->numOfFGroups >= pFileH->maxFGroups) return -1;
+/**
+ * Create the file group if the file group not exists.
+ *
+ * @return A pointer to
+ */
+SFileGroup *tsdbCreateFGroup(STsdbFileH *pFileH, char *dataDir, int fid, int maxTables) {
+  if (pFileH->numOfFGroups >= pFileH->maxFGroups) return NULL;
 
   SFileGroup  fGroup;
   SFileGroup *pFGroup = &fGroup;
-  if (tsdbSearchFGroup(pFileH, fid) == NULL) { // if not exists, create one
+
+  SFileGroup *pGroup = tsdbSearchFGroup(pFileH, fid);
+  if (pGroup == NULL) {  // if not exists, create one
     pFGroup->fileId = fid;
     for (int type = TSDB_FILE_TYPE_HEAD; type < TSDB_FILE_TYPE_MAX; type++) {
-      if (tsdbCreateFile(dataDir, fid, tsdbFileSuffix[type], maxTables, &(pFGroup->files[type]), type == TSDB_FILE_TYPE_HEAD ? 1 : 0, 1) < 0) {
-        // TODO: deal with the ERROR here, remove those creaed file
-        return -1;
-      }
+      if (tsdbCreateFile(dataDir, fid, tsdbFileSuffix[type], maxTables, &(pFGroup->files[type]),
+                         type == TSDB_FILE_TYPE_HEAD ? 1 : 0, 1) < 0)
+        goto _err;
     }
 
     pFileH->fGroup[pFileH->numOfFGroups++] = fGroup;
     qsort((void *)(pFileH->fGroup), pFileH->numOfFGroups, sizeof(SFileGroup), compFGroup);
+    return tsdbSearchFGroup(pFileH, fid);
   }
-  return 0;
+
+  return pGroup;
+
+_err:
+  // TODO: deal with the err here
+  return NULL;
 }
 
 int tsdbRemoveFileGroup(STsdbFileH *pFileH, int fid) {
@@ -183,27 +196,27 @@ SFileGroup *tsdbGetFileGroupNext(SFileGroupIter *pIter) {
   return ret;
 }
 
-int tsdbLoadDataBlock(SFile *pFile, SCompBlock *pStartBlock, int numOfBlocks, SDataCols *pCols, SCompData *pCompData) {
-  SCompBlock *pBlock = pStartBlock;
-  for (int i = 0; i < numOfBlocks; i++) {
-    if (tsdbLoadCompCols(pFile, pBlock, (void *)pCompData) < 0) return -1;
-    pCols->numOfPoints += (pCompData->cols[0].len / 8);
-    for (int iCol = 0; iCol < pBlock->numOfCols; iCol++) {
-      SCompCol *pCompCol = &(pCompData->cols[iCol]);
-      // pCols->numOfPoints += pBlock->numOfPoints;
-      int k = 0;
-      for (; k < pCols->numOfCols; k++) {
-        if (pCompCol->colId == pCols->cols[k].colId) break;
-      }
-
-      if (tsdbLoadColData(pFile, pCompCol, pBlock->offset,
-                          (void *)((char *)(pCols->cols[k].pData) + pCols->cols[k].len)) < 0)
-        return -1;
-    }
-    pStartBlock++;
-  }
-  return 0;
-}
+// int tsdbLoadDataBlock(SFile *pFile, SCompBlock *pStartBlock, int numOfBlocks, SDataCols *pCols, SCompData *pCompData) {
+//   SCompBlock *pBlock = pStartBlock;
+//   for (int i = 0; i < numOfBlocks; i++) {
+//     if (tsdbLoadCompCols(pFile, pBlock, (void *)pCompData) < 0) return -1;
+//     pCols->numOfPoints += (pCompData->cols[0].len / 8);
+//     for (int iCol = 0; iCol < pBlock->numOfCols; iCol++) {
+//       SCompCol *pCompCol = &(pCompData->cols[iCol]);
+//       // pCols->numOfPoints += pBlock->numOfPoints;
+//       int k = 0;
+//       for (; k < pCols->numOfCols; k++) {
+//         if (pCompCol->colId == pCols->cols[k].colId) break;
+//       }
+
+//       if (tsdbLoadColData(pFile, pCompCol, pBlock->offset,
+//                           (void *)((char *)(pCols->cols[k].pData) + pCols->cols[k].len)) < 0)
+//         return -1;
+//     }
+//     pStartBlock++;
+//   }
+//   return 0;
+// }
 
 int tsdbCopyBlockDataInFile(SFile *pOutFile, SFile *pInFile, SCompInfo *pCompInfo, int idx, int isLast, SDataCols *pCols) {
   SCompBlock *pSuperBlock = TSDB_COMPBLOCK_AT(pCompInfo, idx);
@@ -239,42 +252,42 @@ int tsdbCopyBlockDataInFile(SFile *pOutFile, SFile *pInFile, SCompInfo *pCompInf
   return 0;
 }
 
-int tsdbLoadCompIdx(SFileGroup *pGroup, void *buf, int maxTables) {
-  SFile *pFile = &(pGroup->files[TSDB_FILE_TYPE_HEAD]);
-  if (lseek(pFile->fd, TSDB_FILE_HEAD_SIZE, SEEK_SET) < 0) return -1;
+// int tsdbLoadCompIdx(SFileGroup *pGroup, void *buf, int maxTables) {
+//   SFile *pFile = &(pGroup->files[TSDB_FILE_TYPE_HEAD]);
+//   if (lseek(pFile->fd, TSDB_FILE_HEAD_SIZE, SEEK_SET) < 0) return -1;
 
-  if (read(pFile->fd, buf, sizeof(SCompIdx) * maxTables) < 0) return -1;
-  // TODO: need to check the correctness
-  return 0;
-}
+//   if (read(pFile->fd, buf, sizeof(SCompIdx) * maxTables) < 0) return -1;
+//   // TODO: need to check the correctness
+//   return 0;
+// }
 
-int tsdbLoadCompBlocks(SFileGroup *pGroup, SCompIdx *pIdx, void *buf) {
-  SFile *pFile = &(pGroup->files[TSDB_FILE_TYPE_HEAD]);
+// int tsdbLoadCompBlocks(SFileGroup *pGroup, SCompIdx *pIdx, void *buf) {
+//   SFile *pFile = &(pGroup->files[TSDB_FILE_TYPE_HEAD]);
 
-  if (lseek(pFile->fd, pIdx->offset, SEEK_SET) < 0) return -1;
+//   if (lseek(pFile->fd, pIdx->offset, SEEK_SET) < 0) return -1;
 
-  if (read(pFile->fd, buf, pIdx->len) < 0) return -1;
+//   if (read(pFile->fd, buf, pIdx->len) < 0) return -1;
 
-  // TODO: need to check the correctness
+//   // TODO: need to check the correctness
 
-  return 0;
-}
+//   return 0;
+// }
 
-int tsdbLoadCompCols(SFile *pFile, SCompBlock *pBlock, void *buf) {
-  // assert(pBlock->numOfSubBlocks == 0 || pBlock->numOfSubBlocks == 1);
+// int tsdbLoadCompCols(SFile *pFile, SCompBlock *pBlock, void *buf) {
+//   // assert(pBlock->numOfSubBlocks == 0 || pBlock->numOfSubBlocks == 1);
 
-  if (lseek(pFile->fd, pBlock->offset, SEEK_SET) < 0) return -1;
-  size_t size = sizeof(SCompData) + sizeof(SCompCol) * pBlock->numOfCols;
-  if (read(pFile->fd, buf, size) < 0) return -1;
+//   if (lseek(pFile->fd, pBlock->offset, SEEK_SET) < 0) return -1;
+//   size_t size = sizeof(SCompData) + sizeof(SCompCol) * pBlock->numOfCols;
+//   if (read(pFile->fd, buf, size) < 0) return -1;
 
-  return 0;
-}
+//   return 0;
+// }
 
-int tsdbLoadColData(SFile *pFile, SCompCol *pCol, int64_t blockBaseOffset, void *buf) {
-  if (lseek(pFile->fd, blockBaseOffset + pCol->offset, SEEK_SET) < 0) return -1;
-  if (read(pFile->fd, buf, pCol->len) < 0) return -1;
-  return 0;
-}
+// int tsdbLoadColData(SFile *pFile, SCompCol *pCol, int64_t blockBaseOffset, void *buf) {
+//   if (lseek(pFile->fd, blockBaseOffset + pCol->offset, SEEK_SET) < 0) return -1;
+//   if (read(pFile->fd, buf, pCol->len) < 0) return -1;
+//   return 0;
+// }
 
 static int compFGroupKey(const void *key, const void *fgroup) {
   int         fid = *(int *)key;
@@ -299,7 +312,7 @@ static int tsdbWriteFileHead(SFile *pFile) {
 }
 
 static int tsdbWriteHeadFileIdx(SFile *pFile, int maxTables) {
-  int   size = sizeof(SCompIdx) * maxTables;
+  int   size = sizeof(SCompIdx) * maxTables + sizeof(TSCKSUM);
   void *buf = calloc(1, size);
   if (buf == NULL) return -1;
 
@@ -308,6 +321,8 @@ static int tsdbWriteHeadFileIdx(SFile *pFile, int maxTables) {
     return -1;
   }
 
+  taosCalcChecksumAppend(0, (uint8_t *)buf, size);
+
   if (write(pFile->fd, buf, size) < 0) {
     free(buf);
     return -1;
@@ -319,7 +334,7 @@ static int tsdbWriteHeadFileIdx(SFile *pFile, int maxTables) {
   return 0;
 }
 
-static int tsdbGetFileName(char *dataDir, int fileId, const char *suffix, char *fname) {
+int tsdbGetFileName(char *dataDir, int fileId, const char *suffix, char *fname) {
   if (dataDir == NULL || fname == NULL) return -1;
 
   sprintf(fname, "%s/f%d%s", dataDir, fileId, suffix);

src/tsdb/src/tsdbMeta.c

@@ -225,7 +225,7 @@ STSchema * tsdbGetTableTagSchema(STsdbMeta *pMeta, STable *pTable) {
   }
 }
 
-int32_t tsdbGetTableTagVal(tsdb_repo_t* repo, STableId id, int32_t colId, int16_t* type, int16_t* bytes, char** val) {
+int32_t tsdbGetTableTagVal(TsdbRepoT* repo, STableId id, int32_t colId, int16_t* type, int16_t* bytes, char** val) {
   STsdbMeta* pMeta = tsdbGetMeta(repo);
   STable* pTable = tsdbGetTableByUid(pMeta, id.uid);
   

src/tsdb/src/tsdbRead.c

@@ -120,6 +120,7 @@ typedef struct STsdbQueryHandle {
   SFileGroup*    pFileGroup;
   SFileGroupIter fileIter;
   SCompIdx*      compIndex;
+  SRWHelper rhelper;
 } STsdbQueryHandle;
 
 static void tsdbInitDataBlockLoadInfo(SDataBlockLoadInfo* pBlockLoadInfo) {
@@ -134,7 +135,7 @@ static void tsdbInitCompBlockLoadInfo(SLoadCompBlockInfo* pCompBlockLoadInfo) {
   pCompBlockLoadInfo->fileListIndex = -1;
 }
 
-tsdb_query_handle_t* tsdbQueryTables(tsdb_repo_t* tsdb, STsdbQueryCond* pCond, STableGroupInfo* groupList) {
+TsdbQueryHandleT* tsdbQueryTables(TsdbRepoT* tsdb, STsdbQueryCond* pCond, STableGroupInfo* groupList) {
   // todo 1. filter not exist table
   // todo 2. add the reference count for each table that is involved in query
 
@@ -142,7 +143,8 @@ tsdb_query_handle_t* tsdbQueryTables(tsdb_repo_t* tsdb, STsdbQueryCond* pCond, S
   pQueryHandle->order  = pCond->order;
   pQueryHandle->window = pCond->twindow;
   pQueryHandle->pTsdb  = tsdb;
-  pQueryHandle->compIndex = calloc(10000, sizeof(SCompIdx)),
+  pQueryHandle->compIndex = calloc(10000, sizeof(SCompIdx));
+  tsdbInitReadHelper(&pQueryHandle->rhelper, (STsdbRepo*) tsdb);
 
   pQueryHandle->cur.fid = -1;
 
@@ -196,7 +198,7 @@ tsdb_query_handle_t* tsdbQueryTables(tsdb_repo_t* tsdb, STsdbQueryCond* pCond, S
   tsdbInitDataBlockLoadInfo(&pQueryHandle->dataBlockLoadInfo);
   tsdbInitCompBlockLoadInfo(&pQueryHandle->compBlockLoadInfo);
 
-  return (tsdb_query_handle_t)pQueryHandle;
+  return (TsdbQueryHandleT)pQueryHandle;
 }
 
 static bool hasMoreDataInCache(STsdbQueryHandle* pHandle) {
@@ -288,14 +290,10 @@ static int32_t getFileCompInfo(STsdbQueryHandle* pQueryHandle, int32_t* numOfBlo
   SFileGroup* fileGroup = pQueryHandle->pFileGroup;
   
   assert(fileGroup->files[TSDB_FILE_TYPE_HEAD].fname > 0);
-  if (fileGroup->files[TSDB_FILE_TYPE_HEAD].fd == FD_INITIALIZER) {
-    fileGroup->files[TSDB_FILE_TYPE_HEAD].fd = open(fileGroup->files[TSDB_FILE_TYPE_HEAD].fname, O_RDONLY);
-  } else {
-    assert(FD_VALID(fileGroup->files[TSDB_FILE_TYPE_HEAD].fd));
-  }
+  tsdbSetAndOpenHelperFile(&pQueryHandle->rhelper, fileGroup);
 
   // load all the comp offset value for all tables in this file
-  tsdbLoadCompIdx(fileGroup, pQueryHandle->compIndex, 10000);  // todo set dynamic max tables
+  // tsdbLoadCompIdx(fileGroup, pQueryHandle->compIndex, 10000);  // todo set dynamic max tables
 
   *numOfBlocks = 0;
   size_t numOfTables = taosArrayGetSize(pQueryHandle->pTableCheckInfo);
@@ -303,7 +301,7 @@ static int32_t getFileCompInfo(STsdbQueryHandle* pQueryHandle, int32_t* numOfBlo
   for (int32_t i = 0; i < numOfTables; ++i) {
     STableCheckInfo* pCheckInfo = taosArrayGet(pQueryHandle->pTableCheckInfo, i);
 
-    SCompIdx* compIndex = &pQueryHandle->compIndex[pCheckInfo->tableId.tid];
+    SCompIdx* compIndex = &pQueryHandle->rhelper.pCompIdx[pCheckInfo->tableId.tid];
     if (compIndex->len == 0 || compIndex->numOfSuperBlocks == 0) {  // no data block in this file, try next file
       continue;//no data blocks in the file belongs to pCheckInfo->pTable
     } else {
@@ -317,8 +315,13 @@ static int32_t getFileCompInfo(STsdbQueryHandle* pQueryHandle, int32_t* numOfBlo
         pCheckInfo->compSize = compIndex->len;
       }
       
-      tsdbLoadCompBlocks(fileGroup, compIndex, pCheckInfo->pCompInfo);
-  
+      // tsdbLoadCompBlocks(fileGroup, compIndex, pCheckInfo->pCompInfo);
+      STable* pTable = tsdbGetTableByUid(tsdbGetMeta(pQueryHandle->pTsdb), pCheckInfo->tableId.uid);
+      assert(pTable != NULL);
+      
+      tsdbSetHelperTable(&pQueryHandle->rhelper, pTable, pQueryHandle->pTsdb);
+
+      tsdbLoadCompInfo(&(pQueryHandle->rhelper), (void *)(pCheckInfo->pCompInfo));
       SCompInfo* pCompInfo = pCheckInfo->pCompInfo;
       
       TSKEY s = MIN(pCheckInfo->lastKey, pQueryHandle->window.ekey);
@@ -409,12 +412,12 @@ static bool doLoadFileDataBlock(STsdbQueryHandle* pQueryHandle, SCompBlock* pBlo
 
   tdInitDataCols(pCheckInfo->pDataCols, tsdbGetTableSchema(tsdbGetMeta(pQueryHandle->pTsdb), pCheckInfo->pTableObj));
 
-  SFile* pFile = &pQueryHandle->pFileGroup->files[TSDB_FILE_TYPE_DATA];
-  if (pFile->fd == FD_INITIALIZER) {
-    pFile->fd = open(pFile->fname, O_RDONLY);
-  }
+  // SFile* pFile = &pQueryHandle->pFileGroup->files[TSDB_FILE_TYPE_DATA];
+  // if (pFile->fd == FD_INITIALIZER) {
+  //   pFile->fd = open(pFile->fname, O_RDONLY);
+  // }
 
-  if (tsdbLoadDataBlock(pFile, pBlock, 1, pCheckInfo->pDataCols, data) == 0) {
+  if (tsdbLoadBlockData(&(pQueryHandle->rhelper), pBlock, NULL) == 0) {
     SDataBlockLoadInfo* pBlockLoadInfo = &pQueryHandle->dataBlockLoadInfo;
 
     pBlockLoadInfo->fileGroup = pQueryHandle->pFileGroup;
@@ -427,9 +430,6 @@ static bool doLoadFileDataBlock(STsdbQueryHandle* pQueryHandle, SCompBlock* pBlo
   taosArrayDestroy(sa);
   tfree(data);
 
-  TSKEY* d = (TSKEY*)pCheckInfo->pDataCols->cols[PRIMARYKEY_TIMESTAMP_COL_INDEX].pData;
-  assert(d[0] == pBlock->keyFirst && d[pBlock->numOfPoints - 1] == pBlock->keyLast);
-
   return blockLoaded;
 }
 
@@ -596,9 +596,10 @@ static void filterDataInDataBlock(STsdbQueryHandle* pQueryHandle, STableCheckInf
       SColumnInfoData* pCol = taosArrayGet(pQueryHandle->pColumns, j);
 
       if (pCol->info.colId == colId) {
-        SDataCol* pDataCol = &pCols->cols[i];
-        memmove(pCol->pData, pDataCol->pData + pCol->info.bytes * start,
-                pQueryHandle->realNumOfRows * pCol->info.bytes);
+        // SDataCol* pDataCol = &pCols->cols[i];
+//        pCol->pData = pQueryHandle->rhelper.pDataCols[0]->cols[i].pData + pCol->info.bytes * start;
+         memmove(pCol->pData, pQueryHandle->rhelper.pDataCols[0]->cols[i].pData + pCol->info.bytes * start,
+                 pQueryHandle->realNumOfRows * pCol->info.bytes);
         break;
       }
     }
@@ -909,7 +910,7 @@ static bool doHasDataInBuffer(STsdbQueryHandle* pQueryHandle) {
 }
 
 // handle data in cache situation
-bool tsdbNextDataBlock(tsdb_query_handle_t* pqHandle) {
+bool tsdbNextDataBlock(TsdbQueryHandleT* pqHandle) {
   STsdbQueryHandle* pQueryHandle = (STsdbQueryHandle*) pqHandle;
   
   size_t numOfTables = taosArrayGetSize(pQueryHandle->pTableCheckInfo);
@@ -937,7 +938,6 @@ bool tsdbNextDataBlock(tsdb_query_handle_t* pqHandle) {
 
     return getDataBlocksInFiles(pQueryHandle);
   }
-  
 }
 
 static int tsdbReadRowsFromCache(SSkipListIterator* pIter, TSKEY maxKey, int maxRowsToRead, TSKEY* skey, TSKEY* ekey,
@@ -1009,7 +1009,7 @@ static int tsdbReadRowsFromCache(SSkipListIterator* pIter, TSKEY maxKey, int max
 }
 
 // copy data from cache into data block
-SDataBlockInfo tsdbRetrieveDataBlockInfo(tsdb_query_handle_t* pQueryHandle) {
+SDataBlockInfo tsdbRetrieveDataBlockInfo(TsdbQueryHandleT* pQueryHandle) {
   STsdbQueryHandle* pHandle = (STsdbQueryHandle*)pQueryHandle;
 
   STable* pTable = NULL;
@@ -1067,12 +1067,12 @@ SDataBlockInfo tsdbRetrieveDataBlockInfo(tsdb_query_handle_t* pQueryHandle) {
 }
 
 // return null for data block in cache
-int32_t tsdbRetrieveDataBlockStatisInfo(tsdb_query_handle_t* pQueryHandle, SDataStatis** pBlockStatis) {
+int32_t tsdbRetrieveDataBlockStatisInfo(TsdbQueryHandleT* pQueryHandle, SDataStatis** pBlockStatis) {
   *pBlockStatis = NULL;
   return TSDB_CODE_SUCCESS;
 }
 
-SArray* tsdbRetrieveDataBlock(tsdb_query_handle_t* pQueryHandle, SArray* pIdList) {
+SArray* tsdbRetrieveDataBlock(TsdbQueryHandleT* pQueryHandle, SArray* pIdList) {
   /**
    * In the following two cases, the data has been loaded to SColumnInfoData.
    * 1. data is from cache, 2. data block is not completed qualified to query time range
@@ -1109,17 +1109,17 @@ SArray* tsdbRetrieveDataBlock(tsdb_query_handle_t* pQueryHandle, SArray* pIdList
   }
 }
 
-int32_t tsdbResetQuery(tsdb_query_handle_t* pQueryHandle, STimeWindow* window, tsdbpos_t position, int16_t order) {
+int32_t tsdbResetQuery(TsdbQueryHandleT* pQueryHandle, STimeWindow* window, TsdbPosT position, int16_t order) {
   return 0;
 }
 
-SArray* tsdbRetrieveDataRow(tsdb_query_handle_t* pQueryHandle, SArray* pIdList, SQueryRowCond* pCond) { return NULL; }
+SArray* tsdbRetrieveDataRow(TsdbQueryHandleT* pQueryHandle, SArray* pIdList, SQueryRowCond* pCond) { return NULL; }
 
-tsdb_query_handle_t* tsdbQueryFromTagConds(STsdbQueryCond* pCond, int16_t stableId, const char* pTagFilterStr) {
+TsdbQueryHandleT* tsdbQueryFromTagConds(STsdbQueryCond* pCond, int16_t stableId, const char* pTagFilterStr) {
   return NULL;
 }
 
-SArray* tsdbGetTableList(tsdb_query_handle_t* pQueryHandle) { return NULL; }
+SArray* tsdbGetTableList(TsdbQueryHandleT* pQueryHandle) { return NULL; }
 
 static int32_t getAllTableIdList(STsdbRepo* tsdb, int64_t uid, SArray* list) {
   STable* pTable = tsdbGetTableByUid(tsdbGetMeta(tsdb), uid);
@@ -1432,7 +1432,7 @@ static int32_t doQueryTableList(STable* pSTable, SArray* pRes, tExprNode* pExpr)
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t tsdbQueryByTagsCond(tsdb_repo_t* tsdb, int64_t uid, const char* pTagCond, size_t len, STableGroupInfo* pGroupInfo,
+int32_t tsdbQueryByTagsCond(TsdbRepoT* tsdb, int64_t uid, const char* pTagCond, size_t len, STableGroupInfo* pGroupInfo,
     SColIndex* pColIndex, int32_t numOfCols) {
   
   STable* pSTable = tsdbGetTableByUid(tsdbGetMeta(tsdb), uid);
@@ -1474,7 +1474,7 @@ int32_t tsdbQueryByTagsCond(tsdb_repo_t* tsdb, int64_t uid, const char* pTagCond
   return ret;
 }
 
-int32_t tsdbGetOneTableGroup(tsdb_repo_t* tsdb, int64_t uid, STableGroupInfo* pGroupInfo) {
+int32_t tsdbGetOneTableGroup(TsdbRepoT* tsdb, int64_t uid, STableGroupInfo* pGroupInfo) {
   STable* pTable = tsdbGetTableByUid(tsdbGetMeta(tsdb), uid);
   if (pTable == NULL) {
     return TSDB_CODE_INVALID_TABLE_ID;
@@ -1491,7 +1491,7 @@ int32_t tsdbGetOneTableGroup(tsdb_repo_t* tsdb, int64_t uid, STableGroupInfo* pG
   
   return TSDB_CODE_SUCCESS;
 }
-void tsdbCleanupQueryHandle(tsdb_query_handle_t queryHandle) {
+void tsdbCleanupQueryHandle(TsdbQueryHandleT queryHandle) {
   STsdbQueryHandle* pQueryHandle = (STsdbQueryHandle*)queryHandle;
   if (pQueryHandle == NULL) {
     return;
@@ -1514,14 +1514,15 @@ void tsdbCleanupQueryHandle(tsdb_query_handle_t queryHandle) {
   taosArrayDestroy(pQueryHandle->pTableCheckInfo);
   tfree(pQueryHandle->compIndex);
 
-  size_t cols = taosArrayGetSize(pQueryHandle->pColumns);
-  for (int32_t i = 0; i < cols; ++i) {
-    SColumnInfoData* pColInfo = taosArrayGet(pQueryHandle->pColumns, i);
-    tfree(pColInfo->pData);
-  }
+   size_t cols = taosArrayGetSize(pQueryHandle->pColumns);
+   for (int32_t i = 0; i < cols; ++i) {
+     SColumnInfoData* pColInfo = taosArrayGet(pQueryHandle->pColumns, i);
+      tfree(pColInfo->pData);
+   }
 
   taosArrayDestroy(pQueryHandle->pColumns);
   
   tfree(pQueryHandle->pDataBlockInfo);
+  tsdbDestroyHelper(&pQueryHandle->rhelper);
   tfree(pQueryHandle);
 }

src/tsdb/tests/tsdbTests.cpp

@@ -5,12 +5,84 @@
 #include "dataformat.h"
 #include "tsdbMain.h"
 
-double getCurTime() {
+static double getCurTime() {
   struct timeval tv;
   gettimeofday(&tv, NULL);
   return tv.tv_sec + tv.tv_usec * 1E-6;
 }
 
+typedef struct {
+  TsdbRepoT *pRepo;
+  int          tid;
+  int64_t      uid;
+  int          sversion;
+  TSKEY        startTime;
+  TSKEY        interval;
+  int          totalRows;
+  int          rowsPerSubmit;
+  STSchema *   pSchema;
+} SInsertInfo;
+
+static int insertData(SInsertInfo *pInfo) {
+  SSubmitMsg *pMsg =
+      (SSubmitMsg *)malloc(sizeof(SSubmitMsg) + sizeof(SSubmitBlk) + tdMaxRowBytesFromSchema(pInfo->pSchema) * pInfo->rowsPerSubmit);
+  if (pMsg == NULL) return -1;
+  TSKEY start_time = pInfo->startTime;
+
+  // Loop to write data
+  double stime = getCurTime();
+
+  for (int k = 0; k < pInfo->totalRows/pInfo->rowsPerSubmit; k++) {
+    memset((void *)pMsg, 0, sizeof(SSubmitMsg));
+    SSubmitBlk *pBlock = pMsg->blocks;
+    pBlock->uid = pInfo->uid;
+    pBlock->tid = pInfo->tid;
+    pBlock->sversion = pInfo->sversion;
+    pBlock->len = 0;
+    for (int i = 0; i < pInfo->rowsPerSubmit; i++) {
+      // start_time += 1000;
+      start_time += pInfo->interval;
+      SDataRow row = (SDataRow)(pBlock->data + pBlock->len);
+      tdInitDataRow(row, pInfo->pSchema);
+
+      for (int j = 0; j < schemaNCols(pInfo->pSchema); j++) {
+        if (j == 0) {  // Just for timestamp
+          tdAppendColVal(row, (void *)(&start_time), schemaColAt(pInfo->pSchema, j));
+        } else {  // For int
+          int val = 10;
+          tdAppendColVal(row, (void *)(&val), schemaColAt(pInfo->pSchema, j));
+        }
+      }
+      pBlock->len += dataRowLen(row);
+    }
+    pMsg->length = pMsg->length + sizeof(SSubmitBlk) + pBlock->len;
+    pMsg->numOfBlocks = 1;
+
+    pBlock->len = htonl(pBlock->len);
+    pBlock->numOfRows = htonl(pBlock->numOfRows);
+    pBlock->uid = htobe64(pBlock->uid);
+    pBlock->tid = htonl(pBlock->tid);
+
+    pBlock->sversion = htonl(pBlock->sversion);
+    pBlock->padding = htonl(pBlock->padding);
+
+    pMsg->length = htonl(pMsg->length);
+    pMsg->numOfBlocks = htonl(pMsg->numOfBlocks);
+    pMsg->compressed = htonl(pMsg->numOfBlocks);
+
+    if (tsdbInsertData(pInfo->pRepo, pMsg) < 0) {
+      tfree(pMsg);
+      return -1;
+    }
+  }
+
+  double etime = getCurTime();
+
+  printf("Spent %f seconds to write %d records\n", etime - stime, pInfo->totalRows);
+  tfree(pMsg);
+  return 0;
+}
+
 TEST(TsdbTest, DISABLED_tableEncodeDecode) {
 // TEST(TsdbTest, tableEncodeDecode) {
   STable *pTable = (STable *)malloc(sizeof(STable));
@@ -48,135 +120,132 @@ TEST(TsdbTest, DISABLED_tableEncodeDecode) {
   ASSERT_EQ(memcmp(pTable->schema, tTable->schema, sizeof(STSchema) + sizeof(STColumn) * nCols), 0);
 }
 
-TEST(TsdbTest, DISABLED_createRepo) {
-// TEST(TsdbTest, createRepo) {
-  // STsdbCfg config;
-
-  // // 1. Create a tsdb repository
-  // tsdbSetDefaultCfg(&config);
-  // tsdb_repo_t *pRepo = tsdbCreateRepo("/home/ubuntu/work/ttest/vnode0", &config, NULL);
-  // ASSERT_NE(pRepo, nullptr);
-
-  // // 2. Create a normal table
-  // STableCfg tCfg;
-  // ASSERT_EQ(tsdbInitTableCfg(&tCfg, TSDB_SUPER_TABLE, 987607499877672L, 0), -1);
-  // ASSERT_EQ(tsdbInitTableCfg(&tCfg, TSDB_NORMAL_TABLE, 987607499877672L, 0), 0);
-
-  // int       nCols = 5;
-  // STSchema *schema = tdNewSchema(nCols);
-
-  // for (int i = 0; i < nCols; i++) {
-  //   if (i == 0) {
-  //     tdSchemaAppendCol(schema, TSDB_DATA_TYPE_TIMESTAMP, i, -1);
-  //   } else {
-  //     tdSchemaAppendCol(schema, TSDB_DATA_TYPE_INT, i, -1);
-  //   }
-  // }
-
-  // tsdbTableSetSchema(&tCfg, schema, true);
-
-  // tsdbCreateTable(pRepo, &tCfg);
-
-  // // // 3. Loop to write some simple data
-  // int nRows = 1;
-  // int rowsPerSubmit = 1;
-  // int64_t start_time = 1584081000000;
-
-  // SSubmitMsg *pMsg = (SSubmitMsg *)malloc(sizeof(SSubmitMsg) + sizeof(SSubmitBlk) + tdMaxRowBytesFromSchema(schema) * rowsPerSubmit);
-
-  // double stime = getCurTime();
-
-  // for (int k = 0; k < nRows/rowsPerSubmit; k++) {
-  //   memset((void *)pMsg, 0, sizeof(SSubmitMsg));
-  //   SSubmitBlk *pBlock = pMsg->blocks;
-  //   pBlock->uid = 987607499877672L;
-  //   pBlock->tid = 0;
-  //   pBlock->sversion = 0;
-  //   pBlock->len = 0;
-  //   for (int i = 0; i < rowsPerSubmit; i++) {
-  //     // start_time += 1000;
-  //     start_time += 1000;
-  //     SDataRow row = (SDataRow)(pBlock->data + pBlock->len);
-  //     tdInitDataRow(row, schema);
-
-  //     for (int j = 0; j < schemaNCols(schema); j++) {
-  //       if (j == 0) {  // Just for timestamp
-  //         tdAppendColVal(row, (void *)(&start_time), schemaColAt(schema, j));
-  //       } else {  // For int
-  //         int val = 10;
-  //         tdAppendColVal(row, (void *)(&val), schemaColAt(schema, j));
-  //       }
-  //     }
-  //     pBlock->len += dataRowLen(row);
-  //   }
-  //   pMsg->length = pMsg->length + sizeof(SSubmitBlk) + pBlock->len;
-  //   pMsg->numOfBlocks = 1;
-
-  //   pBlock->len = htonl(pBlock->len);
-  //   pBlock->numOfRows = htonl(pBlock->numOfRows);
-  //   pBlock->uid = htobe64(pBlock->uid);
-  //   pBlock->tid = htonl(pBlock->tid);
-
-  //   pBlock->sversion = htonl(pBlock->sversion);
-  //   pBlock->padding = htonl(pBlock->padding);
-
-  //   pMsg->length = htonl(pMsg->length);
-  //   pMsg->numOfBlocks = htonl(pMsg->numOfBlocks);
-  //   pMsg->compressed = htonl(pMsg->numOfBlocks);
-
-  //   tsdbInsertData(pRepo, pMsg);
-  // }
-
-  // double etime = getCurTime();
-
-  // void *ptr = malloc(150000);
-  // free(ptr);
-
-  // printf("Spent %f seconds to write %d records\n", etime - stime, nRows);
-
-  // tsdbCloseRepo(pRepo);
+// TEST(TsdbTest, DISABLED_createRepo) {
+TEST(TsdbTest, createRepo) {
+  STsdbCfg config;
+  STsdbRepo *repo;
 
-}
+  // 1. Create a tsdb repository
+  tsdbSetDefaultCfg(&config);
+  ASSERT_EQ(tsdbCreateRepo("/home/ubuntu/work/ttest/vnode0", &config, NULL), 0);
 
-// TEST(TsdbTest, DISABLED_openRepo) {
-TEST(TsdbTest, openRepo) {
-  tsdb_repo_t *repo = tsdbOpenRepo("/home/ubuntu/work/build/test/data/vnode/vnode2/tsdb", NULL);
-  ASSERT_NE(repo, nullptr);
+  TsdbRepoT *pRepo = tsdbOpenRepo("/home/ubuntu/work/ttest/vnode0", NULL);
+  ASSERT_NE(pRepo, nullptr);
 
-  STsdbRepo *pRepo = (STsdbRepo *)repo;
+  // 2. Create a normal table
+  STableCfg tCfg;
+  ASSERT_EQ(tsdbInitTableCfg(&tCfg, TSDB_SUPER_TABLE, 987607499877672L, 0), -1);
+  ASSERT_EQ(tsdbInitTableCfg(&tCfg, TSDB_NORMAL_TABLE, 987607499877672L, 0), 0);
 
-  SFileGroup *pGroup = tsdbSearchFGroup(pRepo->tsdbFileH, 1655);
+  int       nCols = 5;
+  STSchema *schema = tdNewSchema(nCols);
 
-  for (int type = TSDB_FILE_TYPE_HEAD; type < TSDB_FILE_TYPE_MAX; type++) {
-    tsdbOpenFile(&pGroup->files[type], O_RDONLY);
+  for (int i = 0; i < nCols; i++) {
+    if (i == 0) {
+      tdSchemaAppendCol(schema, TSDB_DATA_TYPE_TIMESTAMP, i, -1);
+    } else {
+      tdSchemaAppendCol(schema, TSDB_DATA_TYPE_INT, i, -1);
+    }
   }
 
-  SCompIdx *pIdx = (SCompIdx *)calloc(pRepo->config.maxTables, sizeof(SCompIdx));
-  tsdbLoadCompIdx(pGroup, (void *)pIdx, pRepo->config.maxTables);
+  tsdbTableSetSchema(&tCfg, schema, true);
 
-  SCompInfo *pCompInfo = (SCompInfo *)malloc(sizeof(SCompInfo) + pIdx[1].len);
+  tsdbCreateTable(pRepo, &tCfg);
 
-  tsdbLoadCompBlocks(pGroup, &pIdx[1], (void *)pCompInfo);
+  // Insert Some Data
+  SInsertInfo iInfo = {
+    .pRepo = pRepo,
+    .tid = tCfg.tableId.tid,
+    .uid = tCfg.tableId.uid,
+    .sversion = tCfg.sversion,
+    .startTime = 1584081000000,
+    .interval = 1000,
+    .totalRows = 50,
+    .rowsPerSubmit = 1,
+    .pSchema = schema
+  };
 
-  int blockIdx = 0;
-  SCompBlock *pBlock = &(pCompInfo->blocks[blockIdx]);
+  ASSERT_EQ(insertData(&iInfo), 0);
 
-  SCompData *pCompData = (SCompData *)malloc(sizeof(SCompData) + sizeof(SCompCol) * pBlock->numOfCols);
+  // Close the repository
+  tsdbCloseRepo(pRepo);
 
-  tsdbLoadCompCols(&pGroup->files[TSDB_FILE_TYPE_DATA], pBlock, (void *)pCompData);
+  // Open the repository again
+  pRepo = tsdbOpenRepo("/home/ubuntu/work/ttest/vnode0", NULL);
+  repo = (STsdbRepo *)pRepo;
+  ASSERT_NE(pRepo, nullptr);
 
-  STable *pTable = tsdbGetTableByUid(pRepo->tsdbMeta, pCompData->uid);
-  SDataCols *pDataCols = tdNewDataCols(tdMaxRowBytesFromSchema(tsdbGetTableSchema(pRepo->tsdbMeta, pTable)), 5, 10);
-  tdInitDataCols(pDataCols, tsdbGetTableSchema(pRepo->tsdbMeta, pTable));
+  // Insert more data
+  iInfo.startTime = iInfo.startTime + iInfo.interval * iInfo.totalRows;
+  iInfo.totalRows = 10;
+  iInfo.pRepo = pRepo;
+  ASSERT_EQ(insertData(&iInfo), 0);
 
-  tsdbLoadDataBlock(&pGroup->files[TSDB_FILE_TYPE_DATA], pBlock, 1, pDataCols, pCompData);
+  // Close the repository
+  tsdbCloseRepo(pRepo);
 
-  tdResetDataCols(pDataCols);
+  // Open the repository again
+  pRepo = tsdbOpenRepo("/home/ubuntu/work/ttest/vnode0", NULL);
+  repo = (STsdbRepo *)pRepo;
+  ASSERT_NE(pRepo, nullptr);
 
-  tsdbLoadDataBlock(&pGroup->files[TSDB_FILE_TYPE_DATA], pBlock + 1, 1, pDataCols, pCompData);
+  // Read from file
+  SRWHelper rhelper;
+  tsdbInitReadHelper(&rhelper, repo);
 
+  SFileGroup *pFGroup = tsdbSearchFGroup(repo->tsdbFileH, 1833);
+  ASSERT_NE(pFGroup, nullptr);
+  ASSERT_GE(tsdbSetAndOpenHelperFile(&rhelper, pFGroup), 0);
+
+  STable *pTable = tsdbGetTableByUid(repo->tsdbMeta, tCfg.tableId.uid);
+  ASSERT_NE(pTable, nullptr);
+  tsdbSetHelperTable(&rhelper, pTable, repo);
+
+  ASSERT_EQ(tsdbLoadCompInfo(&rhelper, NULL), 0);
+  ASSERT_EQ(tsdbLoadBlockData(&rhelper, blockAtIdx(&rhelper, 0), NULL), 0);
 
   int k = 0;
+}
+
+TEST(TsdbTest, DISABLED_openRepo) {
+// TEST(TsdbTest, openRepo) {
+  // tsdb_repo_t *repo = tsdbOpenRepo("/home/ubuntu/work/build/test/data/vnode/vnode2/tsdb", NULL);
+  // ASSERT_NE(repo, nullptr);
+
+  // STsdbRepo *pRepo = (STsdbRepo *)repo;
+
+  // SFileGroup *pGroup = tsdbSearchFGroup(pRepo->tsdbFileH, 1655);
+
+//   for (int type = TSDB_FILE_TYPE_HEAD; type < TSDB_FILE_TYPE_MAX; type++) {
+//     tsdbOpenFile(&pGroup->files[type], O_RDONLY);
+//   }
+
+//   SCompIdx *pIdx = (SCompIdx *)calloc(pRepo->config.maxTables, sizeof(SCompIdx));
+//   tsdbLoadCompIdx(pGroup, (void *)pIdx, pRepo->config.maxTables);
+
+//   SCompInfo *pCompInfo = (SCompInfo *)malloc(sizeof(SCompInfo) + pIdx[1].len);
+
+  // tsdbLoadCompBlocks(pGroup, &pIdx[1], (void *)pCompInfo);
+
+//   int blockIdx = 0;
+//   SCompBlock *pBlock = &(pCompInfo->blocks[blockIdx]);
+
+//   SCompData *pCompData = (SCompData *)malloc(sizeof(SCompData) + sizeof(SCompCol) * pBlock->numOfCols);
+
+//   tsdbLoadCompCols(&pGroup->files[TSDB_FILE_TYPE_DATA], pBlock, (void *)pCompData);
+
+  // STable *pTable = tsdbGetTableByUid(pRepo->tsdbMeta, pCompData->uid);
+  // SDataCols *pDataCols = tdNewDataCols(tdMaxRowBytesFromSchema(tsdbGetTableSchema(pRepo->tsdbMeta, pTable)), 5, 10);
+  // tdInitDataCols(pDataCols, tsdbGetTableSchema(pRepo->tsdbMeta, pTable));
+
+//   tsdbLoadDataBlock(&pGroup->files[TSDB_FILE_TYPE_DATA], pBlock, 1, pDataCols, pCompData);
+
+  // tdResetDataCols(pDataCols);
+
+  // tsdbLoadDataBlock(&pGroup->files[TSDB_FILE_TYPE_DATA], pBlock + 1, 1, pDataCols, pCompData);
+
+
+//   int k = 0;
 
 }
 

src/util/inc/tutil.h

@@ -176,6 +176,13 @@ uint32_t ip2uint(const char *const ip_addr);
 void taosSetAllocMode(int mode, const char* path, bool autoDump);
 void taosDumpMemoryLeak();
 
+void * tmalloc(size_t size);
+void * tcalloc(size_t nmemb, size_t size);
+size_t tsizeof(void *ptr);
+void   tmemset(void *ptr, int c);
+void * trealloc(void *ptr, size_t size);
+void   tzfree(void *ptr);
+
 #ifdef TAOS_MEM_CHECK
 
 void *  taos_malloc(size_t size, const char *file, uint32_t line);

src/util/src/tutil.c

@@ -618,3 +618,48 @@ char *taosCharsetReplace(char *charsetstr) {
 
   return strdup(charsetstr);
 }
+
+void *tmalloc(size_t size) {
+  if (size <= 0) return NULL;
+
+  void *ret = malloc(size + sizeof(size_t));
+  if (ret == NULL) return NULL;
+
+  *(size_t *)ret = size;
+
+  return (void *)((char *)ret + sizeof(size_t));
+}
+
+void *tcalloc(size_t nmemb, size_t size) {
+  size_t tsize = nmemb * size;
+  void * ret = tmalloc(tsize);
+  if (ret == NULL) return NULL;
+
+  tmemset(ret, 0);
+  return ret;
+}
+
+size_t tsizeof(void *ptr) { return (ptr) ? (*(size_t *)((char *)ptr - sizeof(size_t))) : 0; }
+
+void tmemset(void *ptr, int c) { memset(ptr, c, tsizeof(ptr)); }
+
+void * trealloc(void *ptr, size_t size) {
+  if (ptr == NULL) return tmalloc(size);
+
+  if (size <= tsizeof(ptr)) return ptr;
+
+  void * tptr = (void *)((char *)ptr - sizeof(size_t));
+  size_t tsize = size + sizeof(size_t);
+  tptr = realloc(tptr, tsize);
+  if (tptr == NULL) return NULL;
+
+  *(size_t *)tptr = size;
+
+  return (void *)((char *)tptr + sizeof(size_t));
+}
+
+void tzfree(void *ptr) {
+  if (ptr) {
+    free((void *)((char *)ptr - sizeof(size_t)));
+  }
+}
\ No newline at end of file

src/vnode/src/vnodeRead.c

@@ -56,8 +56,7 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, void *pCont, int32_t cont
   
   qinfo_t pQInfo = NULL;
   if (contLen != 0) {
-    void* tsdb = vnodeGetTsdb(pVnode);
-    pRet->code = qCreateQueryInfo(tsdb, pQueryTableMsg, &pQInfo);
+    pRet->code = qCreateQueryInfo(pVnode->tsdb, pQueryTableMsg, &pQInfo);
   
     SQueryTableRsp *pRsp = (SQueryTableRsp *) rpcMallocCont(sizeof(SQueryTableRsp));
     pRsp->qhandle = htobe64((uint64_t) (pQInfo));

tests/test/c/CMakeLists.txt

@@ -14,4 +14,7 @@ IF ((TD_LINUX_64) OR (TD_LINUX_32 AND TD_ARM))
 
   add_executable(importOneRow importOneRow.c)
   target_link_libraries(importOneRow taos_static pthread)
+
+  add_executable(importPerTabe importPerTabe.c)
+  target_link_libraries(importPerTabe taos_static pthread)
 ENDIF()