From a086f83c27e52deaf6da9527195bc81df91d2079 Mon Sep 17 00:00:00 2001
From: Hongze Cheng <hzcheng@taosdata.com>
Date: Wed, 8 Feb 2023 09:46:48 +0800
Subject: [PATCH] fix: slow TDB page defragment

---
 source/libs/tdb/src/db/tdbPage.c | 48 ++++++++++++++++++++++++++++++--
 1 file changed, 46 insertions(+), 2 deletions(-)

diff --git a/source/libs/tdb/src/db/tdbPage.c b/source/libs/tdb/src/db/tdbPage.c
index 50dc8e0a65..757a7f1ba0 100644
--- a/source/libs/tdb/src/db/tdbPage.c
+++ b/source/libs/tdb/src/db/tdbPage.c
@@ -404,7 +404,52 @@ static int tdbPageFree(SPage *pPage, int idx, SCell *pCell, int szCell) {
   return 0;
 }
 
+typedef struct {
+  int32_t iCell;
+  int32_t offset;
+} SCellIdx;
+static int32_t tCellIdxCmprFn(const void *p1, const void *p2) {
+  if (((SCellIdx *)p1)->offset < ((SCellIdx *)p2)->offset) {
+    return -1;
+  } else if (((SCellIdx *)p1)->offset > ((SCellIdx *)p2)->offset) {
+    return 1;
+  } else {
+    return 0;
+  }
+}
 static int tdbPageDefragment(SPage *pPage) {
+#if 1
+  int32_t nFree = TDB_PAGE_NFREE(pPage);
+  int32_t nCell = TDB_PAGE_NCELLS(pPage);
+
+  SCellIdx *aCellIdx = (SCellIdx *)tdbOsMalloc(sizeof(SCellIdx) * nCell);
+  if (aCellIdx == NULL) return -1;
+  for (int32_t iCell = 0; iCell < nCell; iCell++) {
+    aCellIdx[iCell].iCell = iCell;
+    aCellIdx[iCell].offset = TDB_PAGE_CELL_OFFSET_AT(pPage, iCell);
+  }
+  taosSort(aCellIdx, nCell, sizeof(SCellIdx), tCellIdxCmprFn);
+
+  SCell *pNextCell = (u8 *)pPage->pPageFtr;
+  for (int32_t iCell = nCell - 1; iCell >= 0; iCell--) {
+    SCell  *pCell = TDB_PAGE_CELL_AT(pPage, aCellIdx[iCell].iCell);
+    int32_t szCell = pPage->xCellSize(pPage, pCell, 0, NULL, NULL);
+
+    ASSERT(pNextCell - szCell >= pCell);
+
+    pNextCell -= szCell;
+    if (pNextCell > pCell) {
+      memmove(pNextCell, pCell, szCell);
+      TDB_PAGE_CELL_OFFSET_AT_SET(pPage, aCellIdx[iCell].iCell, pNextCell - pPage->pData);
+    }
+  }
+  pPage->pFreeEnd = pNextCell;
+  TDB_PAGE_CCELLS_SET(pPage, pPage->pFreeEnd - pPage->pData);
+  TDB_PAGE_FCELL_SET(pPage, 0);
+  tdbOsFree(aCellIdx);
+
+  ASSERT(pPage->pFreeEnd - pPage->pFreeStart == nFree);
+#else
   int    nFree;
   int    nCells;
   SCell *pCell;
@@ -457,10 +502,9 @@ static int tdbPageDefragment(SPage *pPage) {
     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);
-
+#endif
   return 0;
 }