more work

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

@@ -181,21 +181,15 @@ static int32_t tsdbCommitDelStart(SCommitter *pCommitter) {
   // load old
   if (pDelFileR) {
     code = tsdbDelFReaderOpen(&pCommitter->pDelFReader, pDelFileR, pTsdb, NULL);
-    if (code) {
+    if (code) goto _err;
-      goto _err;
-    }
 
-    code = tsdbReadDelIdx(pCommitter->pDelFReader, &pCommitter->oDelIdxMap, &pCommitter->pBuf1);
+    code = tsdbReadDelIdx(pCommitter->pDelFReader, &pCommitter->oDelIdxMap, NULL);
-    if (code) {
+    if (code) goto _err;
-      goto _err;
-    }
   }
 
   // prepare new
   code = tsdbDelFWriterOpen(&pCommitter->pDelFWriter, pDelFileW, pTsdb);
-  if (code) {
+  if (code) goto _err;
-    goto _err;
-  }
 
 _exit:
   tsdbDebug("vgId:%d commit del start", TD_VID(pTsdb->pVnode));
@@ -206,64 +200,89 @@ _err:
   return code;
 }
 
-static int32_t tsdbCommitTableDel(SCommitter *pCommitter);
-
 static int32_t tsdbCommitDelImpl(SCommitter *pCommitter) {
-  int32_t      code = 0;
+  int32_t    code = 0;
-  STsdb       *pTsdb = pCommitter->pTsdb;
+  STsdb     *pTsdb = pCommitter->pTsdb;
-  SMemTable   *pMemTable = pTsdb->imem;
+  SMemTable *pMemTable = pTsdb->imem;
-  int32_t      c;
+  int32_t    iDelIdx = 0;
-  int32_t      iTbData = 0;
+  int32_t    nDelIdx = pCommitter->oDelIdxMap.nItem;
-  int32_t      nTbData = taosArrayGetSize(pMemTable->aTbData);
+  int32_t    iTbData = 0;
-  int32_t      iDelIdxItem = 0;
+  int32_t    nTbData = taosArrayGetSize(pMemTable->aTbData);
-  int32_t      nDelIdxItem = pCommitter->delIdxOld.offset.nOffset;
+  STbData   *pTbData;
-  STbData     *pTbData = NULL;
+  SDelIdx   *pDelIdx;
-  SDelIdxItem *pDelIdxItem = NULL;
+  SDelIdx    delIdx;
-  SDelIdxItem  item;
+  int32_t    c;
 
-  while (iTbData < nTbData || iDelIdxItem < nDelIdxItem) {
+  ASSERT(nTbData > 0);
-    pTbData = NULL;
-    pDelIdxItem = NULL;
-    if (iTbData < nTbData) {
-      pTbData = (STbData *)taosArrayGetP(pMemTable->aTbData, iTbData);
-    }
-    if (iDelIdxItem < nDelIdxItem) {
-      tDelIdxGetItemByIdx(&pCommitter->delIdxOld, &item, iDelIdxItem);
-      pDelIdxItem = &item;
-    }
 
-    if (pTbData && pDelIdxItem) {
+  pTbData = (STbData *)taosArrayGetP(pMemTable->aTbData, iTbData);
-      c = tTABLEIDCmprFn(pTbData, pDelIdxItem);
+  if (iDelIdx < nDelIdx) {
+    code = tMapDataGetItemByIdx(&pCommitter->oDelIdxMap, iDelIdx, &delIdx, tGetDelIdx);
+    if (code) goto _err;
+    pDelIdx = &delIdx;
+  } else {
+    pDelIdx = NULL;
+  }
+
+  while (true) {
+    if (pTbData == NULL && pDelIdx == NULL) break;
+
+    if (pTbData && pDelIdx) {
+      c = tTABLEIDCmprFn(pTbData, pDelIdx);
       if (c == 0) {
-        iTbData++;
+        goto _commit_mem_and_disk_del;
-        iDelIdxItem++;
       } else if (c < 0) {
-        iTbData++;
+        goto _commit_mem_del;
-        pDelIdxItem = NULL;
       } else {
-        iDelIdxItem++;
+        goto _commit_disk_del;
-        pTbData = NULL;
       }
     } else {
-      if (pTbData) {
+      if (pTbData) goto _commit_mem_del;
-        iTbData++;
+      if (pDelIdx) goto _commit_disk_del;
-      }
-      if (pDelIdxItem) {
-        iDelIdxItem++;
-      }
     }
 
-    if (pTbData && pTbData->pHead == NULL) {
+  _commit_mem_del:
+    code = tsdbCommitTableDel(pCommitter, pTbData, NULL);
+    if (code) goto _err;
+    iTbData++;
+    if (iTbData < nTbData) {
+      pTbData = (STbData *)taosArrayGetP(pMemTable->aTbData, iTbData);
+    } else {
       pTbData = NULL;
     }
+    continue;
 
-    if (pTbData == NULL && pDelIdxItem == NULL) continue;
+  _commit_disk_del:
-
+    code = tsdbCommitTableDel(pCommitter, NULL, pDelIdx);
-    // do merge
-    pCommitter->pTbData = pTbData;
-    pCommitter->pDelIdxItem = pDelIdxItem;
-    code = tsdbCommitTableDel(pCommitter);
     if (code) goto _err;
+    iDelIdx++;
+    if (iDelIdx < nDelIdx) {
+      code = tMapDataGetItemByIdx(&pCommitter->oDelIdxMap, iDelIdx, &delIdx, tGetDelIdx);
+      if (code) goto _err;
+      pDelIdx = &delIdx;
+    } else {
+      pDelIdx = NULL;
+    }
+    continue;
+
+  _commit_mem_and_disk_del:
+    code = tsdbCommitTableDel(pCommitter, pTbData, pDelIdx);
+    if (code) goto _err;
+    iTbData++;
+    iDelIdx++;
+    if (iTbData < nTbData) {
+      pTbData = (STbData *)taosArrayGetP(pMemTable->aTbData, iTbData);
+    } else {
+      pTbData = NULL;
+    }
+    if (iDelIdx < nDelIdx) {
+      code = tMapDataGetItemByIdx(&pCommitter->oDelIdxMap, iDelIdx, &delIdx, tGetDelIdx);
+      if (code) goto _err;
+      pDelIdx = &delIdx;
+    } else {
+      pDelIdx = NULL;
+    }
+    continue;
   }
 
   return code;
@@ -276,29 +295,20 @@ _err:
 static int32_t tsdbCommitDelEnd(SCommitter *pCommitter) {
   int32_t code = 0;
 
-  code = tsdbWriteDelIdx(pCommitter->pDelFWriter, &pCommitter->delIdxNew, NULL);
+  code = tsdbWriteDelIdx(pCommitter->pDelFWriter, &pCommitter->nDelIdxMap, NULL);
-  if (code) {
+  if (code) goto _err;
-    goto _err;
-  }
 
   code = tsdbUpdateDelFileHdr(pCommitter->pDelFWriter, NULL);
-  if (code) {
+  if (code) goto _err;
-    goto _err;
-  }
 
   code = tsdbDelFWriterClose(pCommitter->pDelFWriter, 1);
-  if (code) {
+  if (code) goto _err;
-    goto _err;
-  }
 
   if (pCommitter->pDelFReader) {
     code = tsdbDelFReaderClose(pCommitter->pDelFReader);
     if (code) goto _err;
   }
 
-  tDelDataClear(&pCommitter->delDataNew);
-  tDelIdxClear(&pCommitter->delIdxNew);
-
   return code;
 
 _err:
@@ -609,129 +619,89 @@ static int32_t tsdbCommitTableDataEnd(SCommitter *pCommitter) {
   return code;
 }
 
-static int32_t tsdbCommitTableDelStart(SCommitter *pCommitter) {
+static int32_t tsdbCommitTableDel(SCommitter *pCommitter, STbData *pTbData, SDelIdx *pDelIdx) {
-  int32_t  code = 0;
+  int32_t      code = 0;
-  tb_uid_t suid;
+  SDelData     delData;
-  tb_uid_t uid;
+  SDelOp      *pDelOp;
+  tb_uid_t     suid;
+  tb_uid_t     uid;
+  SDelIdx      delIdx;  // TODO
+  SDelDataInfo info;    // TODO
 
-  if (pCommitter->pTbData) {
+  // check no del data, just return
-    suid = pCommitter->pTbData->suid;
+  if (pTbData && pTbData->pHead == NULL) {
-    uid = pCommitter->pTbData->uid;
+    pTbData = NULL;
   }
+  if (pTbData == NULL && pDelIdx == NULL) goto _exit;
 
-  // load old
+  // prepare
-  pCommitter->delDataOld = (SDelData){0};
+  if (pTbData) {
-  if (pCommitter->pDelIdxItem) {
+    info.suid = pTbData->suid;
-    suid = pCommitter->pDelIdxItem->suid;
+    info.uid = pTbData->uid;
-    uid = pCommitter->pDelIdxItem->uid;
+  } else {
-    code =
+    info.suid = pDelIdx->suid;
-        tsdbReadDelData(pCommitter->pDelFReader, pCommitter->pDelIdxItem, &pCommitter->delDataOld, &pCommitter->pBuf5);
+    info.uid = pDelIdx->uid;
+  }
+  delIdx.suid = info.suid;
+  delIdx.uid = info.uid;
+  delIdx.minKey = TSKEY_MAX;
+  delIdx.maxKey = TSKEY_MIN;
+  delIdx.minVersion = INT64_MAX;
+  delIdx.maxVersion = -1;
+
+  // start
+  tMapDataReset(&pCommitter->oDelDataMap);
+  tMapDataReset(&pCommitter->nDelDataMap);
+
+  if (pDelIdx) {
+    code = tsdbReadDelData(pCommitter->pDelFReader, pDelIdx, &pCommitter->oDelDataMap, NULL);
     if (code) goto _err;
   }
 
-  // prepare new
+  // disk
-  pCommitter->delDataNew.suid = suid;
+  for (int32_t iDelData = 0; iDelData < pCommitter->oDelDataMap.nItem; iDelData++) {
-  pCommitter->delDataNew.uid = uid;
+    code = tMapDataGetItemByIdx(&pCommitter->oDelDataMap, iDelData, &delData, tGetDelData);
-  pCommitter->delDataNew.offset.flag = 0;
+    if (code) goto _err;
-  pCommitter->delDataNew.offset.nOffset = 0;
-  pCommitter->delDataNew.nData = 0;
-  pCommitter->delIdxItem = (SDelIdxItem){
-      .suid = suid,
-      .uid = uid,
-      .minKey = TSKEY_MAX,
-      .maxKey = TSKEY_MIN,
-      .minVersion = INT64_MAX,
-      .maxVersion = INT64_MIN,
-      .offset = -1,
-      .size = -1,
-  };
 
-  return code;
+    code = tMapDataPutItem(&pCommitter->nDelDataMap, &delData, tPutDelData);
+    if (code) goto _err;
 
-_err:
+    if (delIdx.minKey > delData.sKey) delIdx.minKey = delData.sKey;
-  tsdbError("vgId:%d commit table del start failed since %s", TD_VID(pCommitter->pTsdb->pVnode), tstrerror(code));
+    if (delIdx.maxKey < delData.eKey) delIdx.maxKey = delData.eKey;
-  return code;
+    if (delIdx.minVersion > delData.version) delIdx.minVersion = delData.version;
-}
+    if (delIdx.maxVersion < delData.version) delIdx.maxVersion = delData.version;
-
-static int32_t tsdbCommitTableDelImpl(SCommitter *pCommitter) {
-  int32_t      code = 0;
-  SDelDataItem item;
-
-  // old
-  if (pCommitter->pDelIdxItem) {
-    for (int32_t iDelIdxItem = 0; iDelIdxItem < pCommitter->delDataOld.offset.nOffset; iDelIdxItem++) {
-      code = tDelDataGetItemByIdx(&pCommitter->delDataOld, &item, iDelIdxItem);
-      if (code) goto _err;
-
-      code = tDelDataPutItem(&pCommitter->delDataNew, &item);
-      if (code) goto _err;
-
-      // update index
-      if (item.version < pCommitter->delIdxItem.minVersion) pCommitter->delIdxItem.minVersion = item.version;
-      if (item.version > pCommitter->delIdxItem.maxVersion) pCommitter->delIdxItem.maxVersion = item.version;
-      if (item.sKey < pCommitter->delIdxItem.minKey) pCommitter->delIdxItem.minKey = item.sKey;
-      if (item.eKey > pCommitter->delIdxItem.maxKey) pCommitter->delIdxItem.maxKey = item.eKey;
-    }
   }
 
-  // new
+  // memory
-  if (pCommitter->pTbData) {
+  pDelOp = pTbData ? pTbData->pHead : NULL;
-    for (SDelOp *pDelOp = pCommitter->pTbData->pHead; pDelOp; pDelOp = pDelOp->pNext) {
+  for (; pDelOp; pDelOp = pDelOp->pNext) {
-      item = (SDelDataItem){.version = pDelOp->version, .sKey = pDelOp->sKey, .eKey = pDelOp->eKey};
+    delData.version = pDelOp->version;
+    delData.sKey = pDelOp->sKey;
+    delData.eKey = pDelOp->eKey;
 
-      code = tDelDataPutItem(&pCommitter->delDataNew, &item);
+    code = tMapDataPutItem(&pCommitter->nDelDataMap, &delData, tPutDelData);
-      if (code) goto _err;
+    if (code) goto _err;
 
-      // update index
+    if (delIdx.minKey > delData.sKey) delIdx.minKey = delData.sKey;
-      if (item.version < pCommitter->delIdxItem.minVersion) pCommitter->delIdxItem.minVersion = item.version;
+    if (delIdx.maxKey < delData.eKey) delIdx.maxKey = delData.eKey;
-      if (item.version > pCommitter->delIdxItem.maxVersion) pCommitter->delIdxItem.maxVersion = item.version;
+    if (delIdx.minVersion > delData.version) delIdx.minVersion = delData.version;
-      if (item.sKey < pCommitter->delIdxItem.minKey) pCommitter->delIdxItem.minKey = item.sKey;
+    if (delIdx.maxVersion < delData.version) delIdx.maxVersion = delData.version;
-      if (item.eKey > pCommitter->delIdxItem.maxKey) pCommitter->delIdxItem.maxKey = item.eKey;
-    }
   }
 
+  ASSERT(pCommitter->nDelDataMap.nItem > 0);
+
+  // write
+  code = tsdbWriteDelData(pCommitter->pDelFWriter, &info, &pCommitter->nDelDataMap, NULL, &delIdx.offset, &delIdx.size);
+  if (code) goto _err;
+
+  // put delIdx
+  code = tMapDataPutItem(&pCommitter->nDelIdxMap, &delIdx, tPutDelIdx);
+  if (code) goto _err;
+
+_exit:
   return code;
 
 _err:
-  return code;
+  tsdbError("vgId:%d commit table del failed since %s", TD_VID(pCommitter->pTsdb->pVnode), tstrerror(code));
-}
-
-static int32_t tsdbCommitTableDelEnd(SCommitter *pCommitter) {
-  int32_t code = 0;
-
-  // write table del data
-  code = tsdbWriteDelData(pCommitter->pDelFWriter, &pCommitter->delDataNew, NULL, &pCommitter->delIdxItem.offset,
-                          &pCommitter->delIdxItem.size);
-  if (code) goto _err;
-
-  // add SDelIdxItem
-  code = tDelIdxPutItem(&pCommitter->delIdxNew, &pCommitter->delIdxItem);
-  if (code) goto _err;
-
-  return code;
-
-_err:
-  tsdbError("vgId:%d commit table del end failed since %s", TD_VID(pCommitter->pTsdb->pVnode), tstrerror(code));
-  return code;
-}
-
-static int32_t tsdbCommitTableDel(SCommitter *pCommitter) {
-  int32_t code = 0;
-
-  // start
-  code = tsdbCommitTableDelStart(pCommitter);
-  if (code) goto _err;
-
-  // impl
-  code = tsdbCommitTableDelImpl(pCommitter);
-  if (code) goto _err;
-
-  // end
-  code = tsdbCommitTableDelEnd(pCommitter);
-  if (code) goto _err;
-
-  return code;
-
-_err:
   return code;
 }