SQLite4  Check-in [d54af93981]

    memset(pNew, 0, sizeof(KVLsm));
    pNew->base.pStoreVfunc = &kvlsmMethods;
    pNew->base.pEnv = pEnv;
    rc = lsm_new(0, &pNew->pDb);
    if( rc==SQLITE4_OK ){
      int i;
      int bMmap = 0;
      lsm_config(pNew->pDb, LSM_CONFIG_MMAP, &bMmap);
      for(i=0; i<ArraySize(aConfig); i++){
        const char *zVal = sqlite4_uri_parameter(zName, aConfig[i].zParam);
        if( zVal ){
          int nVal = sqlite4Atoi(zVal);
          lsm_config(pNew->pDb, aConfig[i].eParam, &nVal);
        }
      }

typedef struct Hierarchy Hierarchy;

struct Hierarchy {
  Page **apHier;
  int nHier;
};

/*
** aSave:
**   When mergeWorkerNextPage() is called to advance to the next page in
**   the output segment, if the bStore flag for an element of aSave[] is
**   true, it is cleared and the corresponding iPgno value is set to the 
**   page number of the page just completed.
**
**   aSave[0] is used to record the pointer value to be pushed into the
**   b-tree hierarchy. aSave[1] is used to save the page number of the
**   page containing the indirect key most recently written to the b-tree.
**   see mergeWorkerPushHierarchy() for details.
*/
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() */
  Pgno *aGobble;                  /* Gobble point for each input segment */

  Pgno iIndirect;
  struct SavedPgno {
    Pgno iPgno;
    int bStore;
  } aSave[2];
};

#ifdef LSM_DEBUG_EXPENSIVE
static int assertPointersOk(lsm_db *, Segment *, Segment *, int);
static int assertBtreeOk(lsm_db *, Segment *);
static void assertRunInOrder(lsm_db *pDb, Segment *pSeg);
#else
#define assertRunInOrder(x,y)
#define assertBtreeOk(x,y)
#endif


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

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

    if( rc==LSM_OK ){
      u8 *aData;
      int nData;
      aData = fsPageData(apHier[0], &nData);
      pMW->aSave[0].iPgno = pageGetPtr(aData, nData);
      p->nHier = nHier;
      p->apHier = apHier;
      rc = mergeWorkerMoveHierarchy(pMW, 0);
    }else{
      int i;
      for(i=0; i<nHier; i++){
        lsmFsPageRelease(apHier[i]);
      }
      lsmFree(pEnv, apHier);
    }
  }

  return rc;
}

/*



** B-tree pages use almost the same format as regular pages. The 
** differences are:
**
**   1. The record format is (usually, see below) as follows:
**
**         + Type byte (always SORTED_SEPARATOR or SORTED_SYSTEM_SEPARATOR),
**         + Absolute pointer value (varint),

**
**         + 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 mergeWorkerBtreeWrite(
  MergeWorker *pMW,

  u8 eType, 
  Pgno iPtr,
  Pgno iKeyPg,

  void *pKey,
  int nKey
){
  Segment *pSeg = &pMW->pLevel->lhs;
  Hierarchy *p = &pMW->hier;
  lsm_db *pDb = pMW->pDb;         /* Database handle */
  int rc = LSM_OK;                /* Return Code */
  int iLevel;                     /* Level of b-tree hierachy to write to */
  int nData;                      /* Size of aData[] in bytes */
  u8 *aData;                      /* Page data for level iLevel */
  int iOff;                       /* Offset on b-tree page to write record to */
  int nRec;                       /* Initial number of records on b-tree page */





  /* iKeyPg should be zero for an ordinary b-tree key, or non-zero for an





  ** indirect key. The flags byte for an indirect key is 0x00.  */









  assert( (eType==0)==(iKeyPg!=0) );








  /* The MergeWorker.apHier[] array contains the right-most leaf of the b-tree
  ** hierarchy, the root node, and all nodes that lie on the path between.
  ** apHier[0] is the right-most leaf and apHier[pMW->nHier-1] is the current
  ** root page.
  **
  ** This loop searches for a node with enough space to store the key on,

    if( iLevel==p->nHier ){
      /* Extend the array and allocate a new root page. */
      Page **aNew;
      aNew = (Page **)lsmRealloc(
          pMW->pDb->pEnv, p->apHier, sizeof(Page *)*(p->nHier+1)
      );
      if( !aNew ){
        return LSM_NOMEM_BKPT;

      }
      p->apHier = aNew;
    }else{
      Page *pOld;
      int nFree;

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

      /* Otherwise, this page is full. Set the right-hand-child pointer
      ** to iPtr and release it.  */
      lsmPutU64(&aData[SEGMENT_POINTER_OFFSET(nData)], iPtr);
      rc = lsmFsPagePersist(pOld);
      if( rc==LSM_OK ){
        iPtr = lsmFsPageNumber(pOld);
        lsmFsPageRelease(pOld);
      }
    }

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

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

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

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

  iOff = mergeWorkerPageOffset(aData, nData);

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

    aData[iOff++] = 0x00;
    iOff += lsmVarintPut32(&aData[iOff], iPtr);
    iOff += lsmVarintPut32(&aData[iOff], iKeyPg);
  }else{
    aData[iOff++] = eType;
    iOff += lsmVarintPut32(&aData[iOff], iPtr);
    iOff += lsmVarintPut32(&aData[iOff], nKey);
    memcpy(&aData[iOff], pKey, nKey);
  }

  if( iLevel>0 ){
    Pgno iRight = lsmFsPageNumber(p->apHier[iLevel-1]);
    lsmPutU64(&aData[SEGMENT_POINTER_OFFSET(nData)], iRight);
  }

  return rc;
}

