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;
................................................................................
  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 abHierReadonly[2];          /* True if b-tree heirarchies are 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 *);
FileSystem *lsmPageFS(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 *);
................................................................................
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 *);

/*

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 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;
................................................................................
  void *pLogCtx;

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

struct Segment {
  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 */
};






/*
** 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.
**
** iOutputOff:
**   The byte offset to write to next within the last page of the 
**   output segment.


*/
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 iOutputOff;                 /* Write offset on output page */
  int bHierReadonly;              /* True if b-tree heirarchies are 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. */


void lsmFsGobble(Snapshot *, Segment *, Page *);
int lsmFsSortedDelete(FileSystem *, Snapshot *, int, Segment *);
int lsmFsSortedFinish(FileSystem *, Segment *);
int lsmFsSortedAppend(FileSystem *, Snapshot *, Segment *, Page **);
int lsmFsPhantomMaterialize(FileSystem *, Snapshot *, Segment *);

/* Functions to retrieve the lsm_env pointer from a FileSystem or Page object */
lsm_env *lsmFsEnv(FileSystem *);
lsm_env *lsmPageEnv(Page *);
FileSystem *lsmPageFS(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(Segment *, Page *, int eDir, Page **);

int lsmFsPageWrite(Page *);
u8 *lsmFsPageData(Page *, int *);
int lsmFsPageRelease(Page *);
int lsmFsPagePersist(Page *);
void lsmFsPageRef(Page *);
Pgno lsmFsPageNumber(Page *);
................................................................................
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 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.iRoot, 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);
    ckptSetValue(p, iOut++, pSeg->sep.nSize, pRc);
  }else{
    ckptSetValue(p, iOut++, 0, pRc);
    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.iRoot = aIn[iIn++];
  pSegment->run.nSize = aIn[iIn++];
  pSegment->sep.iFirst = aIn[iIn++];
  pSegment->sep.iLast = aIn[iIn++];
  pSegment->sep.iRoot = aIn[iIn++];
  pSegment->sep.nSize = aIn[iIn++];
  if( pSegment->sep.iFirst && pSegment->sep.nSize==0 ) 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->abHierReadonly[0] = 1;
  pMerge->abHierReadonly[1] = 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]);
................................................................................
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 = 8;     /* 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 

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


*/

/*
** 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->iFirst, pRc);
  ckptSetValue(p, iOut++, pSeg->iLast, pRc);
  ckptSetValue(p, iOut++, pSeg->iRoot, pRc);

  ckptSetValue(p, iOut++, pSeg->nSize, 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->iFirst==0 && pSegment->iLast==0 );
  assert( pSegment->nSize==0 && pSegment->iRoot==0 );



  pSegment->iFirst = aIn[iIn++];
  pSegment->iLast = aIn[iIn++];
  pSegment->iRoot = aIn[iIn++];
  pSegment->nSize = aIn[iIn++];






  *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->iOutputOff = -1;
  pMerge->bHierReadonly = 1;
  pMerge->nSkip = (int)aInt[iIn++];


  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;

    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]);
................................................................................
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 = 4;     /* 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 */
................................................................................

/*
** 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;
................................................................................
    *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.
    */
................................................................................
  *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
................................................................................
**
** 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);

*/
#include "lsmInt.h"

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






































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


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

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







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

/*
** Remove page pPg from the hash table.
*/
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;
................................................................................
    *piNext = fsPageToBlock(pFS, lsmGetU32(&pLast->aData[pFS->nPagesize-4]));
    lsmFsPageRelease(pLast);
  }
  return rc;
}

static int fsRunEndsBetween(
  Segment *pRun, 
  Segment *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, 
  Segment *pIgnore, 
  int iFirst, 
  int iLast
){
  int i;

  if( fsRunEndsBetween(&pLevel->lhs, pIgnore, iFirst, iLast) ){


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


      return 1;
    }
  }

  return 0;
}

static int fsFreeBlock(
  FileSystem *pFS, 
  Snapshot *pSnapshot, 
  Segment *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 Segment structure */
  Segment *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(Segment));
  }
  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,
  Segment *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(Segment *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.iLast, &rc);

    }else{
      int i;
      for(i=0; i<pLvl->nRight; i++){
        addAppendPoint(db, pLvl->aRhs[i].iLast, &rc);

      }
    }
  }

  for(pLvl=lsmDbSnapshotLevel(db->pWorker); pLvl; pLvl=pLvl->pNext){
    int i;
    subAppendPoint(db, pLvl->lhs.iFirst);

    for(i=0; i<pLvl->nRight; i++){
      subAppendPoint(db, pLvl->aRhs[i].iFirst);

    }
  }









































































  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,
  Segment *p, 
  Page **ppOut
){
  int rc = LSM_OK;
  Page *pPg = 0;
  *ppOut = 0;
































  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, Segment *p){
  int rc = LSM_OK;
  if( p ){
    const int nPagePerBlock = (pFS->nBlocksize / pFS->nPagesize);



    /* 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.
    */
................................................................................
  *piParam = pFS->bUseMmap;
  return LSM_OK;
}

/*
** Helper function for lsmInfoArrayStructure().
*/
static Segment *startsWith(Segment *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
................................................................................
**
** 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 */
  Segment *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, iFirst)) ){


      int i;
      for(i=0; i<pLvl->nRight; i++){
        if( (pArray = startsWith(&pLvl->aRhs[i], 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;
}





























#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++){
      Segment *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);

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

  va_end(ap);
  return rc;
}

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

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

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

................................................................................
typedef struct BtreePg BtreePg;
typedef struct BtreeCursor BtreeCursor;
struct BtreePg {
  Page *pPage;
  int iCell;
};
struct BtreeCursor {
  SortedRun *pRun;
  FileSystem *pFS;                /* File system to read pages from */
  int nDepth;                     /* Allocated size of aPg[] */
  int iPg;                        /* Current entry in aPg[]. -1 -> EOF. */
  BtreePg *aPg;                   /* Pages from root to current location */

  /* Cache of current entry. pKey==0 for EOF. */
  void *pKey;
................................................................................
};

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 */
  Hierarchy aHier[2];
  Page *apPage[2];                /* Current output pages (0 is main run) */
  int nWork;                      /* Number of calls to mergeWorkerNextPage() */
};

#ifdef LSM_DEBUG_EXPENSIVE
static int assertPointersOk(lsm_db *, SortedRun *, SortedRun *, int);
static int assertBtreeOk(lsm_db *, SortedRun *);
#endif

struct FilePage { u8 *aData; int nData; };
static u8 *fsPageData(Page *pPg, int *pnData){
  *pnData = ((struct FilePage *)(pPg))->nData;
  return ((struct FilePage *)(pPg))->aData;
}
................................................................................
}

