SQLite4  Check-in [aaac4355c7]

void test_data_1(
  const char *zSystem,            /* Database system name */
  const char *zPattern,           /* Run test cases that match this pattern */
  int *pRc                        /* IN/OUT: Error code */
){
  Datatest1 aTest[] = {
    { {DATA_RANDOM,     500,600,   1000,2000},     1000,  100,  10,  0},
    { {DATA_RANDOM,     20,25,     100,200},       1000,  250, 1000, 1},
    { {DATA_RANDOM,     8,10,      100,200},       1000,  250, 1000, 1},
    { {DATA_RANDOM,     8,10,      10,20},         1000,  250, 1000, 1},
    { {DATA_RANDOM,     8,10,      1000,2000},     1000,  250, 1000, 1},
    { {DATA_RANDOM,     8,100,     10000,20000},    100,   25,  100, 1},
    { {DATA_RANDOM,     80,100,    10,20},         1000,  250, 1000, 1},
    { {DATA_RANDOM,     5000,6000, 10,20},          100,   25,  100, 1},

  int nData,
  int iSeed,
  TestDb *pControl,
  TestDb *pDb,
  int *pRc
){
  int i;

  static int nCall = 0;
  nCall++;

  testScanCompare(pControl, pDb, 0, 0, 0,         0, 0,         pRc);
  testScanCompare(pControl, pDb, 1, 0, 0,         0, 0,         pRc);

  if( *pRc==0 ){
    int iKey1;
    int iKey2;

  sqlite4_buffer output;
};

#define BT_INFO_PAGEDUMP       1
#define BT_INFO_FILENAME       2
#define BT_INFO_HDRDUMP        3
#define BT_INFO_PAGEDUMP_ASCII 4
#define BT_INFO_BLOCK_FREELIST 5
#define BT_INFO_PAGE_FREELIST  6
#define BT_INFO_PAGE_LEAKS     7

typedef struct bt_logsizecb bt_logsizecb;
struct bt_logsizecb {
  void *pCtx;                     /* A copy of this is passed to xLogsize() */
  void (*xLogsize)(void*, int);   /* Callback function */
};

  u32 blksz;                      /* Block size in bytes */
  u32 nPg;                        /* Size of database file in pages */

  u32 iRoot;                      /* B-tree root page */
  u32 iMRoot;                     /* Root page of meta-tree */
  u32 iSRoot;                     /* Root page of schedule-tree */

  u32 iSubBlock;                  /* Block containing current sub-tree */
  u32 nSubPg;                     /* Number of non-overflow pages in sub-tree */

  u32 iCookie;                    /* Current value of schema cookie */
  u32 iFreePg;                    /* First page in free-page list trunk */
  u32 iFreeBlk;                   /* First page in free-block list trunk */
};


/*
** This struct defines the format of database "schedule" pages.
**
** eBusy:
*/
typedef struct BtSchedule BtSchedule;
struct BtSchedule {
  u32 eBusy;                      /* One of the BT_SCHEDULE_XXX constants */
  u32 iAge;                       /* Age of input segments */
  u32 iMinLevel;                  /* Minimum level of input segments */
  u32 iMaxLevel;                  /* Maximum level of input segments */

/*
** Allocate new database pages or blocks.
*/
int sqlite4BtPageAllocate(BtPager*, BtPage **ppPage);
int sqlite4BtBlockAllocate(BtPager*, int nBlk, u32 *piBlk);

/* Block trim */
int sqlite4BtBlockTrim(BtPager*, u32);

/*
** Query page references.
*/
u32 sqlite4BtPagePgno(BtPage*);
void *sqlite4BtPageData(BtPage*);

/*

  /* These are used only by the bt_lock module. */
  BtShared *pShared;              /* Shared by all handles on this file */
  BtLock *pNext;                  /* Next connection using pShared */
  u32 mExclLock;                  /* Mask of exclusive locks held */
  u32 mSharedLock;                /* Mask of shared locks held */
  BtFile *pBtFile;                /* Used to defer close if necessary */

#ifndef NDEBUG
  u8 *aUsed;
#endif
};

struct BtReadSlot {
  u32 iFirst;
  u32 iLast;
};

*************************************************************************/

#ifdef NDEBUG
# define sqlite4BtDebugReadPage(a,b,c,d)
# define sqlite4BtDebugKV(a,b,c,d,e,f)
# define sqlite4BtDebugReadlock(a,b,c)
# define sqlite4BtDebugPageAlloc(a,b,c)
# define sqlite4BtDebugPageFree(a,b,c,d)
# define btErrorBkpt(x) x
#else
void sqlite4BtDebugReadPage(BtLock *pLock, u32 pgno, u8 *aData, int pgsz);
void sqlite4BtDebugKV(BtLock*, const char*,u8 *pK, int nK, u8 *pV, int nV);
void sqlite4BtDebugReadlock(BtLock *pLock, u32 iFirst, u32 iLast);
void sqlite4BtDebugPageAlloc(BtLock *pLock, const char*, u32);
void sqlite4BtDebugPageFree(BtLock *pLock, int, const char*, u32);
int btErrorBkpt(int rc);
#endif

    btLogChecksum(nativeCksum, a, 8, aIn, aOut);
    btLogChecksum(nativeCksum, &a[4], nByte-4, aOut, aOut);
  }else{
    btLogChecksum(nativeCksum, a, nByte, aIn, aOut);
  }
}

#define BT_ALLOC_DEBUG 0
#define BT_PAGE_DEBUG  0
#define BT_VAL_DEBUG   0
#define BT_HDR_DEBUG   0

