alter pages

source/dnode/vnode/src/inc/vnodeInt.h

@@ -111,6 +111,7 @@ SSchemaWrapper* metaGetTableSchema(SMeta* pMeta, tb_uid_t uid, int32_t sver, boo
 STSchema*       metaGetTbTSchema(SMeta* pMeta, tb_uid_t uid, int32_t sver);
 int32_t         metaGetTbTSchemaEx(SMeta* pMeta, tb_uid_t suid, tb_uid_t uid, int32_t sver, STSchema** ppTSchema);
 int             metaGetTableEntryByName(SMetaReader* pReader, const char* name);
+int             metaAlterCache(SMeta* pMeta, int32_t nPage);
 
 tb_uid_t      metaGetTableEntryUidByName(SMeta* pMeta, const char* name);
 int64_t       metaGetTbNum(SMeta* pMeta);

source/dnode/vnode/src/meta/metaOpen.c

@@ -197,6 +197,18 @@ int metaClose(SMeta *pMeta) {
   return 0;
 }
 
+int metaAlterCache(SMeta *pMeta, int32_t nPage) {
+  metaWLock(pMeta);
+
+  if (tdbAlter(pMeta->pEnv, nPage) < 0) {
+    metaULock(pMeta);
+    return -1;
+  }
+
+  metaULock(pMeta);
+  return 0;
+}
+
 int32_t metaRLock(SMeta *pMeta) {
   int32_t ret = 0;
 

source/dnode/vnode/src/vnd/vnodeSvr.c

@@ -1047,7 +1047,14 @@ static int32_t vnodeProcessAlterConfigReq(SVnode *pVnode, int64_t version, void
   }
 
   if (pVnode->config.szCache != alterReq.pages) {
-    // TODO
+    if (metaAlterCache(pVnode->pMeta, alterReq.pages) < 0) {
+      vError("vgId:%d failed to change vnode pages from %d to %d failed since %s", TD_VID(pVnode),
+             pVnode->config.szCache, alterReq.pages, tstrerror(errno));
+      return errno;
+    } else {
+      vInfo("vgId:%d vnode pages is changed from %d to %d", TD_VID(pVnode), pVnode->config.szCache, alterReq.pages);
+      pVnode->config.szCache = alterReq.pages;
+    }
   }
 
   if (pVnode->config.cacheLast != alterReq.cacheLast) {

source/libs/tdb/inc/tdb.h

@@ -36,6 +36,7 @@ int32_t tdbClose(TDB *pDb);
 int32_t tdbBegin(TDB *pDb, TXN *pTxn);
 int32_t tdbCommit(TDB *pDb, TXN *pTxn);
 int32_t tdbAbort(TDB *pDb, TXN *pTxn);
+int32_t tdbAlter(TDB *pDb, int pages);
 
 // TTB
 int32_t tdbTbOpen(const char *tbname, int keyLen, int valLen, tdb_cmpr_fn_t keyCmprFn, TDB *pEnv, TTB **ppTb);

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

@@ -97,6 +97,8 @@ int tdbClose(TDB *pDb) {
   return 0;
 }
 
+int32_t tdbAlter(TDB *pDb, int pages) { return tdbPCacheAlter(pDb->pCache, pages); }
+
 int32_t tdbBegin(TDB *pDb, TXN *pTxn) {
   SPager *pPager;
   int     ret;

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

@@ -61,7 +61,11 @@ int tdbPCacheOpen(int pageSize, int cacheSize, SPCache **ppCache) {
 
   pCache->szPage = pageSize;
   pCache->nPages = cacheSize;
-  pCache->aPage = (SPage **)&pCache[1];
+  pCache->aPage = (SPage **)tdbOsCalloc(cacheSize, sizeof(SPage *));
+  if (pCache->aPage == NULL) {
+    tdbOsFree(pCache);
+    return -1;
+  }
 
   if (tdbPCacheOpenImpl(pCache) < 0) {
     tdbOsFree(pCache);
@@ -75,11 +79,78 @@ int tdbPCacheOpen(int pageSize, int cacheSize, SPCache **ppCache) {
 int tdbPCacheClose(SPCache *pCache) {
   if (pCache) {
     tdbPCacheCloseImpl(pCache);
+    tdbOsFree(pCache->aPage);
     tdbOsFree(pCache);
   }
   return 0;
 }
 
+// TODO:
+// if (pPage->id >= pCache->nPages) {
+//   free(pPage);
+//   pCache->aPage[pPage->id] = NULL;
+// } else {
+//   add to free list
+// }
+
+static int tdbPCacheAlterImpl(SPCache *pCache, int32_t nPage) {
+  if (pCache->nPages == nPage) {
+    return 0;
+  } else if (pCache->nPages < nPage) {
+    SPage **aPage = tdbOsCalloc(nPage, sizeof(SPage *));
+    if (aPage == NULL) {
+      return -1;
+    }
+
+    for (int32_t iPage = pCache->nPage; iPage < nPage; iPage++) {
+      if (tdbPageCreate(pCache->szPage, &aPage[iPage], tdbDefaultMalloc, NULL) < 0) {
+        // TODO: handle error
+        return -1;
+      }
+
+      // pPage->pgid = 0;
+      aPage[iPage]->isAnchor = 0;
+      aPage[iPage]->isLocal = 1;
+      aPage[iPage]->nRef = 0;
+      aPage[iPage]->pHashNext = NULL;
+      aPage[iPage]->pLruNext = NULL;
+      aPage[iPage]->pLruPrev = NULL;
+      aPage[iPage]->pDirtyNext = NULL;
+
+      // add page to free list
+      aPage[iPage]->pFreeNext = pCache->pFree;
+      pCache->pFree = aPage[iPage];
+      pCache->nFree++;
+
+      // add to local list
+      aPage[iPage]->id = iPage;
+    }
+
+    for (int32_t iPage = 0; iPage < pCache->nPage; iPage++) {
+      aPage[iPage] = pCache->aPage[iPage];
+    }
+
+    tdbOsFree(pCache->aPage);
+    pCache->nFree = nPage - pCache->nPage;
+    pCache->aPage = aPage;
+  }
+
+  pCache->nPages = nPage;
+  return 0;
+}
+
+int tdbPCacheAlter(SPCache *pCache, int32_t nPage) {
+  int ret = 0;
+
+  tdbPCacheLock(pCache);
+
+  ret = tdbPCacheAlterImpl(pCache, nPage);
+
+  tdbPCacheUnlock(pCache);
+
+  return ret;
+}
+
 SPage *tdbPCacheFetch(SPCache *pCache, const SPgid *pPgid, TXN *pTxn) {
   SPage *pPage;
   i32    nRef;
@@ -310,8 +381,7 @@ static int tdbPCacheOpenImpl(SPCache *pCache) {
   pCache->nFree = 0;
   pCache->pFree = NULL;
   for (int i = 0; i < pCache->nPages; i++) {
-    ret = tdbPageCreate(pCache->szPage, &pPage, tdbDefaultMalloc, NULL);
-    if (ret < 0) {
+    if (tdbPageCreate(pCache->szPage, &pPage, tdbDefaultMalloc, NULL) < 0) {
       // TODO: handle error
       return -1;
     }

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

@@ -216,6 +216,7 @@ int  tdbPagerRestore(SPager *pPager, SBTree *pBt);
 
 int    tdbPCacheOpen(int pageSize, int cacheSize, SPCache **ppCache);
 int    tdbPCacheClose(SPCache *pCache);
+int    tdbPCacheAlter(SPCache *pCache, int32_t nPage);
 SPage *tdbPCacheFetch(SPCache *pCache, const SPgid *pPgid, TXN *pTxn);
 void   tdbPCacheRelease(SPCache *pCache, SPage *pPage, TXN *pTxn);
 int    tdbPCacheGetPageSize(SPCache *pCache);