static int btreeCursorFirst(BtreeCursor *pCsr){
  int rc;

  Page *pPg = 0;
  FileSystem *pFS = pCsr->pFS;
  int iPg = pCsr->pRun->iRoot;

  do {
    rc = lsmFsDbPageGet(pFS, iPg, &pPg);
    assert( (rc==LSM_OK)==(pPg!=0) );
    if( rc==LSM_OK ){
      u8 *aData;
      int nData;
................................................................................
      Blob blob = {0,0,0};
      void *pSeek;
      int nSeek;
      int iTopicSeek;
      int dummy;

      int iPg = 0;
      int iLoad = pCsr->pRun->iRoot;

      rc = pageGetBtreeKey(pCsr->aPg[nDepth-1].pPage, 
          0, &dummy, &iTopicSeek, &pSeek, &nSeek, &pCsr->blob
      );

      do {
        Page *pPg;
................................................................................
    }
  }
  return rc;
}

static int btreeCursorNew(
  lsm_db *pDb,
  SortedRun *pRun,
  BtreeCursor **ppCsr
){
  int rc = LSM_OK;
  BtreeCursor *pCsr;
  
  assert( pRun->iRoot );
  pCsr = lsmMallocZeroRc(pDb->pEnv, sizeof(BtreeCursor), &rc);
  if( pCsr ){
    pCsr->pFS = pDb->pFS;
    pCsr->pRun = pRun;
    pCsr->iPg = -1;
  }

  *ppCsr = pCsr;
  return rc;
}

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

................................................................................
  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 segmentPtrLoadCell(
  SegmentPtr *pPtr,              /* Load page into this SegmentPtr object */
................................................................................
  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->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
................................................................................
  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 = fsPageData(pTest, &nData);
................................................................................
    ** 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. */
................................................................................
  ** loading (iPg+1) and testing that pKey/nKey is smaller than all
  ** the keys it houses.  */
#if 1
  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;

................................................................................
  }

  pCsr->iCurrentPtr = iBest;
}