static void btDebugTopology(BtLock *pLock, char *zStr, int iSide, u32 *aLog){
#if BT_PAGE_DEBUG
  fprintf(stderr, "%d:%s: (side=%d) %d..%d  %d..%d  %d..%d\n", 
      pLock->iDebugId, zStr, iSide,
      (int)aLog[0], (int)aLog[1], (int)aLog[2], 
      (int)aLog[3], (int)aLog[4], (int)aLog[5]

  fflush(stderr);
#endif
}
#endif

#ifndef NDEBUG
void sqlite4BtDebugPageAlloc(BtLock *pLock, const char *zStr, u32 pgno){
#if BT_ALLOC_DEBUG
  static int nCall = 0;
  fprintf(stderr, "%d:%d: allocate page %d from %s\n", 
      pLock->iDebugId, nCall++, pgno, zStr
  );
  fflush(stderr);
#endif
}
#endif

#ifndef NDEBUG
void sqlite4BtDebugPageFree(
  BtLock *pLock, int bBlock, const char *zStr, u32 pgno
){
#if BT_ALLOC_DEBUG
  static int nCall = 0;
  fprintf(stderr, "%d:%d: trim %s %d (%s)\n", 
      pLock->iDebugId, nCall++, bBlock ? "block" : "page", pgno, zStr
  );
  fflush(stderr);
  assert( pgno<1000000 );
#endif
}
#endif

static int fiLoadSummary(bt_db *, BtCursor *, const u8 **, int *);
static void btReadSummary(const u8 *, int, u16 *, u16 *, u16 *);
static int btCsrData(BtCursor *, int, int, const void **, int *);
static int btReadSchedule(bt_db *, u8 *, BtSchedule *);
static int btCsrEnd(BtCursor *pCsr, int bLast);
static int btCsrStep(BtCursor *pCsr, int bNext);
static int btCsrKey(BtCursor *pCsr, const void **ppK, int *pnK);


/* 
** Meta-table summary key.
*/
static const u8 aSummaryKey[] = {0xFF, 0xFF, 0xFF, 0xFF};

#ifndef btErrorBkpt

    sqlite4BtBufAppendf(pBuf, ")\n");

    for(i=0; i<nCell; i++){
      u8 *pCell;          /* Cell i */
      int nKey;           /* Number of bytes of key to output */
      u8 *pKey;           /* Buffer containing key. */
      int nVal;           /* Number of bytes of value to output */
      int nDummy;         /* Unused */
      u8 *pVal = 0;       /* Buffer containing value. */
      char celltype = 'A';

      pCell = btCellFind(aData, nData, i);
      pCell += sqlite4BtVarintGet32(pCell, &nKey);
      if( flags & BT_PGFLAGS_INTERNAL ){
        celltype = 'I';
      }else{
        if( nKey==0 ){
          celltype = 'C';
          pCell += sqlite4BtVarintGet32(pCell, &nKey);
        }else if( pCell[nKey]==0 ){
          celltype = 'B';
        }
      }
      sqlite4BtBufAppendf(pBuf, "  Cell %d: [%c] ", i, celltype);


      pKey = pCell;
      pCell += nKey;
      btBufferAppendBlob(pBuf, bAscii, pKey, nKey);
      sqlite4BtBufAppendf(pBuf, "  ");



      switch( celltype ){
        case 'I':
          sqlite4BtBufAppendf(pBuf, "child=%d ", (int)btGetU32(pCell));
          break;

        case 'A':
          pCell += sqlite4BtVarintGet32(pCell, &nVal);
          if( nVal>=2 ){
            nVal -= 2;
            pVal = pCell;
          }else{
            sqlite4BtBufAppendf(pBuf, "delete-key");
          }
          break;

        case 'B':
          assert( pCell[0]==0x00 );
          pCell++;
          pCell += sqlite4BtVarintGet32(pCell, &nVal);
          pVal = pCell;
          pCell += nVal;
          pCell += sqlite4BtVarintGet32(pCell, &nDummy);
          break;

        case 'C':
          assert( 0 );
      }

      if( pVal ){
        btBufferAppendBlob(pBuf, bAscii, pVal, nVal);
        if( flags & BT_PGFLAGS_METATREE ){
          /* Interpret the meta-tree entry */
          if( nKey==sizeof(aSummaryKey) && 0==memcmp(pKey, aSummaryKey, nKey) ){
            u16 iMin, nLvl, iMerge;
            int j;
            sqlite4BtBufAppendf(pBuf, "  [summary:");
            for(j=0; j<(nVal/6); j++){
              btReadSummary(pVal, j, &iMin, &nLvl, &iMerge);
              sqlite4BtBufAppendf(pBuf, " %d/%d/%d", 
                  (int)iMin, (int)nLvl, (int)iMerge
              );
            }
            sqlite4BtBufAppendf(pBuf, "]");
            
          }else{
            int nPgPerBlk = (pHdr->blksz / pHdr->pgsz);
            u32 iAge = btGetU32(&pKey[0]);
            u32 iLevel = ~btGetU32(&pKey[4]);
            u32 iRoot = btGetU32(pVal);
            sqlite4BtBufAppendf(pBuf, "  [age=%d level=%d root=%d]", 
                (int)iAge, (int)iLevel, (int)iRoot
            );
            sqlite4BtBufAppendf(pBuf, "  (blk=%d)", 1 + (iRoot / nPgPerBlk));
          }
        }
      }

      if( celltype=='B' ){
        int j;
        u8 ctrl = *pCell++;
        int nDirect = (ctrl & 0x0F);
        int nDepth = ((ctrl>>4) & 0x0F);
        sqlite4BtBufAppendf(
            pBuf, "  [overflow: direct=%d depth=%d]", nDirect, nDepth);
        for(j=0; j<=nDirect; j++){
          sqlite4BtBufAppendf(pBuf, " %d", btGetU32(&pCell[j*4]));
        }
      }
      sqlite4BtBufAppendf(pBuf, "\n");
    }

    if( pPager && btFlags(aData) & BT_PGFLAGS_INTERNAL ){
      for(i=0; i<=btCellCount(aData, nData); i++){
        BtPage *pChild;
        u8 *aChild;
        u32 child;

        child = btChildPgno(aData, nData, i);
        sqlite4BtPageGet(pPager, child, &pChild);
        aChild = sqlite4BtPageData(pChild);
        btPageToAscii(child, bAscii, pPager, aChild, nData, pBuf);
        sqlite4BtPageRelease(pChild);
      }
    }
  }
  sqlite4BtBufAppendf(pBuf, "\n");
}

