SQLite4  Check-in [4431ab2769]

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, nData));



  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");


}
#endif


/*
** This function compares the key passed via parameters pK and nK to the
** key stored as part of cell iCell on the database page stored in buffer

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);
  p += nKey;

  p += sqlite4BtVarintGet32(p, &nKey);
  p += nKey;




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

/*
** Allocate a new page buffer.

  int i;                          /* Used to iterate through cells */
  int bLeaf;                      /* True if pPg is a leaf page */
  int nHdr;                       /* Bytes in header of this page */

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

  aData = sqlite4BtPageData(pPg);
  assert( (btFlags(aData, pgsz) & BT_PGFLAGS_INTERNAL)==0 );
  nCell = btCellCount(aData, pgsz);

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

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

    i += 4;
  }

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
























static int btGatherSiblings(bt_cursor *pCsr, BtPage **apPg, int *pnPg){

  bt_db * const pDb = pCsr->pDb; 
  const int pgsz = sqlite4BtPagerPagesize(pDb->pPager);

  int rc = SQLITE4_OK;
  int nCell;                      /* Number of cells in parent page */
  u8 *aParent;                    /* Buffer of parent page */
  int iChild;                     /* Index of child page */

  }else{
    nSib = 3;
  }

  if( iChild==0 ){
    iSib = 0;
  }else if( iChild==nCell ){
    iSib = nCell-2;
  }else{
    iSib = iChild-1;
  }

  for(i=0; i<nSib && rc==SQLITE4_OK; i++){
    u32 pgno = btChildPgno(aParent, pgsz, iSib+i);
    rc = sqlite4BtPageGet(pDb->pPager, pgno, &apPg[i]);
  }
  *pnPg = nSib;

  pCsr->aiCell[pCsr->nPg-2] = iSib;
  return rc;
}

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

      btPutU32(pCell, pgno);
    }
  }

  return rc;
}

/* Called recursively by btInsertAndRebalance(). todo: Fix this! */

static int btModifyPage(bt_cursor *, int, int, KeyValue *);

#if 0
/*
** An instance of this structure describes all the cells that will be
** redistributed by a single btRebalance() operation.
*/
struct BtRBIter {
  /* Input parameters */
  int nKV;                        /* Number of KV pairs to add */
  int iKVSib;                     /* Sibling to add KV pairs to */
  int iKVCell;                    /* Cell offset at which to insert KV pairs */

  int nSib;                       /* Number of sibling pages */
  /* Output parameters */
  int iSib;                       /* Sibling page to read cell from (or -ve) */
  int iCell;                      /* Index of cell to read (or -ve) */
};
static void btIterFirst(BtRBIter *p){
  p->iSib = 0;
  p->iCell = 0;
  if( p->iKVSib==0 && p->iKVCell==0 ){
    p->iSib = -1;
  }
}
static void btIterNext(BtRBIter *p){
  p->iCell++;
  /* Just finished the KV pairs */
  if( p->iSib<0 && p->iCell==p->nKV ){
    p->iSib = p->iKVSib;
    p->iCell = p->iKVCell;
  }
  if( p->iSib && p->iCell==p->nKV ){
    p->iSib = p->iKVSib;
    p->iCell = p->iKVCell;
  }
}

static void btIterValid(BtRBIter *p){
  return (p->iSib>=0 || p->iCell>=0);
}
#endif


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 */
  KeyValue *pKV;                  /* New KV pair being inserted */
  int nPg;                        /* Number of sibling pages */
  BtPage **apPg;                  /* Array of sibling pages */
  
  int *anOut;                     /* Cell counts for output pages */

  /* Variables used by btBalanceOutput */
  int nOut;                       /* Number of output pages */
  int iOut;                       /* Current output page */
  u8 **apOut;                     /* Buffers to assemble output in */


};

