more TDB

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

@@ -75,6 +75,7 @@ struct SPage {
   int       vLen;  // value length of the page, -1 for unknown
   int       maxLocal;
   int       minLocal;
+  int (*xCellSize)(SCell *pCell);
   // Fields used by SPCache
   TDB_PCACHE_PAGE
 };

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

@@ -256,8 +256,62 @@ static int tdbPageFree(SPage *pPage, int idx, SCell *pCell, int szCell) {
 }
 
 static int tdbPageDefragment(SPage *pPage) {
-  // TODO
-  ASSERT(0);
+  int    nFree;
+  int    nCells;
+  SCell *pCell;
+  SCell *pNextCell;
+  SCell *pTCell;
+  int    szCell;
+  int    idx;
+  int    iCell;
+
+  ASSERT(pPage->pFreeEnd - pPage->pFreeStart < nFree);
+
+  nFree = TDB_PAGE_NFREE(pPage);
+  nCells = TDB_PAGE_NCELLS(pPage);
+
+  // Loop to compact the page content
+  // Here we use an O(n^2) algorithm to do the job since
+  // this is a low frequency job.
+  pNextCell = (u8 *)pPage->pPageFtr;
+  pCell = NULL;
+  for (iCell = 0;; iCell++) {
+    // compact over
+    if (iCell == nCells) {
+      pPage->pFreeEnd = pNextCell;
+      break;
+    }
+
+    for (int i = 0; i < nCells; i++) {
+      if (TDB_PAGE_CELL_OFFSET_AT(pPage, i) < pNextCell - pPage->pData) {
+        pTCell = TDB_PAGE_CELL_AT(pPage, i);
+        if (pCell == NULL || pCell < pTCell) {
+          pCell = pTCell;
+          idx = i;
+        }
+      } else {
+        continue;
+      }
+    }
+
+    ASSERT(pCell != NULL);
+
+    szCell = (*pPage->xCellSize)(pCell);
+
+    ASSERT(pCell + szCell <= pNextCell);
+    if (pCell + szCell < pNextCell) {
+      memmove(pNextCell - szCell, pCell, szCell);
+    }
+
+    pCell = NULL;
+    pNextCell = pNextCell - szCell;
+    TDB_PAGE_CELL_OFFSET_AT_SET(pPage, idx, pNextCell - pPage->pData);
+  }
+
+  ASSERT(pPage->pFreeEnd - pPage->pFreeStart == nFree);
+  TDB_PAGE_CCELLS_SET(pPage, pPage->pFreeEnd - pPage->pData);
+  TDB_PAGE_FCELL_SET(pPage, 0);
+
   return 0;
 }