static int btFreelistToAscii(bt_db *db, u32 iFirst, sqlite4_buffer *pBuf){
  int rc = SQLITE4_OK;
  u32 iTrunk = iFirst;
  while( iTrunk && rc==SQLITE4_OK ){
    BtPage *pPg = 0;
    rc = sqlite4BtPageGet(db->pPager, iTrunk, &pPg);
    if( rc==SQLITE4_OK ){
      u8 *aData = sqlite4BtPageData(pPg);
      u32 nFree = btGetU32(aData);
      u32 iNext = btGetU32(&aData[4]);
      int i;

      sqlite4BtBufAppendf(pBuf, "iTrunk=%d ", (int)iTrunk);
      sqlite4BtBufAppendf(pBuf, "nFree=%d iNext=%d (", (int)nFree, (int)iNext);
      for(i=0; i<(int)nFree; i++){
        u32 pgnoFree = btGetU32(&aData[8 + i*sizeof(u32)]);
        sqlite4BtBufAppendf(pBuf, "%s%d", (i==0)?"": " ", (int)pgnoFree);
      }
      sqlite4BtBufAppendf(pBuf, ")\n");

      sqlite4BtPageRelease(pPg);
      iTrunk = iNext;
    }
  }
  return rc;
}

#ifndef NDEBUG
#include <stdio.h>

static void printPage(FILE *f, u32 pgno, u8 *aData, int nData){
  sqlite4_buffer buf;

    }else{
      /* Type (a) */
      pVLocal = pCell;
      nVLocal -= 2;
    }
  }

  /* A delete-key */
  if( nVLocal<0 ) nVLocal = 0;

  pCsr->ovfl.nKey = nKLocal + nKOvfl;
  pCsr->ovfl.nVal = nVLocal + nVOvfl;

  nReq = pCsr->ovfl.nKey + pCsr->ovfl.nVal;
  assert( nReq>0 );
  rc = sqlite4_buffer_resize(&pCsr->ovfl.buf, nReq);
  if( rc!=SQLITE4_OK ) return rc;

  pSub = &pCsr->aSub[pCsr->iBt];
  aData = sqlite4BtPageData(pSub->csr.apPage[pSub->csr.nPg-1]);
  iCell = pSub->csr.aiCell[pSub->csr.nPg-1];

  *ppCell = btCellFindSize(aData, pgsz, iCell, pnCell);
}

/*
** Return true if the cell that the cursor currently points to contains 
** pointers to one or more overflow pages. Or false otherwise.
*/
static int btCsrOverflow(BtCursor *pCsr){
  const int pgsz = sqlite4BtPagerPagesize(pCsr->base.pDb->pPager);
  u8 *aData;                      /* Current page data */
  int nKey;                       /* First varint in cell */
  int res;                       /* First varint in cell */

  aData = sqlite4BtPageData(pCsr->apPage[pCsr->nPg-1]);
  aData = btCellFind(aData, pgsz, pCsr->aiCell[pCsr->nPg-1]);

  aData += sqlite4BtVarintGet32(aData, &nKey);
  res = (nKey==0 || aData[nKey]==0);
  return res;
}

