void *pVal, int nVal
){
  ScanResult *p = (ScanResult *)pCtx;
  u8 *aKey = (u8 *)pKey;
  u8 *aVal = (u8 *)pVal;
  int i;

  if( test_scan_debug ) printf("%s ", (char *)pKey);

#if 0
  /* Check tdb_fetch() matches */
  int rc = 0;
  testFetch(p->pDb, pKey, nKey, pVal, nVal, &rc);
  assert( rc==0 );
#endif

    res1.bReverse = bReverse;
    res2.pDb = pDb2;
    res2.nKey1 = nKey1; res2.pKey1 = pKey1;
    res2.nKey2 = nKey2; res2.pKey2 = pKey2;
    res2.bReverse = bReverse;

    tdb_scan(pDb1, pRes1, bReverse, pKey1, nKey1, pKey2, nKey2, scanCompareCb);
if( test_scan_debug ) printf("\n");
    tdb_scan(pDb2, pRes2, bReverse, pKey1, nKey1, pKey2, nKey2, scanCompareCb);
if( test_scan_debug ) printf("\n");

    if( res1.nRow!=res2.nRow 
     || res1.cksum1!=res2.cksum1 
     || res1.cksum2!=res2.cksum2
    ){
      printf("expected: %d %X %X\n", res1.nRow, res1.cksum1, res1.cksum2);
      printf("got:      %d %X %X\n", res2.nRow, res2.cksum1, res2.cksum2);

  if( nArg==1 ){
    zPattern = azArg[0];
  }

  do_crash_test(zPattern, &rc);
  return rc;
}










int do_speed_tests(int nArg, char **azArg){

  struct DbSystem {
    const char *zLibrary;
    const char *zColor;
  } aSys[] = {

      if( bSleep && nSleep ) sqlite3_sleep(nSleep);
      bSleep = 1;

      testCaseBegin(&rc, 0, "speed.insert.%s", aSys[j].zLibrary);

      rc = tdb_open(aSys[j].zLibrary, 0, 1, &pDb);
      if( rc ) return rc;
      pLsm = tdb_lsm(pDb);

      if( pLsm ){
        int bMmap = 0;
        int nLimit = 2 * 1024 * 1024;
        int eSafety = 1;
        int bUseLog = 1;

        lsm_config(pLsm, LSM_CONFIG_WRITE_BUFFER, &nLimit);
        lsm_config(pLsm, LSM_CONFIG_SAFETY, &eSafety);
        lsm_config(pLsm, LSM_CONFIG_MMAP, &bMmap);
        lsm_config(pLsm, LSM_CONFIG_USE_LOG, &bUseLog);
      }
  
      testTimeInit();
      for(i=0; i<nRow; i+=nStep){
        int iStep;
        int nWrite1, nWrite2;
        testCaseProgress(i, nRow, testCaseNDot(), &iDot);
        if( pLsm ) lsm_info(pLsm, LSM_INFO_NWRITE, &nWrite1);

      bSleep = 1;

      testCaseBegin(&rc, 0, "speed.select.%s", aSys[j].zLibrary);

      if( doWriteTest ){
        rc = tdb_open(aSys[j].zLibrary, 0, 1, &pDb);
        if( rc ) return rc;


        for(i=0; i<nRow; i+=nSelStep){
          int iStep;
          int iSel;
          testCaseProgress(i, nRow, testCaseNDot(), &iDot);
          for(iStep=0; iStep<nSelStep; iStep++){
            u32 aKey[4];                  /* 16-byte key */

          }
          aSelTime[(j*nRow+i)/nSelStep] = testTimeGet();
          tdb_fetch(pDb, 0, 0, 0, 0);
        }
      }else{
        int t;
        int iSel;
        int bMmap = 0;

        rc = tdb_open(aSys[j].zLibrary, 0, 0, &pDb);
        if( tdb_lsm(pDb) ) lsm_config(tdb_lsm(pDb), LSM_CONFIG_MMAP, &bMmap);

        testTimeInit();
        for(iSel=0; rc==LSM_OK && iSel<nSelTest; iSel++){
          void *pDummy;
          int nDummy;
          u32 iKey;
          u32 aKey[4];                  /* 16-byte key */

  fclose(pInput);
  pEnv->xClose(pOut);

  return rc;
}

static int do_insert(int nArg, char **azArg){

  const char *zDb = "lsm";
  TestDb *pDb = 0;
  int i;
  int rc;
  const int nRow = 1 * 1000 * 1000;

  DatasourceDefn defn = { TEST_DATASOURCE_RANDOM, 8, 15, 80, 150 };
  Datasource *pData = 0;

  if( nArg>1 ){
    testPrintError("Usage: insert ?DATABASE?\n");
    return 1;
  }
  if( nArg==1 ){
    zDb = azArg[0];
  }

  testMallocUninstall(tdb_lsm_env());
  rc = tdb_open(zDb, 0, 1, &pDb);
  if( rc!=0 ){
    testPrintError("Error opening db \"%s\": %d\n", zDb, rc);
  }else{

    InsertWriteHook hook;
    memset(&hook, 0, sizeof(hook));
    hook.pOut = fopen("writelog.txt", "w");

    pData = testDatasourceNew(&defn);
    tdb_lsm_config_work_hook(pDb, do_insert_work_hook, 0);
    tdb_lsm_write_hook(pDb, do_insert_write_hook, (void *)&hook);





    for(i=0; i<nRow; i++){
      void *pKey; int nKey;         /* Database key to insert */
      void *pVal; int nVal;         /* Database value to insert */

      testDatasourceEntry(pData, i, &pKey, &nKey, &pVal, &nVal);
      tdb_write(pDb, pKey, nKey, pVal, nVal);

    }

    testDatasourceFree(pData);
    tdb_close(pDb);
    flushHook(&hook);
    fclose(hook.pOut);
  }

#ifdef LSM_MUTEX_PTHREADS

static void *worker_main(void *pArg){
  LsmWorker *p = (LsmWorker *)pArg;
  lsm_db *pWorker;                /* Connection to access db through */

  pthread_mutex_lock(&p->worker_mutex);
  pWorker = p->pWorker;
  pthread_mutex_unlock(&p->worker_mutex);

  while( pWorker ){
    int nWrite = 0;
    int rc;

    /* Do some work. If an error occurs, exit. */

    rc = lsm_work(pWorker, p->lsm_work_flags, p->lsm_work_npage, &nWrite);

    if( rc!=LSM_OK ){
      p->worker_rc = rc;
      break;
    }

    /* If the call to lsm_work() indicates that there is nothing more
    ** to do at this point, wait on the condition variable. The thread will
    ** wake up when it is signaled either because the client thread has
    ** flushed an in-memory tree into the db file or when the connection
    ** is being closed.  */
    if( nWrite==0 ){
      pthread_mutex_lock(&p->worker_mutex);
      if( p->pWorker && p->bDoWork==0 ){
        pthread_cond_wait(&p->worker_cond, &p->worker_mutex);
      }
      p->bDoWork = 0;
      pWorker = p->pWorker;
      pthread_mutex_unlock(&p->worker_mutex);
    }
  }

  
  return 0;
}


/*
** Signal worker thread iWorker that there may be work to do.

  if( rc==0 ){
    pDb->aWorker = (LsmWorker *)testMalloc(sizeof(LsmWorker) * nWorker);
    memset(pDb->aWorker, 0, sizeof(LsmWorker) * nWorker);
    pDb->nWorker = nWorker;

    rc = mt_start_worker(pDb, 0, zFilename, LSM_WORK_CHECKPOINT, 
        nWorker==1 ? 32 : 0
    );
  }

  if( rc==0 && nWorker==2 ){
    rc = mt_start_worker(pDb, 1, zFilename, 0, 32);
  }

  return rc;
}

int test_lsm_mt2(const char *zFilename, int bClear, TestDb **ppDb){
  return test_lsm_mt(zFilename, 1, bClear, ppDb);
}

int test_lsm_mt3(const char *zFilename, int bClear, TestDb **ppDb){
  return test_lsm_mt(zFilename, 2, bClear, ppDb);
}






















































































#else
static void mt_shutdown(LsmDb *pDb) { 
  unused_parameter(pDb); 
}
int test_lsm_mt(const char *zFilename, int bClear, TestDb **ppDb){
  unused_parameter(zFilename);

**   LSM_CONFIG_MMAP
**     A read/write integer parameter. True to use mmap() to access the 
**     database file. False otherwise.
**
**   LSM_CONFIG_USE_LOG
**     A read/write boolean parameter. True (the default) to use the log
**     file normally. False otherwise.




*/
#define LSM_CONFIG_WRITE_BUFFER  1
#define LSM_CONFIG_PAGE_SIZE     2
#define LSM_CONFIG_SAFETY        3
#define LSM_CONFIG_BLOCK_SIZE    4
#define LSM_CONFIG_AUTOWORK      5
#define LSM_CONFIG_LOG_SIZE      6
#define LSM_CONFIG_MMAP          7
#define LSM_CONFIG_USE_LOG       8


#define LSM_SAFETY_OFF    0
#define LSM_SAFETY_NORMAL 1
#define LSM_SAFETY_FULL   2


/*

*/
#define LSM_PAGE_SIZE   4096
#define LSM_BLOCK_SIZE  (2 * 1024 * 1024)
#define LSM_TREE_BYTES  (2 * 1024 * 1024)
#define LSM_ECOLA       4

#define LSM_DEFAULT_LOG_SIZE (128*1024)


/* Places where a NULL needs to be changed to a real lsm_env pointer
** are marked with NEED_ENV */
#define NEED_ENV ((lsm_env*)0)

/* Initial values for log file checksums. These are only used if the 
** database file does not contain a valid checkpoint.  */

typedef struct Mempool Mempool;
typedef struct MetaPage MetaPage;
typedef struct MultiCursor MultiCursor;
typedef struct Page Page;
typedef struct Segment Segment;
typedef struct SegmentMerger SegmentMerger;
typedef struct Snapshot Snapshot;
typedef struct SortedRun SortedRun;
typedef struct TransMark TransMark;
typedef struct Tree Tree;
typedef struct TreeMark TreeMark;
typedef struct TreeVersion TreeVersion;
typedef struct TreeCursor TreeCursor;
typedef struct Merge Merge;
typedef struct MergeInput MergeInput;

  /* Database handle configuration */
  lsm_env *pEnv;                            /* runtime environment */
  int (*xCmp)(void *, int, void *, int);    /* Compare function */
  int nTreeLimit;                 /* Maximum size of in-memory tree in bytes */
  int bAutowork;                  /* True to do auto-work after writing */
  int eSafety;                    /* LSM_SAFETY_OFF, NORMAL or FULL */


  int nLogSz;                     /* Configured by LSM_CONFIG_LOG_SIZE */
  int bUseLog;                    /* Configured by LSM_CONFIG_USE_LOG */
  int nDfltPgsz;                  /* Configured by LSM_CONFIG_PAGE_SIZE */
  int nDfltBlksz;                 /* Configured by LSM_CONFIG_BLOCK_SIZE */

  /* Sub-system handles */
  FileSystem *pFS;                /* On-disk portion of database */

  void *pLogCtx;

  /* Work done notification callback */
  void (*xWork)(lsm_db *, void *);
  void *pWorkCtx;
};

struct SortedRun {
  int iFirst;                     /* First page of this run */
  int iLast;                      /* Last page of this run */
  Pgno iRoot;                     /* Root page number (if any) */
  int nSize;                      /* Size of this run in pages */
};

struct Segment {
  SortedRun run;                  /* Main array */
  SortedRun sep;                  /* If sep.iFirst!=0, the separators array */
};

/*
** iSplitTopic/pSplitKey/nSplitKey:
**   If nRight>0, this buffer contains a copy of the largest key that has
**   already been written to the left-hand-side of the level.
*/
struct Level {
  Segment lhs;                    /* Left-hand (main) segment */

**
** It is assumed that code that uses an instance of this structure has
** access to the associated Level struct.
**
** bHierReadonly:
**   True if the b-tree hierarchy is currently read-only.
**
** aiOutputOff:
**   The byte offset to write to next within the last page of the output
**   segments main run (aiOutputOff[0]) or separators run (aiOutputOff[1]).
**   If either page is read-only, then the associated aiOutputOff[] entry
**   is set to a negative value.
*/
struct Merge {
  int nInput;                     /* Number of input runs being merged */
  MergeInput *aInput;             /* Array nInput entries in size */
  int nSkip;                      /* Number of separators entries to skip */
  int aiOutputOff[2];             /* Write offsets on run output pages */
  int bHierReadonly;              /* True if b-tree heirarchy is read-only */
};
struct MergeInput {
  Pgno iPg;                       /* Page on which next input is stored */
  int iCell;                      /* Cell containing next input to merge */
};

/* 

int lsmFsPageSize(FileSystem *);
void lsmFsSetPageSize(FileSystem *, int);

int lsmFsFileid(lsm_db *pDb, void **ppId, int *pnId);

/* Creating, populating, gobbling and deleting sorted runs. */
int lsmFsPhantom(FileSystem *, SortedRun *);
void lsmFsPhantomFree(FileSystem *pFS);
void lsmFsGobble(Snapshot *, SortedRun *, Page *);
int lsmFsSortedDelete(FileSystem *, Snapshot *, int, SortedRun *);
int lsmFsSortedFinish(FileSystem *, SortedRun *);
int lsmFsSortedAppend(FileSystem *, Snapshot *, SortedRun *, Page **);
int lsmFsPhantomMaterialize(FileSystem *, Snapshot *, SortedRun *);

/* Functions to retrieve the lsm_env pointer from a FileSystem or Page object */
lsm_env *lsmFsEnv(FileSystem *);
lsm_env *lsmPageEnv(Page *);


int lsmFsSectorSize(FileSystem *);

void lsmSortedSplitkey(lsm_db *, Level *, int *);
int lsmFsSetupAppendList(lsm_db *db);

/* Reading sorted run content. */
int lsmFsDbPageGet(FileSystem *, Pgno, Page **);
int lsmFsDbPageNext(SortedRun *, Page *, int eDir, Page **);

int lsmFsPageWrite(Page *);
u8 *lsmFsPageData(Page *, int *);
int lsmFsPageRelease(Page *);
int lsmFsPagePersist(Page *);
void lsmFsPageRef(Page *);
Pgno lsmFsPageNumber(Page *);

/* 
** Functions from file "lsm_sorted.c".
*/
int lsmInfoPageDump(lsm_db *, Pgno, int, char **);
int lsmSortedFlushTree(lsm_db *, int *);
void lsmSortedCleanup(lsm_db *);
int lsmSortedAutoWork(lsm_db *, int nUnit);



void lsmSortedFreeLevel(lsm_env *pEnv, Level *);

int lsmSortedFlushDb(lsm_db *);
int lsmSortedAdvanceAll(lsm_db *pDb);

int lsmSortedLoadMerge(lsm_db *, Level *, u32 *, int *);

int lsmMCursorType(MultiCursor *, int *);
lsm_db *lsmMCursorDb(MultiCursor *);

int lsmSaveCursors(lsm_db *pDb);
int lsmRestoreCursors(lsm_db *pDb);

void lsmSortedDumpStructure(lsm_db *pDb, Snapshot *, int, int, const char *);
void lsmFsDumpBlockmap(lsm_db *, SortedRun *);
void lsmFsDumpBlocklists(lsm_db *);


void lsmPutU32(u8 *, u32);
u32 lsmGetU32(u8 *);

/*

**   The checksum:
**
**     1. Checksum value 1.
**     2. Checksum value 2.
**
** In the above, a segment record is:
**
**     1. First page of main array,
**     2. Last page of main array,
**     3. Size of main array in pages,
**     4. First page of separators array (or 0),
**     5. Last page of separators array (or 0),
**     6. Root page of separators array (or 0).
*/

/*
** OVERSIZED CHECKPOINT BLOBS:
**
** There are two slots allocated for checkpoints at the start of each 
** database file. Each are 4096 bytes in size, so may accommodate 

  Segment *pSeg, 
  CkptBuffer *p, 
  int *piOut, 
  int *pRc
){
  int iOut = *piOut;

  ckptSetValue(p, iOut++, pSeg->run.iFirst, pRc);
  ckptSetValue(p, iOut++, pSeg->run.iLast, pRc);
  ckptSetValue(p, iOut++, pSeg->run.nSize, pRc);
  if( segmentHasSeparators(pSeg) ){
    ckptSetValue(p, iOut++, pSeg->sep.iFirst, pRc);
    ckptSetValue(p, iOut++, pSeg->sep.iLast, pRc);
    ckptSetValue(p, iOut++, pSeg->sep.iRoot, pRc);
  }else{
    ckptSetValue(p, iOut++, 0, pRc);
    ckptSetValue(p, iOut++, 0, pRc);
    ckptSetValue(p, iOut++, 0, pRc);
  }

  *piOut = iOut;
}

static void ckptExportLevel(
  Level *pLevel,
  CkptBuffer *p,

static void ckptNewSegment(
  u32 *aIn,
  int *piIn,
  Segment *pSegment               /* Populate this structure */
){
  int iIn = *piIn;

  assert( pSegment->run.iFirst==0 && pSegment->run.iLast==0 );
  assert( pSegment->run.nSize==0 && pSegment->run.iRoot==0 );
  assert( pSegment->sep.iFirst==0 && pSegment->sep.iLast==0 );
  assert( pSegment->sep.nSize==0 && pSegment->sep.iRoot==0 );

  pSegment->run.iFirst = aIn[iIn++];
  pSegment->run.iLast = aIn[iIn++];
  pSegment->run.nSize = aIn[iIn++];
  pSegment->sep.iFirst = aIn[iIn++];
  pSegment->sep.iLast = aIn[iIn++];
  pSegment->sep.iRoot = aIn[iIn++];
  if( pSegment->sep.iFirst ) pSegment->sep.nSize = 1;

  *piIn = iIn;
}

static int ckptSetupMerge(lsm_db *pDb, u32 *aInt, int *piIn, Level *pLevel){
  Merge *pMerge;                  /* Allocated Merge object */
  int nInput;                     /* Number of input segments in merge */

  pMerge = (Merge *)lsmMallocZero(pDb->pEnv, nByte);
  if( !pMerge ) return LSM_NOMEM_BKPT;
  pLevel->pMerge = pMerge;

  /* Populate the Merge object. */
  pMerge->aInput = (MergeInput *)&pMerge[1];
  pMerge->nInput = nInput;
  pMerge->aiOutputOff[0] = -1;
  pMerge->aiOutputOff[1] = -1;
  pMerge->nSkip = (int)aInt[iIn++];
  pMerge->bHierReadonly = 1;
  for(i=0; i<nInput; i++){
    pMerge->aInput[i].iPg = (Pgno)aInt[iIn++];
    pMerge->aInput[i].iCell = (int)aInt[iIn++];
  }

  /* Set *piIn and return LSM_OK. */
  *piIn = iIn;

  return rc;
}

static int ckptImport(lsm_db *pDb, void *pCkpt, int nInt, int *pRc){
  int ret = 0;
  if( *pRc==LSM_OK ){
    Snapshot *pSnap = pDb->pWorker;
    FileSystem *pFS = pDb->pFS;
    u32 cksum[2] = {0, 0};
    u32 *aInt = (u32 *)pCkpt;

    lsmChecksumBytes((u8 *)aInt, sizeof(u32)*(nInt-2), 0, cksum);
    if( LSM_LITTLE_ENDIAN ){
      int i;
      for(i=0; i<nInt; i++) aInt[i] = BYTESWAP32(aInt[i]);

      nLevel = (int)aInt[CKPT_HDR_NLEVEL];
      lsmSnapshotSetNBlock(pSnap, (int)aInt[CKPT_HDR_NBLOCK]);
      lsmDbSetPagesize(pDb,(int)aInt[CKPT_HDR_PGSZ],(int)aInt[CKPT_HDR_BLKSZ]);

      /* Import log offset */
      ckptImportLog(aInt, &iIn, lsmDatabaseLog(pDb));

      /* Import each level. This loop runs once for each db level. */
      *pRc = ckptLoadLevels(pDb, aInt, &iIn, nLevel, &pTopLevel);
      lsmDbSnapshotSetLevel(pSnap, pTopLevel);

      /* Import the freelist delta */
      aDelta = lsmFreelistDeltaPtr(pDb);
      for(i=0; i<LSM_FREELIST_DELTA_SIZE; i++){
        aDelta[i] = aInt[iIn++];

int lsmCheckpointLevels(
  lsm_db *pDb,                    /* Database handle */
  int *pnHdrLevel,                /* OUT: Levels to write to db header */
  void **paVal,                   /* OUT: Pointer to LEVELS blob */
  int *pnVal                      /* OUT: Size of LEVELS blob in bytes */
){
  int rc = LSM_OK;                /* Return code */
  const int SEGMENT_SIZE = 6;     /* Size of a checkpoint segment record */
  Level *p;                       /* Used to iterate through levels */
  int nFree;                      /* Free integers remaining in db header */
  int nHdr = 0;                   /* Number of levels stored in db header */
  int nLevels = 0;                /* Number of levels stored in LEVELS */
 
  /* Number of free integers - 1024 less those used by the checkpoint header,
  ** less the 4 used for the log-pointer, less the 3 used for the 

*/
#include "lsmInt.h"

#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>

/* 
** A "phantom" run under construction.
**
** Phantom runs are constructed entirely in memory, then written out to 
** disk. This is distinct from normal runs, for which each page is written
** to disk as soon as it is completely populated. 
**
** They are used when the in-memory tree is flushed to disk. In this case, 
** the main run is written directly to disk and the separators run 
** accumulated in memory as a phantom run and flushed to disk after the main
** run is completed. This allows the separators run to immediately follow
** the main run in the file - making the entire flush operation a single
** contiguous write.
**
** Before they are flushed to disk, the pages of phantom runs do not have
** page numbers. This means it is not possible to obtain pointers to them.
** In practice, this means that when creating a separators run, a phantom
** run is used to accumulate and write all leaf pages to disk, then a
** second pass is made to populate and append the b-tree hierarchy pages.
*/
typedef struct PhantomRun PhantomRun;
struct PhantomRun {
  SortedRun *pRun;                /* Accompanying SortedRun object */
  int nPhantom;                   /* Number of pages in run */
  int bRunFinished;               /* True if the associated run is finished */
  Page *pFirst;                   /* First page in phantom run */
  Page *pLast;                    /* Current last page in phantom run */
};

/*
** Maximum number of pages allowed to accumulate in memory when constructing
** a phantom run. If this limit is exceeded, the phantom run is flushed to
** disk even if it is not finished.
*/
#define FS_MAX_PHANTOM_PAGES 32


/*
** File-system object. Each database connection allocates a single instance
** of the following structure. It is used for all access to the database and
** log files.
**
** pLruFirst, pLruLast:

struct FileSystem {
  lsm_db *pDb;                    /* Database handle that owns this object */
  lsm_env *pEnv;                  /* Environment pointer */
  char *zDb;                      /* Database file name */
  int nMetasize;                  /* Size of meta pages in bytes */
  int nPagesize;                  /* Database page-size in bytes */
  int nBlocksize;                 /* Database block-size in bytes */
  PhantomRun phantom;             /* Phantom run currently under construction */

  /* r/w file descriptors for both files. */
  lsm_file *fdDb;                 /* Database file */
  lsm_file *fdLog;                /* Log file */

  /* mmap() mode things */
  int bUseMmap;                   /* True to use mmap() to access db file */

};

/*
** Database page handle.
*/
struct Page {
  u8 *aData;                      /* Buffer containing page data */

  int iPg;                        /* Page number */
  int nRef;                       /* Number of outstanding references */
  int flags;                      /* Combination of PAGE_XXX flags */
  Page *pHashNext;                /* Next page in hash table slot */
  Page *pLruNext;                 /* Next page in LRU list */
  Page *pLruPrev;                 /* Previous page in LRU list */
  FileSystem *pFS;                /* File system that owns this page */

/*
** Number of pgsz byte pages omitted from the start of block 1. The start
** of block 1 contains two 4096 byte meta pages (8192 bytes in total).
*/
#define BLOCK1_HDR_SIZE(pgsz)  LSM_MAX(1, 8192/(pgsz))

/*
** Return true if the SortedRun passed as the second argument is a phantom
** run currently being constructed by FileSystem object pFS.
*/
#define isPhantom(pFS, pSorted) ((pSorted) && (pFS)->phantom.pRun==(pSorted))

/*
** Wrappers around the VFS methods of the lsm_env object:
**
**     lsmEnvOpen()
**     lsmEnvRead()
**     lsmEnvWrite()

*/
void lsmFsClose(FileSystem *pFS){
  if( pFS ){
    Page *pPg;
    lsm_env *pEnv = pFS->pEnv;

    assert( pFS->nOut==0 );

    pPg = pFS->pLruFirst;
    while( pPg ){
      Page *pNext = pPg->pLruNext;
      if( pPg->flags & PAGE_FREE ) lsmFree(pEnv, pPg->aData);
      lsmFree(pEnv, pPg);
      pPg = pNext;
    }

    pPg->pLruPrev->pLruNext = pPg;
  }else{
    pFS->pLruFirst = pPg;
  }
  pFS->pLruLast = pPg;
}




static void fsPageRemoveFromHash(FileSystem *pFS, Page *pPg){
  int iHash;
  Page **pp;

  iHash = fsHashKey(pFS->nHash, pPg->iPg);
  for(pp=&pFS->apHash[iHash]; *pp!=pPg; pp=&(*pp)->pHashNext);
  *pp = pPg->pHashNext;

    int rc;
    rc = lsmEnvRemap(pFS->pEnv, pFS->fdDb, iSz, &pFS->pMap, &pFS->nMap);
    if( rc==LSM_OK ){
      u8 *aData = (u8 *)pFS->pMap;
      for(pFix=pFS->pLruFirst; pFix; pFix=pFix->pLruNext){
        pFix->aData = &aData[pFS->nPagesize * (i64)(pFix->iPg-1)];
      }


    }
    *pRc = rc;
  }
}

/*
** Return a handle for a database page.

      }

      /* If the xRead() call was successful (or not attempted), link the
      ** page into the page-cache hash-table. Otherwise, if it failed,
      ** free the buffer. */
      if( rc==LSM_OK ){
        p->pHashNext = pFS->apHash[iHash];

        pFS->apHash[iHash] = p;
      }else{
        fsPageBufferFree(p);
        p = 0;
      }
    }
  }else if( p->nRef==0 && pFS->bUseMmap==0 ){

    *piNext = fsPageToBlock(pFS, lsmGetU32(&pLast->aData[pFS->nPagesize-4]));
    lsmFsPageRelease(pLast);
  }
  return rc;
}

static int fsRunEndsBetween(
  SortedRun *pRun, 
  SortedRun *pIgnore, 
  int iFirst, 
  int iLast
){
  return (pRun!=pIgnore && (
        (pRun->iFirst>=iFirst && pRun->iFirst<=iLast)
     || (pRun->iLast>=iFirst && pRun->iLast<=iLast)
  ));
}

static int fsLevelEndsBetween(
  Level *pLevel, 
  SortedRun *pIgnore, 
  int iFirst, 
  int iLast
){
  int i;

  if( fsRunEndsBetween(&pLevel->lhs.run, pIgnore, iFirst, iLast)
   || fsRunEndsBetween(&pLevel->lhs.sep, pIgnore, iFirst, iLast)
  ){
    return 1;
  }
  for(i=0; i<pLevel->nRight; i++){
    if( fsRunEndsBetween(&pLevel->aRhs[i].run, pIgnore, iFirst, iLast)
     || fsRunEndsBetween(&pLevel->aRhs[i].sep, pIgnore, iFirst, iLast)
    ){
      return 1;
    }
  }

  return 0;
}

static int fsFreeBlock(
  FileSystem *pFS, 
  Snapshot *pSnapshot, 
  SortedRun *pIgnore,             /* Ignore this run when searching */
  int iBlk
){
  int rc = LSM_OK;                /* Return code */
  int iFirst;                     /* First page on block iBlk */
  int iLast;                      /* Last page on block iBlk */
  int i;                          /* Used to iterate through append points */
  Level *pLevel;                  /* Used to iterate through levels */

/*
** Delete or otherwise recycle the blocks currently occupied by run pDel.
*/
int lsmFsSortedDelete(
  FileSystem *pFS, 
  Snapshot *pSnapshot,
  int bZero,                      /* True to zero the SortedRun structure */
  SortedRun *pDel
){
  if( pDel->iFirst ){
    int rc = LSM_OK;

    int iBlk;
    int iLastBlk;

      }else if( bZero==0 && pDel->iLast!=fsLastPageOnBlock(pFS, iLastBlk) ){
        break;
      }
      rc = fsFreeBlock(pFS, pSnapshot, pDel, iBlk);
      iBlk = iNext;
    }

    if( bZero ) memset(pDel, 0, sizeof(SortedRun));
  }
  return LSM_OK;
}

/*
** The pager reference passed as the only argument must refer to a sorted
** file page (not a log or meta page). This call indicates that the argument
** page is now the first page in its sorted file - all previous pages may
** be considered free.
*/
void lsmFsGobble(
  Snapshot *pSnapshot,
  SortedRun *pRun, 
  Page *pPg
){
  FileSystem *pFS = pPg->pFS;

  if( pPg->iPg!=pRun->iFirst ){
    int rc = LSM_OK;
    int iBlk = fsPageToBlock(pFS, pRun->iFirst);

** If the previous/next page does exist and is successfully loaded, *ppNext
** is set to point to it and LSM_OK is returned. Otherwise, if an error 
** occurs, *ppNext is set to NULL and and lsm error code returned.
**
** Page references returned by this function should be released by the 
** caller using lsmFsPageRelease().
*/
int lsmFsDbPageNext(SortedRun *pRun, Page *pPg, int eDir, Page **ppNext){
  FileSystem *pFS = pPg->pFS;
  int iPg = pPg->iPg;

  assert( eDir==1 || eDir==-1 );

  if( eDir<0 ){
    if( pRun && iPg==pRun->iFirst ){

  assert( db->pWorker );
  for(pLvl=lsmDbSnapshotLevel(db->pWorker); 
      rc==LSM_OK && pLvl; 
      pLvl=pLvl->pNext
  ){
    if( pLvl->nRight==0 ){
      addAppendPoint(db, pLvl->lhs.sep.iLast, &rc);
      addAppendPoint(db, pLvl->lhs.run.iLast, &rc);
    }else{
      int i;
      for(i=0; i<pLvl->nRight; i++){
        addAppendPoint(db, pLvl->aRhs[i].sep.iLast, &rc);
        addAppendPoint(db, pLvl->aRhs[i].run.iLast, &rc);
      }
    }
  }

  for(pLvl=lsmDbSnapshotLevel(db->pWorker); pLvl; pLvl=pLvl->pNext){
    int i;
    subAppendPoint(db, pLvl->lhs.sep.iFirst);
    subAppendPoint(db, pLvl->lhs.run.iFirst);
    for(i=0; i<pLvl->nRight; i++){
      subAppendPoint(db, pLvl->aRhs[i].sep.iFirst);
      subAppendPoint(db, pLvl->aRhs[i].run.iFirst);
    }
  }

  return rc;
}

int lsmFsPhantom(FileSystem *pFS, SortedRun *pRun){
  assert( pFS->phantom.pRun==0 );
  pFS->phantom.pRun = pRun;
  return LSM_OK;
}

void lsmFsPhantomFree(FileSystem *pFS){
  if( pFS->phantom.pRun ){
    Page *pPg;
    Page *pNext;
    for(pPg=pFS->phantom.pFirst; pPg; pPg=pNext){
      pNext = pPg->pHashNext;
      fsPageBufferFree(pPg);
    }
    memset(&pFS->phantom, 0, sizeof(PhantomRun));
  }
}

int lsmFsPhantomMaterialize(
  FileSystem *pFS, 
  Snapshot *pSnapshot, 
  SortedRun *p
){
  int rc = LSM_OK;
  if( isPhantom(pFS, p) ){
    PhantomRun *pPhantom = &pFS->phantom;
    Page *pPg;
    Page *pNext;
    int i;
    Pgno iFirst = 0;

    /* Search for an existing run in the database that this run can be
    ** appended to. See comments surrounding findAppendPoint() for details. */
    iFirst = findAppendPoint(pFS, pPhantom->nPhantom);

    /* If the array can not be written into any partially used block, 
    ** allocate a new block. The first page of the materialized run will
    ** be the second page of the new block (since the first is undersized
    ** and can not be used).  */
    if( iFirst==0 ){
      int iNew;                   /* New block */
      lsmBlockAllocate(pFS->pDb, &iNew);
      iFirst = fsFirstPageOnBlock(pFS, iNew) + 1;
    }

    p->iFirst = iFirst;
    p->iLast = iFirst + pPhantom->nPhantom - 1;
    assert( 0==fsIsFirst(pFS, p->iFirst) && 0==fsIsLast(pFS, p->iFirst) );
    assert( 0==fsIsFirst(pFS, p->iLast) && 0==fsIsLast(pFS, p->iLast) );
    assert( fsPageToBlock(pFS, p->iFirst)==fsPageToBlock(pFS, p->iLast) );

    i = iFirst;
    for(pPg=pPhantom->pFirst; pPg; pPg=pNext){
      int iHash;
      pNext = pPg->pHashNext;
      pPg->iPg = i++;
      pPg->nRef++;

      iHash = fsHashKey(pFS->nHash, pPg->iPg);
      pPg->pHashNext = pFS->apHash[iHash];
      pFS->apHash[iHash] = pPg;
      pFS->nOut++;
      lsmFsPageRelease(pPg);
    }
    assert( i==p->iLast+1 );

    p->nSize = pPhantom->nPhantom;
    memset(&pFS->phantom, 0, sizeof(PhantomRun));
  }
  return rc;
}

/*
** Append a page to file iFile. Return a reference to it. lsmFsPageWrite()
** has already been called on the returned reference.
*/
int lsmFsSortedAppend(
  FileSystem *pFS, 
  Snapshot *pSnapshot,
  SortedRun *p, 
  Page **ppOut
){
  int rc = LSM_OK;
  Page *pPg = 0;
  *ppOut = 0;

  if( isPhantom(pFS, p) ){
    const int nPagePerBlock = (pFS->nBlocksize / pFS->nPagesize);
    int nLimit = (nPagePerBlock - 2 - (fsFirstPageOnBlock(pFS, 1)-1) );

    if( pFS->phantom.nPhantom>=nLimit ){ 
      rc = lsmFsPhantomMaterialize(pFS, pSnapshot, p);
      if( rc!=LSM_OK ) return rc;
    }
  }

  if( isPhantom(pFS, p) ){
    rc = fsPageBuffer(pFS, 1, &pPg);
    if( rc==LSM_OK ){
      PhantomRun *pPhantom = &pFS->phantom;
      pPg->iPg = 0;
      pPg->nRef = 1;
      pPg->flags |= PAGE_DIRTY;
      pPg->pHashNext = 0;
      pPg->pLruNext = 0;
      pPg->pLruPrev = 0;
      pPg->pFS = pFS;
      if( pPhantom->pFirst ){
        assert( pPhantom->pLast );
        pPhantom->pLast->pHashNext = pPg;
      }else{
        pPhantom->pFirst = pPg;
      }
      pPhantom->pLast = pPg;
      pPhantom->nPhantom++;
    }
  }else{
    int iApp = 0;
    int iNext = 0;
    int iPrev = p->iLast;

    if( iPrev==0 ){
      iApp = findAppendPoint(pFS, 0);
    }else if( fsIsLast(pFS, iPrev) ){
      Page *pLast = 0;
      rc = fsPageGet(pFS, iPrev, 0, &pLast);
      if( rc!=LSM_OK ) return rc;
      iApp = lsmGetU32(&pLast->aData[pFS->nPagesize-4]);
      lsmFsPageRelease(pLast);
    }else{
      iApp = iPrev + 1;
    }

    /* If this is the first page allocated, or if the page allocated is the
     ** last in the block, allocate a new block here.  */
    if( iApp==0 || fsIsLast(pFS, iApp) ){
      int iNew;                     /* New block number */

      lsmBlockAllocate(pFS->pDb, &iNew);
      if( iApp==0 ){
        iApp = fsFirstPageOnBlock(pFS, iNew);
      }else{
        iNext = fsFirstPageOnBlock(pFS, iNew);
      }
    }

    /* Grab the new page. */
    pPg = 0;
    rc = fsPageGet(pFS, iApp, 1, &pPg);
    assert( rc==LSM_OK || pPg==0 );

    /* If this is the first or last page of a block, fill in the pointer 
     ** value at the end of the new page. */
    if( rc==LSM_OK ){
      p->nSize++;
      p->iLast = iApp;
      if( p->iFirst==0 ) p->iFirst = iApp;
      pPg->flags |= PAGE_DIRTY;

      if( fsIsLast(pFS, iApp) ){
        lsmPutU32(&pPg->aData[pFS->nPagesize-4], iNext);
      }else 
        if( fsIsFirst(pFS, iApp) ){
          lsmPutU32(&pPg->aData[pFS->nPagesize-4], iPrev);
        }
    }
  }

  *ppOut = pPg;
  return rc;
}

/*
** Mark the sorted run passed as the second argument as finished. 
*/
int lsmFsSortedFinish(FileSystem *pFS, SortedRun *p){
  int rc = LSM_OK;
  if( p ){
    const int nPagePerBlock = (pFS->nBlocksize / pFS->nPagesize);

    if( pFS->phantom.pRun ) pFS->phantom.bRunFinished = 1;

    /* Check if the last page of this run happens to be the last of a block.
    ** If it is, then an extra block has already been allocated for this run.
    ** Shift this extra block back to the free-block list. 
    **
    ** Otherwise, add the first free page in the last block used by the run
    ** to the lAppend list.
    */

/*
** Return a copy of the environment pointer used by the file-system object
** to which this page belongs.
*/
lsm_env *lsmPageEnv(Page *pPg) { 
  return pPg->pFS->pEnv; 
}





/*
** Return the sector-size as reported by the log file handle.
*/
int lsmFsSectorSize(FileSystem *pFS){
  return lsmEnvSectorSize(pFS->pEnv, pFS->fdLog);
}

  *piParam = pFS->bUseMmap;
  return LSM_OK;
}

/*
** Helper function for lsmInfoArrayStructure().
*/
static SortedRun *startsWith(SortedRun *pRun, Pgno iFirst){
  return (iFirst==pRun->iFirst) ? pRun : 0;
}

/*
** This function implements the lsm_info(LSM_INFO_ARRAY_STRUCTURE) request.
** If successful, *pzOut is set to point to a nul-terminated string 
** containing the array structure and LSM_OK is returned. The caller should
** eventually free the string using lsmFree().
**
** If an error occurs, *pzOut is set to NULL and an LSM error code returned.
*/
int lsmInfoArrayStructure(lsm_db *pDb, Pgno iFirst, char **pzOut){
  int rc = LSM_OK;
  Snapshot *pWorker;              /* Worker snapshot */
  Snapshot *pRelease = 0;         /* Snapshot to release */
  SortedRun *pArray = 0;          /* Array to report on */
  Level *pLvl;                    /* Used to iterate through db levels */

  *pzOut = 0;
  if( iFirst==0 ) return LSM_ERROR;

  /* Obtain the worker snapshot */
  pWorker = pDb->pWorker;
  if( !pWorker ){
    pRelease = pWorker = lsmDbSnapshotWorker(pDb);
  }

  /* Search for the array that starts on page iFirst */
  for(pLvl=lsmDbSnapshotLevel(pWorker); pLvl && pArray==0; pLvl=pLvl->pNext){
    if( 0==(pArray = startsWith(&pLvl->lhs.sep, iFirst))
     && 0==(pArray = startsWith(&pLvl->lhs.run, iFirst))
    ){
      int i;
      for(i=0; i<pLvl->nRight; i++){
        if( (pArray = startsWith(&pLvl->aRhs[i].sep, iFirst)) ) break;
        if( (pArray = startsWith(&pLvl->aRhs[i].run, iFirst)) ) break;
      }
    }
  }

  if( pArray==0 ){
    /* Could not find the requested array. This is an error. */
    *pzOut = 0;

    *pzOut = str.z;
  }

  lsmDbSnapshotRelease(pDb->pEnv, pRelease);
  return rc;
}

void lsmFsDumpBlockmap(lsm_db *pDb, SortedRun *p){
  if( p ){
    FileSystem *pFS = pDb->pFS;
    int iBlk;
    int iLastBlk;
    char *zMsg = 0;
    LsmString zBlk;

    lsmStringInit(&zBlk, pDb->pEnv);
    iBlk = fsPageToBlock(pFS, p->iFirst);
    iLastBlk = fsPageToBlock(pFS, p->iLast);

    while( iBlk ){
      lsmStringAppendf(&zBlk, " %d", iBlk);
      if( iBlk!=iLastBlk ){
        fsBlockNext(pFS, iBlk, &iBlk);
      }else{
        iBlk = 0;
      }
    }

    zMsg = lsmMallocPrintf(pDb->pEnv, "%d..%d: ", p->iFirst, p->iLast);
    lsmLogMessage(pDb, LSM_OK, "    % -15s %s", zMsg, zBlk.z);
    lsmFree(pDb->pEnv, zMsg);
    lsmStringClear(&zBlk);
  }
} 

#ifdef LSM_EXPENSIVE_DEBUG
/*
** Helper function for lsmFsIntegrityCheck()
*/
static void checkBlocks(
  FileSystem *pFS, 
  Segment *pSeg, 
  int bExtra,
  u8 *aUsed
){
  if( pSeg ){
    int i;
    for(i=0; i<2; i++){
      SortedRun *p = (i ? pSeg->pRun : pSeg->pSep);

      if( p && p->nSize>0 ){
        const int nPagePerBlock = (pFS->nBlocksize / pFS->nPagesize);

        int iBlk;
        int iLastBlk;
        iBlk = fsPageToBlock(pFS, p->iFirst);

  pDb->nTreeLimit = LSM_TREE_BYTES;
  pDb->bAutowork = 1;
  pDb->eSafety = LSM_SAFETY_NORMAL;
  pDb->xCmp = xCmp;
  pDb->nLogSz = LSM_DEFAULT_LOG_SIZE;
  pDb->nDfltPgsz = LSM_PAGE_SIZE;
  pDb->nDfltBlksz = LSM_BLOCK_SIZE;

  pDb->bUseLog = 1;

  return LSM_OK;
}

lsm_env *lsm_get_env(lsm_db *pDb){
  assert( pDb->pEnv );

      int *piVal = va_arg(ap, int *);
      if( pDb->nTransOpen==0 && (*piVal==0 || *piVal==1) ){
        pDb->bUseLog = *piVal;
      }
      *piVal = pDb->bUseLog;
      break;
    }








    default:
      rc = LSM_MISUSE;
      break;
  }

  va_end(ap);
  return rc;
}

void lsmAppendSegmentList(LsmString *pStr, char *zPre, Segment *pSeg){
  lsmStringAppendf(pStr, "%s{%d %d %d %d %d %d}", zPre, 
        pSeg->sep.iFirst, pSeg->sep.iLast, pSeg->sep.iRoot,
        pSeg->run.iFirst, pSeg->run.iLast, pSeg->run.nSize
  );
}

int lsmStructList(
  lsm_db *pDb,                    /* Database handle */
  char **pzOut                    /* OUT: Nul-terminated string (tcl list) */
){

    lsmMutexEnter(pDb->pEnv, p->pClientMutex);
    assertSnapshotListOk(p);
    pOld = p->pClient;
    pNew->pSnapshotNext = pOld;
    p->pClient = pNew;
    assertSnapshotListOk(p);
    if( pDb->pClient ){
      assert( pDb->pClient==pOld );
      pDb->pClient = p->pClient;
      p->pClient->nRef++;
    }
    lsmMutexLeave(pDb->pEnv, p->pClientMutex);

    lsmDbSnapshotRelease(pDb->pEnv, pOld);
    p->bDirty = 0;

    /* Upgrade the user connection to the new client snapshot */

/*
** 2011-08-14
**
** The author disclaims copyright to this source code.  In place of
** a legal notice, here is a blessing:
**
**    May you do good and not evil.
**    May you find forgiveness for yourself and forgive others.
**    May you share freely, never taking more than you give.
**
*************************************************************************
**
** SORTED FILE FORMAT:
**
**   A sorted file is divided into pages. The page-size is not stored anywhere
**   within the sorted file itself - it must be known in advance in order to
**   read the file. The maximum allowed page-size is 64KB.
**
** PAGE FORMAT:
**
**   The maximum page size is 65536 bytes.
**
**   Since all records are equal to or larger than 2 bytes in size, and 
**   some space within the page is consumed by the page footer, there must
**   be less than 2^15 records on each page.

**
**   * To iterate and/or seek within a single Segment (the combination of a 
**     main run and an optional sorted run).
**
**   * To iterate through the separators array of a segment.
*/
struct SegmentPtr {
  Segment *pSeg;                  /* Segment to access */
  SortedRun *pRun;                /* Points to either pSeg->run or pSeg->sep */

  /* Current page. See segmentPtrLoadPage(). */
  Page *pPg;                    /* Current page */
  u16 flags;                    /* Copy of page flags field */
  int nCell;                    /* Number of cells on pPg */
  int iPtr;                     /* Base cascade pointer */

  int bIgnoreSeparators;          /* True to ignore SORTED_SEPARATOR records */
  int bIgnoreSystem;              /* True to ignore records for topic!=0 */
  int iCurrentPtr;                /* Current entry in aPtr[] */
  int nPtr;                       /* Size of aPtr[] array */
  SegmentPtr *aPtr;               /* Array of segment pointers */
  Level *pLevel;                  /* Pointer to Level object (if nPtr>1) */
};




































/*
** A cursor used for merged searches or iterations through up to one
** Tree structure and any number of sorted files.
**
**   lsmMCursorNew()
**   lsmMCursorSeek()

  lsm_db *pDb;                    /* Connection that owns this cursor */
  MultiCursor *pNext;             /* Next cursor owned by connection pDb */

  int flags;                      /* Mask of CURSOR_XXX flags */
  int (*xCmp)(void *, int, void *, int);         /* Compare function */
  int eType;                      /* Cache of current key type */
  Blob key;                       /* Cache of current key (or NULL) */


  TreeCursor *pTreeCsr;           /* Single tree cursor */
  int nSegCsr;                    /* Size of aSegCsr[] array */
  LevelCursor *aSegCsr;           /* Array of cursors open on sorted files */
  int nTree;
  int *aTree;


  int *pnHdrLevel;
  void *pSystemVal;
  Snapshot *pSnap;
};






/*
** CURSOR_IGNORE_DELETE
**   If set, this cursor will not visit SORTED_DELETE keys.
**
** CURSOR_NEW_SYSTEM
**   If set, then after all user data from the in-memory tree and any other

#define CURSOR_AT_FREELIST      0x00000004
#define CURSOR_AT_LEVELS        0x00000008
#define CURSOR_IGNORE_SYSTEM    0x00000010
#define CURSOR_NEXT_OK          0x00000020
#define CURSOR_PREV_OK          0x00000040

typedef struct MergeWorker MergeWorker;







struct MergeWorker {
  lsm_db *pDb;                    /* Database handle */
  Level *pLevel;                  /* Worker snapshot Level being merged */
  MultiCursor *pCsr;              /* Cursor to read new segment contents from */
  int bFlush;                     /* True if this is an in-memory tree flush */
  Page **apHier;                  /* Separators array b-tree internal nodes */
  int nHier;                      /* Number of entries in apHier[] */
  Page *apPage[2];                /* Current output pages (0 is main run) */
  int nWork;                      /* Number of calls to mergeWorkerNextPage() */
};

#ifdef LSM_DEBUG_EXPENSIVE
static void assertAllPointersOk(int rc, lsm_db *pDb);
static void assertAllBtreesOk(int rc, lsm_db *);
#else
# define assertAllPointersOk(y, z)
# define assertAllBtreesOk(y, z)
#endif










/*
** Write nVal as a 16-bit unsigned big-endian integer into buffer aOut.
*/
void lsmPutU16(u8 *aOut, u16 nVal){
  aOut[0] = (u8)((nVal>>8) & 0xFF);
  aOut[1] = (u8)(nVal & 0xFF);

static void sortedBlobFree(Blob *pBlob){
  assert( pBlob->pEnv || pBlob->pData==0 );
  if( pBlob->pData ) lsmFree(pBlob->pEnv, pBlob->pData);
  memset(pBlob, 0, sizeof(Blob));
}


static int pageGetNRec(u8 *aData, int nData){
  return (int)lsmGetU16(&aData[SEGMENT_NRECORD_OFFSET(nData)]);
}

static int pageGetPtr(u8 *aData, int nData){
  return (int)lsmGetU32(&aData[SEGMENT_POINTER_OFFSET(nData)]);
}

static int pageGetFlags(u8 *aData, int nData){
  return (int)lsmGetU16(&aData[SEGMENT_FLAGS_OFFSET(nData)]);
}

static u8 *pageGetCell(u8 *aData, int nData, int iCell){
  return &aData[lsmGetU16(&aData[SEGMENT_CELLPTR_OFFSET(nData, iCell)])];
}

/*
** Return the decoded (possibly relative) pointer value stored in cell 
** iCell from page aData/nData.
*/
static int pageGetRecordPtr(u8 *aData, int nData, int iCell){
  int iRet;                       /* Return value */
  u8 *aCell;                      /* Pointer to cell iCell */
  aCell = pageGetCell(aData, nData, iCell);
  lsmVarintGet32(&aCell[1], &iRet);
  return iRet;
}

static void segmentPtrSetPage(SegmentPtr *pPtr, Page *pNext){
  lsmFsPageRelease(pPtr->pPg);
  if( pNext ){
    int nData;
    u8 *aData = lsmFsPageData(pNext, &nData);
    pPtr->nCell = pageGetNRec(aData, nData);
    pPtr->flags = pageGetFlags(aData, nData);
    pPtr->iPtr = pageGetPtr(aData, nData);
  }
  pPtr->pPg = pNext;
}

/*
** Load a new page into the SegmentPtr object pPtr.
*/
static int segmentPtrLoadPage(
  FileSystem *pFS,
  SegmentPtr *pPtr,              /* Load page into this SegmentPtr object */
  int iNew                       /* Page number of new page */
){
  Page *pPg = 0;                 /* The new page */
  int rc;                        /* Return Code */

  assert( pPtr->pSeg==0 
       || pPtr->pRun==&pPtr->pSeg->run 
       || pPtr->pRun==&pPtr->pSeg->sep 
  );
  rc = lsmFsDbPageGet(pFS, iNew, &pPg);
  assert( rc==LSM_OK || pPg==0 );
  segmentPtrSetPage(pPtr, pPg);

  return rc;
}

static int segmentPtrNextPage(
  SegmentPtr *pPtr,              /* Load page into this SegmentPtr object */
  int eDir                       /* +1 for next(), -1 for prev() */
){
  Page *pNext;                   /* New page to load */
  int rc;                        /* Return code */

  assert( eDir==1 || eDir==-1 );
  assert( pPtr->pPg );
  assert( (pPtr->pSeg==0 && eDir>0)
       || pPtr->pRun==&pPtr->pSeg->run 
       || pPtr->pRun==&pPtr->pSeg->sep 
  );

  rc = lsmFsDbPageNext(pPtr->pRun, pPtr->pPg, eDir, &pNext);
  assert( rc==LSM_OK || pNext==0 );
  segmentPtrSetPage(pPtr, pNext);
  return rc;
}

static int sortedReadData(
  Page *pPg,
  int iOff,
  int nByte,
  void **ppData,
  Blob *pBlob
){
  int rc = LSM_OK;
  int iEnd;
  int nData;
  int nCell;
  u8 *aData;

  aData = lsmFsPageData(pPg, &nData);
  nCell = lsmGetU16(&aData[SEGMENT_NRECORD_OFFSET(nData)]);
  iEnd = SEGMENT_EOF(nData, nCell);
  assert( iEnd>0 && iEnd<nData );

  if( iOff+nByte<=iEnd ){
    *ppData = (void *)&aData[iOff];
  }else{

        rc = lsmFsDbPageNext(0, pPg, 1, &pNext);
        if( rc==LSM_OK && pNext==0 ){
          rc = LSM_CORRUPT_BKPT;
        }
        if( rc ) break;
        lsmFsPageRelease(pPg);
        pPg = pNext;
        aData = lsmFsPageData(pPg, &nData);
        flags = lsmGetU16(&aData[SEGMENT_FLAGS_OFFSET(nData)]);
      }while( flags&SEGMENT_BTREE_FLAG );

      iEnd = SEGMENT_EOF(nData, lsmGetU16(&aData[nData-2]));
      assert( iEnd>0 && iEnd<nData );
    }

    lsmFsPageRelease(pPg);
  }

  return rc;
}

static int segmentPtrReadData(


  SegmentPtr *pPtr,



  int iOff,
  int nByte,


  void **ppData,



  Blob *pBlob







){



  return sortedReadData(pPtr->pPg, iOff, nByte, ppData, pBlob);
}

static u8 *pageGetKey(
  Page *pPg,                      /* Page to read from */
  int iCell,                      /* Index of cell on page to read */
  int *piTopic,                   /* OUT: Topic associated with this key */
  int *pnKey,                     /* OUT: Size of key in bytes */
  Blob *pBlob                     /* If required, use this for dynamic memory */
){
  u8 *pKey;
  int nDummy;
  int eType;
  u8 *aData;
  int nData;

  aData = lsmFsPageData(pPg, &nData);

  assert( !(pageGetFlags(aData, nData) & SEGMENT_BTREE_FLAG) );


  pKey = pageGetCell(aData, nData, iCell);
  eType = *pKey++;
  pKey += lsmVarintGet32(pKey, &nDummy);
  pKey += lsmVarintGet32(pKey, pnKey);
  if( rtIsWrite(eType) ){
    pKey += lsmVarintGet32(pKey, &nDummy);

  if( (void *)aKey!=pBlob->pData ){
    rc = sortedBlobSet(pEnv, pBlob, aKey, nKey);
  }

  return rc;
}









































































































































































































































































































































































































































static int segmentPtrLoadCell(
  SegmentPtr *pPtr,              /* Load page into this SegmentPtr object */
  int iNew                       /* Cell number of new cell */
){
  int rc = LSM_OK;
  if( pPtr->pPg ){
    u8 *aData;                    /* Pointer to page data buffer */
    int iOff;                     /* Offset in aData[] to read from */
    int nPgsz;                    /* Size of page (aData[]) in bytes */

    assert( iNew<pPtr->nCell );
    pPtr->iCell = iNew;
    aData = lsmFsPageData(pPtr->pPg, &nPgsz);
    iOff = lsmGetU16(&aData[SEGMENT_CELLPTR_OFFSET(nPgsz, pPtr->iCell)]);
    pPtr->eType = aData[iOff];
    iOff++;
    iOff += lsmVarintGet32(&aData[iOff], &pPtr->iPgPtr);
    iOff += lsmVarintGet32(&aData[iOff], &pPtr->nKey);
    if( rtIsWrite(pPtr->eType) ){
      iOff += lsmVarintGet32(&aData[iOff], &pPtr->nVal);
    }

    rc = segmentPtrReadData(
        pPtr, iOff, pPtr->nKey, &pPtr->pKey, &pPtr->blob1
    );
    if( rc==LSM_OK && rtIsWrite(pPtr->eType) ){
      rc = segmentPtrReadData(
          pPtr, iOff+pPtr->nKey, pPtr->nVal, &pPtr->pVal, &pPtr->blob2
      );



    }
  }

  return rc;
}

static int sortedKeyCompare(
  int (*xCmp)(void *, int, void *, int),
  int iLhsTopic, void *pLhsKey, int nLhsKey,
  int iRhsTopic, void *pRhsKey, int nRhsKey
){
  int res = iLhsTopic - iRhsTopic;
  if( res==0 ){
    res = xCmp(pLhsKey, nLhsKey, pRhsKey, nRhsKey);
  }
  return res;
}

void lsmSortedSplitkey(lsm_db *pDb, Level *pLevel, int *pRc){
  lsm_env *pEnv = pDb->pEnv;      /* Environment handle */
  int rc = *pRc;
  int i;
  Merge *pMerge = pLevel->pMerge;

  for(i=0; rc==LSM_OK && i<pLevel->nRight; i++){
    Page *pPg = 0;
    int iTopic;
    Blob blob = {0, 0, 0, 0};
    SortedRun *pRun = &pLevel->aRhs[i].run;

    assert( pRun->iFirst!=0 );

    rc = lsmFsDbPageGet(pDb->pFS, pMerge->aInput[i].iPg, &pPg);
    if( rc==LSM_OK ){
      rc = pageGetKeyCopy(pEnv, pPg, pMerge->aInput[i].iCell, &iTopic, &blob);
    }
    if( rc==LSM_OK ){
      int res = -1;
      if( pLevel->pSplitKey ){

    int i;
    pCsr->aPtr[0].pSeg = &pLevel->lhs;
    pCsr->nPtr = nPtr;

    for(i=0; i<pLevel->nRight; i++){
      pCsr->aPtr[i+1].pSeg = &pLevel->aRhs[i];
    }
    for(i=0; i<pCsr->nPtr; i++){
      pCsr->aPtr[i].pRun = &pCsr->aPtr[i].pSeg->run;
    }
  }

  return rc;
}

static int levelCursorInitRun(
  lsm_db *pDb,
  SortedRun *pRun, 
  int (*xCmp)(void *, int, void *, int),
  LevelCursor *pCsr              /* Cursor structure to initialize */
){
  int rc = LSM_OK;

  memset(pCsr, 0, sizeof(LevelCursor));
  pCsr->pFS = pDb->pFS;
  pCsr->bIgnoreSeparators = 1;
  pCsr->xCmp = xCmp;
  pCsr->nPtr = 1;
  pCsr->aPtr = (SegmentPtr*)lsmMallocZeroRc(pDb->pEnv, 
      sizeof(SegmentPtr)*pCsr->nPtr, &rc
  );

  if( rc==LSM_OK ){
    pCsr->aPtr[0].pRun = pRun;
  }

  return rc;
}

static void segmentPtrReset(SegmentPtr *pPtr){
  lsmFsPageRelease(pPtr->pPg);

  FileSystem *pFS, 
  SegmentPtr *pPtr, 
  int bLast, 
  int *pRc
){
  if( *pRc==LSM_OK ){
    Page *pNew = 0;
    Pgno iPg = (bLast ? pPtr->pRun->iLast : pPtr->pRun->iFirst);
    *pRc = lsmFsDbPageGet(pFS, iPg, &pNew);
    segmentPtrSetPage(pPtr, pNew);
  }
}

/*
** Try to move the segment pointer passed as the second argument so that it

  int bLast,                      /* True for last, false for first */
  int *pRc                        /* IN/OUT error code */
){
  if( *pRc==LSM_OK ){
    int rc = LSM_OK;

    segmentPtrEndPage(pCsr->pFS, pPtr, bLast, &rc);
    while( rc==LSM_OK && pPtr->pPg && pPtr->nCell==0 ){



      rc = segmentPtrNextPage(pPtr, (bLast ? -1 : 1));
    }
    if( rc==LSM_OK && pPtr->pPg ){
      rc = segmentPtrLoadCell(pPtr, bLast ? (pPtr->nCell-1) : 0);
    }

    if( rc==LSM_OK && pPtr->pPg && (

  for(eDir=-1; eDir<=1; eDir+=2){
    Page *pTest = pPtr->pPg;

    lsmFsPageRef(pTest);
    while( pTest ){
      Page *pNext;

      int rc = lsmFsDbPageNext(pPtr->pRun, pTest, eDir, &pNext);
      lsmFsPageRelease(pTest);
      pTest = pNext;
      assert( rc==LSM_OK );

      if( pTest ){
        int nData;
        u8 *aData = lsmFsPageData(pTest, &nData);
        int nCell = pageGetNRec(aData, nData);
        int flags = pageGetFlags(aData, nData);
        if( nCell && 0==(flags&SEGMENT_BTREE_FLAG) ){
          int nPgKey;
          int iPgTopic;
          u8 *pPgKey;
          int res;

    ** page.  */
    res = iLastTopic - iTopic;
    if( res==0 ) res = pCsr->xCmp(pLastKey, nLastKey, pKey, nKey);
    if( res>=0 ) break;

    /* Advance to the next page that contains at least one key. */
    do {
      rc = lsmFsDbPageNext(pPtr->pRun, pPtr->pPg, 1, &pNext);
      if( pNext==0 ) break;
      assert( rc==LSM_OK );
      segmentPtrSetPage(pPtr, pNext);
    }while( (pPtr->nCell==0 || (pPtr->flags & SEGMENT_BTREE_FLAG)) );
    if( pNext==0 ) break;

    /* This should probably be an LSM_CORRUPT error. */
    assert( rc!=LSM_OK || (pPtr->flags & PGFTR_SKIP_THIS_FLAG) );
  }

  iPtrOut = pPtr->iPtr;

  /* Assert that this page is the right page of this segment for the key
  ** that we are searching for. Do this by loading page (iPg-1) and testing
  ** that pKey/nKey is greater than all keys on that page, and then by 
  ** loading (iPg+1) and testing that pKey/nKey is smaller than all
  ** the keys it houses.  */
#if 0
  assert( assertKeyLocation(pCsr, pPtr, pKey, nKey) );
#endif

  assert( pPtr->nCell>0 
       || pPtr->pRun->nSize==1 
       || lsmFsPageNumber(pPtr->pPg)==pPtr->pRun->iLast
  );
  if( pPtr->nCell==0 ){
    segmentPtrReset(pPtr);
  }else{
    iMin = 0;
    iMax = pPtr->nCell-1;

    );
    if( res ) iBest = i;
  }

  pCsr->iCurrentPtr = iBest;
}

static int seekInSeparators(
  LevelCursor *pCsr,
  SegmentPtr *pPtr,               /* Segment to seek within */
  void *pKey, int nKey,           /* Key to seek to */
  int *piPtr                      /* OUT: FC pointer */
){
  int rc;
  int iPg;

  Blob blob = {0, 0, 0};
  int iTopic = 0;                 /* TODO: Fix me */
  SortedRun *pSep = &pPtr->pSeg->sep;

  iPg = pSep->iRoot;
  do {
    Page *pPg;
    rc = lsmFsDbPageGet(pCsr->pFS, iPg, &pPg);

    if( rc==LSM_OK ){
      u8 *aData;                  /* Buffer containing page data */
      int nData;                  /* Size of aData[] in bytes */
      int iMin;
      int iMax;
      int nRec;
      int flags;

      aData = lsmFsPageData(pPg, &nData);
      flags = pageGetFlags(aData, nData);
      if( (flags & SEGMENT_BTREE_FLAG)==0 ){
        lsmFsPageRelease(pPg);
        break;
      }

      iPg = pageGetPtr(aData, nData);
      nRec = pageGetNRec(aData, nData);

      iMin = 0;
      iMax = nRec-1;
      while( iMax>=iMin ){
        Page *pRef = 0;
        int iTry = (iMin+iMax)/2;
        void *pKeyT; int nKeyT;       /* Key for cell iTry */
        int iTopicT;                  /* Topic for key pKeyT/nKeyT */
        int iPtr;                     /* Pointer associated with cell iTry */
        u8 *aCell;                    /* Pointer to cell iTry */
        int res;                      /* (pKey - pKeyT) */
        int eType;

        aCell = pageGetCell(aData, nData, iTry);
        eType = *aCell++;
        aCell += lsmVarintGet32(aCell, &iPtr);
        if( eType==0 ){
          /* If eType==0, then this b-tree cell does not contain a key. 
          ** Instead, it is a reference to another cell in the same separators
          ** array that does contain a key. */
          Pgno iRef;
          aCell += lsmVarintGet32(aCell, &iRef);
          rc = lsmFsDbPageGet(pCsr->pFS, iRef, &pRef);
          if( rc!=LSM_OK ) break;
          pKeyT = pageGetKey(pRef, 0, &iTopicT, &nKeyT, &blob);
        }else{
          aCell += lsmVarintGet32(aCell, &nKeyT);
          pKeyT = (void *)aCell;
          iTopicT = rtTopic(eType);
        }

        res = iTopic - iTopicT;
        if( res==0 ) res = pCsr->xCmp(pKey, nKey, pKeyT, nKeyT);

        if( res<0 ){
          iPg = iPtr;
          iMax = iTry-1;
        }else{
          iMin = iTry+1;
        }
        lsmFsPageRelease(pRef);
      }
      lsmFsPageRelease(pPg);

    }
  }while( rc==LSM_OK );

  if( rc==LSM_OK ){
    assert( pPtr->pRun==&pPtr->pSeg->run );
    pPtr->pRun = pSep;
    rc = segmentPtrLoadPage(pCsr->pFS, pPtr, iPg);
    if( rc==LSM_OK ){
      rc = segmentPtrSeek(pCsr, pPtr, pKey, nKey, 0, piPtr);
    }
    pPtr->pRun = &pPtr->pSeg->run;
  }

  sortedBlobFree(&blob);


  return rc;
}

static int seekInSegment(
  LevelCursor *pCsr, 
  SegmentPtr *pPtr,
  void *pKey, int nKey,
  int iPg,                        /* Page to search */
  int eSeek,                      /* Search bias - see above */
  int *piPtr                      /* OUT: FC pointer */
){
  int iPtr = iPg;
  int rc = LSM_OK;

  assert( pPtr->pRun==&pPtr->pSeg->run );

  if( segmentHasSeparators(pPtr->pSeg) ){
    rc = seekInSeparators(pCsr, pPtr, pKey, nKey, &iPtr);


  }else if( iPtr==0 ){
    iPtr = pPtr->pSeg->run.iFirst;
  }

  if( rc==LSM_OK ){
    rc = segmentPtrLoadPage(pCsr->pFS, pPtr, iPtr);
  }


  if( rc==LSM_OK ){
    rc = segmentPtrSeek(pCsr, pPtr, pKey, nKey, eSeek, piPtr);
  }
  return rc;
}

/*

  lsmTreeCursorDestroy(pCsr->pTreeCsr);

  /* Close the sorted file cursors */
  for(i=0; i<pCsr->nSegCsr; i++){
    segmentCursorClose(pEnv, &pCsr->aSegCsr[i]);
  }




  /* Free allocations */
  lsmFree(pEnv, pCsr->aSegCsr);
  lsmFree(pEnv, pCsr->aTree);
  lsmFree(pEnv, pCsr->pSystemVal);

  /* Zero fields */
  pCsr->nSegCsr = 0;
  pCsr->aSegCsr = 0;
  pCsr->nTree = 0;
  pCsr->aTree = 0;
  pCsr->pSystemVal = 0;
  pCsr->pSnap = 0;
  pCsr->pTreeCsr = 0;

}

void lsmMCursorClose(MultiCursor *pCsr){
  if( pCsr ){
    lsm_db *pDb = pCsr->pDb;
    MultiCursor **pp;             /* Iterator variable */

    /* The cursor may or may not be currently part of the linked list 
    ** starting at lsm_db.pCsr. If it is, extract it.  */
    for(pp=&pDb->pCsr; *pp; pp=&((*pp)->pNext)){
      if( *pp==pCsr ){
        *pp = pCsr->pNext;
        break;
      }
    }

    /* Free the allocation used to cache the current key, if any. */
    sortedBlobFree(&pCsr->key);


    /* Free the component cursors */
    mcursorFreeComponents(pCsr);

    /* Free the cursor structure itself */
    lsmFree(pDb->pEnv, pCsr);
  }
}

#define MULTICURSOR_ADDLEVEL_ALL 1
#define MULTICURSOR_ADDLEVEL_RHS 2
#define MULTICURSOR_ADDLEVEL_LHS_SEP 3
#define MULTICURSOR_ADDLEVEL_RHS_SEP 4

/*
** Add segments belonging to level pLevel to the multi-cursor pCsr. The
** third argument must be one of the following:
**
**   MULTICURSOR_ADDLEVEL_ALL
**     Add all segments in the level to the cursor.
**
**   MULTICURSOR_ADDLEVEL_RHS
**     Add only the rhs segments in the level to the cursor.
**
**   MULTICURSOR_ADDLEVEL_LHS_SEP
**     Add only the lhs segment. And iterate through its separators array,
**     not the main run array.
**
**   MULTICURSOR_ADDLEVEL_RHS_SEP
**     Add only the first segment from the rhs. And iterate through its 
**     separators array, not the main run array.
**
** RHS and SEP are only used by cursors created to use as data sources when
** creating new segments (either when flushing the in-memory tree to disk or
** when merging existing runs).
*/
int multiCursorAddLevel(
  MultiCursor *pCsr,              /* Multi-cursor to add segment to */ 
  Level *pLevel,                  /* Level to add to multi-cursor merge */
  int eMode                       /* A MULTICURSOR_ADDLEVEL_*** constant */
){
  int rc = LSM_OK;
  int i;
  int nAdd = (eMode==MULTICURSOR_ADDLEVEL_RHS ? pLevel->nRight : 1);

  assert( eMode==MULTICURSOR_ADDLEVEL_ALL
       || eMode==MULTICURSOR_ADDLEVEL_RHS
       || eMode==MULTICURSOR_ADDLEVEL_LHS_SEP
  );










  for(i=0; i<nAdd; i++){
    LevelCursor *pNew;
    lsm_db *pDb = pCsr->pDb;

    /* Grow the pCsr->aSegCsr array if required */
    if( 0==(pCsr->nSegCsr % 16) ){
      int nByte;
      LevelCursor *aNew;
      nByte = sizeof(LevelCursor) * (pCsr->nSegCsr+16);
      aNew = (LevelCursor *)lsmRealloc(pDb->pEnv, pCsr->aSegCsr, nByte);
      if( aNew==0 ) return LSM_NOMEM_BKPT;
      memset(&aNew[pCsr->nSegCsr], 0, sizeof(LevelCursor)*16);
      pCsr->aSegCsr = aNew;
    }
    pNew = &pCsr->aSegCsr[pCsr->nSegCsr];

    switch( eMode ){
      case MULTICURSOR_ADDLEVEL_ALL:
        rc = levelCursorInit(pDb, pLevel, pCsr->xCmp, pNew);
        break;

      case MULTICURSOR_ADDLEVEL_RHS:
        rc = levelCursorInitRun(pDb, &pLevel->aRhs[i].run, pCsr->xCmp, pNew);
        break;

      case MULTICURSOR_ADDLEVEL_LHS_SEP:
        rc = levelCursorInitRun(pDb, &pLevel->lhs.sep, pCsr->xCmp, pNew);
        break;
    }
    if( pCsr->flags & CURSOR_IGNORE_SYSTEM ){
      pNew->bIgnoreSystem = 1;
    }
    if( rc==LSM_OK ) pCsr->nSegCsr++;

  }

  return rc;
}


static int multiCursorNew(
  lsm_db *pDb,                    /* Database handle */
  Snapshot *pSnap,                /* Snapshot to use for this cursor */
  int useTree,                    /* If true, search the in-memory tree */
  int bUserOnly,                  /* If true, ignore all system data */
  MultiCursor **ppCsr             /* OUT: Allocated cursor */
){
  int rc = LSM_OK;                /* Return Code */
  MultiCursor *pCsr = *ppCsr;     /* Allocated multi-cursor */

  if( pCsr==0 ){
    pCsr = (MultiCursor *)lsmMallocZeroRc(pDb->pEnv, sizeof(MultiCursor), &rc);




  }

  if( rc==LSM_OK ){
    if( useTree ){
      assert( pDb->pTV );
      rc = lsmTreeCursorNew(pDb, &pCsr->pTreeCsr);
    }

  if( rc!=LSM_OK ){
    lsmMCursorClose(pCsr);
    pCsr = 0;
  }
  *ppCsr = pCsr;
  return rc;
}


















static void multiCursorReadSeparators(MultiCursor *pCsr){
  if( pCsr->nSegCsr>0 ){
    pCsr->aSegCsr[pCsr->nSegCsr-1].bIgnoreSeparators = 0;
  }
}

  lsm_db *pDb,                    /* Database handle */
  MultiCursor **ppCsr             /* OUT: Allocated cursor */
){
  MultiCursor *pCsr = 0;
  int rc;

  rc = multiCursorAllocate(pDb, 0, &pCsr);
  if( rc==LSM_OK ){
    pCsr->pNext = pDb->pCsr;
    pDb->pCsr = pCsr;
  }

  assert( (rc==LSM_OK)==(pCsr!=0) );
  *ppCsr = pCsr;
  return rc;
}

#define CURSOR_DATA_TREE      0
#define CURSOR_DATA_SYSTEM    1
#define CURSOR_DATA_SEGMENT   2

static void multiCursorGetKey(
  MultiCursor *pCsr, 
  int iKey,
  int *peType,                    /* OUT: Key type (SORTED_WRITE etc.) */
  void **ppKey,                   /* OUT: Pointer to buffer containing key */
  int *pnKey                      /* OUT: Size of *ppKey in bytes */
){

        nKey = 6;
        eType = SORTED_SYSTEM_WRITE;
      }
      break;

    default: {
      int iSeg = iKey - CURSOR_DATA_SEGMENT;




      if( iSeg<pCsr->nSegCsr && segmentCursorValid(&pCsr->aSegCsr[iSeg]) ){
        segmentCursorKey(&pCsr->aSegCsr[iSeg], &pKey, &nKey);
        segmentCursorType(&pCsr->aSegCsr[iSeg], &eType);
      }
      break;
    }
  }

         && segmentCursorValid(&pCsr->aSegCsr[iVal-CURSOR_DATA_SEGMENT]) 
  ){
    segmentCursorValue(&pCsr->aSegCsr[iVal-CURSOR_DATA_SEGMENT], ppVal, pnVal);
  }else{
    *ppVal = 0;
    *pnVal = 0;
  }

  return rc;
}

int lsmSortedLoadSystem(lsm_db *pDb){
  MultiCursor *pCsr = 0;          /* Cursor used to retreive free-list */
  int rc;                         /* Return Code */

  pCsr->aTree[iOut] = iRes;
}

static int multiCursorAllocTree(MultiCursor *pCsr){
  int rc = LSM_OK;
  if( pCsr->aTree==0 ){
    int nByte;                    /* Bytes of space to allocate */



    pCsr->nTree = 2;
    while( pCsr->nTree<(CURSOR_DATA_SEGMENT+pCsr->nSegCsr) ){
      pCsr->nTree = pCsr->nTree*2;
    }

    nByte = sizeof(int)*pCsr->nTree*2;
    pCsr->aTree = (int *)lsmMallocZeroRc(pCsr->pDb->pEnv, nByte, &rc);
  }
  return rc;

  if( pCsr->flags & CURSOR_NEW_SYSTEM ){
    assert( bLast==0 );
    pCsr->flags |= CURSOR_AT_FREELIST;
  }
  for(i=0; rc==LSM_OK && i<pCsr->nSegCsr; i++){
    rc = segmentCursorEnd(&pCsr->aSegCsr[i], bLast);
  }






  if( rc==LSM_OK ){
    rc = multiCursorAllocTree(pCsr);
  }

  if( rc==LSM_OK ){
    for(i=pCsr->nTree-1; i>0; i--){

        if( pCsr->flags & CURSOR_AT_FREELIST ){
          pCsr->flags &= ~CURSOR_AT_FREELIST;
          pCsr->flags |= CURSOR_AT_LEVELS;
        }else{
          pCsr->flags &= ~CURSOR_AT_LEVELS;
        }



      }else{
        LevelCursor *pLevel = &pCsr->aSegCsr[iKey-CURSOR_DATA_SEGMENT];
        rc = segmentCursorAdvance(pLevel, bReverse);
      }
      if( rc==LSM_OK ){
        int i;
        for(i=(iKey+pCsr->nTree)/2; i>0; i=i/2){

    }
    *pnKey = nKey; 
  }
  return LSM_OK;
}

int lsmMCursorValue(MultiCursor *pCsr, void **ppVal, int *pnVal){




  assert( pCsr->aTree );
  assert( rtIsDelete(pCsr->eType)==0 || !(pCsr->flags & CURSOR_IGNORE_DELETE) );

  return multiCursorGetVal(pCsr, pCsr->aTree[1], ppVal, pnVal);












}

int lsmMCursorType(MultiCursor *pCsr, int *peType){
  assert( pCsr->aTree );
  multiCursorGetKey(pCsr, pCsr->aTree[1], peType, 0, 0);
  return LSM_OK;
}

** This function copies all such b-tree pages to new locations, so that
** they can be modified as required.
**
** The complication is that not all database pages are the same size - due
** to the way the file.c module works some (the first and last in each block)
** are 4 bytes smaller than the others.
*/
static int mergeWorkerMoveHierarchy(MergeWorker *pMW){



  SortedRun *pSep;                /* Separators run being modified */
  lsm_db *pDb = pMW->pDb;         /* Database handle */
  int rc = LSM_OK;                /* Return code */
  int i;
  int iRight = 0;
  int nHier = pMW->nHier;
  Page **apHier = pMW->apHier;


  assert( nHier>0 && pMW->pLevel->pMerge->bHierReadonly );

  pSep = &pMW->pLevel->lhs.sep;

  for(i=0; rc==LSM_OK && i<nHier; i++){
    Page *pNew = 0;
    rc = lsmFsSortedAppend(pDb->pFS, pDb->pWorker, pSep, &pNew);
    assert( rc==LSM_OK );

    if( rc==LSM_OK ){
      u8 *a1; int n1;
      u8 *a2; int n2;

      a1 = lsmFsPageData(pNew, &n1);
      a2 = lsmFsPageData(apHier[i], &n2);
      assert( n1==n2 || n1+4==n2 || n2+4==n1 );

      if( n1>=n2 ){
        /* If n1 (size of the new page) is equal to or greater than n2 (the
        ** size of the old page), then copy the data into the new page. If
        ** n1==n2, this could be done with a single memcpy(). However, 
        ** since sometimes n1>n2, the page content and footer must be copied 

        lsmFsPageRelease(pNew);
      }
    }
  }

#ifdef LSM_DEBUG
  if( rc==LSM_OK ){
    for(i=0; i<nHier; i++) assert( lsmFsPageWritable(pMW->apHier[i]) );
  }
#endif

  if( rc==LSM_OK ){
    pMW->pLevel->pMerge->bHierReadonly = 0;
  }
  return rc;
}

/*
** Allocate and populate the MergeWorker.apHier[] array.
*/
static int mergeWorkerLoadHierarchy(MergeWorker *pMW){
  int rc = LSM_OK;



  SortedRun *pSep = &pMW->pLevel->lhs.sep;


  if( pMW->apHier==0 && pSep->iRoot!=0 ){
    int bHierReadonly = pMW->pLevel->pMerge->bHierReadonly;
    FileSystem *pFS = pMW->pDb->pFS;
    lsm_env *pEnv = pMW->pDb->pEnv;
    Page **apHier = 0;
    int nHier = 0;
    int iPg = pSep->iRoot;

    do {
      Page *pPg = 0;
      u8 *aData;
      int nData;
      int flags;

      rc = lsmFsDbPageGet(pFS, iPg, &pPg);
      if( rc!=LSM_OK ) break;

      aData = lsmFsPageData(pPg, &nData);
      flags = pageGetFlags(aData, nData);
      if( flags&SEGMENT_BTREE_FLAG ){
        Page **apNew = (Page **)lsmRealloc(
            pEnv, apHier, sizeof(Page *)*(nHier+1)
        );
        if( apNew==0 ){
          rc = LSM_NOMEM_BKPT;

      }else{
        lsmFsPageRelease(pPg);
        break;
      }
    }while( 1 );

    if( rc==LSM_OK ){
      pMW->nHier = nHier;
      pMW->apHier = apHier;
    }else{
      int i;
      for(i=0; i<nHier; i++){
        lsmFsPageRelease(apHier[i]);
      }
      lsmFree(pEnv, apHier);
    }

** The reason for having the page footer pointer point to the right-child
** (instead of the left) is that doing things this way makes the 
** segWriterMoveHierarchy() operation less complicated (since the pointers 
** that need to be updated are all stored as fixed-size integers within the 
** page footer, not varints in page records).
**
** Records may not span b-tree pages. If this function is called to add a
** record larger than (page-size / 4) bytes, then a pointer to the separators
** array page that contains the main record is added to the b-tree instead.
** In this case the record format is:
**
**         + 0x00 byte (1 byte) 
**         + Absolute pointer value (varint),
**         + Absolute page number of page containing key (varint).
**
** See function seekInSeparators() for the code that traverses b-tree pages.
*/
static int mergeWorkerPushHierarchy(
  MergeWorker *pMW,               /* Merge worker object */

  Pgno iKeyPg,                    /* Page that will contain pKey/nKey */
  int iTopic,                     /* Topic value for this key */
  void *pKey,                     /* Pointer to key buffer */
  int nKey                        /* Size of pKey buffer in bytes */
){
  lsm_db *pDb = pMW->pDb;         /* Database handle */
  int rc;                         /* Return Code */
  int iLevel;                     /* Level of b-tree hierachy to write to */
  int nData;                      /* Size of aData[] in bytes */
  u8 *aData;                      /* Page data for level iLevel */
  int iOff;                       /* Offset on b-tree page to write record to */
  int nRec;                       /* Initial number of records on b-tree page */
  Pgno iPtr;                      /* Pointer value to accompany pKey/nKey */
  int bIndirect;                  /* True to use an indirect record */




  /* If there exists a b-tree hierarchy and it is not loaded into 
  ** memory, load it now.  */


  rc = mergeWorkerLoadHierarchy(pMW);

  /* TODO: What the heck does this do? */



  if( pMW->nHier ){
    aData = lsmFsPageData(pMW->apHier[0], &nData);
    iPtr = lsmGetU32(&aData[SEGMENT_POINTER_OFFSET(nData)]);
  }else{
    iPtr = pMW->pLevel->lhs.sep.iFirst;
  }

  if( pMW->nHier && pMW->pLevel->pMerge->bHierReadonly ){
    rc = mergeWorkerMoveHierarchy(pMW);
    if( rc!=LSM_OK ) goto push_hierarchy_out;
  }

  /* Determine if the indirect format should be used. */
  bIndirect = (nKey*4 > lsmFsPageSize(pMW->pDb->pFS));

  /* The MergeWorker.apHier[] array contains the right-most leaf of the b-tree
  ** hierarchy, the root node, and all nodes that lie on the path between.
  ** apHier[0] is the right-most leaf and apHier[pMW->nHier-1] is the current
  ** root page.
  **
  ** This loop searches for a node with enough space to store the key on,
  ** starting with the leaf and iterating up towards the root. When the loop
  ** exits, the key may be written to apHier[iLevel].
  */
  for(iLevel=0; iLevel<=pMW->nHier; iLevel++){
    int nByte;                    /* Number of free bytes required */
    int iRight;                   /* Right hand pointer from aData[]/nData */

    if( iLevel==pMW->nHier ){
      /* Extend the array and allocate a new root page. */
      Page **aNew;
      aNew = (Page **)lsmRealloc(
          pMW->pDb->pEnv, pMW->apHier, sizeof(Page *)*(pMW->nHier+1)
      );
      if( !aNew ){
        rc = LSM_NOMEM_BKPT;
        goto push_hierarchy_out;
      }
      pMW->apHier = aNew;
    }else{
      int nFree;

      /* If the key will fit on this page, break out of the loop. */
      assert( lsmFsPageWritable(pMW->apHier[iLevel]) );
      aData = lsmFsPageData(pMW->apHier[iLevel], &nData);
      iRight = lsmGetU32(&aData[SEGMENT_POINTER_OFFSET(nData)]);
      if( bIndirect ){
        nByte = 2 + 1 + lsmVarintLen32(iRight) + lsmVarintLen32(iKeyPg);
      }else{
        nByte = 2 + 1 + lsmVarintLen32(iRight) + lsmVarintLen32(nKey) + nKey;
      }
      nRec = pageGetNRec(aData, nData);
      nFree = SEGMENT_EOF(nData, nRec) - mergeWorkerPageOffset(aData, nData);
      if( nByte<=nFree ) break;

      /* Otherwise, it is full. Release it. */
      iPtr = lsmFsPageNumber(pMW->apHier[iLevel]);
      rc = lsmFsPageRelease(pMW->apHier[iLevel]);
    }

    /* Allocate a new page for apHier[iLevel]. */
    pMW->apHier[iLevel] = 0;
    if( rc==LSM_OK ){
      rc = lsmFsSortedAppend(
          pDb->pFS, pDb->pWorker, &pMW->pLevel->lhs.sep, &pMW->apHier[iLevel]
      );
    }
    if( rc!=LSM_OK ) goto push_hierarchy_out;

    aData = lsmFsPageData(pMW->apHier[iLevel], &nData);
    memset(aData, 0, nData);
    lsmPutU16(&aData[SEGMENT_FLAGS_OFFSET(nData)], SEGMENT_BTREE_FLAG);
    lsmPutU16(&aData[SEGMENT_NRECORD_OFFSET(nData)], 0);
    if( iLevel>0 ){
      iRight = lsmFsPageNumber(pMW->apHier[iLevel-1]);
      lsmPutU32(&aData[SEGMENT_POINTER_OFFSET(nData)], iRight);
    }

    if( iLevel==pMW->nHier ){
      pMW->nHier++;
      break;
    }
  }

  /* Write the key into page apHier[iLevel]. */
  aData = lsmFsPageData(pMW->apHier[iLevel], &nData);

  iOff = mergeWorkerPageOffset(aData, nData);

  nRec = pageGetNRec(aData, nData);
  lsmPutU16(&aData[SEGMENT_CELLPTR_OFFSET(nData, nRec)], iOff);
  lsmPutU16(&aData[SEGMENT_NRECORD_OFFSET(nData)], nRec+1);

  if( bIndirect ){
    aData[iOff++] = 0x00;
    iOff += lsmVarintPut32(&aData[iOff], iPtr);
    iOff += lsmVarintPut32(&aData[iOff], iKeyPg);
  }else{
    aData[iOff++] = (u8)(iTopic | SORTED_SEPARATOR);
    iOff += lsmVarintPut32(&aData[iOff], iPtr);
    iOff += lsmVarintPut32(&aData[iOff], nKey);
    memcpy(&aData[iOff], pKey, nKey);
  }

  if( iLevel>0 ){
    int iRight = lsmFsPageNumber(pMW->apHier[iLevel-1]);
    lsmPutU32(&aData[SEGMENT_POINTER_OFFSET(nData)], iRight);
  }

  /* Write the right-hand pointer of the right-most leaf page of the 
  ** b-tree heirarchy. */
  aData = lsmFsPageData(pMW->apHier[0], &nData);
  lsmPutU32(&aData[SEGMENT_POINTER_OFFSET(nData)], iKeyPg);

  /* Ensure that the SortedRun.iRoot field is correct. */
  pMW->pLevel->lhs.sep.iRoot = lsmFsPageNumber(pMW->apHier[pMW->nHier-1]);

push_hierarchy_out:
  return rc;
}

/*
** The merge-worker object passed as the first argument to this function
** was used for an in-memory tree flush. If one was required, the separators 
** array has been assembled in-memory (as a "phantom"). In this case it
** consists of leaf nodes only, there are no b-tree nodes. This function 
** materializes the phantom run (writes it into the db file) and appends
** any required b-tree nodes.
*/
static int mergeWorkerBuildHierarchy(MergeWorker *pMW){
  int rc = LSM_OK;

  assert( pMW->bFlush );
  assert( pMW->pLevel->lhs.sep.iRoot==0 );

  if( pMW->apPage[1] ){
    SortedRun *pRun;              /* Separators run to materialize */
    lsm_db *db = pMW->pDb;
    Blob blob = {0, 0, 0};
    Page *pPg;
    int iLast;

    /* Write the leaf pages into the file. They now have page numbers,
    ** which can be used as pointers in the b-tree hierarchy.  */
    pRun = &pMW->pLevel->lhs.sep;
    rc = lsmFsPhantomMaterialize(db->pFS, db->pWorker, pRun);

    if( rc==LSM_OK ){
      rc = lsmFsDbPageGet(db->pFS, pRun->iFirst, &pPg);
    }

    iLast = pRun->iLast;
    while( rc==LSM_OK && lsmFsPageNumber(pPg)!=iLast ){
      Page *pNext = 0;

      rc = lsmFsDbPageNext(pRun, pPg, 1, &pNext);
      lsmFsPageRelease(pPg);
      pPg = pNext;

      if( rc==LSM_OK ){
        u8 *aData;
        int nData;
        aData = lsmFsPageData(pPg, &nData);
        if( pageGetNRec(aData, nData)>0 ){
          u8 *pKey;
          int nKey;
          int iTopic;
          Pgno iPg = lsmFsPageNumber(pPg);

          pKey = pageGetKey(pPg, 0, &iTopic, &nKey, &blob);
          rc = mergeWorkerPushHierarchy(pMW, iPg, iTopic, pKey, nKey);
        }
      }
    }

    if( pMW->nHier>0 ){
      Page *pRoot = pMW->apHier[pMW->nHier-1];
      pRun->iRoot = lsmFsPageNumber(pRoot);
    }else{
      pRun->iRoot = pRun->iFirst;
    }

    lsmFsPageRelease(pPg);
    sortedBlobFree(&blob);
  }
  return rc;
}

static int keyszToSkip(FileSystem *pFS, int nKey){
  int nPgsz;                /* Nominal database page size */
  nPgsz = lsmFsPageSize(pFS);
  return LSM_MIN(((nKey * 4) / nPgsz), 3);
}

/*
** Advance to the next page of an output run being populated by merge-worker
** pMW. If bSep is true, the separators run output is advanced by one page.
** Otherwise, the main run.
**
** The footer of the new page is initialized to indicate that it contains
** zero records. The flags field is cleared. The page footer pointer field
** is set to iFPtr.
**
** If successful, LSM_OK is returned. Otherwise, an error code.
*/
static int mergeWorkerNextPage(
  MergeWorker *pMW,               /* Merge worker object to append page to */
  int bSep,                       /* True to append to the separators array */
  int iFPtr                       /* Pointer value for footer of new page */
){
  int rc = LSM_OK;                /* Return code */
  Page *pNext = 0;                /* New page appended to run */
  lsm_db *pDb = pMW->pDb;         /* Database handle */
  SortedRun *pRun;                /* Run to append to */

  assert( bSep==0 || bSep==1 );

  pRun = (bSep ? &pMW->pLevel->lhs.sep : &pMW->pLevel->lhs.run);
  rc = lsmFsSortedAppend(pDb->pFS, pDb->pWorker, pRun, &pNext);
  assert( rc!=LSM_OK || bSep || pRun->iFirst>0 );

  if( rc==LSM_OK ){
    u8 *aData;                    /* Data buffer belonging to page pNext */
    int nData;                    /* Size of aData[] in bytes */

    lsmFsPageRelease(pMW->apPage[bSep]);
    pMW->apPage[bSep] = pNext;
    pMW->pLevel->pMerge->aiOutputOff[bSep] = 0;

    aData = lsmFsPageData(pNext, &nData);
    lsmPutU16(&aData[SEGMENT_NRECORD_OFFSET(nData)], 0);
    lsmPutU16(&aData[SEGMENT_FLAGS_OFFSET(nData)], 0);
    lsmPutU32(&aData[SEGMENT_POINTER_OFFSET(nData)], iFPtr);

    if( bSep==0 ) pMW->nWork++;
  }

  return rc;
}

/*
** Write a blob of data into an output segment being populated by a 

    Merge *pMerge = pMW->pLevel->pMerge;
    int nCopy;                    /* Number of bytes to copy */
    u8 *aData;                    /* Pointer to buffer of current output page */
    int nData;                    /* Size of aData[] in bytes */
    int nRec;                     /* Number of records on current output page */
    int iOff;                     /* Offset in aData[] to write to */

    assert( lsmFsPageWritable(pMW->apPage[bSep]) );
   
    aData = lsmFsPageData(pMW->apPage[bSep], &nData);
    nRec = pageGetNRec(aData, nData);
    iOff = pMerge->aiOutputOff[bSep];
    nCopy = LSM_MIN(nRem, SEGMENT_EOF(nData, nRec) - iOff);

    memcpy(&aData[iOff], &aWrite[nWrite-nRem], nCopy);
    nRem -= nCopy;

    if( nRem>0 ){
      rc = mergeWorkerNextPage(pMW, bSep, iFPtr);
    }else{
      pMerge->aiOutputOff[bSep] = iOff + nCopy;
    }
  }

  return rc;
}


static int mergeWorkerWrite(
  MergeWorker *pMW,               /* Merge worker object to write into */
  int bSep,                       /* True to write to separators array */
  int eType,                      /* One of SORTED_SEPARATOR, WRITE or DELETE */
  void *pKey, int nKey,           /* Key value */
  void *pVal, int nVal,           /* Accompanying value, if any */
  int iPtr,                       /* Absolute value of page pointer, or 0 */
  int *piPtrOut                   /* OUT: Pointer to write to separators */
){
  int rc = LSM_OK;                /* Return code */
  Merge *pMerge;                  /* Persistent part of level merge state */
  int nHdr;                       /* Space required for this record header */
  Page *pPg;                      /* Page to write to */
  u8 *aData;                      /* Data buffer for page pWriter->pPage */
  int nData;                      /* Size of buffer aData[] in bytes */
  int nRec;                       /* Number of records on page pPg */
  int iFPtr;                      /* Value of pointer in footer of pPg */
  int iRPtr;                      /* Value of pointer written into record */
  int iOff;                       /* Current write offset within page pPg */
  SortedRun *pRun;                /* Run being written to */
  int flags = 0;                  /* If != 0, flags value for page footer */

  assert( bSep==0 || bSep==1 );
  assert( bSep==0 || rtIsSeparator(eType) );

  pMerge = pMW->pLevel->pMerge;    
  pRun = (bSep ? &pMW->pLevel->lhs.sep : &pMW->pLevel->lhs.run);

  pPg = pMW->apPage[bSep];
  aData = lsmFsPageData(pPg, &nData);
  nRec = pageGetNRec(aData, nData);
  iFPtr = pageGetPtr(aData, nData);

  /* If iPtr is 0, set it to the same value as the absolute pointer 
  ** stored as part of the previous record.  */
  if( iPtr==0 ){
    iPtr = iFPtr;

  ** The header space is:
  **
  **     1) record type - 1 byte.
  **     2) Page-pointer-offset - 1 varint
  **     3) Key size - 1 varint
  **     4) Value size - 1 varint (SORTED_WRITE only)
  */


  nHdr = 1 + lsmVarintLen32(iRPtr) + lsmVarintLen32(nKey);
  if( rtIsWrite(eType) ) nHdr += lsmVarintLen32(nVal);

  /* If the entire header will not fit on page pPg, or if page pPg is 
  ** marked read-only, advance to the next page of the output run. */
  iOff = pMerge->aiOutputOff[bSep];
  if( iOff<0 || iOff+nHdr > SEGMENT_EOF(nData, nRec+1) ){
    iFPtr = iFPtr + (nRec ? pageGetRecordPtr(aData, nData, nRec-1) : 0);
    iRPtr = iPtr - iFPtr;
    iOff = 0;
    nRec = 0;
    rc = mergeWorkerNextPage(pMW, bSep, iFPtr);
    pPg = pMW->apPage[bSep];
    aData = lsmFsPageData(pPg, &nData);

  }

  /* If this record header will be the first on the page, and the page is 
  ** not the very first in the entire run, special actions may need to be 
  ** taken:
  **
  **   * If currently writing the main run, *piPtrOut should be set to
  **     the current page number. The caller will add a key to the separators
  **     array that points to the current page.
  **
  **   * If currently writing the separators array, push a copy of the key
  **     into the b-tree hierarchy.
  */
  if( rc==LSM_OK && nRec==0 && pRun->iFirst!=pRun->iLast ){
    assert( pMerge->nSkip>=0 );

    if( bSep ){
      if( pMW->bFlush==0 ){
        Pgno iPg = lsmFsPageNumber(pPg);
        rc = mergeWorkerPushHierarchy(pMW, iPg, rtTopic(eType), pKey, nKey);
      }
    }else{
      if( pMerge->nSkip ){
        pMerge->nSkip--;
        flags = PGFTR_SKIP_THIS_FLAG;
      }else{
        *piPtrOut = lsmFsPageNumber(pPg);
        pMerge->nSkip = keyszToSkip(pMW->pDb->pFS, nKey);
      }
      if( pMerge->nSkip ) flags |= PGFTR_SKIP_NEXT_FLAG;
    }
  }

  /* Update the output segment */
  if( rc==LSM_OK ){


    /* Update the page footer. */
    lsmPutU16(&aData[SEGMENT_NRECORD_OFFSET(nData)], nRec+1);
    lsmPutU16(&aData[SEGMENT_CELLPTR_OFFSET(nData, nRec)], iOff);
    if( flags ) lsmPutU16(&aData[SEGMENT_FLAGS_OFFSET(nData)], flags);

    /* Write the entry header into the current page. */
    aData[iOff++] = eType;                                               /* 1 */
    iOff += lsmVarintPut32(&aData[iOff], iRPtr);                         /* 2 */
    iOff += lsmVarintPut32(&aData[iOff], nKey);                          /* 3 */
    if( rtIsWrite(eType) ) iOff += lsmVarintPut32(&aData[iOff], nVal);   /* 4 */
    pMerge->aiOutputOff[bSep] = iOff;

    /* Write the key and data into the segment. */
    assert( iFPtr==pageGetPtr(aData, nData) );
    rc = mergeWorkerData(pMW, bSep, iFPtr+iRPtr, pKey, nKey);
    if( rc==LSM_OK && rtIsWrite(eType) ){


      rc = mergeWorkerData(pMW, bSep, iFPtr+iRPtr, pVal, nVal);

    }
  }

  return rc;
}

  MultiCursor *pCsr = pMW->pCsr;

  /* Unless the merge has finished, save the cursor position in the
  ** Merge.aInput[] array. See function mergeWorkerInit() for the 
  ** code to restore a cursor position based on aInput[].  */
  if( pCsr ){
    Merge *pMerge = pMW->pLevel->pMerge;


    /* pMerge->nInput==0 indicates that this is a FlushTree() operation. */
    assert( pMerge->nInput==0 || pMW->pLevel->nRight>0 );
    assert( pMerge->nInput==0 || pMerge->nInput==pCsr->nSegCsr );

    for(i=0; i<pMerge->nInput; i++){
      SegmentPtr *pPtr = &pCsr->aSegCsr[i].aPtr[0];
      if( pPtr->pPg ){
        pMerge->aInput[i].iPg = lsmFsPageNumber(pPtr->pPg);
        pMerge->aInput[i].iCell = pPtr->iCell;
      }else{
        pMerge->aInput[i].iPg = 0;
        pMerge->aInput[i].iCell = 0;
      }
    }




  }

  lsmMCursorClose(pCsr);
  lsmFsPageRelease(pMW->apPage[0]);
  lsmFsPageRelease(pMW->apPage[1]);

  for(i=0; i<pMW->nHier; i++){



    lsmFsPageRelease(pMW->apHier[i]);
  }
  lsmFree(pMW->pDb->pEnv, pMW->apHier);




  pMW->pCsr = 0;
  pMW->apHier = 0;
  pMW->nHier = 0;
  pMW->apPage[0] = 0;
  pMW->apPage[1] = 0;
}

static int mergeWorkerFirstPage(MergeWorker *pMW){
  int rc;                         /* Return code */
  SortedRun *pRun;                /* Run containing sep. keys to merge in */
  Page *pPg = 0;                  /* First page of run pRun */



  assert( pMW->apPage[0]==0 );






  pRun = pMW->pCsr->aSegCsr[pMW->pCsr->nSegCsr-1].aPtr[0].pRun;
  rc = lsmFsDbPageGet(pMW->pDb->pFS, pRun->iFirst, &pPg);
  if( rc==LSM_OK ){
    u8 *aData;                    /* Buffer for page pPg */
    int nData;                    /* Size of aData[] in bytes */
    int iFPtr;                    /* Pointer value read from footer of pPg */
    aData = lsmFsPageData(pPg, &nData);
    iFPtr = pageGetPtr(aData, nData);
    lsmFsPageRelease(pPg);




    rc = mergeWorkerNextPage(pMW, 0, iFPtr);
  }

  return rc;
}

static int mergeWorkerStep(MergeWorker *pMW){
  lsm_db *pDb = pMW->pDb;       /* Database handle */
  MultiCursor *pCsr;            /* Cursor to read input data from */
  int rc;                       /* Return code */
  int eType;                    /* SORTED_SEPARATOR, WRITE or DELETE */
  void *pKey; int nKey;         /* Key */
  void *pVal; int nVal;         /* Value */
  Segment *pSeg;                /* Output segment */
  int iPtr = 0;

  pCsr = pMW->pCsr;
  pSeg = &pMW->pLevel->lhs;

  /* Pull the next record out of the source cursor. */
  lsmMCursorKey(pCsr, &pKey, &nKey);
  rc = lsmMCursorValue(pCsr, &pVal, &nVal);
  eType = pCsr->eType;
  if( rc!=LSM_OK ) return rc;

  /* Figure out if the output record may have a different pointer value
  ** than the previous. This is the case if the current key is identical to
  ** a key that appears in the lowest level run being merged. If so, set 
  ** iPtr to the absolute pointer value. If not, leave iPtr set to zero, 
  ** indicating that the output pointer value should be a copy of the pointer 
  ** value written with the previous key.  */









  if( pCsr->nSegCsr ){
    LevelCursor *pPtrs = &pCsr->aSegCsr[pCsr->nSegCsr-1];
    if( segmentCursorValid(pPtrs)
     && 0==pDb->xCmp(pPtrs->aPtr[0].pKey, pPtrs->aPtr[0].nKey, pKey, nKey)
    ){
      iPtr = pPtrs->aPtr[0].iPtr+pPtrs->aPtr[0].iPgPtr;
    }
  }

  /* If this is a separator key and we know that the output pointer has not
  ** changed, there is no point in writing an output record. Otherwise,
  ** proceed. */
  if( rtIsSeparator(eType)==0 || iPtr!=0 ){
    int iSPtr = 0;                /* Separators require a pointer here */

    if( pMW->apPage[0]==0 ){
      rc = mergeWorkerFirstPage(pMW);
    }

    /* Write the record into the main run. */
    if( rc==LSM_OK ){
      rc = mergeWorkerWrite(pMW, 0, eType, pKey, nKey, pVal, nVal, iPtr,&iSPtr);
    }

    /* If the call to mergeWorkerWrite() above started a new page, then
    ** add a SORTED_SEPARATOR key to the separators run.  */
    if( rc==LSM_OK && iSPtr ){

      /* If the separators array has not been started, start it now. */
      if( pMW->apPage[1]==0 ){
        assert( pSeg->run.iFirst!=0 );
        rc = mergeWorkerNextPage(pMW, 1, pSeg->run.iFirst);
        if( !pMW->bFlush ) pSeg->sep.iRoot = pSeg->sep.iFirst;
      }

      if( rc==LSM_OK ){
        int eSType;                 /* Type of record for separators array */

        /* Figure out how many (if any) keys to skip from this point. */
        assert( pMW->apPage[1] && (pSeg->sep.iFirst || pMW->bFlush) );
        pMW->pLevel->pMerge->nSkip = keyszToSkip(pDb->pFS, nKey);

        /* Write the key into the separators array. */
        eSType = rtTopic(eType) | SORTED_SEPARATOR;
        rc = mergeWorkerWrite(pMW, 1, eSType, pKey, nKey, 0, 0, iSPtr, 0);
      }
    }
  }

  /* Advance the cursor to the next input record (assuming one exists). */
  assert( lsmMCursorValid(pMW->pCsr) );
  if( rc==LSM_OK ) rc = lsmMCursorNext(pMW->pCsr);

  /* If the cursor is at EOF, the merge is finished. Release all page
  ** references currently held by the merge worker and inform the 
  ** FileSystem object that no further pages will be appended to either 
  ** the main or separators array. 
  */
  if( rc==LSM_OK && !lsmMCursorValid(pMW->pCsr) ){
    if( pSeg->run.iFirst ){
      rc = lsmFsSortedFinish(pDb->pFS, &pSeg->run);
    }


    if( rc==LSM_OK && pMW->bFlush ){
      rc = mergeWorkerBuildHierarchy(pMW);



    }

    if( rc==LSM_OK && pSeg->sep.iFirst ){
      rc = lsmFsSortedFinish(pDb->pFS, &pSeg->sep);
    }
    mergeWorkerShutdown(pMW);
  }
  return rc;
}

static int mergeWorkerDone(MergeWorker *pMW){
  return pMW->pCsr==0 || !lsmMCursorValid(pMW->pCsr);

  lsm_db *pDb,                    /* Connection handle */
  int *pnHdrLevel                 /* OUT: Number of levels not stored in LSM */
){
  int rc = LSM_OK;                /* Return Code */
  MultiCursor *pCsr = 0;
  Level *pNext = 0;               /* The current top level */
  Level *pNew;                    /* The new level itself */
  SortedRun *pDel = 0;
  int iLeftPtr = 0;

  /* Allocate the new level structure to write to. */
  pNext = lsmDbSnapshotLevel(pDb->pWorker);
  pNew = (Level *)lsmMallocZeroRc(pDb->pEnv, sizeof(Level), &rc);

  /* Create a cursor to gather the data required by the new segment. The new
  ** segment contains everything in the tree and pointers to the next segment
  ** in the database (if any).  */
  if( rc==LSM_OK ){

    pNew->pNext = pNext;
    lsmDbSnapshotSetLevel(pDb->pWorker, pNew);

    rc = multiCursorNew(pDb, pDb->pWorker, (pDb->pTV!=0), 0, &pCsr);
    if( rc==LSM_OK ){
      if( pNext ){
        assert( pNext->pMerge==0 || pNext->nRight>0 );
        if( pNext->pMerge==0 ){
          if( segmentHasSeparators(&pNext->lhs) ){
            rc = multiCursorAddLevel(pCsr, pNext, MULTICURSOR_ADDLEVEL_LHS_SEP);

            /* This call moves any blocks occupied by separators array pDel 
            ** to the pending list. We do this here, even though pDel will be 
            ** read while building the new level, so that the blocks will be 
            ** included in the "FREELIST" entry visited by the cursor (and 
            ** written into the new top level).  */
            if( rc==LSM_OK ){
              pDel = &pNext->lhs.sep;
              rc = lsmFsSortedDelete(pDb->pFS, pDb->pWorker, 0, pDel);
            }
          }
          iLeftPtr = pNext->lhs.run.iFirst;
        }
      }else{
        /* The new level will be the only level in the LSM. There is no reason
         ** to write out delete keys in this case.  */
        multiCursorIgnoreDelete(pCsr);
      }
    }

    pNew->pMerge = &merge;
    mergeworker.pDb = pDb;
    mergeworker.pLevel = pNew;
    mergeworker.pCsr = pCsr;

    /* Mark the separators array for the new level as a "phantom". */
    mergeworker.bFlush = 1;
    lsmFsPhantom(pDb->pFS, &pNew->lhs.sep);

    /* Allocate the first page of the output segment. */
    rc = mergeWorkerNextPage(&mergeworker, 0, iLeftPtr);

    /* Do the work to create the new merged segment on disk */
    if( rc==LSM_OK ) rc = lsmMCursorFirst(pCsr);
    while( rc==LSM_OK && mergeWorkerDone(&mergeworker)==0 ){
      rc = mergeWorkerStep(&mergeworker);
    }

    lsmFsPhantomFree(pDb->pFS);
    mergeWorkerShutdown(&mergeworker);
    pNew->pMerge = 0;
  }
  lsmFreelistDeltaEnd(pDb);

  /* Link the new level into the top of the tree. Delete the separators
  ** array (if any) that was merged into the new level. */
  if( rc==LSM_OK ){
    if( pDel ){
      /* lsmFsSortedDelete() has already been called on pDel. So all
      ** that is required here is to zero it (so that it is not used by
      ** future LSM searches). */
      memset(pDel, 0, sizeof(SortedRun));
    }
  }else{
    lsmDbSnapshotSetLevel(pDb->pWorker, pNext);
    sortedFreeLevel(pDb->pEnv, pNew);
  }

  if( rc==LSM_OK ){

    lsmSortedNewToplevel(pDb, pnHdrLevel);
  }

#if 0
  lsmSortedDumpStructure(pDb, pDb->pWorker, 0, 0, "tree flush");
#endif

  assertAllBtreesOk(rc, pDb);
  assertAllPointersOk(rc, pDb);
  assert( rc!=LSM_OK || lsmFsIntegrityCheck(pDb) );

  lsmFinishFlush(pDb, rc==LSM_OK);
  return rc;
}

/*

    pTopLevel = lsmDbSnapshotLevel(pDb->pWorker);
    pNew->pNext = p;
    for(pp=&pTopLevel; *pp!=pLevel; pp=&((*pp)->pNext));
    *pp = pNew;
    lsmDbSnapshotSetLevel(pDb->pWorker, pTopLevel);

    /* Determine whether or not the next separators will be linked in */
    if( pNext && pNext->pMerge==0 && segmentHasSeparators(&pNext->lhs) ){
      bUseNext = 1;
    }
  }

  /* Allocate the merge object */
  nByte = sizeof(Merge) + sizeof(MergeInput) * (nMerge + bUseNext);
  pMerge = (Merge *)lsmMallocZeroRc(pDb->pEnv, nByte, &rc);
  if( pMerge ){
    pMerge->aInput = (MergeInput *)&pMerge[1];
    pMerge->nInput = nMerge + bUseNext;
    pNew->pMerge = pMerge;
  }

  *ppNew = pNew;
  return rc;
}

static int mergeWorkerLoadOutputPage(MergeWorker *pMW, int bSep){
  int rc = LSM_OK;                /* Return code */
  SortedRun *pRun;                /* Run to load page from */
  Level *pLevel;

  pLevel = pMW->pLevel;
  pRun = (bSep ? &pLevel->lhs.sep : &pLevel->lhs.run);
  if( pRun->iLast ){
    Page *pPg;
    rc = lsmFsDbPageGet(pMW->pDb->pFS, pRun->iLast, &pPg);

    while( rc==LSM_OK ){
      Page *pNext;
      u8 *aData;
      int nData;
      aData = lsmFsPageData(pPg, &nData);
      if( (pageGetFlags(aData, nData) & SEGMENT_BTREE_FLAG)==0 ) break;
      rc = lsmFsDbPageNext(pRun, pPg, -1, &pNext);
      lsmFsPageRelease(pPg);
      pPg = pNext;
    }

    if( rc==LSM_OK ){
      pMW->apPage[bSep] = pPg;
      if( pLevel->pMerge->aiOutputOff[bSep]>=0 ) rc = lsmFsPageWrite(pPg);
    }
  }
  return rc;
}

static int mergeWorkerInit(
  lsm_db *pDb,                    /* Db connection to do merge work */

      Level *pNext = pLevel->pNext;
      rc = multiCursorAddLevel(pCsr, pNext, MULTICURSOR_ADDLEVEL_LHS_SEP);
    }
    multiCursorReadSeparators(pCsr);
  }else{
    multiCursorIgnoreDelete(pCsr);
  }
  assert( pMerge->nInput==pCsr->nSegCsr );
  pMW->pCsr = pCsr;

  /* Load each of the output pages into memory. */
  if( rc==LSM_OK ) rc = mergeWorkerLoadOutputPage(pMW, 0);
  if( rc==LSM_OK ) rc = mergeWorkerLoadOutputPage(pMW, 1);

  /* Position the cursor. */
  if( rc==LSM_OK ){
    if( pMW->apPage[0]==0 ){
      /* The output array is still empty. So position the cursor at the very 
      ** start of the input.  */
      rc = multiCursorEnd(pCsr, 0);
    }else{
      /* The output array is non-empty. Position the cursor based on the
      ** page/cell data saved in the Merge.aInput[] array.  */
      int i;
      for(i=0; rc==LSM_OK && i<pCsr->nSegCsr; i++){
        MergeInput *pInput = &pMerge->aInput[i];
        if( pInput->iPg ){
          SegmentPtr *pPtr;
          assert( pCsr->aSegCsr[i].nPtr==1 );
          assert( pCsr->aSegCsr[i].aPtr[0].pPg==0 );
          pPtr = &pCsr->aSegCsr[i].aPtr[0];
          rc = segmentPtrLoadPage(pDb->pFS, pPtr, pInput->iPg);
          if( rc==LSM_OK && pPtr->nCell>0 ){
            rc = segmentPtrLoadCell(pPtr, pInput->iCell);
          }
        }
      }






      if( rc==LSM_OK ){
        rc = multiCursorSetupTree(pCsr, 0);
      }
    }
    pCsr->flags |= CURSOR_NEXT_OK;
  }

  return rc;
}


int sortedWork(lsm_db *pDb, int nWork, int bOptimize, int *pnWrite){
  int rc = LSM_OK;                /* Return Code */
  int nRemaining = nWork;         /* Units of work to do before returning */
  Snapshot *pWorker = pDb->pWorker;

  assert( lsmFsIntegrityCheck(pDb) );
  assert( pWorker );
  assertAllPointersOk(rc, pDb);

  if( lsmDbSnapshotLevel(pWorker)==0 ) return LSM_OK;
  lsmDatabaseDirty(pDb);

  while( nRemaining>0 ){
    Level *pLevel;
    Level *pTopLevel = lsmDbSnapshotLevel(pWorker);

    /* Find the longest contiguous run of levels not currently undergoing a 
    ** merge with the same age in the structure. Or the level being merged
    ** with the largest number of right-hand segments. Work on it.  */
    Level *pBest = 0;
    int nBest = 4;

    Level *pThis = 0;
    int nThis = 0;

    for(pLevel = pTopLevel; pLevel; pLevel=pLevel->pNext){
      if( pLevel->nRight==0 && pThis && pLevel->iAge==pThis->iAge ){
        nThis++;

      ** into the lhs of the level.
      */
      if( rc==LSM_OK ){
        if( mergeWorkerDone(&mergeworker)==0 ){
          int iGobble = mergeworker.pCsr->aTree[1] - CURSOR_DATA_SEGMENT;
          if( iGobble<pLevel->nRight ){
            SegmentPtr *pGobble = &mergeworker.pCsr->aSegCsr[iGobble].aPtr[0];
            if( (pGobble->flags & PGFTR_SKIP_THIS_FLAG)==0 ){


              lsmFsGobble(pWorker, pGobble->pRun, pGobble->pPg);
            }
          }

        }else if( pLevel->lhs.run.iFirst==0 ){
          /* If the new level is completely empty, remove it from the 
          ** database snapshot. This can only happen if all input keys were
          ** annihilated. Since keys are only annihilated if the new level
          ** is the last in the linked list (contains the most ancient of
          ** database content), this guarantees that pLevel->pNext==0.  */ 

          Level *pTop;          /* Top level of worker snapshot */
          Level **pp;           /* Read/write iterator for Level.pNext list */
          assert( pLevel->pNext==0 );
          assert( segmentHasSeparators(&pLevel->lhs)==0 );

          /* Remove the level from the worker snapshot. */
          pTop = lsmDbSnapshotLevel(pWorker);
          for(pp=&pTop; *pp!=pLevel; pp=&((*pp)->pNext));
          *pp = pLevel->pNext;
          lsmDbSnapshotSetLevel(pWorker, pTop);

          /* Free the Level structure. */
          lsmFsSortedDelete(pDb->pFS, pWorker, 1, &pLevel->lhs.run);
          sortedFreeLevel(pDb->pEnv, pLevel);
        }else{
          int i;

          /* Free the separators of the next level, if required. */
          if( pLevel->pMerge->nInput > pLevel->nRight ){
            assert( pLevel->pNext );
            assert( segmentHasSeparators(&pLevel->pNext->lhs) );
            lsmFsSortedDelete(pDb->pFS, pWorker, 1, &pLevel->pNext->lhs.sep);
          }

          /* Free the right-hand-side of pLevel */
          for(i=0; i<pLevel->nRight; i++){
            lsmFsSortedDelete(pDb->pFS, pWorker, 1, &pLevel->aRhs[i].run);
            lsmFsSortedDelete(pDb->pFS, pWorker, 1, &pLevel->aRhs[i].sep);
          }
          lsmFree(pDb->pEnv, pLevel->aRhs);
          pLevel->nRight = 0;
          pLevel->aRhs = 0;

          /* Free the Merge object */
          lsmFree(pDb->pEnv, pLevel->pMerge);

    }
  }

  if( pnWrite ){
    *pnWrite = (nWork - nRemaining);
  }

  assertAllBtreesOk(rc, pDb);
  assertAllPointersOk(rc, pDb);
  assert( rc!=LSM_OK || lsmFsIntegrityCheck(pDb) );
  return rc;
}

typedef struct Metric Metric;
struct Metric {
  double fAvgHeight;

}

/*
** Return a string representation of the segment passed as the only argument.
** Space for the returned string is allocated using lsmMalloc(), and should
** be freed by the caller using lsmFree().
*/
static char *segToString(lsm_env *pEnv, SortedRun *pRun, int nMin){
  int nSize = pRun->nSize;
  Pgno iRoot = pRun->iRoot;
  Pgno iFirst = pRun->iFirst;
  Pgno iLast = pRun->iLast;
  char *z;

  char *z1;
  char *z2;
  int nPad;

  z1 = lsmMallocPrintf(pEnv, "%d.%d", iFirst, iLast);

}

static int fileToString(
  lsm_env *pEnv,                  /* For xMalloc() */
  char *aBuf, 
  int nBuf, 
  int nMin,
  SortedRun *pRun
){
  int i = 0;
  char *zSeg;

  zSeg = segToString(pEnv, pRun, nMin);
  i += sqlite4_snprintf(&aBuf[i], nBuf-i, "%s", zSeg);
  lsmFree(pEnv, zSeg);

  return i;
}

void sortedDumpPage(lsm_db *pDb, SortedRun *pRun, Page *pPg, int bVals){
  Blob blob = {0, 0, 0};         /* Blob used for keys */
  LsmString s;
  int i;

  int nRec;
  int iPtr;
  int flags;
  u8 *aData;
  int nData;

  aData = lsmFsPageData(pPg, &nData);

  nRec = pageGetNRec(aData, nData);
  iPtr = pageGetPtr(aData, nData);
  flags = pageGetFlags(aData, nData);

  lsmStringInit(&s, pDb->pEnv);
  lsmStringAppendf(&s,"nCell=%d iPtr=%d flags=%d {", nRec, iPtr, flags);

  u8 *aKey; int nKey = 0;         /* Key */
  u8 *aVal; int nVal = 0;         /* Value */
  int eType;
  int iPgPtr;
  Page *pRef = 0;                 /* Pointer to page iRef */
  u8 *aCell;

  aData = lsmFsPageData(pPg, &nData);

  aCell = pageGetCell(aData, nData, iCell);
  eType = *aCell++;
  aCell += lsmVarintGet32(aCell, &iPgPtr);

  if( eType==0 ){
    int dummy;

    LsmString str;
    int nRec;
    int iPtr;
    int flags;
    int iCell;
    u8 *aData; int nData;         /* Page data and size thereof */

    aData = lsmFsPageData(pPg, &nData);
    nRec = pageGetNRec(aData, nData);
    iPtr = pageGetPtr(aData, nData);
    flags = pageGetFlags(aData, nData);

    lsmStringInit(&str, pDb->pEnv);
    lsmStringAppendf(&str, "Page : %d\n", iPg);
    lsmStringAppendf(&str, "nRec : %d\n", nRec);

    lsmFsPageRelease(pPg);
  }

  lsmDbSnapshotRelease(pDb->pEnv, pRelease);
  return rc;
}

void sortedDumpSegment(lsm_db *pDb, SortedRun *pRun, int bVals){
  assert( pDb->xLog );
  if( pRun ){
    char *zSeg;
    Page *pPg;

    zSeg = segToString(pDb->pEnv, pRun, 0);
    lsmLogMessage(pDb, LSM_OK, "Segment: %s", zSeg);
    lsmFree(pDb->pEnv, zSeg);

    lsmLogMessage(pDb, LSM_OK, "Database structure (%s)", zWhy);

    for(pLevel=pTopLevel; pLevel; pLevel=pLevel->pNext){
      char zLeft[1024];
      char zRight[1024];
      int i = 0;

      SortedRun *aLeft[24];  
      SortedRun *aRight[24];

      int nLeft = 0;
      int nRight = 0;

      Segment *pSeg = &pLevel->lhs;
      if( segmentHasSeparators(pSeg) ){
        aLeft[nLeft++] = &pSeg->sep;
      }
      aLeft[nLeft++] = &pSeg->run;

      for(i=0; i<pLevel->nRight; i++){
        if( segmentHasSeparators(&pLevel->aRhs[i]) ){
          aRight[nRight++] = &pLevel->aRhs[i].sep;
        }
        aRight[nRight++] = &pLevel->aRhs[i].run;
      }

      for(i=0; i<nLeft || i<nRight; i++){
        int iPad = 0;
        char zLevel[32];
        zLeft[0] = '\0';
        zRight[0] = '\0';

            zLevel, iPad, "", zLeft, zRight
        );
      }

      iLevel++;
    }

#if 0
    lsmLogMessage(pDb, LSM_OK, "Block map", zWhy);
    for(pLevel=pDb->pLevel; pLevel; pLevel=pLevel->pNext){
      int iRhs;
      lsmFsDumpBlockmap(pDb, pLevel->lhs.pSep);
      lsmFsDumpBlockmap(pDb, pLevel->lhs.pRun);
      for(iRhs=0; iRhs<pLevel->nRight; iRhs++){
        lsmFsDumpBlockmap(pDb, pLevel->aRhs[iRhs].pSep);
        lsmFsDumpBlockmap(pDb, pLevel->aRhs[iRhs].pRun);
      }
    }
    lsmFsDumpBlocklists(pDb);
#endif

    if( bKeys ){
      for(pLevel=pTopLevel; pLevel; pLevel=pLevel->pNext){
        int i;
        sortedDumpSegment(pDb, &pLevel->lhs.sep, 0);
        sortedDumpSegment(pDb, &pLevel->lhs.run, bVals);
        for(i=0; i<pLevel->nRight; i++){
          if( pLevel->aRhs[i].sep.iFirst>0 ){
            sortedDumpSegment(pDb, &pLevel->aRhs[i].sep, 0);
          }
          sortedDumpSegment(pDb, &pLevel->aRhs[i].run, bVals);
        }
      }
    }
  }

  if( pSnap==0 ){
    lsmDbSnapshotRelease(pDb->pEnv, pDump);

  int rc = LSM_OK;
  Level *p;

  assert( pDb->pWorker );
  for(p=lsmDbSnapshotLevel(pDb->pWorker); p && rc==LSM_OK; p=p->pNext){
    Merge *pMerge = p->pMerge;
    if( pMerge ){
      pMerge->aiOutputOff[0] = -1;
      pMerge->aiOutputOff[1] = -1;
      pMerge->bHierReadonly = 1;
    }
  }

  return LSM_OK;
}

void lsmSortedSaveTreeCursors(lsm_db *pDb){
  MultiCursor *pCsr;
  for(pCsr=pDb->pCsr; pCsr; pCsr=pCsr->pNext){
    lsmTreeCursorSave(pCsr->pTreeCsr);
  }
}


#ifdef LSM_DEBUG_EXPENSIVE

/*
** Argument iPg is a page number within a separators run. Assert() that for
** each key K on on the page, (pKey1 >= K > pKey2) is true. 
**
** Also, if page iPg is a BTREE page, call this function recursively to
** check that the keys on each child page fall into the expected range.
*/
static void assertBtreeRanges(
  lsm_db *pDb, 
  SortedRun *pRun, 
  Pgno iPg,                       /* Database page to load */
  void *pKey1, int nKey1,         /* All keys must be >= than this */
  void *pKey2, int nKey2          /* And < than this */
){
  Blob blob = {0, 0, 0};
  u8 *aData;
  int nData;
  Page *pPg;
  int rc;
  int i;
  int nRec;
  int flags;

  int iPrevTopic = 0;             /* Previous topic value */
  u8 *aPrev = 0;                  /* Buffer pointing to previous key */
  int nPrev = 0;                  /* Size of aPrev[] in bytes */

  rc = lsmFsDbPageGet(pDb->pFS, iPg, &pPg);
  assert( rc==LSM_OK );
  aData = lsmFsPageData(pPg, &nData);

  nRec = pageGetNRec(aData, nData);
  flags = pageGetFlags(aData, nData);

  for(i=0; i<nRec; i++){
    u8 *aKey;
    int nKey;
    int iTopic;
    int iPtr;

    if( flags & SEGMENT_BTREE_FLAG ){
      aKey = pageGetCell(aData, nData, i);
      aKey += lsmVarintGet32(aKey, &iPtr);
      aKey += lsmVarintGet32(aKey, &nKey);
    }else{
      aKey = pageGetKey(pPg, i, &iTopic, &nKey, &blob);
    }

    assert( pKey1==0 || pDb->xCmp(aKey, nKey, pKey1, nKey1)>=0 );
    assert( pKey2==0 || pDb->xCmp(aKey, nKey, pKey2, nKey2)<0 );

    if( flags&SEGMENT_BTREE_FLAG ){
      assertBtreeRanges(pDb, pRun, iPtr, aPrev, nPrev, aKey, nKey);
    }
    aPrev = aKey;
    nPrev = nKey;
  }

  if( flags&SEGMENT_BTREE_FLAG ){
    int iRight = pageGetPtr(aData, nData);
    assertBtreeRanges(pDb, pRun, iRight, aPrev, nPrev, 0, 0);
  }

  lsmFsPageRelease(pPg);
  sortedBlobFree(&blob);
}

/*
** Check that the array pOne contains the required pointers to pTwo.
** Array pTwo must be a main array. pOne may be either a separators array
** or another main array. 
**
** If an error is encountered, *pzErr is set to point to a buffer containing
** a nul-terminated error message and this function returns immediately. The
** caller should eventually call lsmFree(*pzErr) to free the allocated
** error message buffer.
*/
static void assertPointersOk(
  lsm_db *pDb,                    /* Database handle */
  SortedRun *pOne,                /* Run containing pointers */
  SortedRun *pTwo,                /* Run containing pointer targets */
  int bRhs,                       /* True if pTwo may have been Gobble()d */
  char **pzErr
){
  int rc = LSM_OK;                /* Error code */
  SegmentPtr ptr1;                /* Iterates through pOne */
  SegmentPtr ptr2;                /* Iterates through pTwo */
  Pgno iPrev;

  assert( pOne && pTwo );

  memset(&ptr1, 0, sizeof(ptr1));
  memset(&ptr2, 0, sizeof(ptr1));
  ptr1.pRun = pOne;
  ptr2.pRun = pTwo;
  segmentPtrEndPage(pDb->pFS, &ptr1, 0, &rc);
  segmentPtrEndPage(pDb->pFS, &ptr2, 0, &rc);

  /* Check that the footer pointer of the first page of pOne points to
  ** the first page of pTwo. */
  iPrev = pTwo->iFirst;
  if( ptr1.iPtr!=iPrev && !bRhs ){

    do{
      rc = segmentPtrNextPage(&ptr2, 1);
      assert( rc==LSM_OK );
    }while( rc==LSM_OK && ptr2.pPg && ptr2.nCell==0 );
    if( rc!=LSM_OK || ptr2.pPg==0 ) break;
    iThis = lsmFsPageNumber(ptr2.pPg);

    if( (ptr2.flags & PGFTR_SKIP_THIS_FLAG)==0 ){

      /* Load the first cell in the array pTwo page. */
      rc = segmentPtrLoadCell(&ptr2, 0);

      /* Iterate forwards through pOne, searching for a key that matches the
      ** key ptr2.pKey/nKey. This key should have a pointer to the page that
      ** ptr2 currently points to. */
      while( rc==LSM_OK ){
        int res = rtTopic(ptr1.eType) - rtTopic(ptr2.eType);
        if( res==0 ){
          res = pDb->xCmp(ptr1.pKey, ptr1.nKey, ptr2.pKey, ptr2.nKey);
        }

        if( res<0 ){
          assert( bRhs || ptr1.iPtr+ptr1.iPgPtr==iPrev );
        }else if( res>0 ){
          assert( 0 );
        }else{
          assert( ptr1.iPtr+ptr1.iPgPtr==iThis );
          iPrev = lsmFsPageNumber(ptr2.pPg);
          break;
        }

        rc = segmentPtrAdvance(0, &ptr1, 0);
        if( ptr1.pPg==0 ){
          assert( 0 );
        }
      }
    }
  }

  segmentPtrReset(&ptr1);
  segmentPtrReset(&ptr2);
}

static int countBtreeKeys(FileSystem *pFS, SortedRun *pRun, Pgno iPg){
#if 0
  int rc;
  Page *pPg;
  u8 *aData;
  int nData;
  int flags;
  int nRet;

  rc = lsmFsDbPageGet(pFs, iPg, &pPg);
  assert( rc==LSM_OK );
  aData = lsmFsPageData(pPg, &nData);
  flags = pageGetFlags(aData, nData);

  if( flags & SEGMENT_BTREE_FLAG ){
    Pgno iRight;
    int nRec;
    int i;

    iRight = pageGetPtr(aData, nData);
    nRec = pageGetNRec(aData, nData);

    nRet = nRec;
    nRet += countBtreeKeys(pFS, pRun, iRight);
    for(i=0; i<nRec; i++){
      Pgno iPtr;
      u8 *aCell = pageGetCell(aData, nData, i);
      aCell += lsmVarintGet32(aCell, &iPtr);
      if( iPtr==0 ){
        lsmVarintGet32(aCell, &iPtr);
      }
      nRet += countBtreeKeys(pFS, pRun, iPtr);
    }
  }else{
    nRet = 0;
  }

  lsmFsPageRelease(pPg);
  return nRet;
#endif
  return 0;
}

static void assertBtreeSize(FileSystem *pFS, SortedRun *pRun, Pgno iRoot){
#if 0
  int nRun = 0;
  int nKey = 0;
  int rc;

  Page *pPg;

  rc = lsmFsDbPageGet(pFS, pRun->iFirst, &pPg);
  assert( rc==LSM_OK );
  while( pPg ){
    Page *pNext = 0;
    u8 *aData;
    int nData;
    int flags;
    int nRec;

    aData = lsmFsPageData(pPg, &nData);
    flags = pageGetFlags(aData, nData);
    nRec = pageGetNRec(aData, nData);

    if( (flags & SEGMENT_BTREE_FLAG)==0 && nRec ){
      nRun++;
    }

    rc = lsmFsDbPageNext(pPg, 1, &pNext);
    assert( rc==LSM_OK );
    lsmFsPageRelease(pPg);
    pPg = pNext;
  }

  nKey = countBtreeKeys(pFS, pRun, iRoot);
  assert( nRun==1+nKey );
#endif
}

static void assertAllBtreesOk(int rc, lsm_db *pDb){
#if 0
  if( rc==LSM_OK ){
    Level *p;
    for(p=pDb->pLevel; p; p=p->pNext){
      SortedRun *pSep = p->lhs.pSep;
      Pgno iRoot = pSep->iRoot;
      if( pSep && iRoot ){
        assertBtreeRanges(pDb, pSep, iRoot, 0, 0, 0, 0);
        assertBtreeSize(pDb->pFS, pSep, iRoot);
      }
    }
  }
#endif
}

/*
** This function is only useful for debugging. 
*/

static void assertAllPointersOk(int rc, lsm_db *pDb){








  assert( rc!=LSM_OK || pDb->pWorker );

  if( rc==LSM_OK ){
    Level *p;
    for(p=lsmDbSnapshotLevel(pDb->pWorker); p; p=p->pNext){
      int i;


      if( segmentHasSeparators(&p->lhs) ){
        assertPointersOk(pDb, &p->lhs.sep, &p->lhs.run, 0, 0);

      }
      for(i=0; i<p->nRight; i++){





        if( segmentHasSeparators(&p->aRhs[i]) ){















          assertPointersOk(pDb, &p->aRhs[i].sep, &p->aRhs[i].run, 1, 0);

        }
      }



    }


  }
}


#endif /* ifdef LSM_DEBUG_EXPENSIVE */