static int mergeWorkerBtreeIndirect(MergeWorker *pMW){
  int rc = LSM_OK;
  if( pMW->iIndirect ){
    Pgno iKeyPg = pMW->aSave[1].iPgno;
    rc = mergeWorkerBtreeWrite(pMW, 0, pMW->iIndirect, iKeyPg, 0, 0);
    pMW->iIndirect = 0;
  }
  return rc;
}

/*
** Append the database key (iTopic/pKey/nKey) to the b-tree under 
** construction. This key has not yet been written to a segment page.
** The pointer that will accompany the new key in the b-tree - that
** points to the completed segment page that contains keys smaller than
** (pKey/nKey) is currently stored in pMW->aSave[0].iPgno.
*/
static int mergeWorkerPushHierarchy(
  MergeWorker *pMW,               /* Merge worker object */
  int iTopic,                     /* Topic value for this key */
  void *pKey,                     /* Pointer to key buffer */
  int nKey                        /* Size of pKey buffer in bytes */
){
  lsm_db *pDb = pMW->pDb;         /* Database handle */
  int rc = LSM_OK;                /* Return Code */
  int iLevel;                     /* Level of b-tree hierachy to write to */
  int nData;                      /* Size of aData[] in bytes */
  u8 *aData;                      /* Page data for level iLevel */
  int iOff;                       /* Offset on b-tree page to write record to */
  int nRec;                       /* Initial number of records on b-tree page */
  Pgno iPtr;                      /* Pointer value to accompany pKey/nKey */
  int bIndirect;                  /* True to use an indirect record */

  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;

  assert( pMW->aSave[0].bStore==0 );
  assert( pMW->aSave[1].bStore==0 );
  rc = mergeWorkerBtreeIndirect(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.  */
  iPtr = pMW->aSave[0].iPgno;
  assert( iPtr!=0 );

  /* Determine if the indirect format should be used. */
  bIndirect = (nKey*4 > lsmFsPageSize(pMW->pDb->pFS));
  if( bIndirect ){
    pMW->iIndirect = iPtr;
    pMW->aSave[1].bStore = 1;
  }else{
    rc = mergeWorkerBtreeWrite(
        pMW, (u8)(iTopic | LSM_SEPARATOR), iPtr, 0, pKey, nKey
    );
  }

  /* Ensure that the SortedRun.iRoot field is correct. */
  return rc;
}

static int mergeWorkerFinishHierarchy(
  MergeWorker *pMW                /* Merge worker object */
){
  if( pMW->hier.nHier>0 ){
    Page *pPg = pMW->hier.apHier[0];
    int nData;                      /* Size of aData[] in bytes */
    u8 *aData;                      /* Page data for pLeaf */
    Pgno iPtr;

    assert( pPg );
    assert( pMW->aSave[0].bStore==0 );
    iPtr = pMW->aSave[0].iPgno;

    aData = fsPageData(pPg, &nData);
    lsmPutU64(&aData[SEGMENT_POINTER_OFFSET(nData)], iPtr);
  }

  return LSM_OK;
}

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

  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 */
    int i;

    lsmFsPagePersist(pMW->pPage);
    for(i=0; i<2; i++){
      if( pMW->aSave[i].bStore ){
        pMW->aSave[i].iPgno = lsmFsPageNumber(pMW->pPage);
        pMW->aSave[i].bStore = 0;
      }
    }

    /* Release the completed output page. */
    lsmFsPageRelease(pMW->pPage);

    pMW->pPage = pNext;
    pMW->pLevel->pMerge->iOutputOff = 0;
    aData = fsPageData(pNext, &nData);

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 rc = LSM_OK;                /* Return code */
  Merge *pMerge;                  /* Persistent part of level merge state */
  int nHdr;                       /* Space required for this record header */
  Page *pPg;                      /* Page to write to */
  u8 *aData;                      /* Data buffer for page pWriter->pPage */
  int nData;                      /* Size of buffer aData[] in bytes */

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

      rc = mergeWorkerPushHierarchy(pMW, rtTopic(eType), pKey, nKey);

      assert( pMW->aSave[0].bStore==0 );
      pMW->aSave[0].bStore = 1;
      pMerge->nSkip = keyszToSkip(pMW->pDb->pFS, nKey);
    }else{
      pMerge->nSkip--;
      flags = PGFTR_SKIP_THIS_FLAG;



    }

    if( pMerge->nSkip ) flags |= PGFTR_SKIP_NEXT_FLAG;
  }

  /* Update the output segment */
  if( rc==LSM_OK ){
    aData = fsPageData(pPg, &nData);

/*
** Free all resources allocated by mergeWorkerInit().
*/
static void mergeWorkerShutdown(MergeWorker *pMW, int *pRc){
  int i;                          /* Iterator variable */
  int rc = *pRc;
  MultiCursor *pCsr = pMW->pCsr;
  Hierarchy *p = &pMW->hier;

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

    }
    
    pMerge->iOutputOff = -1;
    pMerge->bHierReadonly = 1;
  }

  lsmMCursorClose(pCsr);

  /* Persist and release the output page. */
  rc = lsmFsPagePersist(pMW->pPage);
  for(i=0; i<2; i++){
    if( pMW->aSave[i].bStore ){
      pMW->aSave[i].iPgno = lsmFsPageNumber(pMW->pPage);
      pMW->aSave[i].bStore = 0;
    }
  }
  lsmFsPageRelease(pMW->pPage);

  if( rc==LSM_OK ) rc = mergeWorkerBtreeIndirect(pMW);
  if( rc==LSM_OK ) rc = mergeWorkerFinishHierarchy(pMW);
  if( rc==LSM_OK && p->nHier ){
    pMW->pLevel->lhs.iRoot = lsmFsPageNumber(p->apHier[p->nHier-1]);
  }

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

  lsmFree(pMW->pDb->pEnv, pMW->aGobble);
  pMW->aGobble = 0;
  pMW->pCsr = 0;
  pMW->pPage = 0;
  pMW->pPage = 0;
}

