SQLite4  Check-in [590e0410b1]

  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.
*/
typedef struct BtSchedule BtSchedule;
struct BtSchedule {
  u32 bBusy;                      /* True if page is busy */
  u32 iAge;                       /* Age of input segments */
  u32 iMaxLevel;                  /* Maximum level of input segments */
  u32 iOutLevel;                  /* Level at which to write output */
  u32 aBlock[32];                 /* Allocated blocks */

  u32 iNextPg;                    /* Page that contains next input key */
  u32 iNextCell;                  /* Cell that contains next input key */
  u32 nUsed;                      /* Number of output blocks actually written */
  u32 iFreeList;                  /* First page of new free-list (if any) */
};

/*************************************************************************
** Interface to bt_pager.c functionality.
*/
typedef struct BtPage BtPage;
typedef struct BtPager BtPager;

/*

int sqlite4BtPageRelease(BtPage*);
void sqlite4BtPageReference(BtPage*);

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

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

typedef struct BtCursor BtCursor;
typedef struct FiCursor FiCursor;

struct bt_db {
  sqlite4_env *pEnv;              /* SQLite environment */
  BtPager *pPager;                /* Underlying page-based database */
  bt_cursor *pAllCsr;             /* List of all open cursors */
  int nMinMerge;
  int nScheduleAlloc;
  int bFastInsertOp;              /* Set by CONTROL_FAST_INSERT_OP */
};

typedef struct BtOvfl BtOvfl;
struct BtOvfl {
  int nKey;
  int nVal;

struct FiCursor {
  bt_cursor base;                 /* Base cursor class */
  int iBt;                        /* Current sub-tree (or -1 for EOF) */
  int nBt;                        /* Number of entries in aBt[] array */
  BtCursor *aBt;                  /* Array of sub-tree cursors */
};

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

#ifndef btErrorBkpt
int btErrorBkpt(int rc){
  static int error_cnt = 0;
  error_cnt++;
  return rc;
}
#endif

}
#define btPutU32(x,y) sqlite4BtPutU32(x,y)

/*
** Allocate a new database handle.
*/
int sqlite4BtNew(sqlite4_env *pEnv, int nExtra, bt_db **ppDb){
  static const int MIN_MERGE = 4;
  static const int SCHEDULE_ALLOC = 4;

  bt_db *db = 0;                  /* New database object */
  BtPager *pPager = 0;            /* Pager object for this database */
  int nReq;                       /* Bytes of space required for bt_db object */
  int rc;                         /* Return code */

  nReq = sizeof(bt_db);
  rc = sqlite4BtPagerNew(pEnv, nExtra + nReq, &pPager);
  if( rc==SQLITE4_OK ){
    db = (bt_db*)sqlite4BtPagerExtra(pPager);
    db->pPager = pPager;
    db->pEnv = pEnv;

    db->nMinMerge = MIN_MERGE;
    db->nScheduleAlloc = SCHEDULE_ALLOC;
  }

  *ppDb = db;
  return rc;
}