/*
** Allocate a new page buffer.
*/
static int btNewBuffer(bt_db *pDb, u8 **paBuf){
  const int pgsz = sqlite4BtPagerPagesize(pDb->pPager);
  u8 *aBuf;

** tree structure.
**
** If successful, SQLITE4_OK is returned. If an error occurs, an SQLite
** error code.
*/ 
static int btIntegrateMerge(bt_db *db, BtSchedule *p){
  BtDbHdr *pHdr = sqlite4BtPagerDbhdr(db->pPager);
  const int nPgPerBlk = (pHdr->blksz / pHdr->pgsz);
  int rc = SQLITE4_OK;
  BtCursor csr;                   /* Cursor for reading various sub-trees */
  BtCursor mcsr;                  /* Cursor for reading the meta-tree */
  const void *pKey = 0;           /* If not NULL, first key to leave in input */
  int nKey = 0;                   /* Size of pKey in bytes */
  const u8 *aSum; int nSum;       /* Summary value */

  sqlite4_buffer buf;             /* Buffer object used for various purposes */
  u32 iLvl;
  int iBlk;

  /* Save the value of the fast-insert flag. It will be restored before
  ** this function returns. Leaving it set here interferes with page 
  ** allocation if the meta-tree needs to be extended.  */
  const int bFastInsertOp = db->bFastInsertOp;

    btCsrSetup(db, p->iNextPg, &csr);
    rc = btCsrEnd(&csr, 0);
    if( rc==SQLITE4_OK ){
      csr.aiCell[0] = p->iNextCell;
      rc = btCsrKey(&csr, &pKey, &nKey);
    }
  }

  if( p->iFreeList ){
    u32 iTrunk = p->iFreeList;
    BtCursor delcsr;
    memset(&delcsr, 0, sizeof(BtCursor));
    delcsr.nPg = 1;
    delcsr.base.pDb = db;

    while( rc==SQLITE4_OK && iTrunk!=0 ){
      BtPage *pTrunk = 0;
      rc = sqlite4BtPageGet(db->pPager, iTrunk, &pTrunk);
      if( rc==SQLITE4_OK ){
        u8 *aTData = sqlite4BtPageData(pTrunk);
        int nOvfl = btGetU32(aTData);
        int i;

        for(i=0; i<nOvfl; i++){
          u32 lpgno = btGetU32(&aTData[8 + i*8]);
          delcsr.aiCell[0] = (int)btGetU32(&aTData[8 + i*8 + 4]);
          rc = sqlite4BtPageGet(db->pPager, lpgno, &delcsr.apPage[0]);
          if( rc==SQLITE4_OK ){
            rc = btOverflowDelete(&delcsr);
            sqlite4BtPageRelease(delcsr.apPage[0]);
          }
        }

        iTrunk = btGetU32(&aTData[4]);
        sqlite4BtPageRelease(pTrunk);
      }
    }
  }

  /* The following loop iterates through each of the input levels. Each
  ** level is either removed from the database completely (if the merge
  ** completed) or else modified so that it contains no keys smaller
  ** than (pKey/nKey).  */ 
  for(iLvl=p->iMinLevel; iLvl<=p->iMaxLevel; iLvl++){
    u8 aPrefix[8];
    u32 iRoot = 0;

    /* Seek mcsr to the first sub-tree (smallest keys) in level iLvl. */
    btPutU32(&aPrefix[0], p->iAge);
    btPutU32(&aPrefix[4], ~iLvl);

    rc = btCsrSeek(&mcsr, 0, aPrefix, sizeof(aPrefix), BT_SEEK_GE, 0);
    if( rc==SQLITE4_INEXACT ) rc = SQLITE4_OK;

    /* Loop through the meta-tree entries that compose level iLvl, deleting 
    ** them as they are visited. If (pKey!=0), stop (and do not delete) the 
    ** first sub-tree for which all keys are larger than pKey.
    **
    ** When the loop exits, variable iRoot is left set to the root page of
    ** the last sub-tree deleted.  */
    while( rc==SQLITE4_OK ){
      const u8 *pMKey; int nMKey;
      rc = btCsrKey(&mcsr, (const void **)&pMKey, &nMKey);

      if( rc==SQLITE4_OK ){
        if( nMKey<sizeof(aPrefix) || memcmp(aPrefix, pMKey, sizeof(aPrefix)) ){
          rc = SQLITE4_NOTFOUND;
        }else{
          rc = SQLITE4_OK;
        }
      }
      if( rc==SQLITE4_OK ){
        if( pKey ){

          int res = btKeyCompare(pMKey + 8, nMKey - 8, pKey, nKey);
          if( res>0 ){
            break;
          }
        }
        if( iRoot ){
          rc = sqlite4BtBlockTrim(db->pPager, 1 + (iRoot / nPgPerBlk));
        }
      }

      if( rc==SQLITE4_OK ){
        const void *pData; int nData;
        btCsrData(&mcsr, 0, 4, &pData, &nData);
        iRoot = btGetU32(pData);

        rc = sqlite4BtDelete(&mcsr.base);
      }

      if( rc==SQLITE4_OK ){
        /* rc = btCsrStep(&mcsr, 1); */
        rc = btCsrSeek(&mcsr, 0, aPrefix, sizeof(aPrefix), BT_SEEK_GE, 0);
        if( rc==SQLITE4_INEXACT ) rc = SQLITE4_OK;
      }
    }
    if( rc==SQLITE4_NOTFOUND ) rc = SQLITE4_OK;

    if( rc==SQLITE4_OK && iRoot ){
      if( pKey ){
        int n = sizeof(aPrefix) + nKey;
        rc = sqlite4_buffer_resize(&buf, n);
        if( rc==SQLITE4_OK ){
          u8 aData[4];
          u8 *a = (u8*)buf.p;
          memcpy(a, aPrefix, sizeof(aPrefix));
          memcpy(&a[sizeof(aPrefix)], pKey, nKey);
          btPutU32(aData, iRoot);
          rc = btReplaceEntry(db, pHdr->iMRoot, a, n, aData, sizeof(aData));
        }
      }else{
        rc = sqlite4BtBlockTrim(db->pPager, 1 + (iRoot / nPgPerBlk));
      }
    }
  }

  /* Add new entries for the new output level blocks. */
  for(iBlk=0; 
      rc==SQLITE4_OK && iBlk<array_size(p->aRoot) && p->aRoot[iBlk]; 

      btPutU32(&a[4], ~p->iOutLevel);
      memcpy(&a[8], pKey, nKey);
      btPutU32(aData, p->aRoot[iBlk]);
      rc = btReplaceEntry(db, pHdr->iMRoot, a, nKey+8, aData, sizeof(aData));
    }
  }

  /* Trim any unused blocks */
  while( rc==SQLITE4_OK && iBlk<array_size(p->aBlock) && p->aBlock[iBlk] ){
    rc = sqlite4BtBlockTrim(db->pPager, p->aBlock[iBlk]);
    iBlk++;
  }

  /* Update the summary record with the outcome of the merge operation.  */

  if( rc==SQLITE4_OK ){
    btCsrReset(&csr, 1);
    rc = fiLoadSummary(db, &csr, &aSum, &nSum);
  }
  if( rc==SQLITE4_OK ){
    rc = sqlite4_buffer_resize(&buf, MAX(nSum, (p->iAge+1+1)*6));
    if( rc==SQLITE4_OK ){

      ** of the new segment is 0. The level number is one greater than the
      ** level number of the previous segment.  */
      btPutU32(&aKey[0], 0);
      btPutU32(&aKey[4], ~iLevel);
      btPutU32(&aVal[0], btFirstOfBlock(pHdr, iSubBlock));
      rc = btReplaceEntry(db, pHdr->iMRoot, aKey, 8, aVal, 4);
    }

    if( rc==SQLITE4_OK ){
      u32 iRoot = btFirstOfBlock(pHdr, pHdr->iSubBlock);
      BtPage *pPg = 0;

      rc = sqlite4BtPageGet(db->pPager, iRoot, &pPg);
      if( rc==SQLITE4_OK ) rc = sqlite4BtPageWrite(pPg);
      if( rc==SQLITE4_OK ){
        u8 *aData = sqlite4BtPageData(pPg);
        memset(&aData[pHdr->pgsz-6], 0, 6);
        aData[0] = 0;
      }
      sqlite4BtPageRelease(pPg);
    }
  }

  if( rc==SQLITE4_OK ){
    *piRoot = btFirstOfBlock(pHdr, pHdr->iSubBlock);
  }
  db->bFastInsertOp = 1;
  return rc;

};
struct FiWriter {
  bt_db *db;                      /* Database handle */
  BtSchedule *pSched;             /* Schedule object being implemented */
  int pgsz;                       /* Page size in bytes */
  int nPgPerBlk;                  /* Pages per block in this database */
  u32 iBlk;                       /* Block to write to */
  int nOvflPerPage;               /* Overflow pointers per page */

