Merge branch 'feature/tdb' into feature/meta

source/libs/tdb/CMakeLists.txt

@@ -8,6 +8,7 @@ target_sources(tdb
     "src/db/tdbBtree.c"
     "src/db/tdbDb.c"
     "src/db/tdbEnv.c"
+    "src/db/tdbTxn.c"
     "src/page/tdbPage.c"
     "src/page/tdbPageL.c"
 )

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

@@ -67,7 +67,7 @@ typedef struct {
   u8   *pTmpSpace;
 } SCellDecoder;
 
-static int tdbBtCursorMoveTo(SBTC *pCur, const void *pKey, int kLen, int *pCRst);
+static int tdbBtCursorMoveTo(SBTC *pBtc, const void *pKey, int kLen, int *pCRst);
 static int tdbDefaultKeyCmprFn(const void *pKey1, int keyLen1, const void *pKey2, int keyLen2);
 static int tdbBtreeOpenImpl(SBTree *pBt);
 static int tdbBtreeZeroPage(SPage *pPage, void *arg);
@@ -75,10 +75,10 @@ static int tdbBtreeInitPage(SPage *pPage, void *arg);
 static int tdbBtreeEncodeCell(SPage *pPage, const void *pKey, int kLen, const void *pVal, int vLen, SCell *pCell,
                               int *szCell);
 static int tdbBtreeDecodeCell(SPage *pPage, const SCell *pCell, SCellDecoder *pDecoder);
-static int tdbBtreeBalance(SBTC *pCur);
+static int tdbBtreeBalance(SBTC *pBtc);
 static int tdbBtreeCellSize(const SPage *pPage, SCell *pCell);
 static int tdbBtcMoveToNext(SBTC *pBtc);