static int seekInBtree(
  LevelCursor *pCsr,
  SortedRun *pRun,
  void *pKey, int nKey,           /* Key to seek to */
  Page **ppPg                     /* OUT: Leaf (sorted-run) page reference */
){
  int rc;
  int iPg;
  Page *pPg = 0;
  Blob blob = {0, 0, 0};
  int iTopic = 0;                 /* TODO: Fix me */

  iPg = pRun->iRoot;
  do {
    rc = lsmFsDbPageGet(pCsr->pFS, iPg, &pPg);
    assert( rc==LSM_OK || pPg==0 );
    if( rc==LSM_OK ){
      u8 *aData;                  /* Buffer containing page data */
      int nData;                  /* Size of aData[] in bytes */
      int iMin;
................................................................................

  sortedBlobFree(&blob);
  assert( (rc==LSM_OK)==(pPg!=0) );
  *ppPg = pPg;
  return rc;
}

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;
  Page *pPg;
  SortedRun *pSep = &pPtr->pSeg->sep;

  rc = seekInBtree(pCsr, pSep, pKey, nKey, &pPg);

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

  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( pPtr->pSeg->run.iRoot ){
    Page *pPg;
    assert( pPtr->pSeg->run.iRoot!=0 );
    rc = seekInBtree(pCsr, &pPtr->pSeg->run, pKey, nKey, &pPg);
    if( rc==LSM_OK ) segmentPtrSetPage(pPtr, pPg);
  }else{
    if( iPtr==0 ){
      iPtr = pPtr->pSeg->run.iFirst;
    }
    if( rc==LSM_OK ){
      rc = segmentPtrLoadPage(pCsr->pFS, pPtr, iPtr);
    }
  }

  if( rc==LSM_OK ){
................................................................................
        *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);
  }
................................................................................

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

  if( eMode==MULTICURSOR_ADDLEVEL_LHS_SEP ){
    assert( pLevel->lhs.run.iRoot );
    assert( pCsr->pBtCsr==0 );
    rc = btreeCursorNew(pCsr->pDb, &pLevel->lhs.run, &pCsr->pBtCsr);
    assert( (rc==LSM_OK)==(pCsr->pBtCsr!=0) );
  }else{
    int i;
    int nAdd = (eMode==MULTICURSOR_ADDLEVEL_RHS ? pLevel->nRight : 1);

    for(i=0; i<nAdd; i++){
      LevelCursor *pNew;
................................................................................

      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++;
    }
................................................................................
** 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,               /* Merge worker */
  int bSep                        /* True for separators run */
){
  SortedRun *pRun;                /* Run being modified */
  lsm_db *pDb = pMW->pDb;         /* Database handle */
  int rc = LSM_OK;                /* Return code */
  int i;
  int iRight = 0;
  Page **apHier = pMW->aHier[bSep].apHier;
  int nHier = pMW->aHier[bSep].nHier;

  assert( nHier>0 && pMW->pLevel->pMerge->abHierReadonly[bSep] );
  pRun = (bSep ? &pMW->pLevel->lhs.sep : &pMW->pLevel->lhs.run);

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

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

      a1 = fsPageData(pNew, &n1);
................................................................................
#ifdef LSM_DEBUG
  if( rc==LSM_OK ){
    for(i=0; i<nHier; i++) assert( lsmFsPageWritable(apHier[i]) );
  }
#endif

  if( rc==LSM_OK ){
    pMW->pLevel->pMerge->abHierReadonly[bSep] = 0;
  }
  return rc;
}

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

  Hierarchy *p;
 
  pRun = (bSep ? &pMW->pLevel->lhs.sep : &pMW->pLevel->lhs.run);
  p = &pMW->aHier[bSep];

  if( p->apHier==0 && pRun->iRoot!=0 ){
    int bHierReadonly = pMW->pLevel->pMerge->abHierReadonly[bSep];
    FileSystem *pFS = pMW->pDb->pFS;
    lsm_env *pEnv = pMW->pDb->pEnv;
    Page **apHier = 0;
    int nHier = 0;
    int iPg = pRun->iRoot;

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

................................................................................
** 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 */
  int bSep,                       /* True for separators, false otherwise */
  Pgno iKeyPg,                    /* Page that will contain pKey/nKey */
  int iTopic,                     /* Topic value for this key */
  void *pKey,                     /* Pointer to key buffer */
................................................................................
  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 */

  Hierarchy *p;
  SortedRun *pRun;

  /* If there exists a b-tree hierarchy and it is not loaded into 
  ** memory, load it now.  */
  pRun = (bSep ? &pMW->pLevel->lhs.sep : &pMW->pLevel->lhs.run);
  p = &pMW->aHier[bSep];
  rc = mergeWorkerLoadHierarchy(pMW, bSep);

  /* Obtain the absolute pointer value to store along with the key in the
  ** page body. This pointer points to a page that contains keys that are
  ** smaller than pKey/nKey.  */
  if( p->nHier ){
    aData = fsPageData(p->apHier[0], &nData);
    iPtr = lsmGetU32(&aData[SEGMENT_POINTER_OFFSET(nData)]);
  }else{
    iPtr = pRun->iFirst;
  }

  if( p->nHier && pMW->pLevel->pMerge->abHierReadonly[bSep] ){
    rc = mergeWorkerMoveHierarchy(pMW, bSep);
    if( rc!=LSM_OK ) goto push_hierarchy_out;
  }

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

................................................................................
      rc = lsmFsPageRelease(p->apHier[iLevel]);
    }

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

    aData = fsPageData(p->apHier[iLevel], &nData);
    memset(aData, 0, nData);
    lsmPutU16(&aData[SEGMENT_FLAGS_OFFSET(nData)], SEGMENT_BTREE_FLAG);
................................................................................

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

  /* Ensure that the SortedRun.iRoot field is correct. */
  pRun->iRoot = lsmFsPageNumber(p->apHier[p->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 = fsPageData(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, 1, iPg, iTopic, pKey, nKey);
        }
      }
    }

    if( pMW->aHier[1].nHier>0 ){
      Page *pRoot = pMW->aHier[1].apHier[pMW->aHier[1].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 = fsPageData(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 = fsPageData(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 */
  MultiCursor *pCsr,              /* Read value (if any) from here */
  int iPtr,                       /* Absolute value of page pointer, or 0 */
  int *piPtrOut                   /* OUT: Pointer to write to separators */
){
  int rc = LSM_OK;                /* Return code */
................................................................................
  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 */
  void *pVal;
  int nVal;

  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 = fsPageData(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 ){
................................................................................
  rc = lsmMCursorValue(pCsr, &pVal, &nVal);
  if( rc==LSM_OK ){
    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];
    }
  }

  /* 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 && pMW->bFlush==0) || (bSep==0 && pMerge->nSkip==0) ){
      Pgno iPg = lsmFsPageNumber(pPg);
      rc = mergeWorkerPushHierarchy(pMW, bSep, iPg, rtTopic(eType), pKey, nKey);
    }
    if( bSep==0 ){
      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 ){
    aData = fsPageData(pPg, &nData);

    /* Update the page footer. */
................................................................................
    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) ){
      if( rtTopic(eType)==0 ) rc = lsmMCursorValue(pCsr, &pVal, &nVal);
      if( rc==LSM_OK ){
        rc = mergeWorkerData(pMW, bSep, iFPtr+iRPtr, pVal, nVal);
      }
    }
  }

  return rc;
}