static int btBalanceOutput(
  BalanceCtx *p,                  /* Description of balance operation */
  int iCell,                      /* Cell number in this iteration */
  u8 *pCell, int nByte,           /* Binary cell to copy to output */
  KeyValue *pKV                   /* Key-value cell to write to output */
){

  BalanceCtx *p,
  int (*xVisit)(BalanceCtx*, int, u8*, int, KeyValue*)
){
  const int pgsz = sqlite4BtPagerPagesize(p->pCsr->pDb->pPager);
  int rc = SQLITE4_OK;            /* Return code */
  int iPg;                        /* Current page in apPg[] */
  int iCall = 0;


  BtPage *pIns = p->pCsr->apPage[p->pCsr->nPg-1];
  int iIns = p->pCsr->aiCell[p->pCsr->nPg-1];

  for(iPg=0; iPg<p->nPg && rc==SQLITE4_OK; iPg++){
    BtPage *pPg;                  /* Current page */
    u8 *aData;                    /* Page data */
    int nCell;                    /* Number of cells on page pPg */
    int iCell;                    /* Current cell in pPg */

    pPg = p->apPg[iPg];
    aData = sqlite4BtPageData(pPg);
    nCell = btCellCount(aData, pgsz);

    for(iCell=0; iCell<nCell && rc==SQLITE4_OK; iCell++){
      int nByte;
      u8 *pCell;

      if( pPg==pIns && iCell==iIns ){

        rc = xVisit(p, iCall++, 0, 0, p->pKV);
        if( rc!=SQLITE4_OK ) break;

      }

      pCell = btCellFindSize(aData, pgsz, iCell, &nByte);
      rc = xVisit(p, iCall++, pCell, nByte, 0);
    }

    if( pPg==pIns && iCell==nCell && rc==SQLITE4_OK ){

      rc = xVisit(p, iCall++, 0, 0, p->pKV);

    }
  }

  return rc;
}


