Merge branch 'feature/tdb' of https://github.com/taosdata/TDengine into feature/meta

source/libs/tdb/src/db/tdbBtree.c

@@ -30,7 +30,6 @@ struct SBTree {
   int            valLen;
   SPager        *pPager;
   FKeyComparator kcmpr;
-  u8             fanout;
   int            pageSize;
   int            maxLocal;
   int            minLocal;
@@ -85,6 +84,8 @@ int tdbBtreeOpen(int keyLen, int valLen, SPager *pPager, FKeyComparator kcmpr, S
   SBTree *pBt;
   int     ret;
 
+  ASSERT(keyLen != 0);
+
   *ppBt = NULL;
 
   pBt = (SBTree *)tdbOsCalloc(1, sizeof(*pBt));
@@ -100,21 +101,14 @@ int tdbBtreeOpen(int keyLen, int valLen, SPager *pPager, FKeyComparator kcmpr, S
   pBt->pPager = pPager;
   // pBt->kcmpr
   pBt->kcmpr = kcmpr ? kcmpr : tdbDefaultKeyCmprFn;
-  // pBt->fanout
-  if (keyLen == TDB_VARIANT_LEN) {
-    pBt->fanout = TDB_DEFAULT_FANOUT;
-  } else {
-    ASSERT(0);
-    // TODO: pBt->fanout = 0;
-  }
   // pBt->pageSize
   pBt->pageSize = tdbPagerGetPageSize(pPager);
   // pBt->maxLocal
-  pBt->maxLocal = (pBt->pageSize - 14) / pBt->fanout;
+  pBt->maxLocal = tdbPageCapacity(pBt->pageSize, sizeof(SIntHdr)) / 4;
   // pBt->minLocal: Should not be allowed smaller than 15, which is [nPayload][nKey][nData]
-  pBt->minLocal = (pBt->pageSize - 14) / pBt->fanout / 2;
+  pBt->minLocal = pBt->maxLocal / 2;
   // pBt->maxLeaf
-  pBt->maxLeaf = pBt->pageSize - 14;
+  pBt->maxLeaf = tdbPageCapacity(pBt->pageSize, sizeof(SLeafHdr));
   // pBt->minLeaf
   pBt->minLeaf = pBt->minLocal;
 
@@ -929,7 +923,7 @@ static int tdbBtreeEncodeCell(SPage *pPage, const void *pKey, int kLen, const vo
   }
 
   // 2. Encode payload part
-  if (leaf) {
+  if (leaf && vLen > 0) {
     ret = tdbBtreeEncodePayload(pPage, pCell + nHeader, pKey, kLen, pVal, vLen, &nPayload);
   } else {
     ret = tdbBtreeEncodePayload(pPage, pCell + nHeader, pKey, kLen, NULL, 0, &nPayload);
@@ -968,6 +962,7 @@ static int tdbBtreeDecodePayload(SPage *pPage, const u8 *pPayload, SCellDecoder
   return 0;
 }
 
+// TODO: here has problem
 static int tdbBtreeDecodeCell(SPage *pPage, const SCell *pCell, SCellDecoder *pDecoder) {
   u8  flags;
   u8  leaf;

source/libs/tdb/src/inc/tdbPage.h

@@ -111,6 +111,7 @@ void tdbPageInit(SPage *pPage, u8 szAmHdr, int (*xCellSize)(const SPage *, SCell
 int  tdbPageInsertCell(SPage *pPage, int idx, SCell *pCell, int szCell, u8 asOvfl);
 int  tdbPageDropCell(SPage *pPage, int idx);
 void tdbPageCopy(SPage *pFromPage, SPage *pToPage);
+int  tdbPageCapacity(int pageSize, int amHdrSize);
 
 static inline SCell *tdbPageGetCell(SPage *pPage, int idx) {
   SCell *pCell;

source/libs/tdb/src/page/tdbPage.c

@@ -242,6 +242,18 @@ void tdbPageCopy(SPage *pFromPage, SPage *pToPage) {
   pToPage->nOverflow = pFromPage->nOverflow;
 }
 
+int tdbPageCapacity(int pageSize, int amHdrSize) {
+  int szPageHdr;
+
+  if (pageSize < 65536) {
+    szPageHdr = pageMethods.szPageHdr;
+  } else {
+    szPageHdr = pageLargeMethods.szPageHdr;
+  }
+
+  return pageSize - szPageHdr - amHdrSize;
+}
+
 static int tdbPageAllocate(SPage *pPage, int szCell, SCell **ppCell) {
   SCell *pFreeCell;
   u8    *pOffset;