SQLite4  Check-in [27248a1ebc]

/*
** This struct defines the format of database "schedule" pages.
*/
typedef struct BtSchedule BtSchedule;
struct BtSchedule {
  u32 bBusy;                      /* True if page is busy */
  u32 iAge;                       /* Age of input segments */
  u32 iMinLevel;                  /* Minimum level of input segments */
  u32 iMaxLevel;                  /* Maximum level of input segments */
  u32 iOutLevel;                  /* Level at which to write output */
  u32 aBlock[32];                 /* Allocated blocks */

  u32 iNextPg;                    /* Page that contains next input key */
  u32 iNextCell;                  /* Cell that contains next input key */
  u32 nUsed;                      /* Number of output blocks actually written */
  u32 iFreeList;                  /* First page of new free-list (if any) */
};

int sqlite4BtMerge(bt_db *db, BtDbHdr *pHdr, u8 *aSched);

/*************************************************************************
** Interface to bt_pager.c functionality.
*/
typedef struct BtPage BtPage;
typedef struct BtPager BtPager;

    + (pLog->snapshot.aLog[0]!=0)
    + (int)pLog->snapshot.aLog[3] - (int)pLog->snapshot.aLog[2]
    + (pLog->snapshot.aLog[3]!=0)
    + (int)pLog->snapshot.aLog[5] - (int)pLog->snapshot.aLog[4]
    + (pLog->snapshot.aLog[5]!=0)
  ;
}

static int btLogMerge(BtLog *pLog, u8 *aBuf){
  bt_db *db = (bt_db*)sqlite4BtPagerExtra((BtPager*)pLog->pLock);
  return sqlite4BtMerge(db, &pLog->snapshot.dbhdr, aBuf);
}


int sqlite4BtLogCheckpoint(BtLog *pLog, int nFrameBuffer){
  BtLock *pLock = pLog->pLock;
  int rc;

  /* Take the CHECKPOINTER lock. */
  rc = sqlite4BtLockCkpt(pLock);

      /* Copy data from the log file to the database file. */
      for(i=0; rc==SQLITE4_OK && i<nPgno; i++){
        u32 pgno = aPgno[i];
        rc = btLogRead(pLog, pgno, aBuf, iLast);
        if( rc==SQLITE4_OK ){
          i64 iOff = (i64)pgsz * (pgno-1);
          if( pgno==1 ){
            rc = btLogUpdateDbhdr(pLog, aBuf);
          }else if( pgno==pLog->snapshot.dbhdr.iSRoot ){
            rc = btLogMerge(pLog, aBuf);
          }
          if( rc==SQLITE4_OK ){
            btDebugCkptPage(pLog->pLock, pgno, aBuf, pgsz);
            rc = pVfs->xWrite(pFd, iOff, aBuf, pgsz);
          }
        }else if( rc==SQLITE4_NOTFOUND ){
          rc = SQLITE4_OK;
        }
      }

      /* Sync the database file to disk. */
      if( rc==SQLITE4_OK ){

#define BT_PGFLAGS_METATREE 0x02  /* True for a meta-tree page */
#define BT_PGFLAGS_SCHEDULE 0x04  /* True for a schedule-tree page */

/* #define BT_STDERR_DEBUG 1 */

typedef struct BtCursor BtCursor;
typedef struct FiCursor FiCursor;
typedef struct FiSubCursor FiSubCursor;

struct bt_db {
  sqlite4_env *pEnv;              /* SQLite environment */
  BtPager *pPager;                /* Underlying page-based database */
  bt_cursor *pAllCsr;             /* List of all open cursors */
  int nMinMerge;
  int nScheduleAlloc;

  int bSkipNext;                  /* True if next CsrNext() is a no-op */
  int bSkipPrev;                  /* True if next CsrPrev() is a no-op */
};

/*
** Database f-tree cursor handle.
*/
struct FiSubCursor {
  u8 aPrefix[8];                  /* Meta-tree key prefix for this age/level */
  BtCursor mcsr;                  /* Cursor opened on meta table (scans only) */
  BtCursor csr;                   /* Cursor opened on current sub-tree */
};
struct FiCursor {
  bt_cursor base;                 /* Base cursor class */
  int iBt;                        /* Current sub-tree (or -1 for EOF) */
  int nBt;                        /* Number of entries in aBt[] array */
  FiSubCursor *aSub;              /* Array of sub-tree cursors */
};

/*
** TODO: Rearrange things so these are not required!
*/
static int fiLoadSummary(bt_db *, BtCursor *, const u8 **, int *);
static void btReadSummary(const u8 *, int, u16 *, u16 *, u16 *);
static int btCsrData(BtCursor *, int, int, const void **, int *);

/* 
** Meta-table summary key.
*/
static const u8 aSummaryKey[] = {0xFF, 0xFF, 0xFF, 0xFF};

#ifndef btErrorBkpt
int btErrorBkpt(int rc){

    if( IsBtCsr(pCsr) ){
      BtCursor *p = (BtCursor*)pCsr;
      if( p->nPg>0 ) nExpect += p->nPg;
    }else{
      FiCursor *p = (FiCursor*)pCsr;
      int i;
      for(i=0; i<p->nBt; i++){
        if( p->aSub[i].mcsr.nPg>0 ) nExpect += p->aSub[i].mcsr.nPg;
        if( p->aSub[i].csr.nPg>0 ) nExpect += p->aSub[i].csr.nPg;
      }
    }
  }
  nActual = sqlite4BtPagerRefcount(pDb->pPager);
  assert( nActual==nExpect );
}
#else

  pCsr->bRequireReseek = 0;
}