int btInsertAndRebalance(bt_cursor *pCsr, KeyValue *pKV){
  bt_db * const pDb = pCsr->pDb; 
  const int pgsz = sqlite4BtPagerPagesize(pDb->pPager);
  const int nSpacePerPage = (pgsz - 1 - 6);

  int nIn = 0;                    /* Number of input pages */
  int iPg;                        /* Used to iterate through pages */
  int iCell;                      /* Used to iterate through cells */
  int nCell = 0;                  /* Total number of cells to redistribute */
  int *anCellSz;                  /* Array containing size in bytes of cells */
  int iIns;                       /* Index of new cell */
  int nOut;                       /* Number of output pages */

  BtPage *apPg[5];                /* Input/output pages */
  int anOut[5];                   /* Cell counts for output pages */
  u8 *apOut[5];                   /* Buffers for assembly of output pages */
  KeyValue aPCell[5];             /* Cells to push into the parent page */
  BtPage *pPar;                   /* Parent page */
  int iSib;                       /* Index of left-most sibling */

  int nTotal;                     /* Total bytes of content to distribute */
  int rc = SQLITE4_OK;            /* Return code */

  BalanceCtx ctx;







  memset(apOut, 0, sizeof(apOut));
  memset(apPg, 0, sizeof(apPg));
  memset(aPCell, 0, sizeof(aPCell));

  /* Gather the sibling pages from which cells will be redistributed into
  ** the apPg[] array.  */
  assert( pCsr->nPg>1 );
  rc = btGatherSiblings(pCsr, apPg, &nIn);
  if( rc!=SQLITE4_OK ) goto rebalance_out;
  pPar = pCsr->apPage[pCsr->nPg-2];
  iSib = pCsr->aiCell[pCsr->nPg-2];

  /* Count the cells on the input pages. This loop also sets iNew. */

  for(iPg=0; iPg<nIn; iPg++){
    u8 *aData = sqlite4BtPageData(apPg[iPg]);
    if( apPg[iPg]==pCsr->apPage[pCsr->nPg-1] ){
      iIns = (nCell + pCsr->aiCell[pCsr->nPg-1]);
      nCell++;
    }
    nCell += btCellCount(aData, pgsz);
  }

  /* Allocate and populate the anCellSz[] array */
  anCellSz = (int*)sqlite4_malloc(pDb->pEnv, sizeof(int) * nCell);
  if( anCellSz==0 ){
    rc = btErrorBkpt(SQLITE4_NOMEM);
    goto rebalance_out;
  }
  iCell = 0;
  for(iPg=0; iPg<nIn; iPg++){
    u8 *aData = sqlite4BtPageData(apPg[iPg]);
    int nPgCell;                  /* Number of cells on page apPg[iPg] */
    int iPgCell;                  /* Index of cell under analysis in page */

    nPgCell = btCellCount(aData, pgsz);
    for(iPgCell=0; iPgCell<nPgCell; iPgCell++){
      if( iCell==iIns ) iCell++;
      btCellFindSize(aData, pgsz, iCell, &anCellSz[iCell]);
      iCell++;
    }
  }
  anCellSz[iIns] = btKVCellSize(pKV, pgsz);

  /* Now figure out the number of output pages. Set nOut to this value. */
  iCell = 0;
  for(iPg=0; iCell<nCell; iPg++){
    int nByte = 0;                /* Number of bytes of content on page */
    assert( iPg<array_size(anOut) );
    for(/* noop */; iCell<nCell; iCell++){
      nByte += (anCellSz[iCell] + 2);
      if( nByte>nSpacePerPage ) break;
    }
    anOut[iPg] = iCell;
  }
  nOut = iPg;
  assert( anOut[nOut-1]==nCell );

  /* Calculate the total size of all cells. */
  nTotal = 0;
  for(iCell=0; iCell<nCell; iCell++) nTotal += (anCellSz[iCell] + 2);

  /* The loop in the previous block populated the anOut[] array in such a
  ** way as to make the (nOut-1) leftmost pages completely full but leave
  ** the rightmost page partially empty. This block redistributes cells
  ** a bit more evenly. This block may reduce one or more of the values in 
  ** the anOut[] array, but will not increase any. No values are reduced
  ** to values lower than 1.  */
  iCell = nCell;
  for(iPg=(nOut-2); iPg>=0; iPg--){
    int nByte = 0;                /* Number of bytes of content on page */
    int nGoal = nTotal / (iPg + 2);

    for(/* noop */; iCell>0 && ((nByte<nGoal) || iCell>anOut[iPg]); iCell--){
      int nThis = (anCellSz[iCell-1] + 2);
      if( (nThis + nByte)>nSpacePerPage ) break;
      nByte += nThis;
    }
    assert( iCell<=anOut[iPg] );
    anOut[iPg] = iCell;
    nTotal = nByte;
  }

#ifndef NDEBUG
  {
    int iDbg;
    fprintf(stderr, "btInsertAndRebalance(): nIn=%d anIn[] = ", nIn);
    for(iDbg=0; iDbg<nIn; iDbg++){
      u8 *aData = sqlite4BtPageData(apPg[iDbg]);
      fprintf(stderr, "%d ", btCellCount(aData, pgsz));
    }
    fprintf(stderr, " ->  nOut=%d anOut[] = ", nOut);
    for(iDbg=0; iDbg<nOut; iDbg++){
      fprintf(stderr, "%d ", anOut[iDbg]);
    }
    fprintf(stderr, "\n");
  }
#endif

  /* Allocate buffers for the output leaves */
  for(iPg=0; iPg<nOut; iPg++){
    rc = btNewBuffer(pDb, &apOut[iPg]);
    if( rc!=SQLITE4_OK ) goto rebalance_out;
    memset(apOut[iPg] + pgsz-6, 0, 6);
  }


  memset(&ctx, 0, sizeof(ctx));
  ctx.pCsr = pCsr;
  ctx.pKV = pKV;
  ctx.nPg = nIn;
  ctx.apPg = apPg;
  ctx.pgsz = pgsz;
  ctx.bLeaf = 1; /* todo */

  ctx.anOut = anOut;
  ctx.nOut = nOut;
  ctx.apOut = apOut;
  rc = btBalanceVisitCells(&ctx, btBalanceOutput);

#if 0
  /* Populate buffers for the output leaves */
  iPgOut = 0;                     /* Index of current output page */
  iPgIn = 0;                      /* Index of current input page */
  iPgInFirst = 0;
  for(iCell=0; iCell<nCell; iCell++){
    int iOff;                     /* Output page offset */
    u8 *aOut;                     /* Output page buffer */

    if( iCell==anOut[iPg] ) iPg++;
    aOut = apOut[iPg];
    iOff = btFreeOffset(aOut, pgsz);

    if( iCell==iIns ){
      iOff += btKVCellWrite(pKV, pgsz, &aOut[iOff]);
    }else{
      u8 *aIn = sqlite4BtPageData(apPg[iPgIn]);
      int iPgCell = iCell - iPgInFirst;
      if( iPgCell>=btCellCount(aIn, pgsz) ){
        iPgIn++;
        iPgInFirst = iCell;
        iPgCell = 0;
        aIn = sqlite4BtPageData(apPg[iPgIn]);
      }
      memcpy(&aOut[iOff], btCellFind(aIn, pgsz, iPgCell), anCellSz[iCell]);
      iOff += anCellSz[iCell];
    }
    btPutU16(&aOut[pgsz-6], iOff);
  }
  for(iPg=0; iPg<nOut; iPg++){
    u8 *aData = apOut[iPg];
    btPutU16(&aData[pgsz-2], anOut[iPg] - (iPg==0 ? 0 : anOut[iPg-1]));
  }
#endif

  /* Clobber the old pages with the new buffers */
  for(iPg=0; iPg<nOut; iPg++){
    if( iPg>=nIn ){
      rc = sqlite4BtPageAllocate(pDb->pPager, &apPg[iPg]);
      if( rc!=SQLITE4_OK ) goto rebalance_out;
    }
    btSetBuffer(pDb, apPg[iPg], apOut[iPg]);
    apOut[iPg] = 0;
  }

#ifndef NDEBUG
  {
    int iDbg;
    for(iDbg=0; iDbg<nOut; iDbg++){
      u8 *aData = sqlite4BtPageData(apPg[iDbg]);
      printPage(stderr, sqlite4BtPagePgno(apPg[iDbg]), aData, pgsz);
    }
  }
#endif

  /* The leaves are written. Now gather the keys and page numbers to
  ** push up into the parent page.  */ 
  for(iPg=0; iPg<(nOut-1); iPg++){
    u8 *aData = sqlite4BtPageData(apPg[iPg]);
    u8 *pCell;

    pCell = btCellFind(aData, pgsz, btCellCount(aData, pgsz)-1);
    aPCell[iPg].pgno = sqlite4BtPagePgno(apPg[iPg]);
    pCell += sqlite4BtVarintGet32(pCell, &aPCell[iPg].nK);
    aPCell[iPg].pK = pCell;
  }

  assert( nIn==1 && nOut==2 );
  rc = btSetChildPgno(pDb, pPar, iSib+nIn-1, sqlite4BtPagePgno(apPg[nOut-1]));


  if( rc!=SQLITE4_OK ) goto rebalance_out;

  pCsr->nPg--;
  rc = btModifyPage(pCsr, nIn-1, nOut-1, aPCell);

  if( rc!=SQLITE4_OK ) goto rebalance_out;










 rebalance_out:
  for(iPg=0; iPg<nIn; iPg++){
    sqlite4BtPageRelease(apPg[iPg]);
  }
  return rc;
}