/*

static void btPageToAscii(
  u32 pgno,                       /* Page number */
  int bAscii,                     /* True to print keys and values as ASCII */
  BtPager *pPager,                /* Pager object (or NULL) */
  u8 *aData, int nData,           /* Buffer containing page data */
  sqlite4_buffer *pBuf            /* Output buffer */
){
  BtDbHdr *pHdr = sqlite4BtPagerDbhdr(pPager);
  int i;
  int nCell = (int)btCellCount(aData, nData);
  u8 flags = btFlags(aData);      /* Page flags */

  sqlite4BtBufAppendf(pBuf, "Page %d: ", pgno);

  if( pgno==pHdr->iSRoot ){
    int i;
    BtSchedule s;
    sqlite4BtBufAppendf(pBuf, "(schedule page) ");
    memcpy(&s, aData, sizeof(s));

    sqlite4BtBufAppendf(pBuf, "bBusy=%d ", (int)s.bBusy);
    sqlite4BtBufAppendf(pBuf, "iAge=%d ", (int)s.iAge);
    sqlite4BtBufAppendf(pBuf, "iMaxLevel=%d ", (int)s.iMaxLevel);
    sqlite4BtBufAppendf(pBuf, "iOutLevel=%d ", (int)s.iOutLevel);
    sqlite4BtBufAppendf(pBuf, "blocks=(");
    for(i=0; s.aBlock[i] && i<array_size(s.aBlock); i++){
      sqlite4BtBufAppendf(pBuf, "%s%d", i==0 ? "" : " ", (int)s.aBlock[i]);
    }
    sqlite4BtBufAppendf(pBuf, ") ");

    sqlite4BtBufAppendf(pBuf, "iNextPg=%d ", (int)s.iNextPg);
    sqlite4BtBufAppendf(pBuf, "iNextCell=%d ", (int)s.iNextCell);
    sqlite4BtBufAppendf(pBuf, "nUsed=%d ", (int)s.nUsed);
    sqlite4BtBufAppendf(pBuf, "iFreeList=%d", (int)s.iFreeList);
  }else{
    sqlite4BtBufAppendf(pBuf, "nCell=%d ", nCell);
    sqlite4BtBufAppendf(pBuf, "iFree=%d ", (int)btFreeOffset(aData, nData));
    sqlite4BtBufAppendf(pBuf, "flags=%d ", (int)btFlags(aData));
    if( btFlags(aData) & BT_PGFLAGS_INTERNAL ){
      sqlite4BtBufAppendf(pBuf, "rchild=%d ", (int)btGetU32(&aData[1]));
    }
    sqlite4BtBufAppendf(pBuf, "cell-offsets=(");
    for(i=0; i<nCell; i++){
      u8 *ptr = btCellPtrFind(aData, nData, i);
      sqlite4BtBufAppendf(pBuf, "%s%d", i==0?"":" ", (int)btGetU16(ptr));
    }
    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 */
      u8 *pVal = 0;       /* Buffer containing value. */

      pCell = btCellFind(aData, nData, i);
      sqlite4BtBufAppendf(pBuf, "  Cell %d: ", i);

      pCell += sqlite4BtVarintGet32(pCell, &nKey);
      pKey = pCell;
      pCell += nKey;
      btBufferAppendBlob(pBuf, bAscii, pKey, nKey);

      if( flags & BT_PGFLAGS_INTERNAL ){
        sqlite4BtBufAppendf(pBuf, "  child=%d ", (int)btGetU32(pCell));
      }else{
        sqlite4BtBufAppendf(pBuf, "  ");
        pCell += sqlite4BtVarintGet32(pCell, &nVal);
        if( nVal>=2 ){
          nVal -= 2;
          pVal = pCell;
        }else{
          sqlite4BtBufAppendf(pBuf, "delete-key");
        }
      }
      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) ){
            sqlite4BtBufAppendf(pBuf, "  [summary]");
          }else{
            u32 iAge = btGetU32(&pKey[0]);
            u32 iLevel = ~btGetU32(&pKey[4]);
            u32 iBlk = btGetU32(pVal);
            sqlite4BtBufAppendf(pBuf, "  [age=%d level=%d block=%d]", 
                (int)iAge, (int)iLevel, (int)iBlk
                );
          }
        }
      }
      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");
}

