Merge pull request #792 from taosdata/feature/mergeimport

Feature/mergeimport

src/system/detail/src/vnodeCache.c

@@ -372,13 +372,60 @@ void vnodeCancelCommit(SVnodeObj *pVnode) {
   taosTmrReset(vnodeProcessCommitTimer, pVnode->cfg.commitTime * 1000, pVnode, vnodeTmrCtrl, &pVnode->commitTimer);
 }
 
+/* The vnode cache lock should be hold before calling this interface
+ */
+SCacheBlock *vnodeGetFreeCacheBlock(SVnodeObj *pVnode) {
+  SCachePool  *pPool = (SCachePool *)(pVnode->pCachePool);
+  SVnodeCfg   *pCfg = &(pVnode->cfg);
+  SCacheBlock *pCacheBlock = NULL;
+  int skipped = 0;
+
+  while (1) {
+    pCacheBlock = (SCacheBlock *)(pPool->pMem[((int64_t)pPool->freeSlot)]);
+    if (pCacheBlock->blockId == 0) break;
+
+    if (pCacheBlock->notFree) {
+      pPool->freeSlot++;
+      pPool->freeSlot = pPool->freeSlot % pCfg->cacheNumOfBlocks.totalBlocks;
+      skipped++;
+      if (skipped > pPool->threshold) {
+        vnodeCreateCommitThread(pVnode);
+        pthread_mutex_unlock(&pPool->vmutex);
+        dError("vid:%d committing process is too slow, notFreeSlots:%d....", pVnode->vnode, pPool->notFreeSlots);
+        return NULL;
+      }
+    } else {
+      SMeterObj * pRelObj = pCacheBlock->pMeterObj;
+      SCacheInfo *pRelInfo = (SCacheInfo *)pRelObj->pCache;
+      int firstSlot = (pRelInfo->currentSlot - pRelInfo->numOfBlocks + 1 + pRelInfo->maxBlocks) % pRelInfo->maxBlocks;
+      pCacheBlock = pRelInfo->cacheBlocks[firstSlot];
+      if (pCacheBlock) {
+        pPool->freeSlot = pCacheBlock->index;
+        vnodeFreeCacheBlock(pCacheBlock);
+        break;
+      } else {
+        pPool->freeSlot = (pPool->freeSlot + 1) % pCfg->cacheNumOfBlocks.totalBlocks;
+        skipped++;
+      }
+    }
+  }
+
+  pCacheBlock = (SCacheBlock *)(pPool->pMem[pPool->freeSlot]);
+  pCacheBlock->index = pPool->freeSlot;
+  pCacheBlock->notFree = 1;
+  pPool->freeSlot = (pPool->freeSlot + 1) % pCfg->cacheNumOfBlocks.totalBlocks;
+  pPool->notFreeSlots++;
+
+  return pCacheBlock;
+}
+
 int vnodeAllocateCacheBlock(SMeterObj *pObj) {
   int          index;
   SCachePool * pPool;
   SCacheBlock *pCacheBlock;
   SCacheInfo * pInfo;
   SVnodeObj *  pVnode;
-  int          skipped = 0, commit = 0;
+  int          commit = 0;
 
   pVnode = vnodeList + pObj->vnode;
   pPool = (SCachePool *)pVnode->pCachePool;
@@ -406,45 +453,10 @@ int vnodeAllocateCacheBlock(SMeterObj *pObj) {
     return -1;
   }
 
-  while (1) {
-    pCacheBlock = (SCacheBlock *)(pPool->pMem[((int64_t)pPool->freeSlot)]);
-    if (pCacheBlock->blockId == 0) break;
-
-    if (pCacheBlock->notFree) {
-      pPool->freeSlot++;
-      pPool->freeSlot = pPool->freeSlot % pCfg->cacheNumOfBlocks.totalBlocks;
-      skipped++;
-      if (skipped > pPool->threshold) {
-        vnodeCreateCommitThread(pVnode);
-        pthread_mutex_unlock(&pPool->vmutex);
-        dError("vid:%d sid:%d id:%s, committing process is too slow, notFreeSlots:%d....",
-               pObj->vnode, pObj->sid, pObj->meterId, pPool->notFreeSlots);
-        return -1;
-      }
-    } else {
-      SMeterObj  *pRelObj = pCacheBlock->pMeterObj;
-      SCacheInfo *pRelInfo = (SCacheInfo *)pRelObj->pCache;
-      int firstSlot = (pRelInfo->currentSlot - pRelInfo->numOfBlocks + 1 + pRelInfo->maxBlocks) % pRelInfo->maxBlocks;
-      pCacheBlock = pRelInfo->cacheBlocks[firstSlot];
-      if (pCacheBlock) {
-        pPool->freeSlot = pCacheBlock->index;
-        vnodeFreeCacheBlock(pCacheBlock);
-        break;
-      } else {
-        pPool->freeSlot = (pPool->freeSlot + 1) % pCfg->cacheNumOfBlocks.totalBlocks;
-        skipped++;
-      }
-    }
-  }
-
-  index = pPool->freeSlot;
-  pPool->freeSlot++;
-  pPool->freeSlot = pPool->freeSlot % pCfg->cacheNumOfBlocks.totalBlocks;
-  pPool->notFreeSlots++;
+  if ((pCacheBlock = vnodeGetFreeCacheBlock(pVnode)) == NULL) return -1;
 
+  index = pCacheBlock->index;
   pCacheBlock->pMeterObj = pObj;
-  pCacheBlock->notFree = 1;
-  pCacheBlock->index = index;
 
   pCacheBlock->offset[0] = ((char *)(pCacheBlock)) + sizeof(SCacheBlock) + pObj->numOfColumns * sizeof(char *);
   for (int col = 1; col < pObj->numOfColumns; ++col)

src/system/detail/src/vnodeFile.c

@@ -95,8 +95,8 @@ void vnodeGetDnameFromLname(char *lhead, char *ldata, char *llast, char *dhead,
 }
 
 void vnodeGetHeadTname(char *nHeadName, char *nLastName, int vnode, int fileId) {
-  sprintf(nHeadName, "%s/vnode%d/db/v%df%d.t", tsDirectory, vnode, vnode, fileId);
-  sprintf(nLastName, "%s/vnode%d/db/v%df%d.l", tsDirectory, vnode, vnode, fileId);
+  if (nHeadName != NULL) sprintf(nHeadName, "%s/vnode%d/db/v%df%d.t", tsDirectory, vnode, vnode, fileId);
+  if (nLastName != NULL) sprintf(nLastName, "%s/vnode%d/db/v%df%d.l", tsDirectory, vnode, vnode, fileId);
 }
 
 void vnodeCreateDataDirIfNeeded(int vnode, char *path) {
@@ -180,29 +180,24 @@ int vnodeCreateEmptyCompFile(int vnode, int fileId) {
   return 0;
 }
 
-int vnodeOpenCommitFiles(SVnodeObj *pVnode, int noTempLast) {
-  char        name[TSDB_FILENAME_LEN];
-  char        dHeadName[TSDB_FILENAME_LEN] = "\0";
-  char        dLastName[TSDB_FILENAME_LEN] = "\0";
-  int         len = 0;
-  struct stat filestat;
-  int         vnode = pVnode->vnode;
-  int         fileId, numOfFiles, filesAdded = 0;
-  SVnodeCfg * pCfg = &pVnode->cfg;
+int vnodeCreateNeccessaryFiles(SVnodeObj *pVnode) {
+  int        numOfFiles = 0, fileId, filesAdded = 0;
+  int        vnode = pVnode->vnode;
+  SVnodeCfg *pCfg = &(pVnode->cfg);
 
   if (pVnode->lastKeyOnFile == 0) {
     if (pCfg->daysPerFile == 0) pCfg->daysPerFile = 10;
     pVnode->fileId = pVnode->firstKey / tsMsPerDay[pVnode->cfg.precision] / pCfg->daysPerFile;
     pVnode->lastKeyOnFile = (int64_t)(pVnode->fileId + 1) * pCfg->daysPerFile * tsMsPerDay[pVnode->cfg.precision] - 1;
     pVnode->numOfFiles = 1;
-    vnodeCreateEmptyCompFile(vnode, pVnode->fileId);
+    if (vnodeCreateEmptyCompFile(vnode, pVnode->fileId) < 0) return -1;
   }
 
   numOfFiles = (pVnode->lastKeyOnFile - pVnode->commitFirstKey) / tsMsPerDay[pVnode->cfg.precision] / pCfg->daysPerFile;
   if (pVnode->commitFirstKey > pVnode->lastKeyOnFile) numOfFiles = -1;
 
-  dTrace("vid:%d, commitFirstKey:%ld lastKeyOnFile:%ld numOfFiles:%d fileId:%d vnodeNumOfFiles:%d",
-      vnode, pVnode->commitFirstKey, pVnode->lastKeyOnFile, numOfFiles, pVnode->fileId, pVnode->numOfFiles);
+  dTrace("vid:%d, commitFirstKey:%ld lastKeyOnFile:%ld numOfFiles:%d fileId:%d vnodeNumOfFiles:%d", pVnode->vnode,
+         pVnode->commitFirstKey, pVnode->lastKeyOnFile, numOfFiles, pVnode->fileId, pVnode->numOfFiles);
 
   if (numOfFiles >= pVnode->numOfFiles) {
     // create empty header files backward
@@ -214,7 +209,7 @@ int vnodeOpenCommitFiles(SVnodeObj *pVnode, int noTempLast) {
 #ifdef CLUSTER	  
 	    return vnodeRecoverFromPeer(pVnode, fileId);
 #else
-        return -1;
+      return -1;
 #endif				
     }
   } else if (numOfFiles < 0) {
@@ -224,7 +219,7 @@ int vnodeOpenCommitFiles(SVnodeObj *pVnode, int noTempLast) {
 #ifdef CLUSTER	  
 	    return vnodeRecoverFromPeer(pVnode, pVnode->fileId);
 #else
-        return -1;
+      return -1;
 #endif
     pVnode->lastKeyOnFile += (int64_t)tsMsPerDay[pVnode->cfg.precision] * pCfg->daysPerFile;
     filesAdded = 1;
@@ -238,6 +233,24 @@ int vnodeOpenCommitFiles(SVnodeObj *pVnode, int noTempLast) {
   pVnode->commitFileId = fileId;
   pVnode->numOfFiles = pVnode->numOfFiles + filesAdded;
 
+  return 0;
+}
+
+
+int vnodeOpenCommitFiles(SVnodeObj *pVnode, int noTempLast) {
+  char        name[TSDB_FILENAME_LEN];
+  char        dHeadName[TSDB_FILENAME_LEN] = "\0";
+  char        dLastName[TSDB_FILENAME_LEN] = "\0";
+  int         len = 0;
+  struct stat filestat;
+  int         vnode = pVnode->vnode;
+  int         fileId, numOfFiles, filesAdded = 0;
+  SVnodeCfg * pCfg = &pVnode->cfg;
+
+  if (vnodeCreateNeccessaryFiles(pVnode) < 0) return -1;
+
+  fileId = pVnode->commitFileId;
+
   dTrace("vid:%d, commit fileId:%d, commitLastKey:%ld, vnodeLastKey:%ld, lastKeyOnFile:%ld numOfFiles:%d",
       vnode, fileId, pVnode->commitLastKey, pVnode->lastKey, pVnode->lastKeyOnFile, pVnode->numOfFiles);
 

src/system/detail/src/vnodeMeter.c

@@ -584,12 +584,12 @@ int vnodeInsertPoints(SMeterObj *pObj, char *cont, int contLen, char source, voi
   if (pVnode->cfg.commitLog && source != TSDB_DATA_SOURCE_LOG) {
     if (pVnode->logFd < 0) return TSDB_CODE_INVALID_COMMIT_LOG;
     code = vnodeWriteToCommitLog(pObj, TSDB_ACTION_INSERT, cont, contLen, sversion);
-    if (code != 0) return code;
+    if (code != TSDB_CODE_SUCCESS) return code;
   }
 
   if (source == TSDB_DATA_SOURCE_SHELL && pVnode->cfg.replications > 1) {
     code = vnodeForwardToPeer(pObj, cont, contLen, TSDB_ACTION_INSERT, sversion);
-    if (code != 0) return code;
+    if (code != TSDB_CODE_SUCCESS) return code;
   }
 
   if (pObj->sversion < sversion) {
@@ -601,11 +601,11 @@ int vnodeInsertPoints(SMeterObj *pObj, char *cont, int contLen, char source, voi
   }
 
   pData = pSubmit->payLoad;
-  code = TSDB_CODE_SUCCESS;
 
   TSKEY firstKey = *((TSKEY *)pData);
   TSKEY lastKey = *((TSKEY *)(pData + pObj->bytesPerPoint * (numOfPoints - 1)));
   int cfid = now/pVnode->cfg.daysPerFile/tsMsPerDay[pVnode->cfg.precision];
+  
   TSKEY minAllowedKey = (cfid - pVnode->maxFiles + 1)*pVnode->cfg.daysPerFile*tsMsPerDay[pVnode->cfg.precision];
   TSKEY maxAllowedKey = (cfid + 2)*pVnode->cfg.daysPerFile*tsMsPerDay[pVnode->cfg.precision] - 2;
   if (firstKey < minAllowedKey || firstKey > maxAllowedKey || lastKey < minAllowedKey || lastKey > maxAllowedKey) {
@@ -619,7 +619,7 @@ int vnodeInsertPoints(SMeterObj *pObj, char *cont, int contLen, char source, voi
   }
   
   for (i = 0; i < numOfPoints; ++i) { // meter will be dropped, abort current insertion
-    if (pObj->state >= TSDB_METER_STATE_DELETING) {
+    if (vnodeIsMeterState(pObj, TSDB_METER_STATE_DELETING)) {
       dWarn("vid:%d sid:%d id:%s, meter is dropped, abort insert, state:%d", pObj->vnode, pObj->sid, pObj->meterId,
             pObj->state);
 
@@ -648,6 +648,7 @@ int vnodeInsertPoints(SMeterObj *pObj, char *cont, int contLen, char source, voi
     pData += pObj->bytesPerPoint;
     points++;
   }
+  
   atomic_fetch_add_64(&(pVnode->vnodeStatistic.pointsWritten), points * (pObj->numOfColumns - 1));
   atomic_fetch_add_64(&(pVnode->vnodeStatistic.totalStorage), points * pObj->bytesPerPoint);
 
@@ -660,6 +661,7 @@ int vnodeInsertPoints(SMeterObj *pObj, char *cont, int contLen, char source, voi
   pVnode->version++;
 
   pthread_mutex_unlock(&(pVnode->vmutex));
+  
   vnodeClearMeterState(pObj, TSDB_METER_STATE_INSERT);
 
 _over:

src/system/detail/src/vnodeShell.c

@@ -39,10 +39,21 @@ SShellObj **shellList = NULL;
 int vnodeProcessRetrieveRequest(char *pMsg, int msgLen, SShellObj *pObj);
 int vnodeProcessQueryRequest(char *pMsg, int msgLen, SShellObj *pObj);
 int vnodeProcessShellSubmitRequest(char *pMsg, int msgLen, SShellObj *pObj);
+static void vnodeProcessBatchSubmitTimer(void *param, void *tmrId);
 
 int vnodeSelectReqNum = 0;
 int vnodeInsertReqNum = 0;
 
+typedef struct {
+  int32_t import;
+  int32_t vnode;
+  int32_t numOfSid;
+  int32_t ssid;   // Start sid
+  SShellObj *pObj;
+  int64_t offset; // offset relative the blks
+  char    blks[];
+} SBatchSubmitInfo;
+
 void *vnodeProcessMsgFromShell(char *msg, void *ahandle, void *thandle) {
   int        sid, vnode;
   SShellObj *pObj = (SShellObj *)ahandle;
@@ -249,6 +260,7 @@ int vnodeSendShellSubmitRspMsg(SShellObj *pObj, int code, int numOfPoints) {
   char *pMsg, *pStart;
   int   msgLen;
 
+  dTrace("code:%d numOfTotalPoints:%d", code, numOfPoints);
   pStart = taosBuildRspMsgWithSize(pObj->thandle, TSDB_MSG_TYPE_SUBMIT_RSP, 128);
   if (pStart == NULL) return -1;
   pMsg = pStart;
@@ -280,6 +292,7 @@ int vnodeProcessQueryRequest(char *pMsg, int msgLen, SShellObj *pObj) {
   }
 
   if (pQueryMsg->numOfSids <= 0) {
+    dError("Invalid number of meters to query, numOfSids:%d", pQueryMsg->numOfSids);
     code = TSDB_CODE_INVALID_QUERY_MSG;
     goto _query_over;
   }
@@ -485,10 +498,83 @@ int vnodeProcessRetrieveRequest(char *pMsg, int msgLen, SShellObj *pObj) {
   return msgLen;
 }
 
+static int vnodeCheckSubmitBlockContext(SShellSubmitBlock *pBlocks, SVnodeObj *pVnode) {
+  int32_t  sid = htonl(pBlocks->sid);
+  uint64_t uid = htobe64(pBlocks->uid);
+
+  if (sid >= pVnode->cfg.maxSessions || sid <= 0) {
+    dError("sid:%d is out of range", sid);
+    return TSDB_CODE_INVALID_TABLE_ID;
+  }
+
+  SMeterObj *pMeterObj = pVnode->meterList[sid];
+  if (pMeterObj == NULL) {
+    dError("vid:%d sid:%d, not active table", pVnode->vnode, sid);
+    vnodeSendMeterCfgMsg(pVnode->vnode, sid);
+    return TSDB_CODE_NOT_ACTIVE_TABLE;
+  }
+
+  if (pMeterObj->uid != uid) {
+    dError("vid:%d sid:%d, meterId:%s, uid:%lld, uid in msg:%lld, uid mismatch", pVnode->vnode, sid, pMeterObj->meterId,
+           pMeterObj->uid, uid);
+    return TSDB_CODE_INVALID_SUBMIT_MSG;
+  }
+
+  return TSDB_CODE_SUCCESS;
+}
+
+static int vnodeDoSubmitJob(SVnodeObj *pVnode, int import, int32_t *ssid, int32_t esid, SShellSubmitBlock **ppBlocks,
+                            TSKEY now, SShellObj *pObj) {
+  SShellSubmitBlock *pBlocks = *ppBlocks;
+  int code = TSDB_CODE_SUCCESS;
+  int32_t numOfPoints = 0;
+  int32_t i = 0;
+
+  for (i = *ssid; i < esid; i++) {
+    numOfPoints = 0;
+
+    code = vnodeCheckSubmitBlockContext(pBlocks, pVnode);
+    if (code != TSDB_CODE_SUCCESS) break;
+
+    SMeterObj *pMeterObj = (SMeterObj *)(pVnode->meterList[htonl(pBlocks->sid)]);
+
+    // dont include sid, vid
+    int32_t subMsgLen = sizeof(pBlocks->numOfRows) + htons(pBlocks->numOfRows) * pMeterObj->bytesPerPoint;
+    int32_t sversion = htonl(pBlocks->sversion);
+
+    if (import) {
+      code = vnodeImportPoints(pMeterObj, (char *)&(pBlocks->numOfRows), subMsgLen, TSDB_DATA_SOURCE_SHELL, pObj,
+                               sversion, &numOfPoints, now);
+      pObj->numOfTotalPoints += numOfPoints;
+
+      // records for one table should be consecutive located in the payload buffer, which is guaranteed by client
+      if (code == TSDB_CODE_SUCCESS) {
+        pObj->count--;
+      }
+    } else {
+      code = vnodeInsertPoints(pMeterObj, (char *)&(pBlocks->numOfRows), subMsgLen, TSDB_DATA_SOURCE_SHELL, NULL,
+                               sversion, &numOfPoints, now);
+      pObj->numOfTotalPoints += numOfPoints;
+    }
+
+    if (code != TSDB_CODE_SUCCESS) break;
+
+    pBlocks = (SShellSubmitBlock *)((char *)pBlocks + sizeof(SShellSubmitBlock) +
+                                    htons(pBlocks->numOfRows) * pMeterObj->bytesPerPoint);
+  }
+
+  *ssid = i;
+  *ppBlocks = pBlocks;
+
+  return code;
+}
+
 int vnodeProcessShellSubmitRequest(char *pMsg, int msgLen, SShellObj *pObj) {
   int              code = 0, ret = 0;
+  int32_t          i = 0;
   SShellSubmitMsg  shellSubmit = *(SShellSubmitMsg *)pMsg;
   SShellSubmitMsg *pSubmit = &shellSubmit;
+  SShellSubmitBlock *pBlocks = NULL;
 
   pSubmit->vnode = htons(pSubmit->vnode);
   pSubmit->numOfSid = htonl(pSubmit->numOfSid);
@@ -526,67 +612,69 @@ int vnodeProcessShellSubmitRequest(char *pMsg, int msgLen, SShellObj *pObj) {
 
   pObj->count = pSubmit->numOfSid;  // for import
   pObj->code = 0;                   // for import
-  pObj->numOfTotalPoints = 0;       // for import
-  SShellSubmitBlock *pBlocks = (SShellSubmitBlock *)(pMsg + sizeof(SShellSubmitMsg));
+  pObj->numOfTotalPoints = 0;
 
-  int32_t numOfPoints = 0;
-  int32_t numOfTotalPoints = 0;
-  // We take current time here to avoid it in the for loop.
   TSKEY   now = taosGetTimestamp(pVnode->cfg.precision);
 
-  for (int32_t i = 0; i < pSubmit->numOfSid; ++i) {
-    numOfPoints = 0;
-
-    pBlocks->sid = htonl(pBlocks->sid);
-    pBlocks->uid = htobe64(pBlocks->uid);
+  pBlocks = (SShellSubmitBlock *)(pMsg + sizeof(SShellSubmitMsg));
+  i = 0;
+  code = vnodeDoSubmitJob(pVnode, pSubmit->import, &i, pSubmit->numOfSid, &pBlocks, now, pObj);
 
-    if (pBlocks->sid >= pVnode->cfg.maxSessions || pBlocks->sid <= 0) {
-      dTrace("sid:%d is out of range", pBlocks->sid);
-      code = TSDB_CODE_INVALID_TABLE_ID;
-      goto _submit_over;
+_submit_over:
+  ret = 0;
+  if (pSubmit->import) {  // Import case
+    if (code == TSDB_CODE_ACTION_IN_PROGRESS) {
+
+      SBatchSubmitInfo *pSubmitInfo =
+          (SBatchSubmitInfo *)calloc(1, sizeof(SBatchSubmitInfo) + msgLen - sizeof(SShellSubmitMsg));
+      if (pSubmitInfo == NULL) {
+        code = TSDB_CODE_SERV_OUT_OF_MEMORY;
+        ret = vnodeSendShellSubmitRspMsg(pObj, code, pObj->numOfTotalPoints);
+      } else { // Start a timer to process the next part of request
+        pSubmitInfo->import = 1;
+        pSubmitInfo->vnode = pSubmit->vnode;
+        pSubmitInfo->numOfSid = pSubmit->numOfSid;
+        pSubmitInfo->ssid = i;   // start from this position, not the initial position
+        pSubmitInfo->pObj = pObj;
+        pSubmitInfo->offset = ((char *)pBlocks) - (pMsg + sizeof(SShellSubmitMsg));
+        assert(pSubmitInfo->offset >= 0);
+        memcpy((void *)(pSubmitInfo->blks), (void *)(pMsg + sizeof(SShellSubmitMsg)), msgLen - sizeof(SShellSubmitMsg));
+        taosTmrStart(vnodeProcessBatchSubmitTimer, 10, (void *)pSubmitInfo, vnodeTmrCtrl);
+      }
+    } else {
+      if (code == TSDB_CODE_SUCCESS) assert(pObj->count == 0);
+      ret = vnodeSendShellSubmitRspMsg(pObj, code, pObj->numOfTotalPoints);
     }
+  } else {  // Insert case
+    ret = vnodeSendShellSubmitRspMsg(pObj, code, pObj->numOfTotalPoints);
+  }
 
-    int vnode = pSubmit->vnode;
-    int sid = pBlocks->sid;
-
-    SMeterObj *pMeterObj = vnodeList[vnode].meterList[sid];
-    if (pMeterObj == NULL) {
-      dError("vid:%d sid:%d, no active table", vnode, sid);
-      vnodeSendMeterCfgMsg(vnode, sid);
-      code = TSDB_CODE_NOT_ACTIVE_TABLE;
-      goto _submit_over;
-    }
+  atomic_fetch_add_32(&vnodeInsertReqNum, 1);
+  return ret;
+}
 
-    if (pMeterObj->uid != pBlocks->uid) {
-      dError("vid:%d sid:%d, meterId:%s, uid:%lld, uid in msg:%lld, uid mismatch", vnode, sid, pMeterObj->meterId,
-             pMeterObj->uid, pBlocks->uid);
-      code = TSDB_CODE_INVALID_SUBMIT_MSG;
-      goto _submit_over;
-    }
+static void vnodeProcessBatchSubmitTimer(void *param, void *tmrId) {
+  SBatchSubmitInfo *pSubmitInfo = (SBatchSubmitInfo *)param;
+  assert(pSubmitInfo != NULL && pSubmitInfo->import);
 
-    // dont include sid, vid
-    int subMsgLen = sizeof(pBlocks->numOfRows) + htons(pBlocks->numOfRows) * pMeterObj->bytesPerPoint;
-    int sversion = htonl(pBlocks->sversion);
+  int32_t i = 0;
+  int32_t code = TSDB_CODE_SUCCESS;
 
-    if (pSubmit->import) {
-      code = vnodeImportPoints(pMeterObj, (char *) &(pBlocks->numOfRows), subMsgLen, TSDB_DATA_SOURCE_SHELL, pObj,
-                               sversion, &numOfPoints, now);
-    } else {
-      code = vnodeInsertPoints(pMeterObj, (char *) &(pBlocks->numOfRows), subMsgLen, TSDB_DATA_SOURCE_SHELL, NULL,
-                               sversion, &numOfPoints, now);
-    }
+  SShellObj *        pShell = pSubmitInfo->pObj;
+  SVnodeObj *        pVnode = &vnodeList[pSubmitInfo->vnode];
+  SShellSubmitBlock *pBlocks = (SShellSubmitBlock *)(pSubmitInfo->blks + pSubmitInfo->offset);
+  TSKEY   now = taosGetTimestamp(pVnode->cfg.precision);
+  i = pSubmitInfo->ssid;
 
-    if (code != TSDB_CODE_SUCCESS) {break;}
+  code = vnodeDoSubmitJob(pVnode, pSubmitInfo->import, &i, pSubmitInfo->numOfSid, &pBlocks, now, pShell);
 
-    numOfTotalPoints += numOfPoints;
-    pBlocks = (SShellSubmitBlock *)((char *)pBlocks + sizeof(SShellSubmitBlock) +
-                                    htons(pBlocks->numOfRows) * pMeterObj->bytesPerPoint);
+  if (code == TSDB_CODE_ACTION_IN_PROGRESS) {
+    pSubmitInfo->ssid = i;
+    pSubmitInfo->offset = ((char *)pBlocks) - pSubmitInfo->blks;
+    taosTmrStart(vnodeProcessBatchSubmitTimer, 10, (void *)pSubmitInfo, vnodeTmrCtrl);
+  } else {
+    if (code == TSDB_CODE_SUCCESS) assert(pShell->count == 0);
+    tfree(param);
+    vnodeSendShellSubmitRspMsg(pShell, code, pShell->numOfTotalPoints);
   }
-
-_submit_over:
-  // for import, send the submit response only when return code is not zero
-  if (pSubmit->import == 0 || code != 0) ret = vnodeSendShellSubmitRspMsg(pObj, code, numOfTotalPoints);
-
-  atomic_fetch_add_32(&vnodeInsertReqNum, 1);
-  return ret;
 }