static void fiCsrReset(FiCursor *pCsr){
  int i;
  bt_db *db = pCsr->base.pDb;
  for(i=0; i<pCsr->nBt; i++){
    btCsrReset(&pCsr->aSub[i].csr, 1);
    btCsrReset(&pCsr->aSub[i].mcsr, 1);
  }
  sqlite4_free(db->pEnv, pCsr->aSub);
  pCsr->aSub = 0;
  pCsr->nBt = 0;
  pCsr->iBt = -1;
}

int sqlite4BtCsrClose(bt_cursor *pCsr){
  if( pCsr ){
    bt_cursor **pp;

    int i;
    BtSchedule s;
    sqlite4BtBufAppendf(pBuf, "(schedule page) ");
    memcpy(&s, aData, sizeof(s));

    sqlite4BtBufAppendf(pBuf, "bBusy=%d ", (int)s.bBusy);
    sqlite4BtBufAppendf(pBuf, "iAge=%d ", (int)s.iAge);
    sqlite4BtBufAppendf(pBuf, "iMinLevel=%d ", (int)s.iMinLevel);
    sqlite4BtBufAppendf(pBuf, "iMaxLevel=%d ", (int)s.iMaxLevel);
    sqlite4BtBufAppendf(pBuf, "iOutLevel=%d ", (int)s.iOutLevel);
    sqlite4BtBufAppendf(pBuf, "blocks=(");
    for(i=0; s.aBlock[i] && i<array_size(s.aBlock); i++){
      sqlite4BtBufAppendf(pBuf, "%s%d", i==0 ? "" : " ", (int)s.aBlock[i]);
    }
    sqlite4BtBufAppendf(pBuf, ") ");

**
** In other words, *piRes is +ve, zero or -ve if C is respectively larger, 
** equal to or smaller than K.
*/
static int btCellKeyCompare(
  BtCursor *pCsr,                 /* Cursor handle */
  int bLeaf,                      /* True if cursor currently points to leaf */
  const void *aPrefix,
  const void *pK, int nK,         /* Key to compare against cursor key */
  int *piRes                      /* OUT: Result of comparison */
){
  const void *pCsrKey;
  int nCsrKey;
  int nCmp;
  int nAscend = 0;

        iCell = 0;
      }
      rc = sqlite4BtCsrKey((bt_cursor*)pCsr, &pCsrKey, &nCsrKey);
    }
  }

  if( rc==SQLITE4_OK ){
    if( aPrefix ){
      if( nCsrKey<8 ){
        res = 1;
        goto keycompare_done;
      }
      res = memcmp(pCsrKey, aPrefix, 8);
      if( res ) goto keycompare_done;
      nCsrKey -= 8;
      pCsrKey = (void*)(((u8*)pCsrKey) + 8);
    }

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

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

#define BT_CSRSEEK_SEEK   0
#define BT_CSRSEEK_UPDATE 1
#define BT_CSRSEEK_RESEEK 2

static int btCsrSeek(
  BtCursor *pCsr,                 /* Cursor object to seek */
  u8 *aPrefix,                    /* 8-byte key prefix, or NULL */
  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 eCsrseek
){
  const int pgsz = sqlite4BtPagerPagesize(pCsr->base.pDb->pPager);
  u32 pgno;                       /* Page number for next page to load */

        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, aPrefix, 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{

  }
  return rc;
}

static int btCsrReseek(BtCursor *pCsr){
  int rc = SQLITE4_OK;
  if( pCsr->bRequireReseek ){
    BtOvfl o;                     /* Copy of initial overflow buffer */
    memcpy(&o, &pCsr->ovfl, sizeof(BtOvfl));

    pCsr->ovfl.buf.n = 0;
    pCsr->ovfl.buf.p = 0;
    pCsr->bSkipNext = 0;
    pCsr->bRequireReseek = 0;

    rc = btCsrSeek(pCsr, 0, o.buf.p, o.nKey, BT_SEEK_EQ, BT_CSRSEEK_RESEEK);
    assert( rc!=SQLITE4_INEXACT );
    if( pCsr->ovfl.buf.p==0 ){
      pCsr->ovfl.buf.p = o.buf.p;
    }else{
      sqlite4_buffer_clear(&o.buf);
    }
  }
  return rc;
}

/*
** This function does the work of both sqlite4BtCsrNext() (if parameter

  return rc;
}


static int fiCsrAllocateSubs(bt_db *db, FiCursor *pCsr, int nBt){
  int rc = SQLITE4_OK;            /* Return code */
  if( nBt>pCsr->nBt ){
    int nByte = sizeof(FiSubCursor) * nBt;
    FiSubCursor *aNew;            /* Allocated array */

    aNew = (FiSubCursor*)sqlite4_realloc(db->pEnv, pCsr->aSub, nByte);
    if( aNew ){
      memset(&aNew[pCsr->nBt], 0, sizeof(FiSubCursor)*(nBt-pCsr->nBt));
      pCsr->aSub = aNew;
      pCsr->nBt = nBt;
    }else{
      rc = btErrorBkpt(SQLITE4_NOMEM);
    }
  }

  return rc;

  int rc = SQLITE4_OK;
  int mul;

  assert( pCsr->base.flags & (CSR_NEXT_OK | CSR_PREV_OK) );
  mul = ((pCsr->base.flags & CSR_NEXT_OK) ? 1 : -1);

  for(i=0; i<pCsr->nBt && rc==SQLITE4_OK; i++){
    BtCursor *pSub = &pCsr->aSub[i].csr;
    if( pSub->nPg>0 ){
      int nKey;
      const void *pKey;

      rc = btCsrKey(pSub, &pKey, &nKey);
      if( rc==SQLITE4_OK 
          && (iMin<0 || (mul * btKeyCompare(pKey, nKey, pMin, nMin))<0) 
        ){
        pMin = pKey;
        nMin = nKey;
        iMin = i;
      }
    }
  }

  if( iMin<0 && rc==SQLITE4_OK ) rc = SQLITE4_NOTFOUND;
  pCsr->iBt = iMin;
  return rc;
}