static int btExtendTree(bt_cursor *pCsr){
  bt_db * const pDb = pCsr->pDb;
  const int pgsz = sqlite4BtPagerPagesize(pDb->pPager);

**       deleted from the page.
**
**     * nKV entries are inserted in their place.
**
** The tree balancing routine is called if this causes the page to
** become either overfull or to contain no entries at all.
*/
static int btModifyPage(
  bt_cursor *pCsr,                /* Cursor identifying page to modify */
  int nDel,                       /* Number of entries to delete from page */
  int nKV,                        /* Number of entries in apKV */
  KeyValue *apKV                  /* New cells to insert into the page */
){
  int rc = SQLITE4_OK;
  const int pgsz = sqlite4BtPagerPagesize(pCsr->pDb->pPager);
  u8 *aData;                      /* Page buffer */
  int nCell;                      /* Number of cells on this page already */
  int nFree;                      /* Contiguous free space on this page */
  int nReq;                       /* Space required for type (a) cell */
  int iCell;                      /* Position to insert new key */
  int iWrite;                     /* Byte offset at which to write new cell */

  BtPage *pLeaf;
  KeyValue *pKV;

  assert( nDel==0 && nKV==1 );
  pKV = apKV;

  /* Bytes of space required on the current page. */

  nReq = btKVCellSize(pKV, pgsz) + 2;


  iCell = pCsr->aiCell[pCsr->nPg-1];
  assert( pCsr->nPg>0 );
  pLeaf = pCsr->apPage[pCsr->nPg-1];
  aData = (u8*)sqlite4BtPageData(pLeaf);

  nCell = btCellCount(aData, pgsz);

    }

    iWrite = btFreeOffset(aData, pgsz);
    nFree = btFreeContiguous(aData, pgsz);
  }

  if( nFree>=nReq ){
    /* The new entry will fit on the leaf page. So in this case all there
    ** is to do is update this single page. The easy case. */
    rc = sqlite4BtPageWrite(pLeaf);
    if( rc==SQLITE4_OK ){
      aData = sqlite4BtPageData(pLeaf);

      /* Update the cell pointer array */
      if( iCell!=nCell ){
        u8 *aFrom = btCellPtrFind(aData, pgsz, nCell-1);
        u8 *aTo = btCellPtrFind(aData, pgsz, nCell);
        memmove(aTo, aFrom, (nCell-iCell) * 2);
      }
      btPutU16(btCellPtrFind(aData, pgsz, iCell), iWrite);


      /* Increase cell count */
      btPutU16(&aData[pgsz-2], nCell+1);
      
      /* Write the cell itself */
      iWrite += btKVCellWrite(pKV, pgsz, &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);
      }




      /* Set the offset to the block of empty space */
      btPutU16(&aData[pgsz-6], iWrite);
    }

  }else{
    /* The new entry will not fit on the leaf page. Entries will have
    ** to be shuffled between existing leaves and new leaves may need
    ** to be added to make space for it. */
    if( pCsr->nPg==1 ){
      rc = btExtendTree(pCsr);
    }
    if( rc==SQLITE4_OK ){
      rc = btInsertAndRebalance(pCsr, pKV);
    }
  }

  return rc;
}