/*
** The MergeWorker passed as the only argument is working to merge two or

      lsmFsPageRelease(pPg);
    }
  }

  if( rc==LSM_OK ){
    rc = mergeWorkerNextPage(pMW, iFPtr);
    if( pCsr->pPrevMergePtr ) *pCsr->pPrevMergePtr = iFPtr;
    pMW->aSave[0].bStore = 1;
  }

  return rc;
}

/*
** The cursor passed as the first argument is being used as the input for

      }
    }

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


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

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

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

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

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

#endif
    }

#endif
  }
  return rc;
}

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

    multiCursorVisitFreelist(pCsr, pnOvfl);
    rc = multiCursorAddTree(pCsr, pDb->pWorker, eTree);
    if( rc==LSM_OK && pNext && pNext->pMerge==0 && pNext->lhs.iRoot ){
      pDel = &pNext->lhs;
      rc = btreeCursorNew(pDb, pDel, &pCsr->pBtCsr);
      iLeftPtr = pNext->lhs.iFirst;
    }

    if( pNext==0 ){
      multiCursorIgnoreDelete(pCsr);
    }
  }

  if( rc!=LSM_OK ){
    lsmMCursorClose(pCsr);
  }else{
    Merge merge;                  /* Merge object used to create new level */
    MergeWorker mergeworker;      /* MergeWorker object for the same purpose */

    pCsr->pPrevMergePtr = &iLeftPtr;

    /* 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);
    mergeworker.aSave[0].bStore = 1;

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

    nWrite = mergeworker.nWork;
    mergeWorkerShutdown(&mergeworker, &rc);
    pNew->pMerge = 0;
  }

  /* 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 0
  lsmSortedDumpStructure(pDb, pDb->pWorker, 1, 0, "new-toplevel");
#endif

  if( rc==LSM_OK ){
    assertBtreeOk(pDb, &pNew->lhs);
    sortedInvokeWorkHook(pDb);
  }

  if( pnWrite ) *pnWrite = nWrite;
  pDb->pWorker->nWrite += nWrite;
  return rc;
}

/*

  }else{
    multiCursorIgnoreDelete(pCsr);
  }

  assert( rc!=LSM_OK || pMerge->nInput==(pCsr->nPtr+(pCsr->pBtCsr!=0)) );
  pMW->pCsr = pCsr;

  /* Load the current output page and b-tree hierarchy into memory. */
  if( rc==LSM_OK ) rc = mergeWorkerLoadOutputPage(pMW);
  if( rc==LSM_OK ) rc = mergeWorkerLoadHierarchy(pMW);
  if( rc==LSM_OK && pMW->pPage && pMW->hier.nHier==0 ){
    pMW->aSave[0].iPgno = pLevel->lhs.iFirst;
  }

  /* Set MergeWorker.aSave[0].iPgno to contain the */
  if( rc==LSM_OK ){
  }

  /* Position the cursor. */
  if( rc==LSM_OK ){
    pCsr->pPrevMergePtr = &pMerge->iCurrentPtr;
    if( pMW->pPage==0 ){
      /* The output array is still empty. So position the cursor at the very 
      ** start of the input.  */

    }
    pCsr->flags |= CURSOR_NEXT_OK;
  }

  return rc;
}