/*
** Return SQLITE4_OK if the cursor is successfully stepped, or 
** SQLITE4_NOTFOUND if an EOF is encountered.
**
** If an error occurs (e.g. an IO error or OOM condition), return the
** relevant error code.
*/
static int fiSubCsrStep( 
  FiCursor *pCsr,                 /* Parent cursor */
  FiSubCursor *pSub,              /* Sub-cursor to advance */
  int bNext                       /* True for xNext(), false for xPrev() */
){
  int rc;

  rc = btCsrStep(&pSub->csr, bNext);
  if( rc==SQLITE4_NOTFOUND ){
    const void *pV;
    int nV;
#if 1
    rc = btCsrKey(&pSub->mcsr, &pV, &nV);
    assert( rc==SQLITE4_OK && memcmp(pV, pSub->aPrefix, 8)==0 );
#endif
    rc = btCsrStep(&pSub->mcsr, bNext);

    if( rc==SQLITE4_OK ){
      rc = btCsrKey(&pSub->mcsr, &pV, &nV);
    }
    if( rc==SQLITE4_OK ){
      if( nV<sizeof(pSub->aPrefix) 
       || memcmp(pSub->aPrefix, pV, sizeof(pSub->aPrefix)) 
      ){
        rc = SQLITE4_NOTFOUND;
      }else{
        rc = btCsrData(&pSub->mcsr, 0, 4, &pV, &nV);
      }
    }
    if( rc==SQLITE4_OK ){
      BtDbHdr *pHdr = sqlite4BtPagerDbhdr(pCsr->base.pDb->pPager);
      pSub->csr.iRoot = btBlockToRoot(pHdr, sqlite4BtGetU32((const u8*)pV));
      rc = btCsrEnd(&pSub->csr, !bNext);
    }
  }

  return rc;
}

/*
** Advance the cursor. The direction (xPrev or xNext) is implied by the
** cursor itself - as fast-insert cursors may only be advanced in one
** direction.
*/
static int fiCsrStep(FiCursor *pCsr){
  int rc = SQLITE4_OK;
  int bNext = (0!=(pCsr->base.flags & CSR_NEXT_OK));
  const void *pKey; int nKey;     /* Current key that cursor points to */
  int i;

  assert( pCsr->base.flags & (CSR_NEXT_OK | CSR_PREV_OK) );
  assert( pCsr->iBt>=0 );

  do{
    /* Load the current key in to pKey/nKey. Then advance all sub-cursors 
    ** that share a key with the current sub-cursor. */
    rc = sqlite4BtCsrKey(&pCsr->base, &pKey, &nKey);
    for(i=0; rc==SQLITE4_OK && i<pCsr->nBt; i++){
      FiSubCursor *pSub = &pCsr->aSub[i];
      if( i!=pCsr->iBt && pSub->csr.nPg>0 ){
        const void *p; int n;       /* Key that this sub-cursor points to */
        rc = btCsrKey(&pSub->csr, &p, &n);
        if( rc==SQLITE4_OK && btKeyCompare(p, n, pKey, nKey)==0 ){
          rc = fiSubCsrStep(pCsr, pSub, bNext);
          if( rc==SQLITE4_NOTFOUND ){
            assert( pSub->csr.nPg==0 );
            rc = SQLITE4_OK;
          }
        }
      }
    }

    /* Advance the current sub-cursor */
    if( rc==SQLITE4_OK ){
      rc = fiSubCsrStep(pCsr, &pCsr->aSub[pCsr->iBt], bNext);
      if( rc==SQLITE4_NOTFOUND ){
        assert( pCsr->aSub[pCsr->iBt].csr.nPg==0 );
        rc = SQLITE4_OK;
      }
    }

    /* Figure out a new current bt cursor */
    if( rc==SQLITE4_OK ){
      rc = fiCsrSetCurrent(pCsr);
    }
  }while( rc==SQLITE4_OK && fiCsrIsDelete(&pCsr->aSub[pCsr->iBt].csr) );

  return rc;
}