#ifndef NDEBUG
#include <stdio.h>

        BtCursor *pSub = pCsr->aBt;

        /* Loop through the set of f-tree levels */
        btCsrSetup(db, pHdr->iMRoot, &mcsr);
        for(rc=btCsrEnd(&mcsr, 0); rc==SQLITE4_OK; rc=btCsrStep(&mcsr, 1)){
          const void *pV; int nV; /* Value associated with meta-tree entry */
          u32 iBlk;               /* Block containing sub-tree */

          btCsrKey(&mcsr, &pV, &nV);
          if( nV==sizeof(aSummaryKey) && 0==memcmp(pV, aSummaryKey, nV) ){
            rc = SQLITE4_NOTFOUND;
            break;
          }

          sqlite4BtCsrData(&mcsr.base, 0, 4, &pV, &nV);
          iBlk = sqlite4BtGetU32((const u8*)pV);
          btCsrReset(pSub, 1);
          btCsrSetup(db, btBlockToRoot(pHdr, iBlk), pSub);
          rc = btCsrSeek(pSub, pK, nK, BT_SEEK_EQ, BT_CSRSEEK_SEEK);
          if( rc!=SQLITE4_NOTFOUND && rc!=SQLITE4_INEXACT ){

        u32 iBlk;                 /* Block containing sub-tree */
        BtCursor *pSub;           /* Sub-cursor for this block */

        rc = fiCsrAllocateSubs(db, pCsr, iSub+1);
        if( rc!=SQLITE4_OK ) break;
        pSub = &pCsr->aBt[iSub];
        iSub++;

        btCsrKey(&mcsr, &pV, &nV);
        if( nV==sizeof(aSummaryKey) && 0==memcmp(pV, aSummaryKey, nV) ){
          rc = SQLITE4_NOTFOUND;
          break;
        }

        sqlite4BtCsrData(&mcsr.base, 0, 4, &pV, &nV);
        iBlk = sqlite4BtGetU32((const u8*)pV);

        btCsrReset(pSub, 1);
        btCsrSetup(db, btBlockToRoot(pHdr, iBlk), pSub);
        rc = btCsrSeek(pSub, pK, nK, eSeek, BT_CSRSEEK_SEEK);
        if( rc==SQLITE4_OK ){
          bMatch = 1;
        }else if( rc==SQLITE4_INEXACT ){
          bHit = 1;
        }else if( rc!=SQLITE4_NOTFOUND ){
          break;
        }
      }
      assert( rc!=SQLITE4_OK && rc!=SQLITE4_INEXACT );
      btCsrReset(&mcsr, 1);

      if( rc==SQLITE4_NOTFOUND ){
        pCsr->base.flags |= (eSeek==BT_SEEK_GE ? CSR_NEXT_OK : CSR_PREV_OK);
        rc = fiCsrSetCurrent(pCsr);
        if( rc==SQLITE4_OK ){
          BtCursor *pSub = &pCsr->aBt[pCsr->iBt];
          if( fiCsrIsDelete(pSub) ){

  /* Loop through the set of f-tree levels. Open a cursor on each. */
  btCsrSetup(db, pHdr->iMRoot, &mcsr);
  for(rc=btCsrEnd(&mcsr, 0); rc==SQLITE4_OK; rc=btCsrStep(&mcsr, 1)){
    const void *pV; int nV;   /* Value associated with meta-tree entry */
    u32 iBlk;                 /* Block containing sub-tree */
    BtCursor *pSub;           /* Sub-cursor for this block */

    btCsrKey(&mcsr, &pV, &nV);
    if( nV==sizeof(aSummaryKey) && 0==memcmp(pV, aSummaryKey, nV) ){
      rc = SQLITE4_NOTFOUND;
      break;
    }

    rc = fiCsrAllocateSubs(db, pCsr, iSub+1);
    if( rc!=SQLITE4_OK ) break;
    pSub = &pCsr->aBt[iSub];
    iSub++;

    sqlite4BtCsrData(&mcsr.base, 0, 4, &pV, &nV);
    iBlk = sqlite4BtGetU32((const u8*)pV);
    btCsrSetup(db, btBlockToRoot(pHdr, iBlk), pSub);

    rc = btCsrEnd(pSub, bLast);
    if( rc!=SQLITE4_OK && rc!=SQLITE4_NOTFOUND ) break;
  }
  btCsrReset(&mcsr, 1);

  if( rc==SQLITE4_NOTFOUND ){
    pCsr->base.flags &= ~(CSR_NEXT_OK | CSR_PREV_OK);
    pCsr->base.flags |= (bLast ? CSR_PREV_OK : CSR_NEXT_OK);
    rc = fiCsrSetCurrent(pCsr);
    if( rc==SQLITE4_OK && fiCsrIsDelete(&pCsr->aBt[pCsr->iBt]) ){
      rc = fiCsrStep(pCsr);

    }
  }

  return rc;
}

static int btFastInsertRoot(bt_db *db, BtDbHdr *pHdr, u32 *piRoot);
static int btScheduleMerge(bt_db *db);

static int btReplaceEntry(
  bt_db *db,                      /* Database handle */
  u32 iRoot,                      /* Root page of b-tree to update */
  const void *pK, int nK,         /* Key to insert */
  const void *pV, int nV          /* Value to insert. (nV<0) -> delete */
){

          /* Insert the new KV pair into the current leaf. */
          rc = btInsertAndBalance(&csr, 1, &kv);

          /* Unless this is a block-full error, break out of the loop */
          if( rc!=BT_BLOCKFULL ) break;
          assert( iRoot==0 );

          /* Try to schedule a merge operation */
          rc = btScheduleMerge(db);

          if( rc==SQLITE4_OK ){
            rc = btFastInsertRoot(db, pHdr, &iRootPg);
          }
          if( rc==SQLITE4_OK ){
            btCsrReset(&csr, 1);
            btCsrSetup(db, iRootPg, &csr);
            rc = btCsrSeek(&csr, pK, nK, BT_SEEK_GE, BT_CSRSEEK_UPDATE);
          }
        }while( rc==SQLITE4_NOTFOUND || rc==SQLITE4_INEXACT );
      }

      *paKey = &aK[8];
      *pnKey = nK-8;
    }
  }
  return rc;
}

