From cde7a7c6a1558a8bb76ad3d7bd83fef44fc89172 Mon Sep 17 00:00:00 2001
From: Hongze Cheng <hzcheng@taosdata.com>
Date: Mon, 21 Mar 2022 02:54:00 +0000
Subject: [PATCH] more TDB

---
 source/libs/tdb/src/db/tdbBtree.c  | 14 ++++++++------
 source/libs/tdb/src/inc/tdbPage.h  |  6 +++---
 source/libs/tdb/src/page/tdbPage.c | 24 ++++++++++++++++--------
 3 files changed, 27 insertions(+), 17 deletions(-)

diff --git a/source/libs/tdb/src/db/tdbBtree.c b/source/libs/tdb/src/db/tdbBtree.c
index 621afc9dfb..0c83dd609c 100644
--- a/source/libs/tdb/src/db/tdbBtree.c
+++ b/source/libs/tdb/src/db/tdbBtree.c
@@ -381,19 +381,20 @@ static int tdbBtreeInitPage(SPage *pPage, void *arg) {
   SBTree *pBt;
   u16     flags;
   u8      isLeaf;
+  u8      szAmHdr;
 
   pBt = (SBTree *)arg;
 
   flags = TDB_PAGE_FLAGS(pPage);
   isLeaf = TDB_BTREE_PAGE_IS_LEAF(flags);
   if (isLeaf) {
-    pPage->szAmHdr = 0;
+    szAmHdr = 0;
   } else {
-    pPage->szAmHdr = sizeof(SBtPageHdr);
+    szAmHdr = sizeof(SBtPageHdr);
   }
   pPage->xCellSize = NULL;  // TODO
 
-  tdbPageInit(pPage);
+  tdbPageInit(pPage, szAmHdr);
 
   TDB_BTREE_ASSERT_FLAG(flags);
 
@@ -416,6 +417,7 @@ static int tdbBtreeZeroPage(SPage *pPage, void *arg) {
   u16     flags;
   SBTree *pBt;
   u8      isLeaf;
+  u8      szAmHdr;
 
   flags = ((SBtreeZeroPageArg *)arg)->flags;
   pBt = ((SBtreeZeroPageArg *)arg)->pBt;
@@ -425,13 +427,13 @@ static int tdbBtreeZeroPage(SPage *pPage, void *arg) {
   TDB_PAGE_FLAGS_SET(pPage, flags);
   // Set szAmHdr
   if (isLeaf) {
-    pPage->szAmHdr = 0;
+    szAmHdr = 0;
   } else {
-    pPage->szAmHdr = sizeof(SBtPageHdr);
+    szAmHdr = sizeof(SBtPageHdr);
   }
   pPage->xCellSize = NULL;  // TODO
 
-  tdbPageZero(pPage);
+  tdbPageZero(pPage, szAmHdr);
 
   if (isLeaf) {
     pPage->kLen = pBt->keyLen;
diff --git a/source/libs/tdb/src/inc/tdbPage.h b/source/libs/tdb/src/inc/tdbPage.h
index 82287b0921..58ae1f3963 100644
--- a/source/libs/tdb/src/inc/tdbPage.h
+++ b/source/libs/tdb/src/inc/tdbPage.h
@@ -61,7 +61,6 @@ struct SPage {
   u8                *pData;
   SPageMethods      *pPageMethods;
   // Fields below used by pager and am
-  u8        szAmHdr;
   u8       *pAmHdr;
   u8       *pPageHdr;
   u8       *pCellIdx;
@@ -111,8 +110,8 @@ struct SPage {
 
 int  tdbPageCreate(int pageSize, SPage **ppPage, void *(*xMalloc)(void *, size_t), void *arg);
 int  tdbPageDestroy(SPage *pPage, void (*xFree)(void *arg, void *ptr), void *arg);
-void tdbPageZero(SPage *pPage);
-void tdbPageInit(SPage *pPage);
+void tdbPageZero(SPage *pPage, u8 szAmHdr);
+void tdbPageInit(SPage *pPage, u8 szAmHdr);
 int  tdbPageInsertCell(SPage *pPage, int idx, SCell *pCell, int szCell);
 int  tdbPageDropCell(SPage *pPage, int idx);
 
@@ -134,6 +133,7 @@ static inline SCell *tdbPageGetCell(SPage *pPage, int idx) {
   }
 
   lidx = idx - iOvfl;
+  ASSERT(lidx >= 0 && lidx < pPage->pPageMethods->getCellNum(pPage));
   pCell = pPage->pData + pPage->pPageMethods->getCellOffset(pPage, lidx);
 
   return pCell;
diff --git a/source/libs/tdb/src/page/tdbPage.c b/source/libs/tdb/src/page/tdbPage.c
index 2b0e573aae..11c9a9247e 100644
--- a/source/libs/tdb/src/page/tdbPage.c
+++ b/source/libs/tdb/src/page/tdbPage.c
@@ -32,8 +32,8 @@ extern SPageMethods pageLargeMethods;
 #define TDB_PAGE_NFREE_SET(pPage, NFREE)                (*(pPage)->pPageMethods->setFreeBytes)(pPage, NFREE)
 #define TDB_PAGE_CELL_OFFSET_AT_SET(pPage, idx, OFFSET) (*(pPage)->pPageMethods->setCellOffset)(pPage, idx, OFFSET)
 #define TDB_PAGE_CELL_AT(pPage, idx)                    ((pPage)->pData + TDB_PAGE_CELL_OFFSET_AT(pPage, idx))
-#define TDB_PAGE_MAX_FREE_BLOCK(pPage) \
-  ((pPage)->pageSize - (pPage)->szAmHdr - TDB_PAGE_HDR_SIZE(pPage) - sizeof(SPageFtr))
+#define TDB_PAGE_MAX_FREE_BLOCK(pPage, szAmHdr) \
+  ((pPage)->pageSize - (szAmHdr)-TDB_PAGE_HDR_SIZE(pPage) - sizeof(SPageFtr))
 
 static int tdbPageAllocate(SPage *pPage, int size, SCell **ppCell);
 static int tdbPageDefragment(SPage *pPage);
@@ -79,17 +79,25 @@ int tdbPageDestroy(SPage *pPage, void (*xFree)(void *arg, void *ptr), void *arg)
   return 0;
 }
 
-void tdbPageZero(SPage *pPage) {
+void tdbPageZero(SPage *pPage, u8 szAmHdr) {
+  pPage->pAmHdr = pPage->pData;
+  pPage->pPageHdr = pPage->pAmHdr + szAmHdr;
   TDB_PAGE_NCELLS_SET(pPage, 0);
   TDB_PAGE_CCELLS_SET(pPage, pPage->pageSize - sizeof(SPageFtr));
   TDB_PAGE_FCELL_SET(pPage, 0);
-  TDB_PAGE_NFREE_SET(pPage, TDB_PAGE_MAX_FREE_BLOCK(pPage));
-  tdbPageInit(pPage);
+  TDB_PAGE_NFREE_SET(pPage, TDB_PAGE_MAX_FREE_BLOCK(pPage, szAmHdr));
+  pPage->pCellIdx = pPage->pPageHdr + TDB_PAGE_HDR_SIZE(pPage);
+  pPage->pFreeStart = pPage->pCellIdx;
+  pPage->pFreeEnd = pPage->pData + TDB_PAGE_CCELLS(pPage);
+  pPage->pPageFtr = (SPageFtr *)(pPage->pData + pPage->pageSize - sizeof(SPageFtr));
+  pPage->nOverflow = 0;
+
+  ASSERT((u8 *)pPage->pPageFtr == pPage->pFreeEnd);
 }
 
-void tdbPageInit(SPage *pPage) {
+void tdbPageInit(SPage *pPage, u8 szAmHdr) {
   pPage->pAmHdr = pPage->pData;
-  pPage->pPageHdr = pPage->pAmHdr + pPage->szAmHdr;
+  pPage->pPageHdr = pPage->pAmHdr + szAmHdr;
   pPage->pCellIdx = pPage->pPageHdr + TDB_PAGE_HDR_SIZE(pPage);
   pPage->pFreeStart = pPage->pCellIdx + TDB_PAGE_OFFSET_SIZE(pPage) * TDB_PAGE_NCELLS(pPage);
   pPage->pFreeEnd = pPage->pData + TDB_PAGE_CCELLS(pPage);
@@ -107,7 +115,7 @@ int tdbPageInsertCell(SPage *pPage, int idx, SCell *pCell, int szCell) {
   int    lidx;  // local idx
   SCell *pNewCell;
 
-  ASSERT(szCell <= TDB_PAGE_MAX_FREE_BLOCK(pPage));
+  ASSERT(szCell <= TDB_PAGE_MAX_FREE_BLOCK(pPage, pPage->pPageHdr - pPage->pAmHdr));
 
   nFree = TDB_PAGE_NFREE(pPage);
   nCells = TDB_PAGE_NCELLS(pPage);
-- 
GitLab