typedef struct FiLevelIter FiLevelIter;
struct FiLevelIter {
  /* Used internally */
  BtCursor csr;                   /* Cursor used to read summary blob */
  const u8 *aSum;                 /* Summary blob */
  int nSum;                       /* Size of summary blob in bytes */

  /* Output values */
  int nSub;                       /* Total number of expected levels */
  int iAge;                       /* Current age */
  int iLvl;                       /* Current level */
  int iSub;                       /* Current sub-cursor */
};

static int fiLevelIterNext(FiLevelIter *p){
  u16 iMin, nLevel;

  p->iSub++;
  p->iLvl--;
  btReadSummary(p->aSum, 0, &iMin, &nLevel, 0);
  while( p->iLvl<(int)iMin ){
    p->iAge++;
    if( p->iAge>=(p->nSum)/6 ) return 1;
    btReadSummary(p->aSum, p->iAge, &iMin, &nLevel, 0);
    p->iLvl = (int)iMin + (int)nLevel - 1;
  }

  return 0;
}

static int fiLevelIterInit(bt_db *db, FiLevelIter *p){
  int rc;                         /* Return code */

  memset(p, 0, sizeof(FiLevelIter));
  rc = fiLoadSummary(db, &p->csr, &p->aSum, &p->nSum);
  if( rc==SQLITE4_OK ){
    int iAge;
    for(iAge=0; iAge<(p->nSum/6); iAge++){
      u16 iMin, nLevel;
      btReadSummary(p->aSum, iAge, &iMin, &nLevel, 0);
      p->nSub += nLevel;
      if( iAge==0 ){
        p->iLvl = ((int)iMin + nLevel);
        p->iSub = -1;
      }
    }
  }

  return rc;
}

static void fiLevelIterCleanup(FiLevelIter *p){
  btCsrReset(&p->csr, 1);
}

