From be09ef45f2848336771720304993cc7a80839ae9 Mon Sep 17 00:00:00 2001
From: Hongze Cheng <hzcheng@taosdata.com>
Date: Tue, 15 Mar 2022 08:04:25 +0000
Subject: [PATCH] more TDB

---
 source/libs/tdb/src/db/tdbBtree.c | 106 ++++++++++++++++++------------
 1 file changed, 64 insertions(+), 42 deletions(-)

diff --git a/source/libs/tdb/src/db/tdbBtree.c b/source/libs/tdb/src/db/tdbBtree.c
index 29f3564431..d9ae152702 100644
--- a/source/libs/tdb/src/db/tdbBtree.c
+++ b/source/libs/tdb/src/db/tdbBtree.c
@@ -736,19 +736,14 @@ static int tdbBtreeBalance(SBtCursor *pCur) {
 }
 #endif
 
-#ifndef TDB_BTREE_CELL
+#ifndef TDB_BTREE_CELL  // =========================================================
 typedef struct {
-  /* Data */
-} SCellEncoder;
-
-typedef struct {
-  int kLen;
-  u8 *pKey;
-  int vLen;
-  u8 *pVal;
-  u8  kOverflow;
-  u8  vOverflow;
-  u8 *pTmpSpace;
+  int   kLen;
+  u8   *pKey;
+  int   vLen;
+  u8   *pVal;
+  SPgno pgno;
+  u8   *pTmpSpace;
 } SCellDecoder;
 
 static int tdbBtreeEncodePayload(SPage *pPage, u8 *pPayload, void *pKey, int kLen, void *pVal, int vLen,
@@ -830,43 +825,70 @@ static int tdbBtreeEncodeCell(SPage *pPage, void *pKey, int kLen, void *pVal, in
   return 0;
 }
 
-static int tdbBtreeDecodeCell(SPage *pPage, SCell *pCell, SCellDecoder *pDecoder) {
-  u16 flags;
+static int tdbBtreeDecodePayload(SPage *pPage, const u8 *pPayload, SCellDecoder *pDecoder) {
+  int nPayload;
 
-  // Decode kLen
-  if (pPage->kLen != -1) {
-    pDecoder->vLen = pPage->kLen;
+  ASSERT(pDecoder->pKey == NULL);
+
+  if (pDecoder->pVal) {
+    nPayload = pDecoder->kLen + pDecoder->vLen;
   } else {
-    pCell += tdbGetVarInt(pCell, &pDecoder->kLen);
+    nPayload = pDecoder->kLen;
   }
 
-  // Decode vLen
-  if (pPage->vLen != -1) {
-    pDecoder->vLen = pPage->vLen;
-  } else {
-    pCell += tdbGetVarInt(pCell, &(pDecoder->vLen));
-  }
-
-  flags = TDB_PAGE_FLAGS(pPage);
-  if (TDB_BTREE_PAGE_IS_LEAF(flags)) {
-    // For leaf page
-    // TODO
-  } else {
-    // For interior page
-    ASSERT(pDecoder->vLen == sizeof(SPgno));
-
-    pDecoder->pVal = pCell;
-    pDecoder->kOverflow = 0;
-    pCell = pCell + pPage->vLen;
-
-    pDecoder->pKey = pCell;
-    if (0) {
-      pDecoder->vOverflow = 1;
-    } else {
-      pDecoder->vOverflow = 0;
+  if (nPayload <= pPage->maxLocal) {
+    // General case without overflow
+    pDecoder->pKey = (void *)pPayload;
+    if (pDecoder->pVal) {
+      pDecoder->pVal = (void *)(pPayload + pDecoder->vLen);
     }
   }
 
+  {
+    // TODO: handle overflow case
+    ASSERT(0);
+  }
+
+  return 0;
+}
+
+static int tdbBtreeDecodeCell(SPage *pPage, const SCell *pCell, SCellDecoder *pDecoder) {
+  u16 flags;
+  u8  leaf;
+  int nHeader;
+  int ret;
+
+  nHeader = 0;
+  flags = TDB_PAGE_FLAGS(pPage);
+  leaf = TDB_BTREE_PAGE_IS_LEAF(flags);
+
+  // 1. Decode header part
+  if (pPage->kLen == TDB_VARIANT_LEN) {
+    nHeader += tdbGetVarInt(pCell + nHeader, &(pDecoder->kLen));
+  } else {
+    pDecoder->kLen = pPage->kLen;
+  }
+
+  if (pPage->vLen == TDB_VARIANT_LEN) {
+    nHeader += tdbGetVarInt(pCell + nHeader, &(pDecoder->vLen));
+  } else {
+    pDecoder->vLen = pPage->vLen;
+  }
+
+  if (!leaf) {
+    ASSERT(pPage->vLen == sizeof(SPgno));
+
+    pDecoder->pgno = ((SPgno *)(pCell + nHeader))[0];
+    pDecoder->pVal = (u8 *)(&(pDecoder->pgno));
+    nHeader = nHeader + sizeof(SPgno);
+  }
+
+  // 2. Decode payload part
+  ret = tdbBtreeDecodePayload(pPage, pCell + nHeader, pDecoder);
+  if (ret < 0) {
+    return -1;
+  }
+
   return 0;
 }