................................................................................
    if( bBtree && pMerge->nInput ){
      assert( i==pCsr->nSegCsr );
      btreeCursorPosition(pCsr->pBtCsr, &pMerge->aInput[i]);
    }
  }

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

  for(i=0; i<2; i++){
    Hierarchy *p = &pMW->aHier[i];
    int iPg;
    for(iPg=0; iPg<p->nHier; iPg++){
      lsmFsPageRelease(p->apHier[iPg]);
    }
    lsmFree(pMW->pDb->pEnv, p->apHier);
    p->apHier = 0;
    p->nHier = 0;
  }

  pMW->pCsr = 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 */
  int iFPtr;                      /* Pointer value read from footer of pPg */
  MultiCursor *pCsr = pMW->pCsr;

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

  if( pCsr->pBtCsr ){
    rc = LSM_OK;
    iFPtr = pMW->pLevel->pNext->lhs.run.iFirst;
  }else{

    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 */
      aData = fsPageData(pPg, &nData);
      iFPtr = pageGetPtr(aData, nData);
      lsmFsPageRelease(pPg);
    }
  }

  if( rc==LSM_OK ){
    rc = mergeWorkerNextPage(pMW, 0, iFPtr);
  }

  return rc;
}

static int mergeWorkerStep(MergeWorker *pMW){
  lsm_db *pDb = pMW->pDb;       /* Database handle */
................................................................................

  /* 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, pCsr, iPtr, &iSPtr);
    }

    /* If the call to mergeWorkerWrite() above started a new page, then
    ** add a SORTED_SEPARATOR key to the separators run.  */
#if 0
    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);
      }
    }
#endif
  }

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

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

#ifdef LSM_DEBUG_EXPENSIVE
    if( rc==LSM_OK ){
      rc = assertBtreeOk(pDb, &pSeg->run);
      if( pMW->pCsr->pBtCsr ){
        SortedRun *pNext = &pMW->pLevel->pNext->lhs.run;
        rc = assertPointersOk(pDb, &pSeg->run, pNext, 0);
      }
    }
#endif

    mergeWorkerShutdown(pMW);
  }
  return rc;
................................................................................
    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( pNext->lhs.run.iRoot ){
            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;
              rc = lsmFsSortedDelete(pDb->pFS, pDb->pWorker, 0, &pDel->sep);
            }
          }
          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->sep. So all
      ** that is required here is to zero it (so that it is not used by
      ** future LSM searches). */
      memset(&pDel->sep, 0, sizeof(SortedRun));
      pDel->run.iRoot = 0;
    }
  }else{
    lsmDbSnapshotSetLevel(pDb->pWorker, pNext);
    sortedFreeLevel(pDb->pEnv, pNew);
  }

  if( rc==LSM_OK ){
................................................................................
    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 && pNext->lhs.run.iRoot ){
      bUseNext = 1;
    }
  }

  /* Allocate the merge object */
  nByte = sizeof(Merge) + sizeof(MergeInput) * (nMerge + bUseNext);
  pMerge = (Merge *)lsmMallocZeroRc(pDb->pEnv, nByte, &rc);
................................................................................
    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 = fsPageData(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 */
................................................................................
    multiCursorReadSeparators(pCsr);
  }else{
    multiCursorIgnoreDelete(pCsr);
  }
  assert( rc!=LSM_OK || pMerge->nInput==(pCsr->nSegCsr+(pCsr->pBtCsr!=0)) );
  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;