  int nAlloc;                     /* Pages allocated from current block */
  int nWrite;                     /* Pages written to current block */
  int nHier;                      /* Valid entries in apHier[] array */
  FiWriterPg aHier[MAX_SUBTREE_DEPTH];      /* Path from root to current leaf */

  /* Variables used to collect overflow pages freed by merge operation */
  int nOvfl;                      /* Number of pointers already in buffer */
  u8 *aTrunk;                     /* Buffer for current trunk page */
};

/*
** Write the page out to disk.
*/
static int fiWriterFlush(
  FiWriter *p, 

    int i;

    memset(p, 0, sizeof(FiWriter));
    p->db = db;
    p->pSched = pSched;
    p->pgsz = pHdr->pgsz;
    p->nPgPerBlk = (pHdr->blksz / pHdr->pgsz);
    p->nOvflPerPage = ((pHdr->pgsz / 8) - 1);

    /* Find a block to write to */
    for(i=0; pSched->aBlock[i]; i++){
      if( pSched->aRoot[i]==0 ) break;
    }
    if( pSched->aBlock[i]==0 ){
      *pRc = BT_BLOCKFULL;
    }else{
      p->iBlk = pSched->aBlock[i];
    }
  }
}

static int fiWriterFlushOvfl(FiWriter *p, u32 *pPgno){
  u32 pgno;                       /* Page number for new overflow ptr page */
  int rc;

  /* Set the "number of entries" field */
  btPutU32(p->aTrunk, p->nOvfl);

  /* Write the page to disk */
  pgno = (p->nPgPerBlk * (p->iBlk-1) + 1) + p->nWrite;
  p->nWrite++;
  p->nAlloc++;
  rc = sqlite4BtPagerRawWrite(p->db->pPager, pgno, p->aTrunk);

  btPutU32(&p->aTrunk[4], pgno);
  if( pPgno ) *pPgno = pgno;

  return rc;
}


static int fiWriterFreeOverflow(FiWriter *p, FiCursor *pCsr){
  const void *pKey;               /* Buffer containing current key for pCsr */
  int nKey;                       /* Size of pKey in bytes */
  int rc;
  int i;

  rc = btCsrKey(&pCsr->aSub[pCsr->iBt].csr, &pKey, &nKey);
  for(i=pCsr->iBt+1; i<pCsr->nBt && rc==SQLITE4_OK; i++){
    BtCursor *pSub = &pCsr->aSub[i].csr;
    if( pSub->nPg ){
      const void *pSKey;            /* Current key for pSub */
      int nSKey;                    /* Size of pSKey in bytes */
      rc = btCsrKey(pSub, &pSKey, &nSKey);
      if( rc==SQLITE4_OK 
       && 0==btKeyCompare(pKey, nKey, pSKey, nSKey) 
       && btCsrOverflow(pSub)
      ){
        u32 pgno = sqlite4BtPagePgno(pSub->apPage[pSub->nPg-1]);
        int iCell = pSub->aiCell[pSub->nPg-1];

        if( p->aTrunk==0 ){
          rc = btNewBuffer(p->db, &p->aTrunk);
          if( rc==SQLITE4_OK ) memset(p->aTrunk, 0, 8);
        }else if( p->nOvflPerPage==p->nOvfl ){
          rc = fiWriterFlushOvfl(p, 0);
        }
        if( rc==SQLITE4_OK ){
          btPutU32(&p->aTrunk[8 + p->nOvfl*8], pgno);
          btPutU32(&p->aTrunk[8 + p->nOvfl*8 + 4], iCell);
          p->nOvfl++;
        }
      }
    }
  }

  return rc;
}

/*
** Argument aBuf points to a buffer containing a leaf cell. This function
** returns a pointer to the key prefix embedded within the cell. Before
** returning, *pnKey is set to the size of the key prefix in bytes.
*/
static const u8 *btKeyPrefixFromCell(const u8 *aBuf, int *pnKey){

      const void *pCell;          /* Cell to copy to output */
      int nCell;                  /* Size of cell in bytes */

      /* Read the current cell from the input and push it to the output. */
      fiCsrCell(&fcsr, &pCell, &nCell);
      rc = fiWriterAdd(&writer, pCell, nCell);
      if( rc==BT_BLOCKFULL ){
        int nOvflSaved = writer.nOvfl;
        u8 *aTrunkSaved = writer.aTrunk;
        rc = fiWriterFlushAll(&writer);
        fiWriterInit(db, &s, &writer, &rc);
        writer.nOvfl = nOvflSaved;
        writer.aTrunk = aTrunkSaved;
      }else if( rc==SQLITE4_OK ){
        rc = fiWriterFreeOverflow(&writer, &fcsr);
        if( rc==SQLITE4_OK ) rc = fiCsrStep(&fcsr);
      }
    }

    /* Assuming no error has occurred, update the serialized BtSchedule
    ** structure stored in buffer aSched[]. The caller will write this
    ** buffer to the database file as page (pHdr->iSRoot).  */
    if( rc==BT_BLOCKFULL || rc==SQLITE4_NOTFOUND ){

        assert( pCsr->nPg>0 );
        s.iNextPg = sqlite4BtPagePgno(pCsr->apPage[pCsr->nPg-1]);
        s.iNextCell = pCsr->aiCell[pCsr->nPg-1];
      }
      s.iFreeList = 0;
      s.eBusy = BT_SCHEDULE_DONE;

      assert( s.iFreeList==0 );
      if( writer.aTrunk ){
        rc = fiWriterFlushOvfl(&writer, &s.iFreeList);
      }
      if( rc==SQLITE4_OK ){
        rc = fiWriterFlushAll(&writer);
      }
      if( rc==SQLITE4_OK ){
        btWriteSchedule(aSched, &s, &rc);
      }
    }

    fiWriterCleanup(&writer);
    fiCsrReset(&fcsr);

  return rc;
}