-static int tdbBtcMoveDownward(SBTC *pCur, SPgno pgno);
+static int tdbBtcMoveDownward(SBTC *pBtc, SPgno pgno);
 static int tdbBtcMoveUpward(SBTC *pBtc);
 
 int tdbBtreeOpen(int keyLen, int valLen, SPager *pPager, FKeyComparator kcmpr, SBTree **ppBt) {
@@ -134,7 +134,7 @@ int tdbBtreeClose(SBTree *pBt) {
   return 0;
 }
 
-int tdbBtCursorInsert(SBTC *pCur, const void *pKey, int kLen, const void *pVal, int vLen) {
+int tdbBtCursorInsert(SBTC *pBtc, const void *pKey, int kLen, const void *pVal, int vLen) {
   int     ret;
   int     idx;
   SPager *pPager;
@@ -143,20 +143,20 @@ int tdbBtCursorInsert(SBTC *pCur, const void *pKey, int kLen, const void *pVal,
   int     cret;
   SBTree *pBt;
 
-  ret = tdbBtCursorMoveTo(pCur, pKey, kLen, &cret);
+  ret = tdbBtCursorMoveTo(pBtc, pKey, kLen, &cret);
   if (ret < 0) {
     // TODO: handle error
     return -1;
   }
 
-  if (pCur->idx == -1) {
-    ASSERT(TDB_PAGE_TOTAL_CELLS(pCur->pPage) == 0);
+  if (pBtc->idx == -1) {
+    ASSERT(TDB_PAGE_TOTAL_CELLS(pBtc->pPage) == 0);
     idx = 0;
   } else {
     if (cret > 0) {
-      idx = pCur->idx + 1;
+      idx = pBtc->idx + 1;
     } else if (cret < 0) {
-      idx = pCur->idx;
+      idx = pBtc->idx;
     } else {
       /* TODO */
       ASSERT(0);
@@ -164,7 +164,7 @@ int tdbBtCursorInsert(SBTC *pCur, const void *pKey, int kLen, const void *pVal,
   }
 
   // TODO: refact code here
-  pBt = pCur->pBt;
+  pBt = pBtc->pBt;
   if (!pBt->pTmp) {
     pBt->pTmp = (u8 *)malloc(pBt->pageSize);
     if (pBt->pTmp == NULL) {
@@ -175,20 +175,20 @@ int tdbBtCursorInsert(SBTC *pCur, const void *pKey, int kLen, const void *pVal,
   pCell = pBt->pTmp;
 
   // Encode the cell
-  ret = tdbBtreeEncodeCell(pCur->pPage, pKey, kLen, pVal, vLen, pCell, &szCell);
+  ret = tdbBtreeEncodeCell(pBtc->pPage, pKey, kLen, pVal, vLen, pCell, &szCell);
   if (ret < 0) {
     return -1;
   }
 
   // Insert the cell to the index
-  ret = tdbPageInsertCell(pCur->pPage, idx, pCell, szCell, 0);
+  ret = tdbPageInsertCell(pBtc->pPage, idx, pCell, szCell, 0);
   if (ret < 0) {
     return -1;
   }
 
   // If page is overflow, balance the tree
-  if (pCur->pPage->nOverflow > 0) {
-    ret = tdbBtreeBalance(pCur);
+  if (pBtc->pPage->nOverflow > 0) {
+    ret = tdbBtreeBalance(pBtc);
     if (ret < 0) {
       return -1;
     }
@@ -226,30 +226,30 @@ int tdbBtreeGet(SBTree *pBt, const void *pKey, int kLen, void **ppVal, int *vLen
   return 0;
 }
 
-static int tdbBtCursorMoveTo(SBTC *pCur, const void *pKey, int kLen, int *pCRst) {
+static int tdbBtCursorMoveTo(SBTC *pBtc, const void *pKey, int kLen, int *pCRst) {
   int     ret;
   SBTree *pBt;
   SPager *pPager;
 
-  pBt = pCur->pBt;
+  pBt = pBtc->pBt;
   pPager = pBt->pPager;
 
-  if (pCur->iPage < 0) {
-    ASSERT(pCur->iPage == -1);
-    ASSERT(pCur->idx == -1);
+  if (pBtc->iPage < 0) {
+    ASSERT(pBtc->iPage == -1);
+    ASSERT(pBtc->idx == -1);
 
     // Move from the root
-    ret = tdbPagerFetchPage(pPager, pBt->root, &(pCur->pPage), tdbBtreeInitPage, pBt);
+    ret = tdbPagerFetchPage(pPager, pBt->root, &(pBtc->pPage), tdbBtreeInitPage, pBt);
     if (ret < 0) {
       ASSERT(0);
       return -1;
     }
 
-    pCur->iPage = 0;
+    pBtc->iPage = 0;
 
-    if (TDB_PAGE_TOTAL_CELLS(pCur->pPage) == 0) {
+    if (TDB_PAGE_TOTAL_CELLS(pBtc->pPage) == 0) {
       // Current page is empty
-      // ASSERT(TDB_FLAG_IS(TDB_PAGE_FLAGS(pCur->pPage), TDB_BTREE_ROOT | TDB_BTREE_LEAF));
+      // ASSERT(TDB_FLAG_IS(TDB_PAGE_FLAGS(pBtc->pPage), TDB_BTREE_ROOT | TDB_BTREE_LEAF));
       return 0;
     }
 
@@ -259,7 +259,7 @@ static int tdbBtCursorMoveTo(SBTC *pCur, const void *pKey, int kLen, int *pCRst)
       SPage       *pPage;
       SCellDecoder cd = {0};
 
-      pPage = pCur->pPage;
+      pPage = pBtc->pPage;
       nCells = TDB_PAGE_TOTAL_CELLS(pPage);
       lidx = 0;
       ridx = nCells - 1;
@@ -297,22 +297,22 @@ static int tdbBtCursorMoveTo(SBTC *pCur, const void *pKey, int kLen, int *pCRst)
       u8 flags = TDB_BTREE_PAGE_GET_FLAGS(pPage);
       u8 leaf = TDB_BTREE_PAGE_IS_LEAF(flags);
       if (leaf) {
-        pCur->idx = midx;
+        pBtc->idx = midx;
         *pCRst = c;
         break;
       } else {
         if (c <= 0) {
-          pCur->idx = midx;
-          tdbBtcMoveDownward(pCur, cd.pgno);
+          pBtc->idx = midx;
+          tdbBtcMoveDownward(pBtc, cd.pgno);
         } else {
-          pCur->idx = midx + 1;
+          pBtc->idx = midx + 1;
           if (midx == nCells - 1) {
             /* Move to right-most child */
-            tdbBtcMoveDownward(pCur, ((SIntHdr *)pCur->pPage->pData)->pgno);
+            tdbBtcMoveDownward(pBtc, ((SIntHdr *)pBtc->pPage->pData)->pgno);
           } else {
-            pCell = tdbPageGetCell(pPage, pCur->idx);
+            pCell = tdbPageGetCell(pPage, pBtc->idx);
             tdbBtreeDecodeCell(pPage, pCell, &cd);
-            tdbBtcMoveDownward(pCur, cd.pgno);
+            tdbBtcMoveDownward(pBtc, cd.pgno);
           }
         }
       }
@@ -805,7 +805,7 @@ static int tdbBtreeBalanceNonRoot(SBTree *pBt, SPage *pParent, int idx) {
   return 0;
 }
 
-static int tdbBtreeBalance(SBTC *pCur) {
+static int tdbBtreeBalance(SBTC *pBtc) {
   int    iPage;
   SPage *pParent;
   SPage *pPage;
@@ -816,8 +816,8 @@ static int tdbBtreeBalance(SBTC *pCur) {
 
   // Main loop to balance the BTree
   for (;;) {
-    iPage = pCur->iPage;
-    pPage = pCur->pPage;
+    iPage = pBtc->iPage;
+    pPage = pBtc->pPage;
     flags = TDB_BTREE_PAGE_GET_FLAGS(pPage);
     leaf = TDB_BTREE_PAGE_IS_LEAF(flags);
     root = TDB_BTREE_PAGE_IS_ROOT(flags);
@@ -833,27 +833,27 @@ static int tdbBtreeBalance(SBTC *pCur) {
       // ignore the case of empty
       if (pPage->nOverflow == 0) break;
 
-      ret = tdbBtreeBalanceDeeper(pCur->pBt, pPage, &(pCur->pgStack[1]));
+      ret = tdbBtreeBalanceDeeper(pBtc->pBt, pPage, &(pBtc->pgStack[1]));
       if (ret < 0) {
         return -1;
       }
 
-      pCur->idx = 0;
-      pCur->idxStack[0] = 0;
-      pCur->pgStack[0] = pCur->pPage;
-      pCur->iPage = 1;
-      pCur->pPage = pCur->pgStack[1];
+      pBtc->idx = 0;
+      pBtc->idxStack[0] = 0;
+      pBtc->pgStack[0] = pBtc->pPage;
+      pBtc->iPage = 1;
+      pBtc->pPage = pBtc->pgStack[1];
     } else {
       // Generalized balance step
-      pParent = pCur->pgStack[iPage - 1];
+      pParent = pBtc->pgStack[iPage - 1];
 
-      ret = tdbBtreeBalanceNonRoot(pCur->pBt, pParent, pCur->idxStack[pCur->iPage - 1]);
+      ret = tdbBtreeBalanceNonRoot(pBtc->pBt, pParent, pBtc->idxStack[pBtc->iPage - 1]);
       if (ret < 0) {
         return -1;
       }
 
-      pCur->iPage--;
-      pCur->pPage = pCur->pgStack[pCur->iPage];
+      pBtc->iPage--;
+      pBtc->pPage = pBtc->pgStack[pBtc->iPage];
     }
   }
 
@@ -1050,11 +1050,11 @@ static int tdbBtreeCellSize(const SPage *pPage, SCell *pCell) {
 
 #endif
 
-int tdbBtcOpen(SBTC *pCur, SBTree *pBt) {
-  pCur->pBt = pBt;
-  pCur->iPage = -1;
-  pCur->pPage = NULL;
-  pCur->idx = -1;
+int tdbBtcOpen(SBTC *pBtc, SBTree *pBt) {
+  pBtc->pBt = pBt;
+  pBtc->iPage = -1;
+  pBtc->pPage = NULL;
+  pBtc->idx = -1;
 
   return 0;
 }
@@ -1262,16 +1262,16 @@ int tdbBtcClose(SBTC *pBtc) {
   return 0;
 }
 
-static int tdbBtcMoveDownward(SBTC *pCur, SPgno pgno) {
+static int tdbBtcMoveDownward(SBTC *pBtc, SPgno pgno) {
   int ret;
 
-  pCur->pgStack[pCur->iPage] = pCur->pPage;
-  pCur->idxStack[pCur->iPage] = pCur->idx;
-  pCur->iPage++;
-  pCur->pPage = NULL;
-  pCur->idx = -1;
+  pBtc->pgStack[pBtc->iPage] = pBtc->pPage;
+  pBtc->idxStack[pBtc->iPage] = pBtc->idx;
+  pBtc->iPage++;
+  pBtc->pPage = NULL;
+  pBtc->idx = -1;
 
-  ret = tdbPagerFetchPage(pCur->pBt->pPager, pgno, &pCur->pPage, tdbBtreeInitPage, pCur->pBt);
+  ret = tdbPagerFetchPage(pBtc->pBt->pPager, pgno, &pBtc->pPage, tdbBtreeInitPage, pBtc->pBt);
   if (ret < 0) {
     ASSERT(0);
   }

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

@@ -25,9 +25,7 @@ struct SPager {
   SPCache *pCache;
   SPgno    dbFileSize;
   SPgno    dbOrigSize;
-  int      nDirty;
   SPage   *pDirty;
-  SPage   *pDirtyTail;
   u8       inTran;
 };
 
@@ -46,6 +44,8 @@ TDB_STATIC_ASSERT(sizeof(SFileHdr) == 128, "Size of file header is not correct")
 static int tdbPagerReadPage(SPager *pPager, SPage *pPage);
 static int tdbPagerAllocPage(SPager *pPager, SPgno *ppgno);
 static int tdbPagerInitPage(SPager *pPager, SPage *pPage, int (*initPage)(SPage *, void *), void *arg);
+static int tdbPagerWritePageToJournal(SPager *pPager, SPage *pPage);
+static int tdbPagerWritePageToDB(SPager *pPager, SPage *pPage);
 
 int tdbPagerOpen(SPCache *pCache, const char *fileName, SPager **ppPager) {
   uint8_t *pPtr;
@@ -140,14 +140,25 @@ int tdbPagerWrite(SPager *pPager, SPage *pPage) {
     }
   }
 
-  if (pPage->isDirty == 0) {
-    pPage->isDirty = 1;
-    // TODO: add the page to the dirty list
+  if (pPage->isDirty) return 0;
 
-    // TODO: write the page to the journal
-    if (1 /*actually load from the file*/) {
+  // Set page as dirty
+  pPage->isDirty = 1;
+
+  // Add page to dirty list
+  // TODO: sort the list according to the page number
+  pPage->pDirtyNext = pPager->pDirty;
+  pPager->pDirty = pPage;
+
+  // Write page to journal
+  if (TDB_PAGE_PGNO(pPage) <= pPager->dbOrigSize) {
+    ret = tdbPagerWritePageToJournal(pPager, pPage);
+    if (ret < 0) {
+      ASSERT(0);
+      return -1;
     }
   }
+
   return 0;
 }
 
@@ -170,7 +181,37 @@ int tdbPagerBegin(SPager *pPager) {
 }
 
 int tdbPagerCommit(SPager *pPager) {
-  // TODO
+  SPage *pPage;
+  int    ret;
+
+  // Begin commit
+  {
+    // TODO: Sync the journal file (Here or when write ?)
+  }
+
+  for (;;) {
+    pPage = pPager->pDirty;
+
+    if (pPage == NULL) break;
+
+    ret = tdbPagerWritePageToDB(pPager, pPage);
+    if (ret < 0) {
+      ASSERT(0);
+      return -1;
+    }
+
+    pPager->pDirty = pPage->pDirtyNext;
+    pPage->pDirtyNext = NULL;
+
+    // TODO: release the page
+  }
+
+  fsync(pPager->fd);
+
+  close(pPager->jfd);
+  remove(pPager->jFileName);
+  pPager->jfd = -1;
+
   return 0;
 }
 
@@ -255,9 +296,7 @@ int tdbPagerNewPage(SPager *pPager, SPgno *ppgno, SPage **ppPage, int (*initPage
   return 0;
 }
 
-void tdbPagerReturnPage(SPager *pPager, SPage *pPage) {
-  tdbPCacheRelease(pPager->pCache, pPage);
-}
+void tdbPagerReturnPage(SPager *pPager, SPage *pPage) { tdbPCacheRelease(pPager->pCache, pPage); }
 
 static int tdbPagerAllocFreePage(SPager *pPager, SPgno *ppgno) {
   // TODO: Allocate a page from the free list
@@ -328,5 +367,44 @@ static int tdbPagerInitPage(SPager *pPager, SPage *pPage, int (*initPage)(SPage
     return -1;
   }
 
+  return 0;
+}
+
+// ---------------------------- Journal manipulation
+static int tdbPagerWritePageToJournal(SPager *pPager, SPage *pPage) {
+  int   ret;
+  SPgno pgno;
+
+  pgno = TDB_PAGE_PGNO(pPage);
+
+  ret = tdbWrite(pPager->jfd, &pgno, sizeof(pgno));
+  if (ret < 0) {
+    return -1;
+  }
+
+  ret = tdbWrite(pPager->jfd, pPage->pData, pPage->pageSize);
+  if (ret < 0) {
+    return -1;
+  }
+
+  return 0;
+}
+
+static int tdbPagerWritePageToDB(SPager *pPager, SPage *pPage) {
+  i64 offset;
+  int ret;
+
+  offset = pPage->pageSize * TDB_PAGE_PGNO(pPage);
+  if (lseek(pPager->fd, offset, SEEK_SET) < 0) {
+    ASSERT(0);
+    return -1;
+  }
+
+  ret = tdbWrite(pPager->fd, pPage->pData, pPage->pageSize);
+  if (ret < 0) {
+    ASSERT(0);
+    return -1;
+  }
+
   return 0;
 }
\ No newline at end of file

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

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "tdbInt.h"
+
+int tdbTxnBegin(STEnv *pEnv) {
+  // TODO
+  return 0;
+}
+
+int tdbTxnCommit(STEnv *pEnv) {
+  // TODO
+  return 0;
+}
+
+int tdbTxnRollback(STEnv *pEnv) {
+  // TODO
+  return 0;
+}
\ No newline at end of file

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

@@ -89,4 +89,9 @@ int tdbPRead(int fd, void *pData, int count, i64 offset) {
   }
 
   return count;
+}
+
+int tdbWrite(int fd, void *pData, int count) {
+  // TODO
+  return write(fd, pData, count);
 }
\ No newline at end of file

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

@@ -162,6 +162,8 @@ typedef struct SPage   SPage;
 
 #include "tdbPage.h"
 
+#include "tdbTxn.h"
+
 #ifdef __cplusplus
 }
 #endif

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

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifndef _TDB_TXN_H_
+#define _TDB_TXN_H_
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+typedef struct STxn STXN;
+
+struct STxn {
+  u64 txnId;
+  void *(*xMalloc)(void *, int);
+  void *xArg;
+};
+
+int tdbTxnBegin(STEnv *pEnv);
+int tdbTxnCommit(STEnv *pEnv);
+int tdbTxnRollback(STEnv *pEnv);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /*_TDB_TXN_H_*/
\ No newline at end of file

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

@@ -38,6 +38,7 @@ int tdbGnrtFileID(const char *fname, uint8_t *fileid, bool unique);
 int tdbGetFileSize(const char *fname, int pgSize, SPgno *pSize);
 
 int tdbPRead(int fd, void *pData, int count, i64 offset);
+int tdbWrite(int fd, void *pData, int count);
 
 #define TDB_REALLOC(PTR, SIZE)                                                          \
   ({                                                                                    \

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

@@ -372,11 +372,11 @@ static int tdbPageDefragment(SPage *pPage) {
   int    idx;
   int    iCell;
 
-  ASSERT(pPage->pFreeEnd - pPage->pFreeStart < nFree);
-
   nFree = TDB_PAGE_NFREE(pPage);
   nCells = TDB_PAGE_NCELLS(pPage);
 
+  ASSERT(pPage->pFreeEnd - pPage->pFreeStart < nFree);
+
   // 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.

source/libs/tdb/test/tdbTest.cpp

@@ -39,6 +39,8 @@ static void closePool(SPoolMem *pPool) {
   free(pPool);
 }
 
+#define clearPool closePool
+
 static void *poolMalloc(void *arg, int size) {
   void     *ptr = NULL;
   SPoolMem *pPool = (SPoolMem *)arg;
@@ -116,7 +118,7 @@ TEST(tdb_test, simple_test) {
   STEnv         *pEnv;
   STDB          *pDb;
   FKeyComparator compFunc;
-  int            nData = 10000000;
+  int            nData = 1000000;
 
   // Open Env
   ret = tdbEnvOpen("tdb", 4096, 256000, &pEnv);
@@ -132,13 +134,34 @@ TEST(tdb_test, simple_test) {
     char val[64];
 
     {  // Insert some data
+      int       i = 1;
+      SPoolMem *pPool;
+      int       memPoolCapacity = 16 * 1024;
+
+      pPool = openPool();
+
+      tdbTxnBegin(pEnv);
+
+      for (;;) {
+        if (i > nData) break;
 
-      for (int i = 1; i <= nData; i++) {
         sprintf(key, "key%d", i);
         sprintf(val, "value%d", i);
         ret = tdbDbInsert(pDb, key, strlen(key), val, strlen(val));
         GTEST_ASSERT_EQ(ret, 0);
+
+        if (pPool->size >= memPoolCapacity) {
+          tdbTxnCommit(pEnv);
+
+          clearPool(pPool);
+
+          tdbTxnBegin(pEnv);
+        }
+
+        i++;
       }
+
+      closePool(pPool);
     }
 
     {  // Query the data