SQLite4  Check-in [2bd81969a6]

*/

#include "btInt.h"
#include <string.h>
#include <assert.h>

#define BT_MAX_DEPTH 32           /* Maximum possible depth of tree */
#define BT_MAX_DIRECT_OVERFLOW 8  /* Maximum direct overflow pages per cell */

/*
** Values that make up the single byte flags field at the start of
** b-tree pages. 
*/
#define BT_PGFLAGS_INTERNAL 0x01  /* True for non-leaf nodes */

#ifndef MIN
# define MIN(a,b) (((a)<(b))?(a):(b))
#endif

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

/*
** 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 */

  u8 *pVal;                       /* Buffer used to cache large values. */
  int nVal;                       /* Size of buffer pVal in bytes */
};

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

    }
    if( btFlags(aData, nData) & 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);
  return 0;
}

int printPgToStderr(BtPage *pPg){
  printPage(stderr, sqlite4BtPagePgno(pPg), sqlite4BtPageData(pPg), 1024);
  return 0;
}
#endif


/*

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;

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

    assert( 0 );
#if 0
    if( (btFlags(aData, nData) & BT_PGFLAGS_INTERNAL)==0 ){
      pCell += sqlite4BtVarintGet32(pCell, &nCellKey);
      assert( nCellKey>0 );
    }else{
      assert( 0 );
    }
#endif
  }

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

  sqlite4BtPageRelease(pLeaf);
  *piRes = res;
  return rc;
}

static int btCsrSeek(
  bt_cursor *pCsr, 
  const void *pK,                 /* Key to seek for */

      pCsr->aiCell[pCsr->nPg-1] = iHi;

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

      if( aData[0] & BT_PGFLAGS_INTERNAL ){





        pgno = btChildPgno(aData, pgsz, iHi);




      }else{
        pgno = 0;

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

  }else{
    *ppK = pCell;
    *pnK = nK;
  }

  return SQLITE4_OK;
}

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;
  }
  return SQLITE4_OK;
}