/* TODO: Re-enable this!!! */
static int sortedBtreeGobble(
  lsm_db *pDb, 
  MultiCursor *pCsr, 
  int iGobble
){
  int rc = LSM_OK;
  if( rtTopic(pCsr->eType)==0 ){
    Segment *pSeg = pCsr->aPtr[iGobble].pSeg;
    Blob *p = &pCsr->key;
    Pgno *aPg;
    int nPg;

    assert( pSeg->iRoot>0 );
    aPg = lsmMallocZeroRc(pDb->pEnv, sizeof(Pgno)*32, &rc);
    if( rc==LSM_OK ){
      rc = seekInBtree(pCsr, pSeg, p->pData, p->nData, aPg, 0); 
    }

    for(nPg=0; aPg[nPg]; nPg++);

#if 0
    lsmFsGobble(pDb, pSeg, aPg, nPg);
#endif

    lsmFree(pDb->pEnv, aPg);
  }
  return rc;
}

      ** the database structure has changed. */
      mergeWorkerShutdown(&mergeworker, &rc);
      if( rc==LSM_OK ) sortedInvokeWorkHook(pDb);

#if 0
      lsmSortedDumpStructure(pDb, pDb->pWorker, 1, 0, "work");
#endif
      assertBtreeOk(pDb, &pLevel->lhs);
      assertRunInOrder(pDb, &pLevel->lhs);

      /* If bFlush is true and the database is no longer considered "full",
      ** break out of the loop even if nRemaining is still greater than
      ** zero. The caller has an in-memory tree to flush to disk.  */
      if( bFlush && sortedDbIsFull(pDb)==0 ) break;
    }

#endif
  return rc;
}

/*
** The database connection passed as the first argument must be a worker
** connection. This function checks if there exists an "old" in-memory tree
** ready to be flushed to disk. If so, true is returned. Otherwise false.

**
** If an error occurs, *pRc is set to an LSM error code before returning.
** It is assumed that *pRc is set to LSM_OK when this function is called.
*/
static int sortedTreeHasOld(lsm_db *pDb, int *pRc){
  int rc = LSM_OK;
  int bRet = 0;

  assert( pDb->pWorker );
  if( *pRc==LSM_OK ){
    if( pDb->nTransOpen==0 ){
      rc = lsmTreeLoadHeader(pDb, 0);
    }

    if( rc==LSM_OK 
        && pDb->treehdr.iOldShmid
        && pDb->treehdr.iOldLog!=pDb->pWorker->iLogOff 
      ){
      bRet = 1;
    }else{
      bRet = 0;
    }
    *pRc = rc;
  }
  assert( *pRc==LSM_OK || bRet==0 );
  return bRet;
}

static int doLsmSingleWork(
  lsm_db *pDb, 
  int bShutdown,
  int flags, 
  int nPage,                      /* Number of pages to write to disk */

    nMax = (pDb->nAutockpt/nPgsz) - (nUnsync-nSync);
    if( nMax<nRem ){
      bCkpt = 1;
      nRem = LSM_MAX(nMax, 0);
    }
  }

  /* If there exists in-memory data ready to be flushed to disk, attempt

  ** to flush it now.  */


  if( sortedTreeHasOld(pDb, &rc) ){

    if( sortedDbIsFull(pDb) ){
      int nPg = 0;
      rc = sortedWork(pDb, nRem, 0, 1, &nPg);
      nRem -= nPg;
      assert( rc!=LSM_OK || nRem<=0 || !sortedDbIsFull(pDb) );
      bToplevel = 1;
    }

    if( rc==LSM_OK && nRem>0 ){
      int nPg = 0;
      rc = sortedNewToplevel(pDb, TREE_OLD, &nOvfl, &nPg);
      nRem -= nPg;
      if( rc==LSM_OK && pDb->nTransOpen>0 ){
        lsmTreeDiscardOld(pDb);
      }
      bFlush = 1;
      bToplevel = 0;

    }
  }

  /* If nPage is still greater than zero, do some merging. */
  if( rc==LSM_OK && nRem>0 && bShutdown==0 ){
    int nPg = 0;
    int bOptimize = ((flags & LSM_WORK_OPTIMIZE) ? 1 : 0);

  lsmStringClear(&s);

  sortedBlobFree(&blob);
}