static void btControlTransactionDone(bt_db *db, int iCtx){
  if( iCtx==0 ) sqlite4BtCommit(db, 0);
}

static int btCheckForPageLeaks(bt_db*, sqlite4_buffer*);

static int btControlInfo(bt_db *db, bt_info *pInfo){
  int rc = SQLITE4_OK;

  switch( pInfo->eType ){
    case BT_INFO_PAGEDUMP_ASCII:
    case BT_INFO_PAGEDUMP: {
      int iCtx;                   /* ControlTransaction() context */

      rc = btControlTransaction(db, &iCtx);
      if( rc==SQLITE4_OK ){
        rc = sqlite4BtPagerHdrdump(db->pPager, &pInfo->output);
        btControlTransactionDone(db, iCtx);
      }
      break;
    }

    case BT_INFO_BLOCK_FREELIST: 
    case BT_INFO_PAGE_FREELIST: {
      int iCtx;                   /* ControlTransaction() context */
      rc = btControlTransaction(db, &iCtx);
      if( rc==SQLITE4_OK ){
        BtDbHdr *pHdr = sqlite4BtPagerDbhdr(db->pPager);
        u32 iFirst = (
            pInfo->eType==BT_INFO_BLOCK_FREELIST?pHdr->iFreeBlk:pHdr->iFreePg
        );
        rc = btFreelistToAscii(db, iFirst, &pInfo->output);
        btControlTransactionDone(db, iCtx);
      }
      break;
    }

    case BT_INFO_PAGE_LEAKS: {
      int iCtx;                   /* ControlTransaction() context */
      rc = btControlTransaction(db, &iCtx);
      if( rc==SQLITE4_OK ){
        rc = btCheckForPageLeaks(db, &pInfo->output);
        btControlTransactionDone(db, iCtx);
      }
      break;
    }

    default: {
      rc = SQLITE4_ERROR;
      break;
    }
  }
  sqlite4_buffer_append(&pInfo->output, "\0", 1);
  return rc;
}

int sqlite4BtControl(bt_db *db, int op, void *pArg){
  int rc = SQLITE4_OK;

  switch( op ){

      break;
    }
  }

  return rc;
}

#ifndef NDEBUG

static void markBlockAsUsed(
  bt_db *db,
  u32 iBlk,
  u8 *aUsed
){
  if( iBlk ){
    BtDbHdr *pHdr = sqlite4BtPagerDbhdr(db->pPager);
    int nPgPerBlk = (pHdr->blksz / pHdr->pgsz);
    int i;

    for(i=0; i<nPgPerBlk; i++){
      u32 iPg = (iBlk-1) * nPgPerBlk + 1 + i;
      assert( aUsed[iPg]==0 || aUsed[iPg]==1 );
      aUsed[iPg] = 1;
    }
  }
}

/*
** Iterate through the b-tree with root page iRoot. For each page used
** by the b-tree, set the corresponding entry in the aUsed[] array.
*/
static void assert_pages_used(
  bt_db *db,                      /* Database handle */
  u32 iRoot,                      /* Root page of b-tree to iterate through */
  int *pRc                        /* IN/OUT: Error code */
){
  if( *pRc==SQLITE4_OK && iRoot ){
    BtDbHdr *pHdr = sqlite4BtPagerDbhdr(db->pPager);
    int nPgPerBlk = (pHdr->blksz / pHdr->pgsz);
    int bMeta = (iRoot==pHdr->iMRoot);
    BtLock *pLock = (BtLock*)(db->pPager);
    u8 *aUsed = pLock->aUsed;
    int rc;
    BtCursor csr;

    btCsrSetup(db, iRoot, &csr);
    rc = btCsrEnd(&csr, 0);
    while( rc==SQLITE4_OK ){
      rc = btCsrBuffer(&csr, 1);
      if( bMeta && rc==SQLITE4_OK ){
        u8 *aVal;
        int nVal;

        btCsrKey(&csr, (const void**)&aVal, &nVal);
        if( nVal!=sizeof(aSummaryKey) || memcmp(aVal, aSummaryKey, nVal) ){
          u32 iSubRoot;
          u32 iBlk;
          int i;
          btCsrData(&csr, 0, 4, (const void**)&aVal, &nVal);
          iSubRoot = btGetU32(aVal);
          iBlk = (iSubRoot / nPgPerBlk) + 1;
          assert_pages_used(db, iSubRoot, &rc);
          markBlockAsUsed(db, iBlk, aUsed);
        }
      }
      if( rc==SQLITE4_OK ) rc = btCsrStep(&csr, 1);
    }
  }
}