static int btOverflowArrayRead(
  bt_db *db,
  u8 *pOvfl,
  u8 *aOut,
  int nOut
){
  const int pgsz = sqlite4BtPagerPagesize(db->pPager);
  int rc = SQLITE4_OK;
  int nDirect;                    /* Number of direct overflow pages */
  int iOut;                       /* Bytes of data copied so far */
  int iPg;

  nDirect = (int)(pOvfl[0] & 0x0F);
  nDepth = (int)(pOvfl[0]>>4);

  iOut = 0;

  /* Read from the direct overflow pages. And from the overflow tree, if
  ** it has a depth of zero.  */
  for(iPg=0; rc==SQLITE4_OK && iPg<(nDirect+(nDepth==0)); iPg++){
    BtPage *pPg = 0;
    rc = sqlite4BtPageGet(db->pPager, btGetU32(&pOvfl[1+iPg*4]), &pPg);
    if( rc==SQLITE4_OK ){
      u8 *a = sqlite4BtPageData(pPg);
      memcpy(&aOut[iOut], a, MIN(nOut-iOut, pgsz));
      sqlite4BtPageRelease(pPg);
    }
  }

  /* Read from the overflow tree, if it was not read by the block above. */
  if( nDepth>0 ){
    struct Heir {
      BtPage *pPg;
      int iCell;
    } apHier[8];
    int i;

    memset(apHier, 0, sizeof(apHier));
    pgno = btGetU32(&pOvfl[1+nDirect*4]);
    for(i=0; i<nDepth && rc==SQLITE4_OK; i++){
      u8 *a;
      rc = sqlite4BtPageGet(db->pPager, pgno, &apHier[i].pPg);
      if( rc==SQLITE4_OK ){
        a = sqlite4BtPageData(apHier[i].pPg);
        pgno = btGetU32(a);
      }
    }

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

  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;
  int nV;

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

  if( nV==0 ){
    int nLVal;
    int nOVal;
    int nVal;                     /* Total size of entry data */
    u8 *pOvfl;


    pCell += sqlite4BtVarintGet32(pCell, &nLVal);
    pOvfl = pCell + nLVal;
    pOvfl += sqlite4BtVarintGet32(pOvfl, &nOVal);
    nVal = nLVal + nOVal;

    if( btGrowBuffer(pCsr->pDb, nVal, &pCsr->pVal, &pCsr->nVal) ){
      return SQLITE4_NOMEM;
    }
    memcpy(pCsr->pVal, pCell, nLVal);

    rc = btOverflowArrayRead(pCsr->pDb, pOvfl, &pCsr->pVal[nLVal], nOVal);
    *ppV = pCsr->pVal;
    *pnV = 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 u8 *btCellFindSize(u8 *aData, int nData, int iCell, int *pnByte){
  int nKey;
  u8 *pCell;
  u8 *p;

  p = pCell = btCellFind(aData, nData, iCell);
  p += sqlite4BtVarintGet32(p, &nKey);
  assert( nKey!=0 ); /* todo. type (c) leaf cell */
  p += nKey;
  if( 0==(aData[0] & BT_PGFLAGS_INTERNAL) ){
    p += sqlite4BtVarintGet32(p, &nKey);
    if( nKey==0 ){
      p += sqlite4BtVarintGet32(p, &nKey);
      p += nKey;
      p += sqlite4BtVarintGet32(p, &nKey);
      p += btOverflowArrayLen(p);
    }else{
      p += (nKey-1);
    }
  }else{
    p += 4;
  }

  *pnByte = (p - pCell);
  return pCell;
}

  btPutU16(&aTmp[pgsz-4], pgsz - (3+nCell)*2 - iWrite);
  btPutU16(&aTmp[pgsz-6], iWrite);

  btSetBuffer(pDb, pPg, aTmp);
  return SQLITE4_OK;
}

/*
** The following type is used to represent a single cell or cell value
** by the code that updates and rebalances the tree structure. It is
** usually manipulated using the btKV*() functions and macros.
**
** An instance of type KeyValue may represent three different types
** of cell values:
**
**   * (eType==KV_VALUE && pgno!=0): A value for an internal cell. In this case
**     pK points to a buffer nK bytes in size containing the key prefix and 
**     pgno contains the page number of the cells child page.
**
**   * (eType==KV_VALUE && pgno==0): A value for a leaf cell. The 
**     key is identified by (pK/nK) and the value by (pV/nV).
**
**   * (eType==KV_CELL): A formatted leaf cell stored in (pV/nV). This is 
**     used for cells with overflow arrays.
*/
#define KV_VALUE     0
#define KV_CELL      1
typedef struct KeyValue KeyValue;
struct KeyValue {

  int eType;
  const void *pK; int nK;
  const void *pV; int nV;
  u32 pgno;
};

/*
** Return the number of bytes consumed by a cell generated based on *pKV.

*/
static int btKVCellSize(KeyValue *pKV){
  int nByte;
  assert( pKV->eType==KV_CELL || pKV->eType==KV_VALUE );
  if( pKV->eType==KV_CELL ){
    nByte = pKV->nV;
  }else{
    if( pKV->pgno ){
      nByte = sqlite4BtVarintLen32(pKV->nK) + pKV->nK + 4;
    }else{
      nByte = 
        sqlite4BtVarintLen32(pKV->nK) 
        + sqlite4BtVarintLen32(pKV->nV+1)
        + pKV->nV + pKV->nK;
    }
  }
  return nByte;
}

/*
** Write a cell based on *pKV to buffer aBuffer. Return the number
** of bytes written.
*/
static int btKVCellWrite(KeyValue *pKV, u8 *aBuf){
  int i = 0;
  if( pKV->eType==KV_CELL ){
    i = pKV->nV;
    memcpy(aBuf, pKV->pV, i);
  }else{
    i += sqlite4BtVarintPut32(&aBuf[i], pKV->nK);
    memcpy(&aBuf[i], pKV->pK, pKV->nK); i += pKV->nK;

    if( pKV->pgno==0 ){
      i += sqlite4BtVarintPut32(&aBuf[i], pKV->nV+1);
      memcpy(&aBuf[i], pKV->pV, pKV->nV); i += pKV->nV;
    }else{
      btPutU32(&aBuf[i], pKV->pgno);
      i += 4;
    }
  }

  assert( i==btKVCellSize(pKV) );
  return i;
}

/*
** Return the number of bytes of leaf page space required by an 
** overflow array containing nContent bytes of content, assuming the 
** page size is pgsz.
*/
static int btOverflowArraySz(int pgsz, int nContent){
  int nPg;
  nPg = (nContent + pgsz - 1) / pgsz;
  if( nPg<=BT_MAX_DIRECT_OVERFLOW ){
    return 1 + nPg*4;
  }
  return 1 + (BT_MAX_DIRECT_OVERFLOW+1) * 4;
}

static int btOverflowArrayPopulate(
  bt_db *db, u8 **ppOut,
  u8 *pBuf1, int nBuf1,
  u8 *pBuf2, int nBuf2
){
  const int pgsz = sqlite4BtPagerPagesize(db->pPager);
  const int nPgPtr = pgsz / 4;
  int rc = SQLITE4_OK;
  int n1 = 0;
  int n2 = 0;
  int iOvfl;
  int nDirect = 0;
  int nDepth = 0;
  int nOvfl;
  int i;
  u8 *aOut = *ppOut;

  struct Heir {
    BtPage *pPg;
    int iCell;
  } apHier[8];
  memset(apHier, 0, sizeof(apHier));

  /* Calculate the number of required overflow pages. And the depth of
  ** the overflow tree.  */
  nOvfl = (nBuf1+nBuf2+pgsz-1) / pgsz;
  nOvfl -= BT_MAX_DIRECT_OVERFLOW;
  while( nOvfl>1 ){
    nDepth++;
    nOvfl = (nOvfl / nPgPtr);
  }

  for(i=0; rc==SQLITE4_OK && i<nDepth; i++){
    u32 pgno;
    rc = sqlite4BtPageAllocate(db->pPager, &apHier[i].pPg);
    pgno = sqlite4BtPagePgno(apHier[i].pPg);
    if( i==0 ){
      btPutU32(&aOut[1 + BT_MAX_DIRECT_OVERFLOW*4], pgno);
    }else{
      u8 *a = sqlite4BtPageData(apHier[i-1].pPg);
      btPutU32(a, pgno);
      apHier[i-1].iCell++;
    }
  }

  for(iOvfl=0; rc==SQLITE4_OK && (n1<nBuf1 || n2<nBuf2); iOvfl++){
    int nCopy1, nCopy2;           /* Bytes to copy from pBuf1 and pBuf2 */
    u8 *aData;
    BtPage *pPg;
    u32 pgno;

    rc = sqlite4BtPageAllocate(db->pPager, &pPg);
    if( rc!=SQLITE4_OK ) break;
    aData = sqlite4BtPageData(pPg);
    pgno = sqlite4BtPagePgno(pPg);

    nCopy1 = MIN(pgsz, nBuf1 - n1);
    nCopy2 = MIN(pgsz - nCopy1, nBuf2 - n2);

    memcpy(aData, &pBuf1[n1], nCopy1); n1 += nCopy1;
    memcpy(&aData[nCopy1], &pBuf2[n2], nCopy2); n2 += nCopy2;

    if( iOvfl<(BT_MAX_DIRECT_OVERFLOW+(nDepth==0)) ){
      btPutU32(&aOut[1 + iOvfl*4], pgno);
      nDirect++;
    }else{
      assert( nDepth>0 );
      for(i=nDepth-1; pgno && i>=0; i--){
        u8 *a = sqlite4BtPageData(apHier[i].pPg);
        if( apHier[i].iCell==nPgPtr ){
          BtPage *pNew = 0;
          rc = sqlite4BtPageRelease(apHier[i].pPg);
          if( rc==SQLITE4_OK ){
            rc = sqlite4BtPageAllocate(db->pPager, &pNew);
            if( rc==SQLITE4_OK ){
              u8 *a = sqlite4BtPageData(pNew);
              btPutU32(a, pgno);
              pgno = sqlite4BtPagePgno(pNew);
            }
          }

          if( rc!=SQLITE4_OK ){
            pgno = 0;
          }

          apHier[i].pPg = pNew;
          apHier[i].iCell = 0;
        }else{
          btPutU32(&a[apHier[i].iCell*4], pgno);
          pgno = 0;
        }
      }
    }
  }

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

  if( rc==SQLITE4_OK ){
    if( nDirect>BT_MAX_DIRECT_OVERFLOW ){
      assert( nDirect==BT_MAX_DIRECT_OVERFLOW+1 );
      nDirect = BT_MAX_DIRECT_OVERFLOW;
    }
    assert( nDirect>=1 );
    aOut[0] = (u8)nDirect | (u8)(nDepth<<4) ;
  }

  *ppOut = &aOut[1 + nDirect*4];
  return rc;
}

/*
** 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 btAssignOverflow(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 ){
    u8 *pBuf = 0;                 /* Buffer containing formatted cell */

    int nArraySz = btOverflowArraySz(pgsz, pKV->nK + pKV->nV);
    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 
         + 1 + sqlite4BtVarintLen32(pKV->nV) + nArraySz;

    if( nReq<nMaxSize ){
      /* nSpc is initialized to the amount of space available to store:
      **
      **    * varint containing number of bytes stored locally (nLVal).
      **    * nLVal bytes of content.
      **    * varint containing number of bytes in overflow pages.
      */
      int nLVal;                  /* Bytes of value data on main page */
      int nSpc = (nMaxSize 
          - 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. */
      assert( 0 );
    }

    rc = btNewBuffer(db, &pBuf);
    if( rc==SQLITE4_OK ){
      int nLVal = (pKV->nV - nValOvfl);
      u8 *pOut = pBuf;

      pOut += sqlite4BtVarintPut32(pOut, pKV->nK);
      memcpy(pOut, pKV->pK, pKV->nK);
      pOut += pKV->nK;

      *pOut++ = 0x00;
      pOut += sqlite4BtVarintPut32(pOut, nLVal);
      memcpy(pOut, pKV->pV, nLVal);
      pOut += nLVal;
      pOut += sqlite4BtVarintPut32(pOut, nValOvfl);

      rc = btOverflowArrayPopulate(db, &pOut, 0, 0, 
          (u8*)(pKV->pV) + (pKV->nV - nValOvfl), nValOvfl
      );
      if( rc==SQLITE4_OK ){
        memset(pKV, 0, sizeof(*pKV));
        pKV->pV = pBuf;
        pKV->nV = pOut - pBuf;
        pKV->eType = KV_CELL;
        pBuf = 0;
      }
    }

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

  return rc;
}

typedef struct BalanceCtx BalanceCtx;
struct BalanceCtx {
  int pgsz;                       /* Database page size */
  int bLeaf;                      /* True if we are rebalancing leaf data */
  bt_cursor *pCsr;                /* Cursor identifying where to insert pKV */
  int nKV;                        /* Number of KV pairs */

**     * A page pointer, stored as a 32-bit big-endian unsigned.
*/
static void btInternalCellToKeyValue(u8 *pCell, KeyValue *pKV){
  pKV->pK = pCell + sqlite4BtVarintGet32(pCell, &pKV->nK);
  pKV->pgno = btGetU32(&((u8*)pKV->pK)[pKV->nK]);
  pKV->pV = 0;
  pKV->nV = 0;
  pKV->eType = KV_VALUE;
}

static int btSetChildPgno(bt_db *pDb, BtPage *pPg, int iChild, u32 pgno){
  const int pgsz = sqlite4BtPagerPagesize(pDb->pPager);
  int rc;

  rc = sqlite4BtPageWrite(pPg);

  int iCell,                      /* Cell number in this iteration */
  u8 *pCell, int nByte,           /* Binary cell */
  KeyValue *pKV                   /* Key-value cell */
){
  if( pCell ){
    p->anCellSz[iCell] = nByte;
  }else{
    p->anCellSz[iCell] = btKVCellSize(pKV);
  }
  return SQLITE4_OK;
}

static int btBalanceOutput(
  BalanceCtx *p,                  /* Description of balance operation */
  int iCell,                      /* Cell number in this iteration */

    u32 pgno;
    KeyValue *pPKey = &p->aPCell[p->iOut];

    if( pCell ){
      pKey = pCell + sqlite4BtVarintGet32(pCell, &nKey);
      pgno = btGetU32(&pKey[nKey]);
    }else{
      assert( pKV->eType==KV_VALUE );
      pKey = (u8*)pKV->pK;
      nKey = pKV->nK;
      pgno = pKV->pgno;
    }

    pCopy = &p->pTmp[p->iTmp];
    p->iTmp += nKey;
    memcpy(pCopy, pKey, nKey);

    if( iOff==0 ) iOff = (p->bLeaf ? 1 : 5);
    nCell = btCellCount(aOut, p->pgsz);
    btPutU16(btCellPtrFind(aOut, p->pgsz, nCell), iOff);
    if( pCell ){
      memcpy(&aOut[iOff], pCell, nByte);
      iOff += nByte;
    }else{
      iOff += btKVCellWrite(pKV, &aOut[iOff]);
    }
    btPutU16(&aOut[p->pgsz-2], nCell+1);
    btPutU16(&aOut[p->pgsz-6], iOff);

    if( (iCell+1)==p->anOut[p->iOut] ){
      /* That was the last cell for this page. Fill in the rest of the 
      ** output page footer and the flags byte at the start of the page.  */

  int nReq = 0;                   /* Space required for type (a) cells */
  int iCell;                      /* Position to insert new key */
  int iWrite;                     /* Byte offset at which to write new cell */
  int i;
  int bLeaf;                      /* True if inserting into leaf page */
  BtPage *pLeaf;

  bLeaf = (apKV[0].pgno==0);
  assert( bLeaf==0 || nKV==1 );

  /* Determine the number of bytes of space required on the current page. */
  for(i=0; i<nKV; i++){
    nReq += btKVCellSize(&apKV[i]) + 2;
  }

  iCell = pCsr->aiCell[pCsr->nPg-1];
  assert( pCsr->nPg>0 );
  pLeaf = pCsr->apPage[pCsr->nPg-1];
  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, pgsz) & BT_PGFLAGS_INTERNAL))!=bLeaf ){
    return btErrorBkpt(SQLITE4_CORRUPT);
  }

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

      }

      for(i=0; i<nKV; i++){
        /* Write the cell pointer */
        btPutU16(btCellPtrFind(aData, pgsz, iCell+i), iWrite);
      
        /* Write the cell itself */
        iWrite += btKVCellWrite(&apKV[i], &aData[iWrite]);
      }

      /* Set the new total free space */
      if( nCell==0 ){
        btPutU16(&aData[pgsz-4], nFree - nReq);
      }else{
        btPutU16(&aData[pgsz-4], btFreeSpace(aData, pgsz) - nReq);

    }
  }

  if( nV>=0 && (rc==SQLITE4_NOTFOUND || rc==SQLITE4_INEXACT) ){
    /* 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 = btAssignOverflow(db, &kv);
    if( rc==SQLITE4_OK ){
      rc = btInsertAndBalance(&csr, 1, &kv);
    }
  }

  return rc;
}

int sqlite4BtDelete(bt_cursor *pCsr){
  int rc;
  rc =  btDeleteFromPage(pCsr, 1);
  if( rc==SQLITE4_OK ){
    rc = btBalanceIfUnderfull(pCsr);
  }
  return rc;
}

int sqlite4BtSetCookie(bt_db *db, unsigned int iVal){
  return sqlite4BtPagerSetCookie(db->pPager, iVal);
}

int sqlite4BtGetCookie(bt_db *db, unsigned int *piVal){
  return sqlite4BtPagerGetCookie(db->pPager, piVal);
}

that use overflow pages for the value only, and (c) cells that use overflow
pages for the key and value.

<p>Cell type (a):
<ul>
  <li> Number of bytes of the entries key (nKey), as a varint.
  <li> nKey bytes of key data.
  <li> Number of bytes in entries value plus one (nValue+1), as a varint.
  <li> nValue bytes of value data.
</ul>

<p>Cell type (b):
<ul>
  <li> Number of bytes in entries key (nKey), as a varint.
  <li> nKey bytes of key data.
  <li> Single 0x00 byte.
  <li> Number of bytes in entries value stored locally (nLVal), as a varint.
  <li> nLVal bytes of value data.
  <li> Number of bytes in entries value stored on overflow pages, as a varint.
</ul>

<p>Cell type (c):
<ul>
  <li> Single 0x00 byte.
  <li> Number of bytes of the entries key (nLocalKey) stored locally, 
       as a varint.
  <li> nLocalKey bytes of key data.
  <li> Number of bytes of the entries key stored on overflow pages, as a varint.
  <li> Number of bytes in the entries value, as a varint.
</ul>

<p>Cell types (b) and (c) are followed by an array of pointers to overflow
pages. The overflow data for a single entry is distributed between up to 16
"direct" overflow pages and a single overflow tree. A direct overflow page