................................................................................
      ** from rhs segments for which the content has been completely merged
      ** 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 ){
            Level *pLevel = mergeworker.pCsr->aSegCsr[iGobble].pLevel;
            SegmentPtr *pGobble = &mergeworker.pCsr->aSegCsr[iGobble].aPtr[0];

            if( (pGobble->flags & PGFTR_SKIP_THIS_FLAG)==0 
             && pGobble->pRun->iRoot==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( pLevel->pNext->lhs.run.iRoot );
            lsmFsSortedDelete(pDb->pFS, pWorker, 1, &pLevel->pNext->lhs.sep);
            pLevel->pNext->lhs.run.iRoot = 0;
          }

          /* 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);
................................................................................
}

/*
** 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;
................................................................................
    lsmFsPageRelease(pPg);
  }

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

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

    zSeg = segToString(pDb->pEnv, pRun, 0);
    lsmLogMessage(pDb, LSM_OK, "Segment: %s", 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->abHierReadonly[0] = 1;
      pMerge->abHierReadonly[1] = 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 */
/*
** This function is only included in the build if LSM_DEBUG_EXPENSIVE is 
** defined. Its only purpose is to evaluate various assert() statements to 
** verify that the database is well formed in certain respects.
**
** More specifically, it checks 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 pOne does not contain the 
** correct set of pointers, an assert() statement fails.
*/
static int 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 */
){
  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 ){
................................................................................
** verify that the database is well formed in certain respects.
**
** More specifically, it checks that the b-tree embedded in array pRun
** contains the correct keys. If not, an assert() fails.
*/
static int assertBtreeOk(
  lsm_db *pDb,
  SortedRun *pRun
){
  int rc = LSM_OK;                /* Return code */
  if( pRun->iRoot ){
    Blob blob = {0, 0, 0};        /* Buffer used to cache overflow keys */
    FileSystem *pFS = pDb->pFS;   /* File system to read from */
    Page *pPg = 0;                /* Main run page */
    BtreeCursor *pCsr = 0;        /* Btree cursor */

    rc = btreeCursorNew(pDb, pRun, &pCsr);
    if( rc==LSM_OK ){
      rc = btreeCursorFirst(pCsr);
    }
    if( rc==LSM_OK ){
      rc = lsmFsDbPageGet(pFS, pRun->iFirst, &pPg);
    }

    while( rc==LSM_OK ){
      Page *pNext;
      u8 *aData;
      int nData;
      int flags;

      rc = lsmFsDbPageNext(pRun, pPg, 1, &pNext);
      lsmFsPageRelease(pPg);
      pPg = pNext;
      if( pPg==0 ) break;
      aData = fsPageData(pPg, &nData);
      flags = pageGetFlags(aData, nData);
      if( rc==LSM_OK 
       && 0==((SEGMENT_BTREE_FLAG|PGFTR_SKIP_THIS_FLAG) & flags)

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


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

................................................................................
typedef struct BtreePg BtreePg;
typedef struct BtreeCursor BtreeCursor;
struct BtreePg {
  Page *pPage;
  int iCell;
};
struct BtreeCursor {
  Segment *pSeg;                  /* Iterate through this segments btree */
  FileSystem *pFS;                /* File system to read pages from */
  int nDepth;                     /* Allocated size of aPg[] */
  int iPg;                        /* Current entry in aPg[]. -1 -> EOF. */
  BtreePg *aPg;                   /* Pages from root to current location */

  /* Cache of current entry. pKey==0 for EOF. */
  void *pKey;
................................................................................
};

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 */
  Hierarchy hier;                 /* B-tree hierarchy under construction */
  Page *pPage;                    /* Current output page */
  int nWork;                      /* Number of calls to mergeWorkerNextPage() */
};

#ifdef LSM_DEBUG_EXPENSIVE
static int assertPointersOk(lsm_db *, Segment *, Segment *, int);
static int assertBtreeOk(lsm_db *, Segment *);
#endif

struct FilePage { u8 *aData; int nData; };
static u8 *fsPageData(Page *pPg, int *pnData){
  *pnData = ((struct FilePage *)(pPg))->nData;
  return ((struct FilePage *)(pPg))->aData;
}
................................................................................
}

static int btreeCursorFirst(BtreeCursor *pCsr){
  int rc;

  Page *pPg = 0;
  FileSystem *pFS = pCsr->pFS;
  int iPg = pCsr->pSeg->iRoot;

  do {
    rc = lsmFsDbPageGet(pFS, iPg, &pPg);
    assert( (rc==LSM_OK)==(pPg!=0) );
    if( rc==LSM_OK ){
      u8 *aData;
      int nData;
................................................................................
      Blob blob = {0,0,0};
      void *pSeek;
      int nSeek;
      int iTopicSeek;
      int dummy;

      int iPg = 0;
      int iLoad = pCsr->pSeg->iRoot;

      rc = pageGetBtreeKey(pCsr->aPg[nDepth-1].pPage, 
          0, &dummy, &iTopicSeek, &pSeek, &nSeek, &pCsr->blob
      );

      do {
        Page *pPg;
................................................................................
    }
  }
  return rc;
}

static int btreeCursorNew(
  lsm_db *pDb,
  Segment *pSeg,
  BtreeCursor **ppCsr
){
  int rc = LSM_OK;
  BtreeCursor *pCsr;
  
  assert( pSeg->iRoot );
  pCsr = lsmMallocZeroRc(pDb->pEnv, sizeof(BtreeCursor), &rc);
  if( pCsr ){
    pCsr->pFS = pDb->pFS;
    pCsr->pSeg = pSeg;
    pCsr->iPg = -1;
  }

  *ppCsr = pCsr;
  return rc;
}

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





  rc = lsmFsDbPageGet(pFS, iNew, &pPg);
  assert( rc==LSM_OK || pPg==0 );
  segmentPtrSetPage(pPtr, pPg);

  return rc;
}

................................................................................
  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 || eDir>0 );




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

static int segmentPtrLoadCell(
  SegmentPtr *pPtr,              /* Load page into this SegmentPtr object */
................................................................................
  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};




    assert( pLevel->aRhs[i].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];
    }



  }

  return rc;
}

static int levelCursorInitRun(
  lsm_db *pDb,
  Segment *pSeg, 
  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->xCmp = xCmp;
  pCsr->nPtr = 1;
  pCsr->aPtr = (SegmentPtr*)lsmMallocZeroRc(pDb->pEnv, 
      sizeof(SegmentPtr)*pCsr->nPtr, &rc
  );

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

  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->pSeg->iLast : pPtr->pSeg->iFirst);
    *pRc = lsmFsDbPageGet(pFS, iPg, &pNew);
    segmentPtrSetPage(pPtr, pNew);
  }
}

/*
** Try to move the segment pointer passed as the second argument so that it
................................................................................
  for(eDir=-1; eDir<=1; eDir+=2){
    Page *pTest = pPtr->pPg;

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

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

      if( pTest ){
        int nData;
        u8 *aData = fsPageData(pTest, &nData);
................................................................................
    ** 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->pSeg, 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. */
................................................................................
  ** loading (iPg+1) and testing that pKey/nKey is smaller than all
  ** the keys it houses.  */
#if 1
  assert( assertKeyLocation(pCsr, pPtr, pKey, nKey) );
#endif

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

................................................................................
  }

  pCsr->iCurrentPtr = iBest;
}

