SQLite4  Check-in [bca95ca536]

#endif

struct bt_db {
  sqlite4_env *pEnv;              /* SQLite environment */
  BtPager *pPager;                /* Underlying page-based database */
};

typedef struct BtOvfl BtOvfl;
struct BtOvfl {
  int nKey;
  int nVal;
  u8 *pBuf;
  int nBuf;
};

/*
** Database cursor handle.
*/
struct bt_cursor {
  bt_db *pDb;                     /* Database that owns this cursor */

  int nPg;                        /* Number of valid entries in apPage[] */
  int aiCell[BT_MAX_DEPTH];       /* Current cell of each apPage[] entry */
  BtPage *apPage[BT_MAX_DEPTH];   /* All pages from root to current leaf */


  BtOvfl ovfl;                    /* Overflow cache (see above) */
};

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

  }else{
    btCsrSetup(db, pCsr);
  }

  return rc;
}

static void btCsrReset(bt_cursor *pCsr, int bFreeBuffer){
  int i;
  for(i=0; i<pCsr->nPg; i++){
    sqlite4BtPageRelease(pCsr->apPage[i]);
  }
  if( bFreeBuffer ) sqlite4_free(pCsr->pDb->pEnv, pCsr->ovfl.pBuf);
  pCsr->nPg = 0;
}

int sqlite4BtCsrClose(bt_cursor *pCsr){
  if( pCsr ){
    btCsrReset(pCsr, 1);
    sqlite4_free(pCsr->pDb->pEnv, pCsr);
  }
  return SQLITE4_OK;
}

void *sqlite4BtCsrExtra(bt_cursor *pCsr){
  return (void*)&pCsr[1];

}

/*
** Move the cursor from the current page to the parent. Return 
** SQLITE4_NOTFOUND if the cursor already points to the root page,
** or SQLITE4_OK otherwise.
*/
static int btCsrAscend(bt_cursor *pCsr, int nLvl){
  int i;
  for(i=0; i<nLvl && ( pCsr->nPg>0 ); i++){
    pCsr->nPg--;
    sqlite4BtPageRelease(pCsr->apPage[pCsr->nPg]);
  }
  return (pCsr->nPg==0 ? SQLITE4_NOTFOUND : SQLITE4_OK);
}

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

}

static int btFreeContiguous(const u8 *aData, int nData){
  int nCell = btCellCount(aData, nData);
  return nData - btFreeOffset(aData, nData) - (3+nCell)*2;
}

static u8 btFlags(const u8 *aData){
  return aData[0];
}

static u8 *btCellFind(u8 *aData, int nData, int iCell){
  int iOff = btGetU16(&aData[nData - 6 - iCell*2 - 2]);
  return &aData[iOff];
}