static void infoCellDump(
  lsm_db *pDb,
  int bIndirect,                  /* True to follow indirect refs */
  Page *pPg,
  int iCell,
  int *peType,
  int *piPgPtr,
  u8 **paKey, int *pnKey,
  u8 **paVal, int *pnVal,
  Blob *pBlob

  eType = *aCell++;
  aCell += lsmVarintGet32(aCell, &iPgPtr);

  if( eType==0 ){
    int dummy;
    Pgno iRef;                  /* Page number of referenced page */
    aCell += lsmVarintGet64(aCell, &iRef);
    if( bIndirect ){
      lsmFsDbPageGet(pDb->pFS, iRef, &pRef);
      pageGetKeyCopy(pDb->pEnv, pRef, 0, &dummy, pBlob);
      aKey = (u8 *)pBlob->pData;
      nKey = pBlob->nData;
      lsmFsPageRelease(pRef);
    }else{
      aKey = (u8 *)"<indirect>";
      nKey = 11;
    }
  }else{
    aCell += lsmVarintGet32(aCell, &nKey);
    if( rtIsWrite(eType) ) aCell += lsmVarintGet32(aCell, &nVal);
    sortedReadData(pPg, (aCell-aData), nKey+nVal, (void **)&aKey, pBlob);
    aVal = &aKey[nKey];
  }

    }else{
      lsmStringAppendf(pStr, "%c", isalnum(z[iChar]) ?z[iChar] : '.');
    }
  }
  return LSM_OK;
}

#define INFO_PAGE_DUMP_DATA     0x01
#define INFO_PAGE_DUMP_VALUES   0x02
#define INFO_PAGE_DUMP_HEX      0x04
#define INFO_PAGE_DUMP_INDIRECT 0x08