static void *btMalloc(bt_db *db, int nByte, int *pRc){
  void *pRet = 0;
  if( *pRc==SQLITE4_OK ){
    pRet = sqlite4_malloc(db->pEnv, nByte);
    if( !pRet ) *pRc = btErrorBkpt(SQLITE4_NOMEM);
  }
  return pRet;
}
static void btFree(bt_db *db, void *p){
  sqlite4_free(db->pEnv, p);
}

static int fiLoadSummary(
  bt_db *db, 
  BtCursor *p, 
  const u8 **paSummary, 
  int *pnSummary
){
  static const u8 aZero[] = {0x00, 0x00, 0x00, 0x00, 0x00, 0x00};
  BtDbHdr *pHdr = sqlite4BtPagerDbhdr(db->pPager);
  int rc;

  btCsrSetup(db, pHdr->iMRoot, p);
  rc = btCsrSeek(
      p, aSummaryKey, sizeof(aSummaryKey), BT_SEEK_EQ, BT_CSRSEEK_SEEK
  );
  if( rc==SQLITE4_OK ){
    rc = btCsrData(p, 0, -1, (const void**)paSummary, pnSummary);
  }else if( rc==SQLITE4_NOTFOUND ){
    *paSummary = aZero;
    *pnSummary = sizeof(aZero);
    rc = SQLITE4_OK;
  }

  return rc;
}

static void btReadSummary(
  const u8 *aSum, int iAge, 
  u16 *piMinLevel,
  u16 *pnLevel,
  u16 *piMergeLevel
){
  *piMinLevel = btGetU16(&aSum[iAge * 6]);
  *pnLevel = btGetU16(&aSum[iAge * 6 + 2]);
  *piMergeLevel = btGetU16(&aSum[iAge * 6 + 4]);
}

static void btWriteSummary(
  u8 *aSum, int iAge,
  u16 iMinLevel,
  u16 nLevel,
  u16 iMergeLevel
){
  btPutU16(&aSum[iAge * 6], iMinLevel);
  btPutU16(&aSum[iAge * 6 + 2], nLevel);
  btPutU16(&aSum[iAge * 6 + 4], iMergeLevel);
}

/*
** Allocate a new level for a new age=0 segment. The new level is always
** one greater than the current largest age=0 level number.
*/
static int btAllocateNewLevel(
  bt_db *db, 
  BtDbHdr *pHdr, 
  u32 *piNext
){

  int rc;                         /* Return code */
  BtCursor csr;                   /* Cursor to read meta-table summary */
  int nByte;                      /* Size of buffer aByte[] */
  const u8 *aByte;                /* Summary data */
  u8 *aNew;                       /* New summary data */

  rc = fiLoadSummary(db, &csr, &aByte, &nByte);

  aNew = (u8*)btMalloc(db, nByte, &rc);
  if( rc==SQLITE4_OK ){
    u16 iMin, nLevel, iMerge;
    btReadSummary(aByte, 0, &iMin, &nLevel, &iMerge);

    memcpy(aNew, aByte, nByte);
    btPutU16(&aNew[2], nLevel+1);
    rc = btReplaceEntry(
        db, pHdr->iMRoot, aSummaryKey, sizeof(aSummaryKey), aNew, nByte
    );
    btFree(db, aNew);

    *piNext = (iMin + nLevel);
  }

  btCsrReset(&csr, 1);
  return rc;
}