static void assert_freelist_pages_used(
  bt_db *db, 
  int bBlocklist,                 /* True to examine free-block list */
  u8 *aUsed, 
  int *pRc
){
  if( *pRc==SQLITE4_OK ){
    BtDbHdr *pHdr = sqlite4BtPagerDbhdr(db->pPager);
    u32 iTrunk = (bBlocklist ? pHdr->iFreeBlk : pHdr->iFreePg);
    int rc = SQLITE4_OK;

    while( rc==SQLITE4_OK && iTrunk ){
      BtPage *pPg = 0;
      rc = sqlite4BtPageGet(db->pPager, iTrunk, &pPg);
      if( rc==SQLITE4_OK ){
        int i;
        u32 nFree;
        u8 *aData = sqlite4BtPageData(pPg);

        nFree = btGetU32(aData);
        for(i=0; i<nFree; i++){
          u32 pgno = btGetU32(&aData[8 + i*4]);
          if( bBlocklist ){
            markBlockAsUsed(db, pgno, aUsed);
          }else{
            aUsed[pgno]++;
          }
        }

        iTrunk = btGetU32(&aData[4]);
        sqlite4BtPageRelease(pPg);
      }
    }

    *pRc = rc;
  }
}

static void assert_schedule_page_used(bt_db *db, u8 *aUsed, int *pRc){
  BtDbHdr *pHdr = sqlite4BtPagerDbhdr(db->pPager);
  if( *pRc==SQLITE4_OK && pHdr->iSRoot!=0 ){
    int nPgPerBlk = (pHdr->blksz / pHdr->pgsz);
    BtPage *pPg = 0;
    int rc;

    rc = sqlite4BtPageGet(db->pPager, pHdr->iSRoot, &pPg);
    if( rc==SQLITE4_OK ){
      BtSchedule s;
      int i;

      btReadSchedule(db, sqlite4BtPageData(pPg), &s);
      sqlite4BtPageRelease(pPg);

      assert( s.eBusy!=BT_SCHEDULE_BUSY || s.aRoot[0]==0 );
      if( s.eBusy!=BT_SCHEDULE_EMPTY ){
        for(i=0; rc==SQLITE4_OK && i<array_size(s.aBlock); i++){
          markBlockAsUsed(db, s.aBlock[i], aUsed);
        }
      }
    }

    *pRc = rc;
  }
}

/*
** Check that all pages in the database file are accounted for (not leaked).
** If any problems are detected, append a description of them to the buffer
** passed as the second argument.
*/
static int btCheckForPageLeaks(
  bt_db *db,                      /* Database handle */
  sqlite4_buffer *pBuf            /* Write error messages here */
){
  BtLock *pLock = (BtLock*)(db->pPager);

  static int nCall = 0;
  nCall++;

  if( pLock->aUsed==0 ){
    int rc;
    int iCtx;                     /* ControlTransaction() context */
    rc = btControlTransaction(db, &iCtx);

    if( rc==SQLITE4_OK ){
      BtDbHdr *pHdr = sqlite4BtPagerDbhdr(db->pPager);
      u8 *aUsed;
      int i;

      aUsed = btMalloc(db, pHdr->nPg, &rc);
      if( aUsed ) memset(aUsed, 0, pHdr->nPg);
      aUsed[0] = 1;                 /* Page 1 is always in use */
      pLock->aUsed = &aUsed[-1];

      /* The scheduled-merge page, if it is allocated */
      assert_schedule_page_used(db, pLock->aUsed, &rc);

      /* Walk the main b-tree */
      assert_pages_used(db, pHdr->iRoot, &rc);

      /* Walk the meta-tree */
      assert_pages_used(db, pHdr->iMRoot, &rc);

      /* Walk the free-page list */
      assert_freelist_pages_used(db, 0, pLock->aUsed, &rc);

      /* The free-block list */
      assert_freelist_pages_used(db, 1, pLock->aUsed, &rc);

      for(i=0; i<pHdr->nPg; i++){
        if( aUsed[i]!=1 ){
          sqlite4BtBufAppendf(
              pBuf, "refcount on page %d is %d\n", i+1, (int)aUsed[i]
          );
        }
      }
      btFree(db, aUsed);
      btControlTransactionDone(db, iCtx);
      pLock->aUsed = 0;
    }
  }

  return SQLITE4_OK;
}
#endif