static int seekInBtree(
  LevelCursor *pCsr,
  Segment *pSeg,
  void *pKey, int nKey,           /* Key to seek to */
  Page **ppPg                     /* OUT: Leaf (sorted-run) page reference */
){
  int rc;
  int iPg;
  Page *pPg = 0;
  Blob blob = {0, 0, 0};
  int iTopic = 0;                 /* TODO: Fix me */

  iPg = pSeg->iRoot;
  do {
    rc = lsmFsDbPageGet(pCsr->pFS, iPg, &pPg);
    assert( rc==LSM_OK || pPg==0 );
    if( rc==LSM_OK ){
      u8 *aData;                  /* Buffer containing page data */
      int nData;                  /* Size of aData[] in bytes */
      int iMin;
................................................................................

  sortedBlobFree(&blob);
  assert( (rc==LSM_OK)==(pPg!=0) );
  *ppPg = pPg;
  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;



  if( pPtr->pSeg->iRoot ){
    Page *pPg;
    assert( pPtr->pSeg->iRoot!=0 );
    rc = seekInBtree(pCsr, pPtr->pSeg, pKey, nKey, &pPg);
    if( rc==LSM_OK ) segmentPtrSetPage(pPtr, pPg);
  }else{
    if( iPtr==0 ){
      iPtr = pPtr->pSeg->iFirst;
    }
    if( rc==LSM_OK ){
      rc = segmentPtrLoadPage(pCsr->pFS, pPtr, iPtr);
    }
  }

  if( rc==LSM_OK ){
................................................................................
        *pp = pCsr->pNext;
        break;
      }
    }

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

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

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

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

  if( eMode==MULTICURSOR_ADDLEVEL_LHS_SEP ){
    assert( pLevel->lhs.iRoot );
    assert( pCsr->pBtCsr==0 );
    rc = btreeCursorNew(pCsr->pDb, &pLevel->lhs, &pCsr->pBtCsr);
    assert( (rc==LSM_OK)==(pCsr->pBtCsr!=0) );
  }else{
    int i;
    int nAdd = (eMode==MULTICURSOR_ADDLEVEL_RHS ? pLevel->nRight : 1);

    for(i=0; i<nAdd; i++){
      LevelCursor *pNew;
................................................................................

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

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




          break;
      }
      if( pCsr->flags & CURSOR_IGNORE_SYSTEM ){
        pNew->bIgnoreSystem = 1;
      }
      if( rc==LSM_OK ) pCsr->nSegCsr++;
    }
................................................................................
** 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,               /* Merge worker */
  int bSep                        /* True for separators run */
){
  Segment *pSeg;                  /* Segment being written */
  lsm_db *pDb = pMW->pDb;         /* Database handle */
  int rc = LSM_OK;                /* Return code */
  int i;
  int iRight = 0;
  Page **apHier = pMW->hier.apHier;
  int nHier = pMW->hier.nHier;

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

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

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

      a1 = fsPageData(pNew, &n1);
................................................................................
#ifdef LSM_DEBUG
  if( rc==LSM_OK ){
    for(i=0; i<nHier; i++) assert( lsmFsPageWritable(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;

  Segment *pSeg;
  Hierarchy *p;
 
  pSeg = &pMW->pLevel->lhs;
  p = &pMW->hier;

  if( p->apHier==0 && pSeg->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 = pSeg->iRoot;

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

................................................................................
** 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 seekInBtree() for the code that traverses b-tree pages.
*/
static int mergeWorkerPushHierarchy(
  MergeWorker *pMW,               /* Merge worker object */
  int bSep,                       /* True for separators, false otherwise */
  Pgno iKeyPg,                    /* Page that will contain pKey/nKey */
  int iTopic,                     /* Topic value for this key */
  void *pKey,                     /* Pointer to key buffer */
................................................................................
  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 */

  Hierarchy *p;
  Segment *pSeg;

  /* If there exists a b-tree hierarchy and it is not loaded into 
  ** memory, load it now.  */
  pSeg = &pMW->pLevel->lhs;
  p = &pMW->hier;
  rc = mergeWorkerLoadHierarchy(pMW);

  /* Obtain the absolute pointer value to store along with the key in the
  ** page body. This pointer points to a page that contains keys that are
  ** smaller than pKey/nKey.  */
  if( p->nHier ){
    aData = fsPageData(p->apHier[0], &nData);
    iPtr = lsmGetU32(&aData[SEGMENT_POINTER_OFFSET(nData)]);
  }else{
    iPtr = pSeg->iFirst;
  }

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

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

................................................................................
      rc = lsmFsPageRelease(p->apHier[iLevel]);
    }

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

    aData = fsPageData(p->apHier[iLevel], &nData);
    memset(aData, 0, nData);
    lsmPutU16(&aData[SEGMENT_FLAGS_OFFSET(nData)], SEGMENT_BTREE_FLAG);
................................................................................

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

  /* Ensure that the SortedRun.iRoot field is correct. */
  pSeg->iRoot = lsmFsPageNumber(p->apHier[p->nHier-1]);

push_hierarchy_out:



































































  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. 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 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 */
  Segment *pSeg;                  /* Run to append to */



  pSeg = &pMW->pLevel->lhs;
  rc = lsmFsSortedAppend(pDb->pFS, pDb->pWorker, pSeg, &pNext);
  assert( rc!=LSM_OK || pSeg->iFirst>0 );

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

    lsmFsPageRelease(pMW->pPage);
    pMW->pPage = pNext;
    pMW->pLevel->pMerge->iOutputOff = 0;

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

    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->pPage) );
   
    aData = fsPageData(pMW->pPage, &nData);
    nRec = pageGetNRec(aData, nData);
    iOff = pMerge->iOutputOff;
    nCopy = LSM_MIN(nRem, SEGMENT_EOF(nData, nRec) - iOff);

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

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

  return rc;
}


static int mergeWorkerWrite(
  MergeWorker *pMW,               /* Merge worker object to write into */

  int eType,                      /* One of SORTED_SEPARATOR, WRITE or DELETE */
  void *pKey, int nKey,           /* Key value */
  MultiCursor *pCsr,              /* Read value (if any) from here */
  int iPtr,                       /* Absolute value of page pointer, or 0 */
  int *piPtrOut                   /* OUT: Pointer to write to separators */
){
  int rc = LSM_OK;                /* Return code */
................................................................................
  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 */
  Segment *pSeg;                  /* Segment being written */
  int flags = 0;                  /* If != 0, flags value for page footer */
  void *pVal;
  int nVal;




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

  pPg = pMW->pPage;
  aData = fsPageData(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 ){
................................................................................
  rc = lsmMCursorValue(pCsr, &pVal, &nVal);
  if( rc==LSM_OK ){
    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->iOutputOff;
    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, iFPtr);
      pPg = pMW->pPage;
    }
  }

  /* 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 && pSeg->iFirst!=pSeg->iLast ){
    assert( pMerge->nSkip>=0 );

    if( pMerge->nSkip==0 ){
      Pgno iPg = lsmFsPageNumber(pPg);
      rc = mergeWorkerPushHierarchy(pMW, 0, iPg, rtTopic(eType), pKey, nKey);
    }

    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 ){
    aData = fsPageData(pPg, &nData);

    /* Update the page footer. */
................................................................................
    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->iOutputOff = iOff;

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

  return rc;
}

................................................................................
    if( bBtree && pMerge->nInput ){
      assert( i==pCsr->nSegCsr );
      btreeCursorPosition(pCsr->pBtCsr, &pMerge->aInput[i]);
    }
  }

  lsmMCursorClose(pCsr);
  lsmFsPageRelease(pMW->pPage);
  lsmFsPageRelease(pMW->pPage);

  for(i=0; i<2; i++){
    Hierarchy *p = &pMW->hier;
    int iPg;
    for(iPg=0; iPg<p->nHier; iPg++){
      lsmFsPageRelease(p->apHier[iPg]);
    }
    lsmFree(pMW->pDb->pEnv, p->apHier);
    p->apHier = 0;
    p->nHier = 0;
  }

  pMW->pCsr = 0;
  pMW->pPage = 0;
  pMW->pPage = 0;
}

