SQLite

/*
** 2004 April 6
**
** The author disclaims copyright to this source code.  In place of
** a legal notice, here is a blessing:
**
**    May you do good and not evil.
**    May you find forgiveness for yourself and forgive others.
**    May you share freely, never taking more than you give.
**
*************************************************************************
** $Id: btree.c,v 1.117 2004/05/09 00:40:52 drh Exp $
**
** This file implements a external (disk-based) database using BTrees.
** For a detailed discussion of BTrees, refer to
**
**     Donald E. Knuth, THE ART OF COMPUTER PROGRAMMING, Volume 3:
**     "Sorting And Searching", pages 473-480. Addison-Wesley
**     Publishing Company, Reading, Massachusetts.

  u8 isInit;                     /* True if previously initialized */
  u8 idxShift;                   /* True if Cell indices have changed */
  u8 isOverfull;                 /* Some aCell[] do not fit on page */
  u8 intKey;                     /* True if intkey flag is set */
  u8 leaf;                       /* True if leaf flag is set */
  u8 zeroData;                   /* True if zero data flag is set */
  u8 hdrOffset;                  /* 100 for page 1.  0 otherwise */
  u8 needRelink;                 /* True if need to run relinkCellList() */
  int idxParent;                 /* Index in pParent->aCell[] of this node */
  int nFree;                     /* Number of free bytes on the page */
  int nCell;                     /* Number of entries on this page */
  int nCellAlloc;                /* Number of slots allocated in aCell[] */
  unsigned char **aCell;         /* Pointer to start of each cell */
  struct Btree *pBt;             /* Pointer back to BTree structure */

  maxPayload = pPage->pBt->maxLocal;
  if( nPayload>maxPayload ){
    nPayload = maxPayload + 4;
  }
  return n + nPayload;
}

/*
** Do sanity checking on a page.  Throw an exception if anything is
** not right.
**
** This routine is used for internal error checking only.  It is omitted
** from most builds.
*/
#if defined(BTREE_DEBUG) && !defined(NDEBUG) && 0
static void _pageIntegrity(MemPage *pPage){
  int pageSize;
  u8 *data;
  int i, idx, c, pc, hdr, nFree;
  u8 used[MX_PAGE_SIZE];

  pageSize = pPage->pBt->pageSize;
  assert( pPage->aData==&((unsigned char*)pPage)[-pageSize] );
  hdr = pPage->hdrOffset;
  assert( hdr==(pPage->pgno==1 ? 100 : 0) );
  assert( pPage->pgno==sqlite3pager_pagenumber(pPage->aData) );
  c = pPage->aData[hdr];
  if( pPage->isInit ){
    assert( pPage->leaf == ((c & PTF_LEAF)!=0) );
    assert( pPage->zeroData == ((c & PTF_ZERODATA)!=0) );
    assert( pPage->intKey == ((c & PTF_INTKEY)!=0) );
  }
  data = pPage->aData;
  memset(used, 0, pageSize);
  for(i=0; i<hdr+10-pPage->leaf*4; i++) used[i] = 1;
  nFree = 0;
  pc = get2byte(&data[hdr+1]);
  while( pc ){
    int size;
    assert( pc>0 && pc<pageSize-4 );
    size = get2byte(&data[pc+2]);
    assert( pc+size<=pageSize );
    nFree += size;
    for(i=pc; i<pc+size; i++){
      assert( used[i]==0 );
      used[i] = 1;
    }
    pc = get2byte(&data[pc]);
  }
  assert( pPage->isInit==0 || pPage->nFree==nFree+data[hdr+5] );
  idx = 0;
  pc = get2byte(&data[hdr+3]);
  while( pc ){
    int size;
    assert( pPage->isInit==0 || idx<pPage->nCell );
    assert( pc>0 && pc<pageSize-4 );
    assert( pPage->isInit==0 || pPage->aCell[idx]==&data[pc] );
    size = cellSize(pPage, &data[pc]);
    assert( pc+size<=pageSize );
    for(i=pc; i<pc+size; i++){
      assert( used[i]==0 );
      used[i] = 1;
    }
    pc = get2byte(&data[pc]);
    idx++;
  }
  assert( idx==pPage->nCell );
  nFree = 0;
  for(i=0; i<pageSize; i++){
    assert( used[i]<=1 );
    if( used[i]==0 ) nFree++;
  }
  assert( nFree==data[hdr+5] );
}
#define pageIntegrity(X) _pageIntegrity(X)
#else
# define pageIntegrity(X)
#endif