/*
** Request a reference to page pgno of the database.
*/
int sqlite4BtPageGet(BtPager *p, u32 pgno, BtPage **ppPg){
  int rc = SQLITE4_OK;            /* Return code */
  BtPage *pRet;                   /* Returned page handle */

  if( p->btl.aUsed ){
    p->btl.aUsed[pgno]++;
  }

  /* Search the cache for an existing page. */
  pRet = btHashSearch(p, pgno);

  /* If the page is not in the cache, load it from disk */
  if( pRet==0 ){
    rc = btAllocatePage(p, &pRet);

  return rc;
}

/*
** Add page pgno to the free-page list. If argument pPg is not NULL, then
** it is a reference to page pgno.
*/
static int btFreelistAdd(
  BtPager *p,                     /* Pager object */
  int bBlock,                     /* True if pgno is actually a block number */
  u32 pgno
){
  BtDbHdr *pHdr = sqlite4BtLogDbhdr(p->pLog);
  int rc = SQLITE4_OK;
  int bDone = 0;
  u32 *piFirst = (bBlock ? &pHdr->iFreeBlk : &pHdr->iFreePg);

  /* Check if there is space on the first free-list trunk page. If so,
  ** add the new entry to it. Set variable bDone to indicate that the
  ** page has already been added to the free-list. */
  if( *piFirst ){
    BtPage *pTrunk;
    rc = sqlite4BtPageGet(p, *piFirst, &pTrunk);
    if( rc==SQLITE4_OK ){
      const int nMax = ((pHdr->pgsz - 8) / 4);
      u8 *aData = pTrunk->aData;
      int nFree = (int)sqlite4BtGetU32(aData);

      if( nFree<nMax ){
        rc = sqlite4BtPageWrite(pTrunk);
        if( rc==SQLITE4_OK ){
          sqlite4BtPutU32(&pTrunk->aData[8 + nFree*4], pgno);
          sqlite4BtPutU32(pTrunk->aData, nFree+1);
          bDone = 1;
          sqlite4BtDebugPageFree((BtLock*)p, bBlock, "free-list-leaf", pgno);
        }
      }
      sqlite4BtPageRelease(pTrunk);
    } 
  }

  /* If no error has occurred but the page number has not yet been added 
  ** to the free-list, this page becomes the first trunk in the list.  */
  if( rc==SQLITE4_OK && bDone==0 ){
    BtPage *pTrunk = 0;

    if( bBlock ){
      rc = sqlite4BtPageAllocate(p, &pTrunk);
    }else{
      rc = sqlite4BtPageGet(p, pgno, &pTrunk);
      if( rc==SQLITE4_OK ) rc = sqlite4BtPageWrite(pTrunk);
    }
    if( rc==SQLITE4_OK ){



      sqlite4BtPagerDbhdrDirty(p);
      sqlite4BtPutU32(&pTrunk->aData[0], 0);
      sqlite4BtPutU32(&pTrunk->aData[4], *piFirst);
      *piFirst = pTrunk->pgno;
      sqlite4BtDebugPageFree((BtLock*)p, 0, "free-list-trunk", pTrunk->pgno);
    }
    sqlite4BtPageRelease(pTrunk);
    if( rc==SQLITE4_OK && bBlock ){
      rc = btFreelistAdd(p, 1, pgno);
    }

  }

  return rc;
}

/*
** Attempt to allocate a page from the free-list.
*/
static int btFreelistAlloc(
  BtPager *p,                     /* Pager object */
  int bBlock,                     /* True to allocate a block (not a page) */
  u32 *pPgno                      /* OUT: Page or block number */
){
  BtDbHdr *pHdr = sqlite4BtLogDbhdr(p->pLog);
  int rc = SQLITE4_OK;
  u32 *piFirst = (bBlock ? &pHdr->iFreeBlk : &pHdr->iFreePg);
  u32 pgno = 0;

  assert( *pPgno==0 );

  while( *piFirst && pgno==0 && rc==SQLITE4_OK ){
    BtPage *pTrunk = 0;
    rc = sqlite4BtPageGet(p, *piFirst, &pTrunk);
    if( rc==SQLITE4_OK ){
      rc = sqlite4BtPageWrite(pTrunk);
    }
    if( rc==SQLITE4_OK ){
      u8 *aData = pTrunk->aData;
      u32 nFree = sqlite4BtGetU32(aData);
      if( nFree>0 ){
        pgno = sqlite4BtGetU32(&aData[8 + 4*(nFree-1)]);
        sqlite4BtPutU32(aData, nFree-1);


        sqlite4BtDebugPageAlloc((BtLock*)p, "free-list", pgno);
      }else{
        u32 iNext = sqlite4BtGetU32(&aData[4]);
        sqlite4BtPagerDbhdrDirty(p);
        
        if( bBlock ){
          rc = btFreelistAdd(p, 0, *piFirst);
        }else{
          pgno = *piFirst;
          sqlite4BtDebugPageAlloc((BtLock*)p, "free-list-trunk", pgno);

        }
        *piFirst = iNext;
      }
    }

    sqlite4BtPageRelease(pTrunk);
  }

  *pPgno = pgno;
  return rc;
}

/*
** Decrement the refcount on page pPg. Also, indicate that page pPg is
** no longer in use.
*/
int sqlite4BtPageTrim(BtPage *pPg){
  int rc;                         /* Return code */
  rc = btFreelistAdd(pPg->pPager, 0, pPg->pgno);
  sqlite4BtPageRelease(pPg);
  return rc;
}

/*
** Page number pgno is no longer in use.
*/

  BtPage *pPg = 0;
  int rc;
  u32 pgno = 0;

  /* Find the page number of the new page. There are two ways a page may
  ** be allocated - from the free-list or by appending it to the end of
  ** the database file. */ 
  rc = btFreelistAlloc(p, 0, &pgno);
  if( rc==SQLITE4_OK && pgno==0 ){
    pgno = p->pHdr->nPg+1;
    sqlite4BtDebugPageAlloc((BtLock*)p, "end-of-file", pgno);
  }

  rc = sqlite4BtPageGet(p, pgno, &pPg);
  if( rc==SQLITE4_OK ){

int sqlite4BtBlockAllocate(BtPager *p, int nBlk, u32 *aiBlk){
  int rc = SQLITE4_OK;
  BtDbHdr *pHdr = p->pHdr;
  int nPgPerBlk = (pHdr->blksz / pHdr->pgsz);
  int i;

  for(i=0; rc==SQLITE4_OK && i<nBlk; i++){
    u32 iBlk = 0;
    u32 iRoot;
    u32 iFree;

    rc = btFreelistAlloc(p, 1, &iBlk);
    if( rc==SQLITE4_OK && iBlk==0 ){

      /* Figure out the next block in the file. And its root (first) page. */
      iBlk = 1 + (pHdr->nPg + nPgPerBlk - 1) / nPgPerBlk;
      iRoot = (iBlk-1) * nPgPerBlk + 1;
      assert( iBlk>0 );

      for(iFree = pHdr->nPg+1; rc==SQLITE4_OK && iFree<iRoot; iFree++){
        rc = sqlite4BtPageTrimPgno(p, iFree);
      }

      pHdr->nPg = iBlk * nPgPerBlk;

    }

    aiBlk[i] = iBlk;
  }

  return rc;
}

/*
** Trim a block.
*/
int sqlite4BtBlockTrim(BtPager *p, u32 iBlk){
  return btFreelistAdd(p, 1, iBlk);
}

/*
** Return the current page number of the argument page reference.
*/
u32 sqlite4BtPagePgno(BtPage *pPg){
  return pPg->pgno;
}