/*
** Seek a fast-insert cursor.
*/
static int fiCsrSeek(FiCursor *pCsr, const void *pK, int nK, int eSeek){
  u8 aPrefix[8];
  int rc = SQLITE4_NOTFOUND;      /* Return code */
  bt_db *db = pCsr->base.pDb;     /* Database handle */
  BtDbHdr *pHdr = sqlite4BtPagerDbhdr(db->pPager);

  assert( eSeek==BT_SEEK_LE || eSeek==BT_SEEK_EQ || eSeek==BT_SEEK_GE );
  fiCsrReset(pCsr);

  if( pHdr->iMRoot ){
    FiLevelIter iter;

    /* Initialize the iterator used to skip through database levels */
    rc = fiLevelIterInit(db, &iter);
    if( rc!=SQLITE4_OK ) return rc;

    if( eSeek==BT_SEEK_EQ ){
      FiSubCursor *pSub;
      BtCursor *pM;
      int iAge;

      /* A BT_SEEK_EQ is a special case. There is no need to set up a cursor
      ** that can be advanced (in either direction) in this case. All that
      ** is required is to search each level in order for the requested key 
      ** (or a corresponding delete marker). Once a match is found, there
      ** is no need to search any further. As a result, only a single
      ** sub-cursor is required.  */
      rc = fiCsrAllocateSubs(db, pCsr, 1);



      pSub = pCsr->aSub;
      pM = &pSub->mcsr;


      btCsrSetup(db, pHdr->iMRoot, pM);



      while( 0==fiLevelIterNext(&iter) ){




        u16 iMin;
        u16 nLevel;
        u16 iMerge;
        int iLvl;

        btPutU32(aPrefix, iter.iAge);
        btPutU32(&aPrefix[4], ~(u32)iter.iLvl);
        rc = btCsrSeek(pM, aPrefix, pK, nK, BT_SEEK_LE, BT_CSRSEEK_SEEK);

        if( rc==SQLITE4_NOTFOUND ){
          /* All keys in this level are greater than pK/nK. */
          /* no-op */
        }else if( rc==SQLITE4_OK || rc==SQLITE4_INEXACT ){
          const void *pV;
          int nV;
          int iBlk;
          sqlite4BtCsrData(&pM->base, 0, 4, &pV, &nV);
          iBlk = sqlite4BtGetU32((const u8*)pV);
          btCsrReset(&pSub->csr, 1);
          btCsrSetup(db, btBlockToRoot(pHdr, iBlk), &pSub->csr);

          rc = btCsrSeek(&pSub->csr, 0, pK, nK, BT_SEEK_EQ, BT_CSRSEEK_SEEK);
          assert( rc!=SQLITE4_INEXACT );
          if( rc!=SQLITE4_NOTFOUND ){
            /* A hit on the requested key or an error has occurred. Either
             ** way, break out of the loop. If this is a hit, set iBt to
             ** zero so that the BtCsrKey() and BtCsrData() routines know
             ** to return data from the first (only) sub-cursor. */
            assert( pCsr->iBt<0 );
            if( rc==SQLITE4_OK ){
              if( 0==fiCsrIsDelete(&pSub->csr) ){
                pCsr->iBt = 0;
              }else{
                rc = SQLITE4_NOTFOUND;
              }
            }
            break;
          }
        }

      }
    }else{
      int bMatch = 0;           /* Found an exact match */
      int bHit = 0;             /* Found at least one entry */

      /* Allocate required sub-cursors. */
      if( rc==SQLITE4_OK ){
        rc = fiCsrAllocateSubs(db, pCsr, iter.nSub);
      }

      /* This loop runs once for each sub-cursor */
      while( rc==SQLITE4_OK && 0==fiLevelIterNext(&iter) ){
        FiSubCursor *pSub = &pCsr->aSub[iter.iSub];
        BtCursor *pM = &pSub->mcsr;
        btCsrSetup(db, pHdr->iMRoot, pM);




        btPutU32(&pSub->aPrefix[0], (u32)iter.iAge);
        btPutU32(&pSub->aPrefix[4], ~(u32)iter.iLvl);


        rc = btCsrSeek(pM, pSub->aPrefix, pK, nK, BT_SEEK_LE, BT_CSRSEEK_SEEK);
        if( rc==SQLITE4_NOTFOUND && eSeek==BT_SEEK_GE ){
          rc = btCsrEnd(pM, 0);

        }


        if( rc==SQLITE4_NOTFOUND ){
          /* No keys to visit in this level */
          rc = SQLITE4_OK;
        }else if( rc==SQLITE4_OK || rc==SQLITE4_INEXACT ){

          const void *pV;
          int nV;
          int iBlk;
          sqlite4BtCsrData(&pM->base, 0, 4, &pV, &nV);
          iBlk = sqlite4BtGetU32((const u8*)pV);

          btCsrReset(&pSub->csr, 1);
          btCsrSetup(db, btBlockToRoot(pHdr, iBlk), &pSub->csr);

          rc = btCsrSeek(&pSub->csr, 0, pK, nK, eSeek, BT_CSRSEEK_SEEK);
          if( rc==SQLITE4_OK ) bMatch = 1;

          if( rc==SQLITE4_INEXACT ) bHit = 1;

          if( rc==SQLITE4_INEXACT || rc==SQLITE4_NOTFOUND ) rc = SQLITE4_OK;

        }else{
          /* An error */
        }
      }
      assert( rc!=SQLITE4_OK || iter.iSub==iter.nSub );


      if( rc==SQLITE4_OK ){
        pCsr->base.flags |= (eSeek==BT_SEEK_GE ? CSR_NEXT_OK : CSR_PREV_OK);
        rc = fiCsrSetCurrent(pCsr);
        if( rc==SQLITE4_OK ){
          BtCursor *pCurrent = &pCsr->aSub[pCsr->iBt].csr;
          if( fiCsrIsDelete(pCurrent) ){
            rc = fiCsrStep(pCsr);
            if( rc==SQLITE4_OK ) rc = SQLITE4_INEXACT;
          }else if( bMatch==0 ){
            rc = (bHit ? SQLITE4_INEXACT : SQLITE4_NOTFOUND);
          }
        }
      }
    }

    fiLevelIterCleanup(&iter);
  }

  return rc;
}

static int fiCsrEnd(FiCursor *pCsr, int bLast){
  bt_db *db = pCsr->base.pDb;
  BtDbHdr *pHdr = sqlite4BtPagerDbhdr(db->pPager);

  FiLevelIter iter;         /* Used to iterate through all f-tree levels */
  int rc;                   /* Return code */

  rc = fiLevelIterInit(db, &iter);
  if( rc==SQLITE4_OK ){
    rc = fiCsrAllocateSubs(db, pCsr, iter.nSub);
  }


  while( rc==SQLITE4_OK && 0==fiLevelIterNext(&iter) ){
    FiSubCursor *pSub = &pCsr->aSub[iter.iSub];
    const int n = (int)sizeof(pSub->aPrefix);


    btPutU32(pSub->aPrefix, (u32)iter.iAge);
    btPutU32(&pSub->aPrefix[4], ~(u32)iter.iLvl);

    btCsrSetup(db, pHdr->iMRoot, &pSub->mcsr);
    if( bLast==0 ){
      rc = btCsrSeek(&pSub->mcsr, 0, pSub->aPrefix, n, BT_SEEK_GE, 0);
    }else{
      u8 aPrefix[8];
      btPutU32(aPrefix, (u32)iter.iAge + (iter.iLvl==0 ? 1 : 0));
      btPutU32(&aPrefix[4], ~(u32)(iter.iLvl - (iter.iLvl==0 ? 0 : 1)));
      rc = btCsrSeek(&pSub->mcsr, 0, aPrefix, n, BT_SEEK_LE, 0);
      if( rc==SQLITE4_OK ){

        rc = btCsrStep(&pSub->mcsr, 0);
      }
    }

    if( rc==SQLITE4_INEXACT ) rc = SQLITE4_OK;



    if( rc==SQLITE4_OK ){
      const void *pV;
      int nV;
      int iBlk;
      btCsrData(&pSub->mcsr, 0, 4, &pV, &nV);
      iBlk = sqlite4BtGetU32((const u8*)pV);
      btCsrReset(&pSub->csr, 1);
      btCsrSetup(db, btBlockToRoot(pHdr, iBlk), &pSub->csr);
      rc = btCsrEnd(&pSub->csr, bLast);
    }


  }

  fiLevelIterCleanup(&iter);

  if( rc==SQLITE4_OK ){
    pCsr->base.flags &= ~(CSR_NEXT_OK | CSR_PREV_OK);
    pCsr->base.flags |= (bLast ? CSR_PREV_OK : CSR_NEXT_OK);
    rc = fiCsrSetCurrent(pCsr);
    if( rc==SQLITE4_OK && fiCsrIsDelete(&pCsr->aSub[pCsr->iBt].csr) ){
      rc = fiCsrStep(pCsr);
    }
  }

  return rc;
}