/*
** Defragment the page given.  All Cells are moved to the
** beginning of the page and all free space is collected 
** into one big FreeBlk at the end of the page.
*/
static void defragmentPage(MemPage *pPage){
  int pc, i, n, addr;
  int start, hdr, size;
  int leftover;
  unsigned char *oldPage;
  unsigned char newPage[MX_PAGE_SIZE];

  assert( sqlite3pager_iswriteable(pPage->aData) );
  assert( pPage->pBt!=0 );
  assert( pPage->pBt->pageSize <= MX_PAGE_SIZE );
  assert( !pPage->needRelink );
  assert( !pPage->isOverfull );
  oldPage = pPage->aData;
  hdr = pPage->hdrOffset;
  addr = 3+hdr;
  n = 6+hdr;
  if( !pPage->leaf ){
    n += 4;
  }
  memcpy(&newPage[hdr], &oldPage[hdr], n-hdr);
  start = n;
  pc = get2byte(&oldPage[addr]);
  i = 0;
  while( pc>0 ){
    assert( n<pPage->pBt->pageSize );
    size = cellSize(pPage, &oldPage[pc]);
    memcpy(&newPage[n], &oldPage[pc], size);
    put2byte(&newPage[addr],n);
    assert( pPage->aCell[i]==&oldPage[pc] );
    pPage->aCell[i++] = &oldPage[n];
    addr = n;
    n += size;

    pc = get2byte(&oldPage[pc]);
  }
  assert( i==pPage->nCell );
  leftover = pPage->pBt->pageSize - n;
  assert( leftover>=0 );
  assert( pPage->nFree==leftover );
  if( leftover<4 ){

  int pageSize;
  int sumCell = 0;       /* Total size of all cells */

  assert( pPage->pBt!=0 );
  assert( pParent==0 || pParent->pBt==pPage->pBt );
  assert( pPage->pgno==sqlite3pager_pagenumber(pPage->aData) );
  assert( pPage->aData == &((unsigned char*)pPage)[-pPage->pBt->pageSize] );
  assert( pPage->pParent==0 || pPage->pParent==pParent );

  if( pPage->pParent==0 && pParent!=0 ){
    pPage->pParent = pParent;

    sqlite3pager_ref(pParent->aData);
  }
  if( pPage->isInit ) return SQLITE_OK;
  pPage->nCell = pPage->nCellAlloc = 0;
  assert( pPage->hdrOffset==(pPage->pgno==1 ? 100 : 0) );
  hdr = pPage->hdrOffset;
  data = pPage->aData;
  c = data[hdr];
  pPage->intKey = (c & PTF_INTKEY)!=0;
  pPage->zeroData = (c & PTF_ZERODATA)!=0;
  pPage->leaf = (c & PTF_LEAF)!=0;
  pPage->isOverfull = 0;
  pPage->needRelink = 0;
  pPage->idxShift = 0;
  pageSize = pPage->pBt->pageSize;

  /* Initialize the cell count and cell pointers */
  pc = get2byte(&data[hdr+3]);
  while( pc>0 ){
    if( pc>=pageSize ) return SQLITE_CORRUPT;

  /* Sanity check:  Cells and freespace and header must sum to the size
  ** a page. */
  if( sumCell+pPage->nFree+hdr+10-pPage->leaf*4 != pageSize ){
    return SQLITE_CORRUPT;
  }

  pPage->isInit = 1;
  pageIntegrity(pPage);
  return SQLITE_OK;
}

/*
** Set up a raw page so that it looks like a database page holding
** no entries.
*/
static void zeroPage(MemPage *pPage, int flags){
  unsigned char *data = pPage->aData;
  Btree *pBt = pPage->pBt;
  int hdr = pPage->hdrOffset;
  int first;

  assert( sqlite3pager_pagenumber(data)==pPage->pgno );
  assert( &data[pBt->pageSize] == (unsigned char*)pPage );
  assert( sqlite3pager_iswriteable(data) );
  memset(&data[hdr], 0, pBt->pageSize - hdr);
  data[hdr] = flags;
  first = hdr + 6 + 4*((flags&PTF_LEAF)==0);
  put2byte(&data[hdr+1], first);
  put2byte(&data[first+2], pBt->pageSize - first);
  sqliteFree(pPage->aCell);
  pPage->aCell = 0;
  pPage->nCell = 0;
  pPage->nCellAlloc = 0;
  pPage->nFree = pBt->pageSize - first;
  pPage->intKey = (flags & PTF_INTKEY)!=0;
  pPage->leaf = (flags & PTF_LEAF)!=0;
  pPage->zeroData = (flags & PTF_ZERODATA)!=0;
  pPage->hdrOffset = hdr;
  pPage->isOverfull = 0;
  pPage->needRelink = 0;
  pPage->idxShift = 0;
  pPage->isInit = 1;
  pageIntegrity(pPage);
}

/*
** Get a page from the pager.  Initialize the MemPage.pBt and
** MemPage.aData elements if needed.
*/
static int getPage(Btree *pBt, Pgno pgno, MemPage **ppPage){

/*
** This routine is called when the reference count for a page
** reaches zero.  We need to unref the pParent pointer when that
** happens.
*/
static void pageDestructor(void *pData){
  MemPage *pPage = (MemPage*)&((char*)pData)[SQLITE_PAGE_SIZE];
  assert( pPage->isInit==0 || pPage->needRelink==0 );
  if( pPage->pParent ){
    MemPage *pParent = pPage->pParent;
    pPage->pParent = 0;
    releasePage(pParent);
  }
  sqliteFree(pPage->aCell);
  pPage->aCell = 0;

/*
** Invalidate all cursors
*/
static void invalidateCursors(Btree *pBt){
  BtCursor *pCur;
  for(pCur=pBt->pCursor; pCur; pCur=pCur->pNext){
    MemPage *pPage = pCur->pPage;
    if( pPage /* && !pPage->isInit */ ){
      pageIntegrity(pPage);
      releasePage(pPage);
      pCur->pPage = 0;
      pCur->isValid = 0;
      pCur->status = SQLITE_ABORT;
    }
  }
}

#ifdef SQLITE_TEST
/*
** Print debugging information about all cursors to standard output.
*/
void sqlite3BtreeCursorList(Btree *pBt){
  BtCursor *pCur;
  for(pCur=pBt->pCursor; pCur; pCur=pCur->pNext){
    MemPage *pPage = pCur->pPage;
    char *zMode = pCur->wrFlag ? "rw" : "ro";
    printf("CURSOR %08x rooted at %4d(%s) currently at %d.%d%s\n",
       (int)pCur, pCur->pgnoRoot, zMode,
       pPage ? pPage->pgno : 0, pCur->idx,
       pCur->isValid ? "" : " eof"
    );
  }
}
#endif

/*
** Rollback the transaction in progress.  All cursors will be
** invalided by this operation.  Any attempt to use a cursor
** that was open at the beginning of this operation will result
** in an error.
**
** This will release the write lock on the database file.  If there

  MemPage *pPage;
  unsigned char *cell;

  if( !pCur->isValid ){
    *pSize = 0;
  }else{
    pPage = pCur->pPage;
    pageIntegrity(pPage);
    assert( pPage!=0 );
    assert( pCur->idx>=0 && pCur->idx<pPage->nCell );
    cell = pPage->aCell[pCur->idx];
    cell += 2;   /* Skip the offset to the next cell */
    if( !pPage->leaf ){
      cell += 4;  /* Skip the child pointer */
    }

  u64 nData, nKey;
  int maxLocal, ovflSize;

  assert( pCur!=0 && pCur->pPage!=0 );
  assert( pCur->isValid );
  pBt = pCur->pBt;
  pPage = pCur->pPage;
  pageIntegrity(pPage);
  assert( pCur->idx>=0 && pCur->idx<pPage->nCell );
  aPayload = pPage->aCell[pCur->idx];
  aPayload += 2;  /* Skip the next cell index */
  if( !pPage->leaf ){
    aPayload += 4;  /* Skip the child pointer */
  }
  if( pPage->zeroData ){

  if( !pCur->isValid ){
    return 0;
  }
  assert( pCur->pPage!=0 );
  assert( pCur->idx>=0 && pCur->idx<pCur->pPage->nCell );
  pBt = pCur->pBt;
  pPage = pCur->pPage;
  pageIntegrity(pPage);
  assert( pCur->idx>=0 && pCur->idx<pPage->nCell );
  assert( pPage->intKey==0 );
  aPayload = pPage->aCell[pCur->idx];
  aPayload += 2;  /* Skip the next cell index */
  if( !pPage->leaf ){
    aPayload += 4;  /* Skip the child pointer */
  }

  if( !pCur->isValid ){
    return pCur->status ? pCur->status : SQLITE_INTERNAL;
  }
  pPage = pCur->pPage;
  assert( pPage!=0 );
  assert( pPage->isInit );
  pageIntegrity(pPage);
  if( pPage->zeroData ){
    *pSize = 0;
  }else{
    assert( pCur->idx>=0 && pCur->idx<pPage->nCell );
    cell = pPage->aCell[pCur->idx];
    cell += 2;   /* Skip the offset to the next cell */
    if( !pPage->leaf ){

  MemPage *pNewPage;
  MemPage *pOldPage;
  Btree *pBt = pCur->pBt;

  assert( pCur->isValid );
  rc = getAndInitPage(pBt, newPgno, &pNewPage, pCur->pPage);
  if( rc ) return rc;
  pageIntegrity(pNewPage);
  pNewPage->idxParent = pCur->idx;
  pOldPage = pCur->pPage;
  pOldPage->idxShift = 0;
  releasePage(pOldPage);
  pCur->pPage = pNewPage;
  pCur->idx = 0;
  if( pNewPage->nCell<1 ){

** for the table rooted on page 1, sometime the real root page
** is empty except for the right-pointer.  In such cases the
** virtual root page is the page that the right-pointer of page
** 1 is pointing to.
*/
static int isRootPage(MemPage *pPage){
  MemPage *pParent = pPage->pParent;
  if( pParent==0 ) return 1;
  if( pParent->pgno>1 ) return 0;
  if( get2byte(&pParent->aData[pParent->hdrOffset+3])==0 ) return 1;
  return 0;
}

/*
** Move the cursor up to the parent page.
**
** pCur->idx is set to the cell index that contains the pointer

  MemPage *pPage;
  int idxParent;

  assert( pCur->isValid );
  pPage = pCur->pPage;
  assert( pPage!=0 );
  assert( !isRootPage(pPage) );
  pageIntegrity(pPage);
  pParent = pPage->pParent;
  assert( pParent!=0 );
  pageIntegrity(pParent);
  idxParent = pPage->idxParent;
  sqlite3pager_ref(pParent->aData);
  oldPgno = pPage->pgno;
  releasePage(pPage);
  pCur->pPage = pParent;
  assert( pParent->idxShift==0 );
  if( pParent->idxShift==0 ){

  rc = getAndInitPage(pBt, pCur->pgnoRoot, &pRoot, 0);
  if( rc ){
    pCur->isValid = 0;
    return rc;
  }
  releasePage(pCur->pPage);
  pageIntegrity(pRoot);
  pCur->pPage = pRoot;
  pCur->idx = 0;
  if( pRoot->nCell==0 && !pRoot->leaf ){
    Pgno subpage;
    assert( pRoot->pgno==1 );
    subpage = get4byte(&pRoot->aData[pRoot->hdrOffset+6]);
    assert( subpage>0 );

  for(;;){
    int lwr, upr;
    Pgno chldPg;
    MemPage *pPage = pCur->pPage;
    int c = -1;  /* pRes return if table is empty must be -1 */
    lwr = 0;
    upr = pPage->nCell-1;
    pageIntegrity(pPage);
    while( lwr<=upr ){
      void *pCellKey;
      u64 nCellKey;
      pCur->idx = (lwr+upr)/2;
      sqlite3BtreeKeySize(pCur, &nCellKey);
      if( pPage->intKey ){
        if( nCellKey<nKey ){

static void reparentPage(Btree *pBt, Pgno pgno, MemPage *pNewParent, int idx){
  MemPage *pThis;
  unsigned char *aData;

  if( pgno==0 ) return;
  assert( pBt->pPager!=0 );
  aData = sqlite3pager_lookup(pBt->pPager, pgno);
  if( aData ){
    pThis = (MemPage*)&aData[pBt->pageSize];
    if( pThis->isInit ){
      if( pThis->pParent!=pNewParent ){
        if( pThis->pParent ) sqlite3pager_unref(pThis->pParent->aData);
        pThis->pParent = pNewParent;
        if( pNewParent ) sqlite3pager_ref(pNewParent->aData);
      }
      pThis->idxParent = idx;
    }
    sqlite3pager_unref(aData);
  }
}

/*
** Change the pParent pointer of all children of pPage to point back
** to pPage.

** Remove the i-th cell from pPage.  This routine effects pPage only.
** The cell content is not freed or deallocated.  It is assumed that
** the cell content has been copied someplace else.  This routine just
** removes the reference to the cell from pPage.
**
** "sz" must be the number of bytes in the cell.
**
** Try to maintain the integrity of the linked list of cells.  But if
** the cell being inserted does not fit on the page, this will not be
** possible.  If the linked list is not maintained, then just update
** pPage->aCell[] and set the pPage->needRelink flag so that we will

** know to rebuild the linked list later.
*/
static void dropCell(MemPage *pPage, int idx, int sz){
  int j, pc;
  u8 *data;
  assert( idx>=0 && idx<pPage->nCell );
  assert( sz==cellSize(pPage, pPage->aCell[idx]) );
  assert( sqlite3pager_iswriteable(pPage->aData) );
  assert( pPage->aCell[idx]>=pPage->aData );
  assert( pPage->aCell[idx]<&pPage->aData[pPage->pBt->pageSize-sz] );
  data = pPage->aData;
  pc = Addr(pPage->aCell[idx]) - Addr(data);
  assert( pc>pPage->hdrOffset && pc+sz<=pPage->pBt->pageSize );
  freeSpace(pPage, pc, sz);
  for(j=idx; j<pPage->nCell-1; j++){
    pPage->aCell[j] = pPage->aCell[j+1];
  }
  pPage->nCell--;
  if( !pPage->isOverfull && !pPage->needRelink ){
    u8 *pPrev;
    if( idx==0 ){
      pPrev = &data[pPage->hdrOffset+3];
    }else{
      pPrev = pPage->aCell[idx-1];
    }
    if( idx<pPage->nCell ){
      pc = Addr(pPage->aCell[idx]) - Addr(data);
    }else{
      pc = 0;
    }
    put2byte(pPrev, pc);
    pageIntegrity(pPage);
  }else{
    pPage->needRelink = 1;
  }
  pPage->idxShift = 1;
}

/*
** Insert a new cell on pPage at cell index "i".  pCell points to the
** content of the cell.
**
** If the cell content will fit on the page, then put it there.  If it
** will not fit, then just make pPage->aCell[i] point to the content
** and set pPage->isOverfull.  
**
** Try to maintain the integrity of the linked list of cells.  But if
** the cell being inserted does not fit on the page, this will not be
** possible.  If the linked list is not maintained, then just update
** pPage->aCell[] and set the pPage->needRelink flag so that we will

** know to rebuild the linked list later.
*/
static void insertCell(MemPage *pPage, int i, unsigned char *pCell, int sz){
  int idx, j;
  assert( i>=0 && i<=pPage->nCell );
  assert( sz==cellSize(pPage, pCell) );
  assert( sqlite3pager_iswriteable(pPage->aData) );
  idx = pPage->needRelink ? 0 : allocateSpace(pPage, sz);
  resizeCellArray(pPage, pPage->nCell+1);
  for(j=pPage->nCell; j>i; j--){
    pPage->aCell[j] = pPage->aCell[j-1];
  }
  pPage->nCell++;
  if( idx<=0 ){
    pPage->isOverfull = 1;
    pPage->aCell[i] = pCell;
  }else{
    u8 *data = pPage->aData;
    memcpy(&data[idx], pCell, sz);
    pPage->aCell[i] = &data[idx];
  }
  if( !pPage->isOverfull && !pPage->needRelink ){
    u8 *pPrev;
    int pc;
    if( i==0 ){
      pPrev = &pPage->aData[pPage->hdrOffset+3];
    }else{
      pPrev = pPage->aCell[i-1];
    }
    pc = get2byte(pPrev);
    put2byte(pPrev, idx);
    put2byte(pPage->aCell[i], pc);
    pageIntegrity(pPage);
  }else{
    pPage->needRelink = 1;
  }
  pPage->idxShift = 1;
}

/*
** Rebuild the linked list of cells on a page so that the cells
** occur in the order specified by the pPage->aCell[] array.  
** Invoke this routine once to repair damage after one or more
** invocations of either insertCell() or dropCell().
*/
static void relinkCellList(MemPage *pPage){
  int i, idxFrom;
  assert( sqlite3pager_iswriteable(pPage->aData) );
  if( !pPage->needRelink ) return;
  idxFrom = pPage->hdrOffset+3;
  for(i=0; i<pPage->nCell; i++){
    int idx = Addr(pPage->aCell[i]) - Addr(pPage->aData);
    assert( idx>pPage->hdrOffset && idx<pPage->pBt->pageSize );
    put2byte(&pPage->aData[idxFrom], idx);
    idxFrom = idx;
  }
  put2byte(&pPage->aData[idxFrom], 0);
  pPage->needRelink = 0;
}

/*
** GCC does not define the offsetof() macro so we'll have to do it
** ourselves.
*/
#ifndef offsetof
#define offsetof(STRUCTURE,FIELD) ((int)((char*)&((STRUCTURE*)0)->FIELD))
#endif

/*
** Move the content of the page at pFrom over to pTo.  The pFrom->aCell[]
** pointers that point into pFrom->aData[] must be adjusted to point
** into pTo->aData[] instead.  But some pFrom->aCell[] entries might
** not point to pFrom->aData[].  Those are unchanged.
**
** Over this operation completes, the meta data for pFrom is zeroed.
*/
static void movePage(MemPage *pTo, MemPage *pFrom){
  uptr from, to;
  int i;
  int pageSize;
  int ofst;

  assert( pTo->hdrOffset==0 );
  ofst = pFrom->hdrOffset;

    uptr x = Addr(pTo->aCell[i]);
    if( x>from && x<from+pageSize-ofst ){
      *((uptr*)&pTo->aCell[i]) = x + to - from;
    }
  }
}

/*
** For debugging...
*/
#if 1
# define TRACE(X)   if( pager3_refinfo_enable ) printf X
#else
# define TRACE(X)
#endif

/*
** The following parameters determine how many adjacent pages get involved
** in a balancing operation.  NN is the number of neighbors on either side
** of the page that participate in the balancing operation.  NB is the
** total number of pages that participate, including the target page and
** NN neighbors on either side.
**

  int idx;                     /* Index of pPage in pParent->aCell[] */
  int nxDiv;                   /* Next divider slot in pParent->aCell[] */
  int rc;                      /* The return code */
  int leafCorrection;          /* 4 if pPage is a leaf.  0 if not */
  int usableSpace;             /* Bytes in pPage beyond the header */
  int pageFlags;               /* Value of pPage->aData[0] */
  int subtotal;                /* Subtotal of bytes in cells on one page */
  MemPage *extraUnref = 0;     /* Unref this page if not zero */
  MemPage *apOld[NB];          /* pPage and up to two siblings */
  Pgno pgnoOld[NB];            /* Page numbers for each page in apOld[] */
  MemPage *apCopy[NB];         /* Private copies of apOld[] pages */
  MemPage *apNew[NB+1];        /* pPage and up to NB siblings after balancing */
  Pgno pgnoNew[NB+1];          /* Page numbers for each page in apNew[] */
  int idxDiv[NB];              /* Indices of divider cells in pParent */
  u8 *apDiv[NB];               /* Divider cells in pParent */

  }

  /*
  ** Find the parent of the page to be balanced.  If there is no parent,
  ** it means this page is the root page and special rules apply.
  */
  pParent = pPage->pParent;
  TRACE(("BALANCE: begin page %d\n", pPage->pgno));
  if( pParent==0 ){
    Pgno pgnoChild;
    MemPage *pChild;
    assert( pPage->isInit );
    if( pPage->nCell==0 ){
      if( pPage->leaf ){
        /* The table is completely empty */
        relinkCellList(pPage);
        TRACE(("BALANCE: empty table\n"));
      }else{
        /* The root page is empty but has one child.  Transfer the
        ** information from that one child into the root page if it 
        ** will fit.  This reduces the depth of the BTree by one.
        **
        ** If the root page is page 1, it has less space available than
        ** its child (due to the 100 byte header that occurs at the beginning

            zeroPage(pPage, pChild->aData[0]);
            resizeCellArray(pPage, pChild->nCell);
            for(i=0; i<pChild->nCell; i++){
              insertCell(pPage, i, pChild->aCell[i], 
                        cellSize(pChild, pChild->aCell[i]));
            }
            freePage(pChild);
            TRACE(("BALANCE: child %d transfer to page 1\n", pChild->pgno));
          }else{
            /* The child has more information that will fit on the root.
            ** The tree is already balanced.  Do nothing. */
            TRACE(("BALANCE: child %d will not fit on page 1\n", pChild->pgno));
          }
        }else{
          memcpy(pPage, pChild, pBt->pageSize);
          pPage->isInit = 0;
          pPage->pParent = 0;
          rc = initPage(pPage, 0);
          assert( rc==SQLITE_OK );
          freePage(pChild);
          TRACE(("BALANCE: transfer child %d into root\n", pChild->pgno));
        }
        reparentChildPages(pPage);
        releasePage(pChild);
      }
      return SQLITE_OK;
    }
    if( !pPage->isOverfull ){
      /* It is OK for the root page to be less than half full.
      */
      relinkCellList(pPage);
      TRACE(("BALANCE:  Root page is underfull but that is ok\n"));
      return SQLITE_OK;
    }
    /*
    ** If we get to here, it means the root page is overfull.
    ** When this happens, Create a new child page and copy the
    ** contents of the root into the child.  Then make the root
    ** page an empty page with rightChild pointing to the new
    ** child.  Then fall thru to the code below which will cause
    ** the overfull child page to be split.
    */
    rc = allocatePage(pBt, &pChild, &pgnoChild, pPage->pgno);
    if( rc ) return rc;
    assert( sqlite3pager_iswriteable(pChild->aData) );
    movePage(pChild, pPage);
    assert( pChild->aData[0]==pPage->aData[pPage->hdrOffset] );
    pChild->pParent = pPage;
    sqlite3pager_ref(pPage->aData);
    pChild->idxParent = 0;

    pChild->isOverfull = 1;
    zeroPage(pPage, pChild->aData[0] & ~PTF_LEAF);
    put4byte(&pPage->aData[pPage->hdrOffset+6], pChild->pgno);
    pParent = pPage;
    pPage = pChild;
    extraUnref = pChild;
    TRACE(("BALANCE: Copy root into %d and blance\n", pPage->pgno));
  }
  rc = sqlite3pager_write(pParent->aData);
  if( rc ) return rc;
  assert( pParent->isInit );
  
  /*
  ** Find the cell in the parent page whose left child points back

  ** The rest of this function will use data from the copies rather
  ** that the original pages since the original pages will be in the
  ** process of being overwritten.
  */
  for(i=0; i<nOld; i++){
    MemPage *p = apCopy[i] = (MemPage*)&aCopy[i+1][-sizeof(MemPage)];
    p->aData = &((u8*)p)[-pBt->pageSize];
    p->aCell = 0;
    p->hdrOffset = 0;
    movePage(p, apOld[i]);
  }

  /*
  ** Load pointers to all cells on sibling pages and the divider cells
  ** into the local apCell[] array.  Make copies of the divider cells
  ** into aTemp[] and remove the the divider Cells from pParent.
  **

    MemPage *pOld = apCopy[i];
    for(j=0; j<pOld->nCell; j++){
      apCell[nCell] = pOld->aCell[j];
      szCell[nCell] = cellSize(pOld, apCell[nCell]);
      nCell++;
    }
    if( i<nOld-1 ){
      szCell[nCell] = cellSize(pParent, apDiv[i]);
      memcpy(aTemp[i], apDiv[i], szCell[nCell]);
      apCell[nCell] = &aTemp[i][leafCorrection];
      dropCell(pParent, nxDiv, szCell[nCell]);
      szCell[nCell] -= leafCorrection;
      assert( get4byte(&aTemp[i][2])==pgnoOld[i] );
      if( !pOld->leaf ){
        assert( leafCorrection==0 );
        /* The right pointer of the child page pOld becomes the left
        ** pointer of the divider cell */
        memcpy(&apCell[nCell][2], &pOld->aData[pOld->hdrOffset+6], 4);
      }else{
        assert( leafCorrection==4 );

  /*
  ** Allocate k new pages.  Reuse old pages where possible.
  */
  assert( pPage->pgno>1 );
  pageFlags = pPage->aData[0];
  for(i=0; i<k; i++){
    MemPage *pNew;
    if( i<nOld ){
      pNew = apNew[i] = apOld[i];
      pgnoNew[i] = pgnoOld[i];
      apOld[i] = 0;
      sqlite3pager_write(pNew->aData);
    }else{
      rc = allocatePage(pBt, &pNew, &pgnoNew[i], pgnoNew[i-1]);
      if( rc ) goto balance_cleanup;
      apNew[i] = pNew;
    }
    nNew++;
    zeroPage(pNew, pageFlags);

  }

  /* Free any old pages that were not reused as new pages.
  */
  while( i<nOld ){
    rc = freePage(apOld[i]);
    if( rc ) goto balance_cleanup;
    releasePage(apOld[i]);
    apOld[i] = 0;
    i++;
  }

  /*
  ** Put the new pages in accending order.  This helps to
  ** keep entries in the disk file in order so that a scan

    reparentChildPages(apNew[i]);
  }
  reparentChildPages(pParent);

  /*
  ** balance the parent page.
  */
  assert( pPage->isInit );
  assert( pParent->isInit );
  pageIntegrity(pPage);
  rc = balance(pParent);
  

  /*
  ** Cleanup before returning.
  */
balance_cleanup:
  for(i=0; i<nOld; i++){
    releasePage(apOld[i]);
    if( apCopy[i] ){

      sqliteFree(apCopy[i]->aCell);
    }
  }
  for(i=0; i<nNew; i++){
    releasePage(apNew[i]);
  }
  releasePage(pParent);
  releasePage(extraUnref);
  TRACE(("BALANCE: Finished with %d\n", pPage->pgno));
  return rc;
}

/*
** This routine checks all cursors that point to the same table
** as pCur points to.  If any of those cursors were opened with
** wrFlag==0 then this routine returns SQLITE_LOCKED.  If all

    ** next Cell after the one to be deleted is guaranteed to exist and
    ** to be a leaf so we can use it.
    */
    BtCursor leafCur;
    unsigned char *pNext;
    int szNext;
    int notUsed;
    unsigned char tempbuf[4];
    getTempCursor(pCur, &leafCur);
    rc = sqlite3BtreeNext(&leafCur, &notUsed);
    if( rc!=SQLITE_OK ){
      if( rc!=SQLITE_NOMEM ) rc = SQLITE_CORRUPT;
      return rc;
    }
    rc = sqlite3pager_write(leafCur.pPage->aData);
    if( rc ) return rc;
    dropCell(pPage, pCur->idx, cellSize(pPage, pCell));
    pNext = leafCur.pPage->aCell[leafCur.idx];
    szNext = cellSize(leafCur.pPage, pNext);
    memcpy(tempbuf, &pNext[-2], 4);
    put4byte(&pNext[-2], pgnoChild);
    insertCell(pPage, pCur->idx, &pNext[-4], szNext+4);
    rc = balance(pPage);
    if( rc ) return rc;
    memcpy(&pNext[-2], tempbuf, 4);
    dropCell(leafCur.pPage, leafCur.idx, szNext);
    rc = balance(leafCur.pPage);
    releaseTempCursor(&leafCur);
  }else{
    dropCell(pPage, pCur->idx, cellSize(pPage, pCell));
    rc = balance(pPage);
  }

  }
  hdr = pPage->hdrOffset;
  data = pPage->aData;
  c = data[hdr];
  pPage->intKey = (c & PTF_INTKEY)!=0;
  pPage->zeroData = (c & PTF_ZERODATA)!=0;
  pPage->leaf = (c & PTF_LEAF)!=0;
  printf("PAGE %d:  flags=0x%02x  frag=%d   parent=%d\n", pgno,
    data[hdr], data[hdr+5], 
    (pPage->isInit && pPage->pParent) ? pPage->pParent->pgno : 0);
  i = 0;
  assert( hdr == (pgno==1 ? 100 : 0) );
  idx = get2byte(&data[hdr+3]);
  while( idx>0 && idx<=pBt->pageSize ){
    u64 nData, nKey;
    int nHeader;
    Pgno child;

**   aResult[4] =  Number of free bytes on this page
**   aResult[5] =  Number of free blocks on the page
**   aResult[6] =  Page number of the left child of this entry
**   aResult[7] =  Page number of the right child for the whole page
**
** This routine is used for testing and debugging only.
*/
int sqlite3BtreeCursorInfo(BtCursor *pCur, int *aResult){
  int cnt, idx;
  MemPage *pPage = pCur->pPage;

  pageIntegrity(pPage);
  assert( pPage->isInit );
  aResult[0] = sqlite3pager_pagenumber(pPage->aData);
  assert( aResult[0]==pPage->pgno );
  aResult[1] = pCur->idx;
  aResult[2] = pPage->nCell;
  if( pCur->idx>=0 && pCur->idx<pPage->nCell ){
    aResult[3] = cellSize(pPage, pPage->aCell[pCur->idx]);

# 2001 September 15
#
# The author disclaims copyright to this source code.  In place of
# a legal notice, here is a blessing:
#
#    May you do good and not evil.
#    May you find forgiveness for yourself and forgive others.
#    May you share freely, never taking more than you give.
#
#***********************************************************************
# This file implements regression tests for SQLite library.  The
# focus of this script is btree database backend
#
# $Id: btree.test,v 1.21 2004/05/09 00:40:52 drh Exp $


set testdir [file dirname $argv0]
source $testdir/tester.tcl

# Basic functionality.  Open and close a database.
#

#btree_page_dump $::b1 2
do_test btree-7.15 {
  lindex [btree_pager_stats $::b1] 1
} {1}

# Check to see that data on overflow pages work correctly.
#
#btree_page_dump $::b1 1
do_test btree-8.1 {
  set data "*** This is a very long key "
  while {[string length $data]<1234} {append data $data}
  set ::data $data
  btree_insert $::c1 2020 $data
} {}
#btree_page_dump $::b1 1

} $::data
do_test btree-8.4 {
  btree_delete $::c1
} {}
do_test btree-8.4.1 {
  lindex [btree_get_meta $::b1] 0
} [expr {int(([string length $::data]-238+1019)/1020)}]



do_test btree-8.5 {
  set data "*** This is an even longer key "
  while {[string length $data]<2000} {append data $data}
  append data END
  set ::data $data
  btree_insert $::c1 2030 $data
} {}

#btree_page_dump $::b1 2
do_test btree-8.21 {
  lindex [btree_get_meta $::b1] 0
} {2}
do_test btree-8.22 {
  lindex [btree_pager_stats $::b1] 1
} {2}
do_test btree-8.23 {
  btree_close_cursor $::c1
  btree_drop_table $::b1 2






  set ::c1 [btree_cursor $::b1 2 1]
  lindex [btree_get_meta $::b1] 0
} {5}
do_test btree-8.24 {
  lindex [btree_pager_stats $::b1] 1
} {2}
#btree_pager_ref_dump $::b1




# Check page splitting logic
#
do_test btree-9.1 {
  for {set i 1} {$i<=19} {incr i} {
    set key [format %03d $i]
    set data "*** $key *** $key *** $key *** $key ***"

  }] "*** [format %03d $i] ***"
}
#btree_pager_ref_dump $::b1
do_test btree-9.6 {
  btree_close_cursor $::c1
  lindex [btree_pager_stats $::b1] 1
} {1}


do_test btree-9.7 {
  btree_rollback $::b1
  lindex [btree_pager_stats $::b1] 1
} {0}




# Create a tree of depth two.  That is, there is a single divider entry
# on the root pages and two leaf pages.  Then delete the divider entry
# see what happens.
#
do_test btree-10.1 {
  btree_begin_transaction $::b1

  btree_next $::c1
  btree_data $::c1
} $::data
do_test btree-12.12 {
  btree_next $::c1
  btree_key $::c1
} {402}


#do_test btree-13.1 {
#  btree_integrity_check $::b1 1 2
#} {}

# To Do:
#
#   1.  Do some deletes from the 3-layer tree
#   2.  Commit and reopen the database
#   3.  Read every 15th entry and make sure it works
#   4.  Implement btree_sanity and put it throughout this script