static void btWriteSchedule(BtPage *pPg, BtSchedule *p, int *pRc){
  if( *pRc==SQLITE4_OK ){
    int rc = sqlite4BtPageWrite(pPg);
    if( rc==SQLITE4_OK ){
      /* TODO: convert values to big-endian format */
      u8 *aData = sqlite4BtPageData(pPg);
      memcpy(aData, p, sizeof(BtSchedule));
    }
    *pRc = rc;
  }
}
static int btReadSchedule(bt_db *db, BtPage *pPg, BtSchedule *p){
  /* TODO: convert values to big-endian format */
  u8 *aData = sqlite4BtPageData(pPg);
  memcpy(p, aData, sizeof(BtSchedule));
  return SQLITE4_OK;
}

static int btAllocateBlock(
  bt_db *db, 
  int nBlk,
  u32 *aiBlk
){
  return sqlite4BtBlockAllocate(db->pPager, nBlk, aiBlk);
}

/*
** This is a helper function for btScheduleMerge(). It determines the
** age and range of levels to be used as inputs by the merge (if any).
*/
static int btFindMerge(
  bt_db *db,                      /* Database handle */
  u32 *piAge,                     /* OUT: Age of input segments to merge */
  u32 *piMaxLevel,                /* OUT: Maximum input level value */
  u32 *piOutLevel                 /* OUT: Output level value */
){
  BtCursor csr;                   /* Cursor used to read summary record */
  int rc;                         /* Return code */
  const u8 *aSum;
  int nSum;

  rc = fiLoadSummary(db, &csr, &aSum, &nSum);
  if( rc==SQLITE4_OK ){
    int iAge;
    int iBestAge = -1;            /* Best age to merge levels from */
    int nBest = (db->nMinMerge-1);/* Number of levels merged at iBestAge */
    u16 iMin, nLevel, iMerge;     /* Summary of current age */

    rc = SQLITE4_NOTFOUND;
    for(iAge=0; iAge<(nSum/3); iAge++){
      btReadSummary(aSum, iAge, &iMin, &nLevel, &iMerge);
      if( iMerge ){
        int n = 1 + (iMerge-iMin);
        if( n>nBest ){
          *piMaxLevel = iMerge;
          *piAge = iAge;
          btReadSummary(aSum, iAge+1, &iMin, &nLevel, &iMerge);
          *piOutLevel = (iMin + nLevel - 1);
          rc = SQLITE4_OK;
        }
        break;
      }else{
        if( nLevel>nBest ){
          iBestAge = iAge;
          nBest = nLevel;
        }
      }
    }

    if( rc==SQLITE4_NOTFOUND && iBestAge>=0 ){
      u8 *aNew;
      int nByte = nSum;
      if( iAge+1>=(nSum/3) ) nByte += 6;

      rc = SQLITE4_OK;
      aNew = (u8*)btMalloc(db, nByte, &rc);
      if( rc==SQLITE4_OK ){
        BtDbHdr *pHdr = sqlite4BtPagerDbhdr(db->pPager);

        /* Create a copy of the summary record */
        memcpy(aNew, aSum, nSum);
        if( nByte>nSum ) btWriteSummary(aNew, iBestAge+1, 0, 0, 0);

        /* Find the input age and maximum level */
        btReadSummary(aSum, iBestAge, &iMin, &nLevel, &iMerge);
        *piMaxLevel = (u32)(iMin + nLevel - 1);
        *piAge = iBestAge;

        /* Find the output level */
        btReadSummary(aSum, iBestAge+1, &iMin, &nLevel, &iMerge);
        *piOutLevel = iMin + nLevel;

        /* Update the summary record for the output segment. */
        btWriteSummary(aNew, iBestAge+1, iMin, nLevel+1, iMerge);

        /* Write the updated summary record back to the db. */
        rc = btReplaceEntry(
             db, pHdr->iMRoot, aSummaryKey, sizeof(aSummaryKey), aNew, nByte
        );
      }
    }
  }

  btCsrReset(&csr, 1);
  return rc;
}