int sqlite4BtCsrSeek(
  bt_cursor *pBase, 
  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 rc;
  btCheckPageRefs(pBase->pDb);
  if( IsBtCsr(pBase) ){
    BtCursor *pCsr = (BtCursor*)pBase;
    rc = btCsrSeek(pCsr, 0, pK, nK, eSeek, BT_CSRSEEK_SEEK);
  }else{
    rc = fiCsrSeek((FiCursor*)pBase, pK, nK, eSeek);
  }
  btCheckPageRefs(pBase->pDb);
  return rc;
}

  int rc = SQLITE4_OK;            /* Return code */
  
  if( IsBtCsr(pBase) ){
    rc = btCsrKey((BtCursor*)pBase, ppK, pnK);
  }else{
    FiCursor *pCsr = (FiCursor*)pBase;
    assert( pCsr->iBt>=0 );
    rc = btCsrKey(&pCsr->aSub[pCsr->iBt].csr, ppK, pnK);
  }

  return rc;
}

int sqlite4BtCsrData(
  bt_cursor *pBase,               /* 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 */
){
  int rc = SQLITE4_OK;            /* Return code */
  
  if( IsBtCsr(pBase) ){
    rc = btCsrData((BtCursor*)pBase, iOffset, nByte, ppV, pnV);
  }else{
    FiCursor *pCsr = (FiCursor*)pBase;
    assert( pCsr->iBt>=0 );
    rc = btCsrData(&pCsr->aSub[pCsr->iBt].csr, iOffset, nByte, ppV, pnV);
  }

  return rc;
}

/*
** The argument points to a buffer containing an overflow array. Return

    rc = btFastInsertRoot(db, pHdr, &iRootPg);
  }
  btCsrSetup(db, iRootPg, &csr);

  /* Seek stack cursor csr to the b-tree page that key pK/nK is/would be
  ** stored on.  */
  if( rc==SQLITE4_OK ){
    rc = btCsrSeek(&csr, 0, pK, nK, BT_SEEK_GE, BT_CSRSEEK_UPDATE);
  }

  if( rc==SQLITE4_OK ){
    /* The cursor currently points to an entry with key pK/nK. This call
    ** should therefore replace that entry. So delete it and then re-seek
    ** the cursor.  */
    rc = sqlite4BtDelete(&csr.base);
    if( rc==SQLITE4_OK && (nV>=0 || iRoot==0) ){
      rc = btCsrSeek(&csr, 0, pK, nK, BT_SEEK_GE, BT_CSRSEEK_UPDATE);
      if( rc==SQLITE4_OK ) rc = btErrorBkpt(SQLITE4_CORRUPT);
    }
  }


  if( rc==SQLITE4_NOTFOUND || rc==SQLITE4_INEXACT ){
    if( nV<0 && iRoot!=0 ){

          rc = btInsertAndBalance(&csr, 1, &kv);

          /* Unless this is a block-full error, break out of the loop */
          if( rc!=BT_BLOCKFULL ) break;
          assert( iRoot==0 );

          /* Try to schedule a merge operation */
#if 0
          rc = btScheduleMerge(db);
#endif
          rc = SQLITE4_OK;

          if( rc==SQLITE4_OK ){
            rc = btFastInsertRoot(db, pHdr, &iRootPg);
          }
          if( rc==SQLITE4_OK ){
            btCsrReset(&csr, 1);
            btCsrSetup(db, iRootPg, &csr);
            rc = btCsrSeek(&csr, 0, pK, nK, BT_SEEK_GE, BT_CSRSEEK_UPDATE);
          }
        }while( rc==SQLITE4_NOTFOUND || rc==SQLITE4_INEXACT );
      }

      if( kv.eType==KV_CELL ){
        sqlite4_free(db->pEnv, (void*)kv.pV);
      }

){
  static const u8 aZero[] = {0x00, 0x00, 0x00, 0x00, 0x00, 0x00};
  BtDbHdr *pHdr = sqlite4BtPagerDbhdr(db->pPager);
  int rc;

  btCsrSetup(db, pHdr->iMRoot, p);
  rc = btCsrSeek(
      p, 0, aSummaryKey, sizeof(aSummaryKey), BT_SEEK_EQ, BT_CSRSEEK_SEEK
  );
  if( rc==SQLITE4_OK ){
    rc = btCsrData(p, 0, -1, (const void**)paSummary, pnSummary);
  }else if( rc==SQLITE4_NOTFOUND ){
    *paSummary = aZero;
    *pnSummary = sizeof(aZero);
    rc = SQLITE4_OK;
  }

  return rc;
}