static int mergeWorkerFirstPage(MergeWorker *pMW){
  int rc;                         /* Return code */

  Page *pPg = 0;                  /* First page of run pSeg */
  int iFPtr;                      /* Pointer value read from footer of pPg */
  MultiCursor *pCsr = pMW->pCsr;

  assert( pMW->pPage==0 );

  if( pCsr->pBtCsr ){
    rc = LSM_OK;
    iFPtr = pMW->pLevel->pNext->lhs.iFirst;
  }else{
    Segment *pSeg;
    pSeg = pMW->pCsr->aSegCsr[pMW->pCsr->nSegCsr-1].aPtr[0].pSeg;
    rc = lsmFsDbPageGet(pMW->pDb->pFS, pSeg->iFirst, &pPg);
    if( rc==LSM_OK ){
      u8 *aData;                    /* Buffer for page pPg */
      int nData;                    /* Size of aData[] in bytes */
      aData = fsPageData(pPg, &nData);
      iFPtr = pageGetPtr(aData, nData);
      lsmFsPageRelease(pPg);
    }
  }

  if( rc==LSM_OK ){
    rc = mergeWorkerNextPage(pMW, iFPtr);
  }

  return rc;
}

static int mergeWorkerStep(MergeWorker *pMW){
  lsm_db *pDb = pMW->pDb;       /* Database handle */
................................................................................

  /* 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->pPage==0 ){
      rc = mergeWorkerFirstPage(pMW);
    }

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


























  }

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

  /* 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->iFirst ){
      rc = lsmFsSortedFinish(pDb->pFS, pSeg);






    }

#ifdef LSM_DEBUG_EXPENSIVE
    if( rc==LSM_OK ){
      rc = assertBtreeOk(pDb, pSeg);
      if( pMW->pCsr->pBtCsr ){
        Segment *pNext = &pMW->pLevel->pNext->lhs;
        rc = assertPointersOk(pDb, pSeg, pNext, 0);
      }
    }
#endif

    mergeWorkerShutdown(pMW);
  }
  return rc;
................................................................................
    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( pNext->lhs.iRoot ){
            rc = multiCursorAddLevel(pCsr, pNext, MULTICURSOR_ADDLEVEL_LHS_SEP);






            if( rc==LSM_OK ){
              pDel = &pNext->lhs;

            }
          }
          iLeftPtr = pNext->lhs.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;


    /* Allocate the first page of the output segment. */
    rc = mergeWorkerNextPage(&mergeworker, 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);
    }


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

  /* Link the new level into the top of the tree. */

  if( rc==LSM_OK ){
    if( pDel ){




      pDel->iRoot = 0;
    }
  }else{
    lsmDbSnapshotSetLevel(pDb->pWorker, pNext);
    sortedFreeLevel(pDb->pEnv, pNew);
  }

  if( rc==LSM_OK ){
................................................................................
    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 && pNext->lhs.iRoot ){
      bUseNext = 1;
    }
  }

  /* Allocate the merge object */
  nByte = sizeof(Merge) + sizeof(MergeInput) * (nMerge + bUseNext);
  pMerge = (Merge *)lsmMallocZeroRc(pDb->pEnv, nByte, &rc);