/*
** If possible, schedule a merge operation. 
**
** The merge operation is selected based on the following criteria:
**
**   * The more levels involved in the merge the better, and
**   * It is better to merge younger segments than older ones.
*/
static int btScheduleMerge(bt_db *db){

  BtDbHdr *pHdr = sqlite4BtPagerDbhdr(db->pPager);
  BtPage *pPg = 0;                /* Schedule page */
  u8 *aData;                      /* Schedule page data */

  int rc;
  u32 iAge;
  u32 iLvl;
  u32 iOutLvl;

  /* Find the schedule page. If there is no schedule page, allocate it now. */
  if( pHdr->iSRoot==0 ){
    rc = sqlite4BtPageAllocate(db->pPager, &pPg);
    if( rc==SQLITE4_OK ){
      u8 *aData = sqlite4BtPageData(pPg);
      memset(aData, 0, pHdr->pgsz);
      pHdr->iSRoot = sqlite4BtPagePgno(pPg);
    }
  }else{
    rc = sqlite4BtPageGet(db->pPager, pHdr->iSRoot, &pPg);
  }

  /* Check if the schedule page is busy. If so, no new merge may be 
  ** scheduled. If the schedule page is not busy, call btFindMerge() to
  ** figure out which levels should be scheduled for merge.  */
  if( rc==SQLITE4_OK ){
    aData = sqlite4BtPageData(pPg);
    if( btGetU32(aData) ){
      rc = SQLITE4_NOTFOUND;
    }else{
      rc = btFindMerge(db, &iAge, &iLvl, &iOutLvl);
    }
  }

  if( rc==SQLITE4_OK ){
    BtSchedule s;
    memset(&s, 0, sizeof(BtSchedule));

    s.bBusy = 1;
    s.iAge = iAge;
    s.iMaxLevel = iLvl;
    s.iOutLevel = iOutLvl;
    rc = btAllocateBlock(db, db->nScheduleAlloc, s.aBlock);

    btWriteSchedule(pPg, &s, &rc);
  }

  sqlite4BtPageRelease(pPg);
  if( rc==SQLITE4_NOTFOUND ) rc = SQLITE4_OK;
  return rc;
}