static void btReadSummary(
  const u8 *aSum, int iAge, 
  u16 *piMinLevel,
  u16 *pnLevel,
  u16 *piMergeLevel
){
  if( piMinLevel ) *piMinLevel = btGetU16(&aSum[iAge * 6]);
  if( pnLevel ) *pnLevel = btGetU16(&aSum[iAge * 6 + 2]);
  if( piMergeLevel ) *piMergeLevel = btGetU16(&aSum[iAge * 6 + 4]);
}

static void btWriteSummary(
  u8 *aSum, int iAge,
  u16 iMinLevel,
  u16 nLevel,
  u16 iMergeLevel

    *piNext = (iMin + nLevel);
  }

  btCsrReset(&csr, 1);
  return rc;
}

static void btWriteSchedule(u8 *aData, BtSchedule *p, int *pRc){
  if( *pRc==SQLITE4_OK ){
    /* TODO: convert values to big-endian format */
    memcpy(aData, p, sizeof(BtSchedule));
  }
}
static int btReadSchedule(bt_db *db, u8 *aData, BtSchedule *p){
  /* TODO: convert values to big-endian format */
  memcpy(p, aData, sizeof(BtSchedule));
  return SQLITE4_OK;
}

static void btWriteSchedulePage(BtPage *pPg, BtSchedule *p, int *pRc){
  if( *pRc==SQLITE4_OK ){
    int rc = sqlite4BtPageWrite(pPg);
    if( rc==SQLITE4_OK ){

      u8 *aData = sqlite4BtPageData(pPg);
      btWriteSchedule(aData, p, &rc);
    }
    *pRc = rc;
  }
}







static int btAllocateBlock(
  bt_db *db, 
  int nBlk,
  u32 *aiBlk
){
  return sqlite4BtBlockAllocate(db->pPager, nBlk, aiBlk);
}

