SQLite4  Check-in [a5186d0b0a]

  if( nKLocal==0 ){
    /* Type (c) leaf cell. */
    pCell += sqlite4BtVarintGet32(pCell, &nKLocal);
    pKLocal = pCell;
    pCell += nKLocal;
    pCell += sqlite4BtVarintGet32(pCell, &nKOvfl);
    pCell += sqlite4BtVarintGet32(pCell, &nVOvfl);
    nVOvfl -= 1;

  }else{
    pKLocal = pCell;
    pCell += nKLocal;
    pCell += sqlite4BtVarintGet32(pCell, &nVLocal);
    if( nVLocal==0 ){
      /* Type (b) */
      pCell += sqlite4BtVarintGet32(pCell, &nVLocal);
      pVLocal = pCell;
      pCell += nVLocal;
      pCell += sqlite4BtVarintGet32(pCell, &nVOvfl);
    }else{
      /* Type (a) */
      pVLocal = pCell;
      nVLocal -= 2;
    }
  }

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

  nReq = pCsr->ovfl.nKey + pCsr->ovfl.nVal;

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

  if( n==0 ){
    /* Type (c) cell */
    pCell += sqlite4BtVarintGet32(pCell, &n);
    pCell += n;
    pCell += sqlite4BtVarintGet32(pCell, &n);
    pCell += sqlite4BtVarintGet32(pCell, &n);
    pOvfl = pCell;
  }else{
    pCell += n;
    pCell += sqlite4BtVarintGet32(pCell, &n);
    if( n==0 ){
      /* Type (b) cell */
      pCell += sqlite4BtVarintGet32(pCell, &n);
      pCell += n;
      pCell += sqlite4BtVarintGet32(pCell, &n);
      pOvfl = pCell;
    }
  }

      pCell += sqlite4BtVarintGet32(pCell, &nK);
      if( nK>0 ){
        pCell += nK;
        pCell += sqlite4BtVarintGet32(pCell, &nV);
      }

      if( nV==0 ){
        /* Type (b) or (c) cell */
        rc = btCsrBuffer(pCsr, 1);
        if( rc==SQLITE4_OK ){
          u8 *aBuf = (u8*)pCsr->ovfl.buf.p;
          *ppV = &aBuf[pCsr->ovfl.nKey];
          *pnV = pCsr->ovfl.nVal;
        }
      }else{
        /* Type (a) cell */
        *ppV = pCell;
        *pnV = (nV-2);
      }

#ifndef NDEBUG
      if( rc==SQLITE4_OK ){
        const void *pK; int nK;
        rc = sqlite4BtCsrKey((bt_cursor*)pCsr, &pK, &nK);
        if( rc==SQLITE4_OK ){

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

  p += sqlite4BtVarintGet32(p, &nKey);
  if( bLeaf==0 ){
    /* Internal page cell */
    p += nKey;
    p += 4;
  }else if( nKey==0 ){
    /* Type (c) cell */
    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 ){
      /* Type (b) cell */
      p += sqlite4BtVarintGet32(p, &nKey);
      p += nKey;
      p += sqlite4BtVarintGet32(p, &nKey);
      p += btOverflowArrayLen(p);
    }else{
      /* Type (a) cell */
      p += (nKey-2);
    }
  }

  return (p-pCell);
}

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

  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.
**
** If the KeyValue is not already in KV_CELL form, then assume it will
** be formatted as a type (a) cell.
*/
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+2)
        + 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+2);
      memcpy(&aBuf[i], pKV->pV, pKV->nV); 
      i += pKV->nV;
    }else{
      btPutU32(&aBuf[i], pKV->pgno);
      i += 4;
    }
  }

  assert( i==btKVCellSize(pKV) );

  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 = btKVCellSize(pKV);
  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.  
    **
    ** This expression uses sqlite4BtVarintLen32() to calculate an upper
    ** bound for the size of the varint that indicates the number of bytes
    ** of the value stored locally.  */
    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. */
      int nLKey = nMaxSize 
          - 1                                    /* 0x00 byte */
          - sqlite4BtVarintLen32(pgsz)           /* nLKey */
          - sqlite4BtVarintLen32(pKV->nK)        /* nOKey */
          - sqlite4BtVarintLen32(pKV->nV+1)      /* nVal */
          - nArraySz;                            /* overflow array */

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

    /* Allocate a pager buffer to store the KV_CELL buffer. Using a pager

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

      rc = btOverflowArrayPopulate(db, &pOut,
          (u8*)(pKV->pK) + nLKey, nKeyOvfl,
          (u8*)(pKV->pV) + nLVal, nValOvfl
      );
      if( rc==SQLITE4_OK ){
        memset(pKV, 0, sizeof(*pKV));

    rc = btBalance(pCsr, bLeaf, 0, 0);
  }
  return rc;
}

static int btSaveAllCursor(bt_db *pDb, BtCursor *pCsr){
  int rc = SQLITE4_OK;            /* Return code */
  bt_cursor *pIter;               /* Used to iterate through cursors */

  for(pIter=pDb->pAllCsr; rc==SQLITE4_OK && pIter; pIter=pIter->pNextCsr){
    if( pIter->eType==CSR_TYPE_BT ){
      BtCursor *p = (BtCursor*)pIter;
      if( p->nPg>0 ){
        assert( p->bRequireReseek==0 );
        rc = btCsrBuffer(p, 0);
        if( rc==SQLITE4_OK ){
          assert( p->ovfl.buf.p );
          p->bRequireReseek = 1;
          if( p!=pCsr ) btCsrReleaseAll(p);
        }
      }
    }else{
      /* ?? */
    }

  }

  return rc;
}

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

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+2), as a varint. Or,
       for a delete key, a single 0x01 byte.
  <li> Unless this is a delete key, 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 plus one (nValue+1), as a varint.
       Or, for a delete key, a single 0x00 byte.
</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
is just that - a pointer to an overflow page that contains data. An overflow
"tree" is a tree structure where leaves contain overflow data and nodes