................................................................................
    pNew->pMerge = pMerge;
  }

  *ppNew = pNew;
  return rc;
}

static int mergeWorkerLoadOutputPage(MergeWorker *pMW){
  int rc = LSM_OK;                /* Return code */
  Segment *pSeg;                  /* Run to load page from */
  Level *pLevel;

  pLevel = pMW->pLevel;
  pSeg = &pLevel->lhs;
  if( pSeg->iLast ){
    Page *pPg;
    rc = lsmFsDbPageGet(pMW->pDb->pFS, pSeg->iLast, &pPg);

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

    if( rc==LSM_OK ){
      pMW->pPage = pPg;
      if( pLevel->pMerge->iOutputOff>=0 ) rc = lsmFsPageWrite(pPg);
    }
  }
  return rc;
}

static int mergeWorkerInit(
  lsm_db *pDb,                    /* Db connection to do merge work */
................................................................................
    multiCursorReadSeparators(pCsr);
  }else{
    multiCursorIgnoreDelete(pCsr);
  }
  assert( rc!=LSM_OK || pMerge->nInput==(pCsr->nSegCsr+(pCsr->pBtCsr!=0)) );
  pMW->pCsr = pCsr;

  /* Load the current output page into memory. */
  if( rc==LSM_OK ) rc = mergeWorkerLoadOutputPage(pMW);


  /* Position the cursor. */
  if( rc==LSM_OK ){
    if( pMW->pPage==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;
................................................................................
      ** from rhs segments for which the content has been completely merged
      ** 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 
             && pGobble->pSeg->iRoot==0
            ){
              lsmFsGobble(pWorker, pGobble->pSeg, pGobble->pPg);
            }
          }

        }else if( pLevel->lhs.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 );


          /* 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);
          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->lhs.iRoot );


            pLevel->pNext->lhs.iRoot = 0;
          }

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

          }
          lsmFree(pDb->pEnv, pLevel->aRhs);
          pLevel->nRight = 0;
          pLevel->aRhs = 0;

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

/*
** 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, Segment *pSeg, int nMin){
  int nSize = pSeg->nSize;
  Pgno iRoot = pSeg->iRoot;
  Pgno iFirst = pSeg->iFirst;
  Pgno iLast = pSeg->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,
  Segment *pSeg
){
  int i = 0;
  char *zSeg;

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

  return i;
}

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

  int nRec;
  int iPtr;
  int flags;
................................................................................
    lsmFsPageRelease(pPg);
  }

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

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

    zSeg = segToString(pDb->pEnv, pRun, 0);
    lsmLogMessage(pDb, LSM_OK, "Segment: %s", 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;

      Segment *aLeft[24];  
      Segment *aRight[24];

      int nLeft = 0;
      int nRight = 0;

      Segment *pSeg = &pLevel->lhs;

      aLeft[nLeft++] = pSeg;



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

        aRight[nRight++] = &pLevel->aRhs[i];


      }

      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( bKeys ){
      for(pLevel=pTopLevel; pLevel; pLevel=pLevel->pNext){
        int i;
        sortedDumpSegment(pDb, &pLevel->lhs, bVals);

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

          sortedDumpSegment(pDb, &pLevel->aRhs[i], 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->iOutputOff = -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
/*
** This function is only included in the build if LSM_DEBUG_EXPENSIVE is 
** defined. Its only purpose is to evaluate various assert() statements to 
** verify that the database is well formed in certain respects.
**
** More specifically, it checks 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 pOne does not contain the 
** correct set of pointers, an assert() statement fails.
*/
static int assertPointersOk(
  lsm_db *pDb,                    /* Database handle */
  Segment *pOne,                  /* Segment containing pointers */
  Segment *pTwo,                  /* Segment containing pointer targets */
  int bRhs                        /* True if pTwo may have been Gobble()d */
){
  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.pSeg = pOne;
  ptr2.pSeg = 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 ){
................................................................................
** verify that the database is well formed in certain respects.
**
** More specifically, it checks that the b-tree embedded in array pRun
** contains the correct keys. If not, an assert() fails.
*/
static int assertBtreeOk(
  lsm_db *pDb,
  Segment *pSeg
){
  int rc = LSM_OK;                /* Return code */
  if( pSeg->iRoot ){
    Blob blob = {0, 0, 0};        /* Buffer used to cache overflow keys */
    FileSystem *pFS = pDb->pFS;   /* File system to read from */
    Page *pPg = 0;                /* Main run page */
    BtreeCursor *pCsr = 0;        /* Btree cursor */

    rc = btreeCursorNew(pDb, pSeg, &pCsr);
    if( rc==LSM_OK ){
      rc = btreeCursorFirst(pCsr);
    }
    if( rc==LSM_OK ){
      rc = lsmFsDbPageGet(pFS, pSeg->iFirst, &pPg);
    }

    while( rc==LSM_OK ){
      Page *pNext;
      u8 *aData;
      int nData;
      int flags;

      rc = lsmFsDbPageNext(pSeg, pPg, 1, &pNext);
      lsmFsPageRelease(pPg);
      pPg = pNext;
      if( pPg==0 ) break;
      aData = fsPageData(pPg, &nData);
      flags = pageGetFlags(aData, nData);
      if( rc==LSM_OK 
       && 0==((SEGMENT_BTREE_FLAG|PGFTR_SKIP_THIS_FLAG) & flags)