static int btFastInsertRoot(
  bt_db *db, 
  BtDbHdr *pHdr, 
  u32 *piRoot
){
  int rc = SQLITE4_OK;

  assert( db->bFastInsertOp );
  db->bFastInsertOp = 0;

  /* If the meta-tree has not been created, create it now. */
  if( pHdr->iMRoot==0 ){
    rc = btAllocateNewRoot(db, BT_PGFLAGS_METATREE, &pHdr->iMRoot);
  }

  /* If no writable sub-tree current exists, create one */ 
  if( rc==SQLITE4_OK && pHdr->iSubBlock==0 ){
    u32 iLevel;                   /* Level number for new sub-tree */
    u32 iSubBlock;                /* New block */

    rc = btAllocateNewLevel(db, pHdr, &iLevel);
    if( rc==SQLITE4_OK ){
      rc = btAllocateBlock(db, 1, &iSubBlock);
    }

    if( rc==SQLITE4_OK ){
      u8 aKey[8];
      u8 aVal[4];
      pHdr->iSubBlock = iSubBlock;
      pHdr->nSubPg = 1;           /* Root page is automatically allocated */

      /* The key for the new entry consists of the concatentation of two 
      ** 32-bit big-endian integers - the <age> and <level-no>. The age
      ** 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], iSubBlock);



      rc = btReplaceEntry(db, pHdr->iMRoot, aKey, 8, aVal, 4);

    }
  }

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

/*
** Insert a new key/value pair or replace an existing one.
**
** This function may modify either the b-tree or fast-insert-tree, depending

    case BT_INFO_PAGEDUMP: {
      int iCtx;                   /* ControlTransaction() context */
      rc = btControlTransaction(db, &iCtx);
      if( rc==SQLITE4_OK ){
        BtPage *pPg = 0;
        rc = sqlite4BtPageGet(db->pPager, pInfo->pgno, &pPg);
        if( rc==SQLITE4_OK ){
          BtPager *p = db->pPager;
          int bAscii = (pInfo->eType==BT_INFO_PAGEDUMP_ASCII);
          u8 *aData;
          int nData;
          aData = sqlite4BtPageData(pPg);
          nData = sqlite4BtPagerPagesize(p);
          btPageToAscii(pInfo->pgno, bAscii, p, aData, nData, &pInfo->output);
          sqlite4_buffer_append(&pInfo->output, "", 1);
          sqlite4BtPageRelease(pPg);
        }
        btControlTransactionDone(db, iCtx);
      }
      break;
    }

  fprintf(stderr, "allocated page %d\n", pgno);
#endif

  *ppPg = pPg;
  return rc;
}

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; i<nBlk; i++){
    u32 iBlk;
    u32 iRoot;
    u32 iFree;

    /* 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);
    }
    if( rc==SQLITE4_OK ){
      pHdr->nPg = iBlk * nPgPerBlk;
      aiBlk[i] = iBlk;
    }
  }

  return rc;
}

/*
** Return the current page number of the argument page reference.

  <li> Segment age (big-endian 32-bit integer).
  <li> Level number (ones-compliment of value as a big-endian 32-bit integer).
  <li> Separator key. All keys within the segment are guaranteed to be as
       large or larger than the separator key. The separator key may be
       zero bytes in size.
</ul>

<p> As well as the records that index all segments in the db, the meta-table
contains a single summary record. The key is four 0xFF bytes.

<p>For each age between 0 and the largest age in the db, inclusive:

  * current minimum level number (16-bits).
  * total number of levels.
  * maximum level in active merge (16-bits).

Store this as a separate record in the meta-table. Even if there segments with
age=31 in the db, this is still only 6*32=192 bytes.

<h3> Schedule-Page Format</h3>

<p>We might need more than one schedule page. Perhaps anyway... But say 
worry about this later on.

<p>The writer adds an entry to the schedule-tree. Schedule-tree entries
are always appended to the tree - the integer key is one greater than the

  <li><p> A flag to indicate that the merge has been performed.

  <li><p> A bitmask (or some other indication) indicating which of the output
          blocks were populated.

  <li><p> A pointer to a list of overflow pages that were freed, if any.

  <li><p> A pointer to the next key read from the input levels, if any.
</ul>

<p>Note the implication here: A writer may only modify the page if it currently
resides in the db file (not the log). The writer can tell if the schedule page
resides in the db file or the log by checking the "merge performed" flag.


<p>The page is organized as follows:

<ul>
  <li> Magic number used to identify the state of the page (big-endian u32):
       0x6a846607 if the merge has not yet been performed or 0x1a468c33 if
       it has.

  <li> Age of input segments (big-endian u32).

  <li> Max level of input segments (bit-endian u32). All segments of the
       specified age with a level equal to or less than this are considered
       inputs.

  <li> Level of output segments.

  <li> Number of allocated output blocks (big-endian u32) - <i>nBlock</i>.
       Maximum is (say) 32.

  <li> Block number of each allocated block (array of <i>nBlock</i> big-endian 
       u32 fields).
</ul>

<p>Then, after the merge has taken place:
<ul>
  <li> Page number of input page containing next key to read.
  <li> Cell number of next key to read within the identified page.
  <li> Number of input blocks used (input blocks are always used in the
       order they are specified).
  <li> First page of freed-page list.
</ul>



<h2 id=in-memory_tree>3.2. In-Memory Tree</h2>

<p>This design uses the same multi-version in-memory tree that LSM does. 
The tree structure may be located in heap or shared memory. If shared memory
is used then a separate file (*-shm2 or some such) is used for the data. The