static int btRemoveFromLeaf(bt_cursor *pCsr){
  const int pgsz = sqlite4BtPagerPagesize(pCsr->pDb->pPager);
  int rc = SQLITE4_OK;            /* Return code */
  BtPage *pLeaf;                  /* Leaf page */

  pLeaf = pCsr->apPage[pCsr->nPg-1];
  rc = sqlite4BtPageWrite(pLeaf);
  if( rc==SQLITE4_OK ){

    u8 *aData;                    /* Page buffer */
    int nCell;                    /* Number of cells initially on this page */
    int iDel;                     /* Index of cell to delete */
    int nByte;                    /* Size of cell to delete in bytes */


    iDel = pCsr->aiCell[pCsr->nPg-1];
    aData = (u8*)sqlite4BtPageData(pLeaf);
    nCell = btCellCount(aData, pgsz);



    btCellFindSize(aData, pgsz, iDel, &nByte);


    if( iDel<(nCell-1) ){

      u8 *aTo = btCellPtrFind(aData, pgsz, nCell-2);
      u8 *aFrom = btCellPtrFind(aData, pgsz, nCell-1);
      memmove(aTo, aFrom, 2*(nCell-iDel-1));
    }

    /* Decrease cell count */
    btPutU16(&aData[pgsz-2], nCell-1);

    /* Increase total free space */
    btPutU16(&aData[pgsz-4], btFreeSpace(aData, pgsz) + nByte + 2);
  }
  
  return rc;
}

/*
** Insert a new key/value pair or replace an existing one.

  if( rc==SQLITE4_NOTFOUND || rc==SQLITE4_INEXACT ){
    /* Insert the new KV pair into the current leaf. */
    KeyValue kv;
    kv.pgno = 0;
    kv.pK = pK; kv.nK = nK;
    kv.pV = pV; kv.nV = nV;
    rc = btModifyPage(&csr, 0, 1, &kv);
  }

  return rc;
}

int sqlite4BtDelete(bt_cursor *pCsr){
  return btRemoveFromLeaf(pCsr);
}

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);
}

  SELECT a, length(b) FROM t1 
} {1 200  3 200  4 200  5 200}

do_execsql_test 2.4 {
  INSERT INTO t1 VALUES(6, $val);
}

breakpoint
do_execsql_test 2.5 { 
  SELECT a, length(b) FROM t1 
} {1 200  3 200  4 200  5 200  6 200}






















finish_test