static int infoPageDump(
  lsm_db *pDb,                    /* Database handle */
  Pgno iPg,                       /* Page number of page to dump */
  int flags,
  char **pzOut                    /* OUT: lsmMalloc'd string */
){
  int rc = LSM_OK;                /* Return code */
  Page *pPg = 0;                  /* Handle for page iPg */
  int i, j;                       /* Loop counters */
  const int perLine = 16;         /* Bytes per line in the raw hex dump */

  int bValues = (flags & INFO_PAGE_DUMP_VALUES);
  int bHex = (flags & INFO_PAGE_DUMP_HEX);
  int bData = (flags & INFO_PAGE_DUMP_DATA);
  int bIndirect = (flags & INFO_PAGE_DUMP_INDIRECT);

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

  rc = lsmFsDbPageGet(pDb->pFS, iPg, &pPg);
  if( rc==LSM_OK ){
    Blob blob = {0, 0, 0, 0};

    lsmStringAppendf(&str, "nRec : %d\n", nRec);
    lsmStringAppendf(&str, "iPtr : %d\n", iPtr);
    lsmStringAppendf(&str, "flags: %04x\n", flags);
    lsmStringAppendf(&str, "\n");

    for(iCell=0; iCell<nRec; iCell++){
      int nKey;
      infoCellDump(pDb, bIndirect, pPg, iCell, 0, 0, 0, &nKey, 0, 0, &blob);
      if( nKey>nKeyWidth ) nKeyWidth = nKey;
    }
    if( bHex ) nKeyWidth = nKeyWidth * 2;

    for(iCell=0; iCell<nRec; iCell++){
      u8 *aKey; int nKey = 0;       /* Key */
      u8 *aVal; int nVal = 0;       /* Value */
      int iPgPtr;
      int eType;
      char cType = '?';
      Pgno iAbsPtr;
      char zFlags[8];

      infoCellDump(pDb, bIndirect, pPg, iCell, &eType, &iPgPtr,
          &aKey, &nKey, &aVal, &nVal, &blob
      );
      iAbsPtr = iPgPtr + ((flags & SEGMENT_BTREE_FLAG) ? 0 : iPtr);

      lsmFlagsToString(eType, zFlags);
      lsmStringAppendf(&str, "%s %d (%s) ", 
          zFlags, iAbsPtr, (rtTopic(eType) ? "sys" : "usr")

    TreeKey *pKey = csrGetKey(&csr, &blob, &rc);
    if( rc!=LSM_OK ) break;
    assert( ((prev&LSM_START_DELETE)==0)==((pKey->flags&LSM_END_DELETE)==0) );
    prev = pKey->flags;
  }

  tblobFree(csr.pDb, &csr.blob);
  tblobFree(csr.pDb, &blob);

  return 1;
}

static int treeCountEntries(lsm_db *db){
  TreeCursor csr;               /* Cursor used to iterate through tree */
  int rc;

populate_db
do_execsql_test 2.1 { 
  BEGIN;
    INSERT INTO t1 VALUES(10, 100);
}
do_test 2.2 { 
  sqlite4 db2 ./test.db
  list [catch { db2 eval { BEGIN ; INSERT INTO t1 VALUES(1, 2) } } msg] $msg
} {1 {database is locked}}

do_execsql_test 2.3 { COMMIT }
do_test 2.4 { sqlite4_lsm_work db2 main -flush 0 } {0}
db2 close


#-------------------------------------------------------------------------

  execsql { SELECT count(*) FROM t1 } db2
} {256}
db2 close

reset_db
do_execsql_test 3.5 { CREATE TABLE t1(a, b) }
do_test 3.6 {
  sqlite4_lsm_checkpoint db main
  for {set i 0} {$i < 203} {incr i} {
    execsql { INSERT INTO t1 VALUES(randstr(100,100), randstr(100,100)) }
  }
  execsql { SELECT count(*) FROM t1 }
} {203}

do_test 3.7 {

  INSERT INTO x VALUES(randstr(10,10), randstr(100,100));
  INSERT INTO x VALUES(randstr(10,10), randstr(100,100));
  INSERT INTO x VALUES(randstr(10,10), randstr(100,100));
  INSERT INTO x VALUES(randstr(10,10), randstr(100,100));
  INSERT INTO x VALUES(randstr(10,10), randstr(100,100));
}
do_filesize_test 8.2   0 776
do_test          8.3.1 { sqlite4_lsm_flush db main } {}
do_test          8.3.2 { sqlite4_lsm_work db main } 0
do_execsql_test  8.4 { INSERT INTO x VALUES(randstr(10,10), randstr(100,100)) }
do_filesize_test 8.5   12288 915
do_test          8.6 { sqlite4_lsm_checkpoint db main } {}

do_test 8.7 {
  copy_db_files test.db test.db2
  sqlite4 db2 test.db2
  execsql { SELECT count(*) FROM x ; PRAGMA integrity_check } db2
} {6 ok}

}
do_test 10.7 { sqlite4_lsm_info db main log-structure } {0 0 0 0 0 556}
do_test 10.8 { sqlite4_lsm_work db main -flush } 0
do_execsql_test 10.9 {
  INSERT INTO t1 VALUES(randstr(10,10), randstr(100,100));
}
do_test 10.9  { sqlite4_lsm_info db main log-structure } {0 0 0 0 0 695}
do_test 10.10 { sqlite4_lsm_checkpoint db main } {}
do_test 10.11 { sqlite4_lsm_info db main log-structure } {0 0 0 0 556 695}

#-------------------------------------------------------------------------
#
reset_db
do_test         11.1 { sqlite4_lsm_config db main log-size 800 } 800
do_test         11.2 { sqlite4_lsm_config db main log-size     } 800

do_test 11.4 { sqlite4_lsm_info db main log-structure } {0 0 0 0 0 1335}
do_test 11.5 { sqlite4_lsm_work db main -flush } 0
do_execsql_test 11.6 {
  INSERT INTO t1 VALUES(randstr(10,10), randstr(100,100));
}
do_test 11.7 { sqlite4_lsm_info db main log-structure } {0 0 0 0 0 1474}
do_test 11.8 { sqlite4_lsm_checkpoint db main } {}
do_test 11.9 { sqlite4_lsm_info db main log-structure } {0 0 0 0 1335 1474}
do_execsql_test 11.10 {
  INSERT INTO t1 VALUES(randstr(10,10), randstr(100,100));
}
do_test 11.11 { sqlite4_lsm_info db main log-structure } {1335 1482 0 0 0 139}
do_test 11.12 {
  execsql { SELECT count(*) FROM t1 ; PRAGMA integrity_check } 

db eval {SELECT randstr(5,5)}
do_execsql_test 11.22 {
  INSERT INTO t1 VALUES(randstr(10,10), randstr(100,100));
}
do_test 11.23 { 
  sqlite4_lsm_info db main log-structure 
} {1335 1482 0 1259 1483 1908}
do_test 11.24 { sqlite4_lsm_checkpoint db main } {}
do_test 11.25 { 
  sqlite4_lsm_info db main log-structure 
} {0 0 0 0 1769 1908}

#-------------------------------------------------------------------------
#
reset_db
do_test         12.1 { sqlite4_lsm_config db main log-size 800 } 800
do_execsql_test 12.2 {
  CREATE TABLE t1(a PRIMARY KEY, b);
  CREATE INDEX i1 ON t1(b);
}
for {set iTest 1} {$iTest<=150} {incr iTest} {
  expr srand(0)
  do_test 12.3.$iTest {
    for {set i 0} {$i < 10} {incr i} {
      execsql { INSERT INTO t1 VALUES(randstr(20,20), randstr(100,100)) }
      if { int(rand()*10.0)==0 } { sqlite4_lsm_work db main -flush }
      if { int(rand()*10.0)==0 } { sqlite4_lsm_checkpoint db main }
    }
    copy_db_files test.db test.db2
    sqlite4 db2 test.db2
    set sql "SELECT count(*) FROM t1 ; "
    if {0==($iTest % 25)} {
      append sql "PRAGMA integrity_check"
    } else {

    INSERT INTO t1 VALUES(randstr(50,50), randstr(50,50));
    INSERT INTO t1 VALUES(randstr(50,50), randstr(50,50));
    INSERT INTO t1 VALUES(randstr(50,50), randstr(50,50));
  COMMIT;
} {}
do_filesize_test 2.5   0 2048

do_test         2.6 { sqlite4_lsm_flush db main } {}
do_execsql_test 2.7 { INSERT INTO t1 VALUES(randstr(50,50), randstr(50,50)) }
do_test         2.8 { sqlite4_lsm_checkpoint db main } {}
do_test 2.9 { sqlite4_lsm_info db main log-structure } {0 0 0 0 2560 3072}

for {set i 1} {$i <= 6} {incr i} {
  do_execsql_test 2.10.$i.1 {
    INSERT INTO t1 VALUES(randstr(50,50), randstr(50,50));
  }
  do_execsql_test 2.10.$i.2 { SELECT count(*) FROM t1 } [expr 8 + $i]
  do_recover_test 2.10.$i.3 { SELECT count(*) FROM t1 } [expr 8 + $i]
}

do_test 2.11 { 
  sqlite4_lsm_info db main log-structure 
} {2560 3080 0 2216 3584 4608}


finish_test

#   full
#
lappend ::testsuitelist xxx

test_suite "src4" -prefix "" -description {
} -files {
  simple.test simple2.test
  log3.test 
  csr1.test
  ckpt1.test
  mc1.test

  aggerror.test
  attach.test
  autoindex1.test

do_catchsql_test 70.3 {
  select * from maintable, joinme INDEXED by joinme_id_text_idx
} {1 {cannot use index: joinme_id_text_idx}}

#-------------------------------------------------------------------------
# This is testing that the "phantom" runs feature works.
#
# UPDATE: Said feature was dropped early in development. But the test 
# remains valid.
reset_db
do_execsql_test 71.1 {
  CREATE TABLE t1(x);
  INSERT INTO t1 VALUES(randomblob(1024));           --   1
  INSERT INTO t1 SELECT randomblob(1024) FROM t1;    --   2
  INSERT INTO t1 SELECT randomblob(1024) FROM t1;    --   4
  INSERT INTO t1 SELECT randomblob(1024) FROM t1;    --   8

do_execsql_test 71.3 { SELECT count(*) FROM t1 } 64
do_test 71.4 { 
  expr {[file size test.db] < 256*1024}
} {1}

#-------------------------------------------------------------------------
# This is testing that the "phantom" runs feature works with mmap.
#
# UPDATE: Said feature was dropped early in development. But the test 
# remains valid.
reset_db

#do_test 72.0.1 { sqlite4_lsm_config db main mmap   } 0
#do_test 72.0.2 { sqlite4_lsm_config db main mmap 1 } 1
#do_test 72.0.3 { sqlite4_lsm_config db main mmap   } 1

do_execsql_test 72.1 {
  CREATE TABLE t1(x);
  INSERT INTO t1 VALUES(randomblob(1024));           --   1
  INSERT INTO t1 SELECT randomblob(1024) FROM t1;    --   2
  INSERT INTO t1 SELECT randomblob(1024) FROM t1;    --   4
  INSERT INTO t1 SELECT randomblob(1024) FROM t1;    --   8

  Tcl_Obj *CONST objv[]
){
  struct Switch {
    const char *zSwitch;
    int flags;
  } aSwitch[] = {
    { "-flush",      LSM_WORK_FLUSH }, 

    { "-optimize",   LSM_WORK_OPTIMIZE }, 
    { 0, 0 }
  };

  int flags = 0;
  int nPage = 0;
  const char *zDb;

    Tcl_SetResult(interp, (char *)sqlite4TestErrorName(rc), TCL_STATIC);
    return TCL_ERROR;
  }

  Tcl_SetObjResult(interp, Tcl_NewIntObj(nWork));
  return TCL_OK;
}

/*
** TCLCMD:    sqlite4_lsm_checkpoint DB DBNAME 
*/
static int test_lsm_checkpoint(
  void * clientData,
  Tcl_Interp *interp,
  int objc,
  Tcl_Obj *CONST objv[]
){
  const char *zDb;
  const char *zName;
  int rc;
  sqlite4 *db;
  lsm_db *pLsm;

  if( objc!=3 ){
    Tcl_WrongNumArgs(interp, 1, objv, "DB DBNAME");
    return TCL_ERROR;
  }
  zDb = Tcl_GetString(objv[1]);
  zName = Tcl_GetString(objv[2]);

  rc = getDbPointer(interp, zDb, &db);
  if( rc!=TCL_OK ) return rc;

  rc = sqlite4_kvstore_control(db, zName, SQLITE4_KVCTRL_LSM_HANDLE, &pLsm);
  if( rc==SQLITE4_OK ){
    rc = lsm_checkpoint(pLsm, 0);
  }
  if( rc!=SQLITE4_OK ){
    Tcl_SetResult(interp, (char *)sqlite4TestErrorName(rc), TCL_STATIC);
    return TCL_ERROR;
  }

  Tcl_ResetResult(interp);
  return TCL_OK;
}

/*
** TCLCMD:    sqlite4_lsm_flush DB DBNAME 
*/
static int test_lsm_flush(
  void * clientData,
  Tcl_Interp *interp,
  int objc,
  Tcl_Obj *CONST objv[]
){
  const char *zDb;
  const char *zName;
  int rc;
  sqlite4 *db;
  lsm_db *pLsm;

  if( objc!=3 ){
    Tcl_WrongNumArgs(interp, 1, objv, "DB DBNAME");
    return TCL_ERROR;
  }
  zDb = Tcl_GetString(objv[1]);
  zName = Tcl_GetString(objv[2]);

  rc = getDbPointer(interp, zDb, &db);
  if( rc!=TCL_OK ) return rc;

  rc = sqlite4_kvstore_control(db, zName, SQLITE4_KVCTRL_LSM_HANDLE, &pLsm);
  if( rc==SQLITE4_OK ){
    int nZero = 0;
    int nOrig = -1;
    lsm_config(pLsm, LSM_CONFIG_WRITE_BUFFER, &nOrig);
    lsm_config(pLsm, LSM_CONFIG_WRITE_BUFFER, &nZero);
    rc = lsm_begin(pLsm, 1);
    if( rc==LSM_OK ) rc = lsm_commit(pLsm, 0);
    lsm_config(pLsm, LSM_CONFIG_WRITE_BUFFER, &nOrig);
  }
  if( rc!=SQLITE4_OK ){
    Tcl_SetResult(interp, (char *)sqlite4TestErrorName(rc), TCL_STATIC);
    return TCL_ERROR;
  }

  Tcl_ResetResult(interp);
  return TCL_OK;
}

int SqlitetestLsm_Init(Tcl_Interp *interp){
  struct SyscallCmd {
    const char *zName;
    Tcl_ObjCmdProc *xCmd;
  } aCmd[] = {
    { "sqlite4_lsm_work",       test_lsm_work                },
    { "sqlite4_lsm_checkpoint", test_lsm_checkpoint          },
    { "sqlite4_lsm_flush",      test_lsm_flush               },
    { "sqlite4_lsm_info",       test_lsm_info                },
    { "sqlite4_lsm_config",     test_lsm_config              },
  };
  int i;

  for(i=0; i<sizeof(aCmd)/sizeof(aCmd[0]); i++){
    Tcl_CreateObjCommand(interp, aCmd[i].zName, aCmd[i].xCmd, 0, 0);
  }
  return TCL_OK;
}

# Flush the in-memory tree to disk and merge all runs together into
# a single b-tree structure. Because this annihilates all delete keys,
# the next rowid allocated for each table with an IPK will be as expected
# by SQLite 3 tests.
#
proc optimize_db {} { 
  #catch { 
    sqlite4_lsm_flush db main 
    sqlite4_lsm_work db main -opt -flush 100000 
    sqlite4_lsm_checkpoint db main
  #}
  return ""
}


# If the library is compiled with the SQLITE4_DEFAULT_AUTOVACUUM macro set
# to non-zero, then set the global variable $AUTOVACUUM to 1.
set AUTOVACUUM $sqlite_options(default_autovacuum)