/*
** This is a helper function for btScheduleMerge(). It determines the
** age and range of levels to be used as inputs by the merge (if any).
*/
static int btFindMerge(
  bt_db *db,                      /* Database handle */
  u32 *piAge,                     /* OUT: Age of input segments to merge */
  u32 *piMinLevel,                /* OUT: Minimum input level value */
  u32 *piMaxLevel,                /* OUT: Maximum input level value */
  u32 *piOutLevel                 /* OUT: Output level value */
){
  BtCursor csr;                   /* Cursor used to read summary record */
  int rc;                         /* Return code */
  const u8 *aSum;
  int nSum;

  rc = fiLoadSummary(db, &csr, &aSum, &nSum);
  if( rc==SQLITE4_OK ){
    int iAge;
    int iBestAge = -1;            /* Best age to merge levels from */
    int nBest = (db->nMinMerge-1);/* Number of levels merged at iBestAge */
    u16 iMin, nLevel, iMerge;     /* Summary of current age */

    rc = SQLITE4_NOTFOUND;
    for(iAge=0; iAge<(nSum/6); iAge++){
      btReadSummary(aSum, iAge, &iMin, &nLevel, &iMerge);
      if( iMerge ){
        int n = 1 + (iMerge-iMin);
        if( n>nBest ){
          *piMinLevel = iMin;
          *piMaxLevel = iMerge;
          *piAge = iAge;
          btReadSummary(aSum, iAge+1, &iMin, &nLevel, &iMerge);
          *piOutLevel = (iMin + nLevel - 1);
          rc = SQLITE4_OK;
        }
        break;

        /* Create a copy of the summary record */
        memcpy(aNew, aSum, nSum);
        if( nByte>nSum ) btWriteSummary(aNew, iBestAge+1, 0, 0, 0);

        /* Find the input age and maximum level */
        btReadSummary(aSum, iBestAge, &iMin, &nLevel, &iMerge);
        *piMinLevel = (u32)iMin;
        *piMaxLevel = (u32)(iMin + nLevel - 1);
        *piAge = iBestAge;

        /* Find the output level */
        btReadSummary(aSum, iBestAge+1, &iMin, &nLevel, &iMerge);
        *piOutLevel = iMin + nLevel;

**
** The merge operation is selected based on the following criteria:
**
**   * The more levels involved in the merge the better, and
**   * It is better to merge younger segments than older ones.
*/
static int btScheduleMerge(bt_db *db){

  BtDbHdr *pHdr = sqlite4BtPagerDbhdr(db->pPager);
  BtPage *pPg = 0;                /* Schedule page */
  u8 *aData;                      /* Schedule page data */
  int rc;                         /* Return code */


  /* Details of proposed merge: */
  u32 iAge;                       /* Input age */
  u32 iMin;                       /* Minimum input level number */
  u32 iMax;                       /* Maximum input level number */
  u32 iOutLvl;                    /* Output level number */

  /* Find the schedule page. If there is no schedule page, allocate it now. */
  if( pHdr->iSRoot==0 ){
    rc = sqlite4BtPageAllocate(db->pPager, &pPg);
    if( rc==SQLITE4_OK ){
      u8 *aData = sqlite4BtPageData(pPg);
      memset(aData, 0, pHdr->pgsz);

  ** scheduled. If the schedule page is not busy, call btFindMerge() to
  ** figure out which levels should be scheduled for merge.  */
  if( rc==SQLITE4_OK ){
    aData = sqlite4BtPageData(pPg);
    if( btGetU32(aData) ){
      rc = SQLITE4_NOTFOUND;
    }else{
      rc = btFindMerge(db, &iAge, &iMin, &iMax, &iOutLvl);
    }
  }

  if( rc==SQLITE4_OK ){
    BtSchedule s;
    memset(&s, 0, sizeof(BtSchedule));

    s.bBusy = 1;
    s.iAge = iAge;
    s.iMaxLevel = iMax;
    s.iMinLevel = iMin;
    s.iOutLevel = iOutLvl;
    rc = btAllocateBlock(db, db->nScheduleAlloc, s.aBlock);

    btWriteSchedulePage(pPg, &s, &rc);
  }

  sqlite4BtPageRelease(pPg);
  if( rc==SQLITE4_NOTFOUND ) rc = SQLITE4_OK;
  return rc;
}

  if( rc==SQLITE4_OK ){
    *piRoot = btBlockToRoot(pHdr, pHdr->iSubBlock);
  }
  db->bFastInsertOp = 1;
  return rc;
}

/*
** Set up a fast-insert cursor to read the input data for a merge operation.
*/
static int fiSetupMergeCsr(
  bt_db *db,                      /* Database handle */
  BtDbHdr *pHdr,                  /* Current database header values */
  BtSchedule *p,                  /* Description of merge operation */
  FiCursor *pCsr                  /* Populate this object before returning */
){
  int iBt;                        /* Used to loop through component cursors */
  int rc;                         /* Return code */

  memset(pCsr, 0, sizeof(FiCursor));
  pCsr->base.flags = CSR_TYPE_FAST;
  pCsr->base.pDb = db;

  rc = fiCsrAllocateSubs(db, pCsr, (p->iMaxLevel - p->iMinLevel) + 1);
  assert( rc==SQLITE4_OK || pCsr->nBt==0 );
  for(iBt=0; iBt<pCsr->nBt && rc==SQLITE4_OK; iBt++){
  }

  return rc;
}

/*
** This is called by a checkpointer to handle a schedule page.
*/
int sqlite4BtMerge(bt_db *db, BtDbHdr *pHdr, u8 *aSched){
  BtSchedule s;                   /* Deserialized schedule object */
  FiCursor fcsr;                  /* FiCursor used to read input */
  int rc;                         /* Return code */

  /* Set up the input cursor. */
  btReadSchedule(db, aSched, &s);
  assert( s.bBusy );
  rc = fiSetupMergeCsr(db, pHdr, &s, &fcsr);

  /* The following loop runs once for each key copied from the input to
  ** the output segments. It terminates either when the input is exhausted
  ** or when all available output blocks are full.  */
  while( rc==SQLITE4_OK ){
    rc = fiCsrStep(&fcsr);
  }

  /* Assuming no error has occurred, update the serialized BtSchedule
  ** structure stored in buffer aSched[]. The caller will write this
  ** buffer to the database file as page (pHdr->iSRoot).  */
  if( rc==SQLITE4_OK || rc==SQLITE4_NOTFOUND ){
    rc = SQLITE4_OK;
    btWriteSchedule(aSched, &s, &rc);
  }

  fiCsrReset(&fcsr);
  return rc;
}

/*
** Insert a new key/value pair or replace an existing one.
**
** This function may modify either the b-tree or fast-insert-tree, depending
** on whether or not the db->bFastInsertOp flag is set.
*/

    if( rc==SQLITE4_OK ){
      rc = btBalanceIfUnderfull(pCsr);
    }

    btCsrReleaseAll(pCsr);
  }else{
    FiCursor *pCsr = (FiCursor*)pBase;
    BtCursor *pSub = &pCsr->aSub[pCsr->iBt].csr;

    void *pKey;
    int nKey;

    rc = btCsrBuffer(pSub, 0);
    pKey = pSub->ovfl.buf.p;
    nKey = pSub->ovfl.nKey;

resides in the db file (not the log). The writer can tell if the schedule page
resides in the db file or the log by checking the "merge performed" flag.


<p>The page is organized as follows:

<ul>
  <li> Busy flag.



  <li> Age of input segments (big-endian u32).

  <li> Max level of input segments (bit-endian u32). All segments of the
       specified age with a level equal to or less than this are considered
       inputs.

  <li> Level of output segments.




  <li> Block number of each allocated block (array of <i>nBlock</i> big-endian 
       u32 fields).
</ul>

<p>Then, after the merge has taken place:
<ul>
  <li> Page number of input page containing next key to read.