#include <stdio.h>
static void printPage(FILE *f, u32 pgno, u8 *aData, int nData){
  int i;
  int nCell = (int)btCellCount(aData, nData);
  fprintf(f, "Page %d: ", pgno);
  fprintf(f, "nCell=%d ", nCell);
  fprintf(f, "iFree=%d ", (int)btFreeOffset(aData, nData));
  fprintf(f, "flags=%d ", (int)btFlags(aData));
  if( btFlags(aData) & BT_PGFLAGS_INTERNAL ){
    fprintf(f, "rchild=%d ", (int)btGetU32(&aData[1]));
  }
  fprintf(f, "cell-offsets=(");
  for(i=0; i<nCell; i++){
    u8 *ptr = btCellPtrFind(aData, nData, i);
    fprintf(f, "%s%d", i==0?"":" ", (int)btGetU16(ptr));
  }
  fprintf(f, ")\n");

  for(i=0; i<nCell; i++){
    int nKey;
    int j;
    u8 *pCell = btCellFind(aData, nData, i);
    fprintf(f, "  Key %d: ", i);
    pCell += sqlite4BtVarintGet32(pCell, &nKey);
    for(j=0; j<nKey; j++){
      fprintf(f, "%02X", (int)pCell[j]);
    }
    if( btFlags(aData) & BT_PGFLAGS_INTERNAL ){
      fprintf(f, "  child=%d ", (int)btGetU32(&pCell[j]));
    }
    fprintf(f, "\n");
  }
}
int printPgdataToStderr(u32 pgno, u8 *aData, int nData){
  printPage(stderr, pgno, aData, nData);

** If the cell key is C, and the user key K, then this function sets:
**
**     *piRes = (C - K).
**
** In other workds, *piRes is +ve, zero or -ve if C is respectively larger, 
** equal to or smaller than K.
*/
#if 0
static int btCellKeyCompare(
  bt_cursor *pCsr,                /* Cursor handle */
  const u8 *aData, int nData,     /* Page data (nData is the page size) */
  int iCell,                      /* Cell to compare key from */
  const u8 *pK, int nK,           /* Key to compare to cell key */
  int *piRes                      /* OUT: Result of comparison */
){
  BtPage *pLeaf = 0;              /* Leaf page reference to release */
  int rc = SQLITE4_OK;
  int nCellKey;                   /* Number of bytes in cell key */
  int res;
  u8 *pCell = btCellFind((u8*)aData, nData, iCell);
  int nCmp;
  int nAscend = 0;

  pCell += sqlite4BtVarintGet32(pCell, &nCellKey);
  if( nCellKey==0 ){


    if( (btFlags(aData, nData) & BT_PGFLAGS_INTERNAL)==0 ){
      rc = sqlite4BtCsrKey(pCsr, &pCell, &nCellKey);
    }else{
      rc = btOverflowBuffer(pCsr, 0);
      if( rc!=SQLITE4_OK ) return rc;
      pK = pCsr->ovfl.pBuf;
    }else{
      assert( 0 );
    }

  }

  nCmp = MIN(nCellKey, nK);
  res = memcmp(pCell, pK, nCmp);
  if( res==0 ){
    res = nCellKey - nK;
  }

  *piRes = res;
  return rc;
}
#endif

static int btCellKeyCompare(
  bt_cursor *pCsr, 
  int bLeaf, 
  const void *pK, 
  int nK, 
  int *piRes
){
  const void *pCsrKey;
  int nCsrKey;
  int nCmp;
  int nAscend = 0;
  int rc = SQLITE4_OK;
  int res;

  if( bLeaf ){
    rc = sqlite4BtCsrKey(pCsr, &pCsrKey, &nCsrKey);
  }else{
    const int pgsz = sqlite4BtPagerPagesize(pCsr->pDb->pPager);

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

    pCsrKey = pCell + sqlite4BtVarintGet32(pCell, &nCsrKey);
    if( nCsrKey==0 ){
      while( 1 ){
        u8 *aData = sqlite4BtPageData(pCsr->apPage[pCsr->nPg-1]);
        u32 pgno = btChildPgno(aData, pgsz, pCsr->aiCell[pCsr->nPg-1]);
        nAscend++;
        rc = btCsrDescend(pCsr, pgno);
        if( rc!=SQLITE4_OK ) break;
        aData = sqlite4BtPageData(pCsr->apPage[pCsr->nPg-1]);
        pCsr->aiCell[pCsr->nPg-1] = btCellCount(aData, pgsz);
        if( (btFlags(aData) & BT_PGFLAGS_INTERNAL)==0 ) break;
      }
      pCsr->aiCell[pCsr->nPg-1]--;
      rc = sqlite4BtCsrKey(pCsr, &pCsrKey, &nCsrKey);
    }
  }

  if( rc==SQLITE4_OK ){
    nCmp = MIN(nCsrKey, nK);
    res = memcmp(pCsrKey, pK, nCmp);
    if( res==0 ){
      res = nCsrKey - nK;
    }
  }

  btCsrAscend(pCsr, nAscend);
  *piRes = res;
  return rc;
}

static int btCsrSeek(
  bt_cursor *pCsr, 
  const void *pK,                 /* Key to seek for */
  int nK,                         /* Size of key pK in bytes */
  int eSeek,                      /* Seek mode (a BT_SEEK_XXX constant) */
  int bUpdate
){
  const int pgsz = sqlite4BtPagerPagesize(pCsr->pDb->pPager);
  u32 pgno;                       /* Page number for next page to load */
  int rc = SQLITE4_OK;            /* Return Code */

  /* Reset the cursor */
  btCsrReset(pCsr, 0);

  /* Figure out the root page number */
  assert( pCsr->nPg==0 );
  pgno = sqlite4BtPagerRootpgno(pCsr->pDb->pPager);

  while( rc==SQLITE4_OK && pgno ){
    /* Load page number pgno into the b-tree */
    rc = btCsrDescend(pCsr, pgno);
    if( rc==SQLITE4_OK ){
      int nCell;                  /* Number of cells on this page */
      int iHi;                    /* pK/nK is <= than cell iHi */
      int iLo;                    /* pK/nK is > than cell (iLo-1) */
      int res;                    /* Result of comparison */
      u8 *aData = (u8*)sqlite4BtPageData(pCsr->apPage[pCsr->nPg-1]);
      int bLeaf = ((btFlags(aData) & BT_PGFLAGS_INTERNAL)==0);

      iLo = 0;
      iHi = nCell = btCellCount(aData, pgsz);
      if( nCell==0 ){
        rc = SQLITE4_NOTFOUND;
        break;
      }

      while( iHi>iLo ){
        int iTst = (iHi+iLo)/2;   /* Cell to compare to pK/nK */
        pCsr->aiCell[pCsr->nPg-1] = iTst;
        rc = btCellKeyCompare(pCsr, bLeaf, pK, nK, &res);
        if( rc!=SQLITE4_OK || res==0 ){
          /* Cell iTst is EQUAL to pK/nK */
          iHi = iLo = iTst;
        }else if( res<0 ){
          /* Cell iTst is SMALLER than pK/nK */
          iLo = iTst+1;
        }else{
          /* Cell iTst is LARGER than pK/nK */
          iHi = iTst;
        }
      }
      if( rc!=SQLITE4_OK ) break;
      assert( iHi==iLo );
      pCsr->aiCell[pCsr->nPg-1] = iHi;

#if 0
printPage(stderr, pgno, aData, pgsz);
#endif

      if( bLeaf==0 ){
        pgno = btChildPgno(aData, pgsz, iHi);
      }else{
        pgno = 0;

        if( res!=0 ){
          assert( BT_SEEK_LEFAST<0 && BT_SEEK_LE<0 );
          if( eSeek<0 ){

*/
static int btCsrEnd(bt_cursor *pCsr, int bLast){
  const int pgsz = sqlite4BtPagerPagesize(pCsr->pDb->pPager);
  int rc;                         /* Return Code */
  u32 pgno;                       /* Page number for next page to load */

  /* Reset the cursor */
  btCsrReset(pCsr, 0);

  /* Figure out the root page number */
  assert( pCsr->nPg==0 );
  pgno = sqlite4BtPagerRootpgno(pCsr->pDb->pPager);

  while( rc==SQLITE4_OK ){
    /* Load page number pgno into the b-tree */

    }else{
      if( pCsr->aiCell[iPg]>0 ){
        pCsr->aiCell[iPg]--;
        break;
      }
    }

    rc = btCsrAscend(pCsr, 1);
    bRequireDescent = 1;
  }

  if( bRequireDescent && rc==SQLITE4_OK ){
    u32 pgno;                   /* Child page number */
    u8 *aData = (u8*)sqlite4BtPageData(pCsr->apPage[pCsr->nPg-1]);

    pgno = btChildPgno(aData, pgsz, pCsr->aiCell[pCsr->nPg-1]);

    while( 1 ){
      rc = btCsrDescend(pCsr, pgno);
      if( rc!=SQLITE4_OK ){
        break;
      }else{
        int nCell;
        aData = (u8*)sqlite4BtPageData(pCsr->apPage[pCsr->nPg-1]);
        nCell = btCellCount(aData, pgsz);
        if( btFlags(aData) & BT_PGFLAGS_INTERNAL ){
          pCsr->aiCell[pCsr->nPg-1] = (bNext ? 0 : nCell);
          pgno = btChildPgno(aData, pgsz, pCsr->aiCell[pCsr->nPg-1]);
        }else{
          pCsr->aiCell[pCsr->nPg-1] = (bNext ? 0 : nCell-1);
          break;
        }
      }

/*
** Retreat to the previous entry in the tree.
*/
int sqlite4BtCsrPrev(bt_cursor *pCsr){
  return btCsrStep(pCsr, 0);
}























static int btGrowBuffer(bt_db *db, int nReq, u8 **ppVal, int *pnVal){
  if( nReq>*pnVal ){
    u8 *pNew = sqlite4_malloc(db->pEnv, nReq*2);
    if( pNew==0 ) return btErrorBkpt(SQLITE4_NOMEM);
    sqlite4_free(db->pEnv, *ppVal);
    *ppVal = pNew;
    *pnVal = nReq*2;

      for(iLvl=nDepth-1; iLvl>=0; iLvl--){
        if( apHier[iLvl].iCell<(nPgPtr-1) ) break;
      }
      if( iLvl<0 ) break; /* SQLITE4_CORRUPT? */
      apHier[iLvl].iCell++;

      for(; iLvl<nDepth && rc==SQLITE4_OK; iLvl++){
        a = sqlite4BtPageData(apHier[iLvl].pPg);
        pgno = btGetU32(&a[apHier[iLvl].iCell * 4]);
        if( iLvl<(nDepth-1) ){
          apHier[iLvl+1].iCell = 0;
          sqlite4BtPageRelease(apHier[iLvl+1].pPg);
          apHier[iLvl+1].pPg = 0;
          rc = sqlite4BtPageGet(db->pPager, pgno, &apHier[iLvl+1].pPg);
        }
      }
    }

    for(i=0; i<nDepth && rc==SQLITE4_OK; i++){
      sqlite4BtPageRelease(apHier[i].pPg);
    }
  }

  return rc;
}

/*
** If the current cell is a type (c) leaf cell, load the entire key
** into the pCsr->ovfl buffer. If bVal is true, then also load the
** entries value into the buffer.
*/
static int btOverflowBuffer(bt_cursor *pCsr, int bVal){
  const int pgsz = sqlite4BtPagerPagesize(pCsr->pDb->pPager);
  int rc = SQLITE4_OK;
  u8 *aData;
  u8 *pCell;
  int iCell = pCsr->aiCell[pCsr->nPg-1];

  int nK;
  int nReq;
  u8 *pLocal;                     /* Pointer to local data within leaf page */
  u8 *aOut;                       /* Output buffer for overflow data */
  int nLocal = 0;
  int nOvfl1 = 0;
  int nOvfl2 = 0;

  aData = (u8*)sqlite4BtPageData(pCsr->apPage[pCsr->nPg-1]);
  pCell = btCellFind(aData, pgsz, iCell);
  pCell += sqlite4BtVarintGet32(pCell, &nK);
  if( nK==0 ){
    /* type (c) leaf cell */
    pCell += sqlite4BtVarintGet32(pCell, &nLocal);
    pLocal = pCell;
    pCell += nLocal;
    pCell += sqlite4BtVarintGet32(pCell, &nOvfl1);
    pCell += sqlite4BtVarintGet32(pCell, &nOvfl2);

    pCsr->ovfl.nKey = nLocal + nOvfl1;
    pCsr->ovfl.nVal = nOvfl2;
  }else{
    /* type (b) leaf cell */
    pCell += nK;
    assert( pCell[0]==0x00 );
    pCell++;

    pCell += sqlite4BtVarintGet32(pCell, &nLocal);
    pLocal = pCell;
    pCell += nLocal;
    pCell += sqlite4BtVarintGet32(pCell, &nOvfl1);

    pCsr->ovfl.nKey = 0;
    pCsr->ovfl.nVal = nLocal + nOvfl1;
  }

  nReq = nLocal + nOvfl1 + nOvfl2;
  rc = btGrowBuffer(pCsr->pDb, nReq, &pCsr->ovfl.pBuf, &pCsr->ovfl.nBuf);
  if( rc!=SQLITE4_OK ) return rc;

  /* Copy in local data */
  memcpy(pCsr->ovfl.pBuf, pLocal, nLocal);

  /* Load in overflow data */
  aOut = &pCsr->ovfl.pBuf[nLocal];
  rc = btOverflowArrayRead(pCsr->pDb, pCell, aOut, nOvfl1 + nOvfl2);

  return rc;
}

int sqlite4BtCsrKey(bt_cursor *pCsr, const void **ppK, int *pnK){
  const int pgsz = sqlite4BtPagerPagesize(pCsr->pDb->pPager);
  u8 *aData;
  u8 *pCell;
  int iCell = pCsr->aiCell[pCsr->nPg-1];
  int nK;
  int rc = SQLITE4_OK;
  
  aData = (u8*)sqlite4BtPageData(pCsr->apPage[pCsr->nPg-1]);
  assert( btCellCount(aData, pgsz)>iCell );
  pCell = btCellFind(aData, pgsz, iCell);
  pCell += sqlite4BtVarintGet32(pCell, &nK);

  if( nK==0 ){
    /* type (c) leaf cell */
    rc = btOverflowBuffer(pCsr, 0);
    if( rc==SQLITE4_OK ){
      *ppK = pCsr->ovfl.pBuf;
      *pnK = pCsr->ovfl.nKey;
    }
  }else{
    *ppK = pCell;
    *pnK = nK;
  }

  return rc;
}

int sqlite4BtCsrData(
  bt_cursor *pCsr,                /* Cursor handle */
  int iOffset,                    /* Offset of requested data */
  int nByte,                      /* Bytes requested (or -ve for all avail.) */
  const void **ppV,               /* OUT: Pointer to data buffer */
  int *pnV                        /* OUT: Size of data buffer in bytes */
){
  const int pgsz = sqlite4BtPagerPagesize(pCsr->pDb->pPager);
  int rc = SQLITE4_OK;
  u8 *aData;
  u8 *pCell;
  int iCell = pCsr->aiCell[pCsr->nPg-1];
  int nK = 0;
  int nV = 0;

  aData = (u8*)sqlite4BtPageData(pCsr->apPage[pCsr->nPg-1]);
  pCell = btCellFind(aData, pgsz, iCell);
  pCell += sqlite4BtVarintGet32(pCell, &nK);
  if( nK>0 ){
    pCell += nK;
    pCell += sqlite4BtVarintGet32(pCell, &nV);
  }






  if( nV==0 ){




    rc = btOverflowBuffer(pCsr, 1);

    if( rc==SQLITE4_OK ){

      *ppV = &pCsr->ovfl.pBuf[pCsr->ovfl.nKey];
      *pnV = pCsr->ovfl.nVal;
    }



  }else{
    *ppV = pCell;
    *pnV = (nV-1);
  }

  return rc;
}

/*
** The argument points to a buffer containing an overflow array. Return
** the size of the overflow array in bytes. 
*/
static int btOverflowArrayLen(u8 *p){
  return 1 + ((int)(p[0] & 0x0F) + 1) * 4;
}

static int btCellSize(u8 *pCell, int bLeaf){
  u8 *p = pCell;
  int nKey;



  p += sqlite4BtVarintGet32(p, &nKey);
  if( bLeaf==0 ){
    p += nKey;
    p += 4;
  }else if( nKey==0 ){
    p += sqlite4BtVarintGet32(p, &nKey);
    p += nKey;
    p += sqlite4BtVarintGet32(p, &nKey);
    p += sqlite4BtVarintGet32(p, &nKey);
    p += btOverflowArrayLen(p);
  }else{
    p += nKey;

    p += sqlite4BtVarintGet32(p, &nKey);
    if( nKey==0 ){
      p += sqlite4BtVarintGet32(p, &nKey);
      p += nKey;
      p += sqlite4BtVarintGet32(p, &nKey);
      p += btOverflowArrayLen(p);
    }else{
      p += (nKey-1);
    }
  }


  return (p-pCell);
}

static u8 *btCellFindSize(u8 *aData, int nData, int iCell, int *pnByte){
  u8 *pCell;

  pCell = btCellFind(aData, nData, iCell);
  *pnByte = btCellSize(pCell, 0==(btFlags(aData) & BT_PGFLAGS_INTERNAL));
  return pCell;
}

/*
** Allocate a new page buffer.
*/
static int btNewBuffer(bt_db *pDb, u8 **paBuf){

  int nHdr;                       /* Bytes in header of this page */

  if( btNewBuffer(pDb, &aTmp) ) return SQLITE4_NOMEM;

  aData = sqlite4BtPageData(pPg);
  nCell = btCellCount(aData, pgsz);

  bLeaf = 0==(btFlags(aData) & BT_PGFLAGS_INTERNAL);
  nHdr = bLeaf ? 1 : 5;

  /* Set header bytes of new page */
  memcpy(aTmp, aData, nHdr);

  iWrite = nHdr;
  for(i=0; i<nCell; i++){
    int nByte;
    u8 *pCell;
    pCell = btCellFindSize(aData, pgsz, i, &nByte);

    btPutU16(btCellPtrFind(aTmp, pgsz, i), iWrite);
    memcpy(&aTmp[iWrite], pCell, nByte);
    iWrite += nByte;
  }


  /* Write the rest of the page footer */
  btPutU16(&aTmp[pgsz-2], nCell);
  btPutU16(&aTmp[pgsz-4], pgsz - (3+nCell)*2 - iWrite);
  btPutU16(&aTmp[pgsz-6], iWrite);

  btSetBuffer(pDb, pPg, aTmp);

** Argument pKV contains a key/value pair destined for a leaf page in
** a database with page size pgsz. Currently it is in KV_VALUE form.
** If the key/value pair is too large to fit entirely within a leaf
** page, this function allocates and writes the required overflow
** pages to the database, and converts pKV to a KV_CELL cell (that
** contains the overflow array).
*/
static int btOverflowAssign(bt_db *db, KeyValue *pKV){
  const int pgsz = sqlite4BtPagerPagesize(db->pPager);
  int nMaxSize = (pgsz - 1 - 6 - 2);
  int nReq;
  int rc = SQLITE4_OK;

  assert( pKV->pgno==0 && pKV->eType==KV_VALUE );

  /* Check if this is a type (a) cell - one that can fit entirely on a 
  ** leaf page. If so, do nothing.  */
  nReq = sqlite4BtVarintLen32(pKV->nK) + sqlite4BtVarintLen32(pKV->nV);
  nReq += pKV->nK + pKV->nV;
  if( nReq > nMaxSize ){


    int nArraySz = btOverflowArraySz(pgsz, pKV->nK + pKV->nV);
    u8 *pBuf = 0;                 /* Buffer containing formatted cell */
    int nKeyOvfl;                 /* Bytes of key that overflow */
    int nValOvfl;                 /* Bytes of value that overflow */

    /* Check if the entire key can fit on a leaf page. If so, this is a
    ** type (b) page - entire key and partial value on the leaf page, 
    ** overflow pages contain the rest of the value.  */
    nReq = 1 + sqlite4BtVarintLen32(pKV->nK) + pKV->nK 

          - sqlite4BtVarintLen32(pKV->nK) - pKV->nK - 1 - nArraySz
      );
      nLVal = nSpc - sqlite4BtVarintLen32(pgsz) - sqlite4BtVarintLen32(pKV->nV);
      nKeyOvfl = 0;
      nValOvfl = pKV->nV - nLVal;
    }else{
      /* Type (c) cell. Both the key and value overflow. */
      int nLKey = nMaxSize 
          - 1                                    /* 0x00 byte */
          - sqlite4BtVarintLen32(pgsz)           /* nLKey */
          - sqlite4BtVarintLen32(pKV->nK)        /* nOKey */
          - sqlite4BtVarintLen32(pKV->nV)        /* nVal */
          - nArraySz;                            /* overflow array */

      nValOvfl = pKV->nV;
      nKeyOvfl = pKV->nK - nLKey;
    }

    /* Allocate a pager buffer to store the KV_CELL buffer. Using a pager
    ** buffer is convenient here as (a) it is roughly the right size and
    ** (b) can probably be allocated/deallocated faster than a regular
    ** heap allocation.  */
    rc = btNewBuffer(db, &pBuf);

    if( rc==SQLITE4_OK ){
      int nLVal = (pKV->nV - nValOvfl);
      int nLKey = (pKV->nK - nKeyOvfl);
      u8 *pOut = pBuf;

      if( nKeyOvfl>0 ){
        *pOut++ = 0x00;
      }
      pOut += sqlite4BtVarintPut32(pOut, nLKey);
      memcpy(pOut, pKV->pK, nLKey);
      pOut += nLKey;
      if( nKeyOvfl==0 ){
        *pOut++ = 0x00;
        pOut += sqlite4BtVarintPut32(pOut, nLVal);
        memcpy(pOut, pKV->pV, nLVal);
        pOut += nLVal;
      }else{
        pOut += sqlite4BtVarintPut32(pOut, nKeyOvfl);
      }
      pOut += sqlite4BtVarintPut32(pOut, nValOvfl);

      rc = btOverflowArrayPopulate(db, &pOut,
          (u8*)(pKV->pK) + nLKey, nKeyOvfl,
          (u8*)(pKV->pV) + nLVal, nValOvfl
      );
      if( rc==SQLITE4_OK ){
        memset(pKV, 0, sizeof(*pKV));
        pKV->pV = pBuf;
        pKV->nV = pOut - pBuf;
        pKV->eType = KV_CELL;
        pBuf = 0;
        assert( pKV->nV<=nMaxSize );
        assert( pKV->nV==btCellSize((u8*)pKV->pV, 1) );
      }
    }

    if( pBuf ){
      btFreeBuffer(db, pBuf);
    }
  }

#endif

  /* The leaves are written. Now gather the keys and page numbers to
  ** push up into the parent page. This is only required when rebalancing
  ** b-tree leaves. When internal nodes are balanced, the btBalanceOutput
  ** loop accumulates the cells destined for the parent page.  */
  for(iPg=0; iPg<(ctx.nOut-1); iPg++){
    ctx.aPCell[iPg].pgno = sqlite4BtPagePgno(ctx.apPg[iPg]);
    assert( ctx.aPCell[iPg].nK==0 );

    if( bLeaf ){
#if 0
      u8 *aData = sqlite4BtPageData(ctx.apPg[iPg]);
      u8 *pCell;

      pCell = btCellFind(aData, pgsz, btCellCount(aData, pgsz)-1);

      pCell += sqlite4BtVarintGet32(pCell, &ctx.aPCell[iPg].nK);
      ctx.aPCell[iPg].pK = pCell;
#endif

    }
  }

  rc = btSetChildPgno(
      pDb, pPar, iSib+ctx.nIn-1, sqlite4BtPagePgno(ctx.apPg[ctx.nOut-1])
  );
  if( rc==SQLITE4_OK ){
    btCsrAscend(pCsr, 1);
    rc = btDeleteFromPage(pCsr, ctx.nIn-1);
  }
  iPg = pCsr->nPg;
  if( rc==SQLITE4_OK && ctx.nOut>1 ){
    rc = btInsertAndBalance(pCsr, ctx.nOut-1, ctx.aPCell);
  }
  if( rc==SQLITE4_OK && iPg==pCsr->nPg ){

  aData = (u8*)sqlite4BtPageData(pLeaf);

  /* Set the bLeaf variable to true if inserting into a leaf page, or
  ** false otherwise. Return SQLITE4_CORRUPT if the page is a leaf but
  ** the KeyValue pairs being inserted are suitable for internal nodes,
  ** or vice-versa.  */
  assert( nKV>0 );
  if( (0==(btFlags(aData) & BT_PGFLAGS_INTERNAL))!=bLeaf ){
    return btErrorBkpt(SQLITE4_CORRUPT);
  }

  nCell = btCellCount(aData, pgsz);
  assert( iCell<=btCellCount(aData, pgsz) );

  if( nCell==0 ){

  const int pgsz = sqlite4BtPagerPagesize(pCsr->pDb->pPager);
  int rc = SQLITE4_OK;
  int iPg = pCsr->nPg-1;
  BtPage *pPg = pCsr->apPage[iPg];
  u8 *aData = sqlite4BtPageData(pPg);
  int nCell = btCellCount(aData, pgsz);
  int nFree = btFreeSpace(aData, pgsz);
  int bLeaf = (0==(btFlags(aData) & BT_PGFLAGS_INTERNAL));

  if( iPg==0 ){
    /* Root page. If it contains no cells at all and is not already
    ** a leaf, shorten the tree by one here by copying the contents 
    ** of the only child into the root. */
    if( nCell==0 && bLeaf==0 ){
      BtPager *pPager = pCsr->pDb->pPager;

    /* Insert the new KV pair into the current leaf. */
    KeyValue kv;
    kv.pgno = 0;
    kv.eType = KV_VALUE;
    kv.pK = pK; kv.nK = nK;
    kv.pV = pV; kv.nV = nV;

    rc = btOverflowAssign(db, &kv);
    if( rc==SQLITE4_OK ){
      rc = btInsertAndBalance(&csr, 1, &kv);
    }
    if( kv.eType==KV_CELL ){
      sqlite4_free(db->pEnv, (void*)kv.pV);
    }
  }
  btCsrReset(&csr, 1);

  return rc;
}

int sqlite4BtDelete(bt_cursor *pCsr){
  int rc;
  rc =  btDeleteFromPage(pCsr, 1);

/*
** Read, write and trim existing database pages.
*/
int sqlite4BtPageGet(BtPager *p, u32 pgno, BtPage **ppPg){
  int rc = SQLITE4_OK;            /* Return code */
  BtPage *pRet;                   /* Returned page handle */





  /* 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);
    if( rc==SQLITE4_OK ){

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

db close
forcedelete test.db

proc bigstr {n} {
  set nRep [expr 1+($n/20)]
  string range [string repeat "abcdefghijklmnopqrstuvwxyz" $nRep] 0 [expr $n-1]
}

if 0 {
sqlite4 db file:test.db?kv=bt
foreach {tn nStr} {
  1 3000
  2 30000
  3 300000
  4 3000000
  5 30000000
} {
  set big [bigstr $nStr]
  do_execsql_test 6.$tn.1 {
    DROP TABLE IF EXISTS t6;
    CREATE TABLE t6(a PRIMARY KEY, b VALUE);
    INSERT INTO t6 VALUES($big, '123');
  }

  do_execsql_test 6.$tn.2 {
    SELECT length(a) FROM t6;
  } $nStr

  do_execsql_test 6.$tn.3 {
    SELECT a FROM t6;
  } [list $big]
}

finish_test
}


do_test 1.0 {
  sqlite4 db file:test.db?kv=bt
  db close
} {}

do_test 1.1 { sqlite4 db file:test.db?kv=bt } {}

    CREATE TABLE t5(a PRIMARY KEY, b VALUE);
    INSERT INTO t5 VALUES(1, $big);
  }

  do_execsql_test 5.$tn.2 {
    SELECT length(b) FROM t5;
  } $nStr
  if {[set_test_counter errors]} break

  do_execsql_test 5.$tn.3 {
    SELECT b FROM t5;
  } [list $big]
}

foreach {tn nStr} {
  1 3000
  2 30000
  3 300000
  4 3000000
  5 30000000
} {
  set big [bigstr $nStr]
  do_execsql_test 6.$tn.1 {
    DROP TABLE IF EXISTS t6;
    CREATE TABLE t6(a PRIMARY KEY, b VALUE);
    INSERT INTO t6 VALUES($big, '123');
  }

  do_execsql_test 6.$tn.2 {
    SELECT length(a) FROM t6;
  } $nStr
  if {[set_test_counter errors]} break

  do_execsql_test 6.$tn.3 {
    SELECT a FROM t6;
  } [list $big]
}

finish_test

is instead read dynamically from the leaf page containing the smallest keys
larger than or equal to the omitted key-prefix. This leaf may be found by
following the existing tree pointers. N is selected so as to guarantee a
minimum internal fanout of (say) 4 or 8.

<p>Any real key/value pair that is too large for a single leaf page is
spread across the leaf and one or more overflow pages. Overflow pages are
organized differently to those in SQLite with three goals in mind:

<ul>
  <li> To make random access to offsets within large values more efficient, and
  <li> To make it possible to append data to an existing value without
       rewriting all constituent pages.
  <li> To make it possible to store large objects contiguously within the
       database file. So that if the file is memory mapped, a pointer to
       the object can be returned to the user without copying it into
       heap memory.
</ul>

<i>
  <p> Inode-style overflow pages (instead of a linked list)?
</i>

<i>