SQLite4  Check-in [63d8eea506]

** Macros to help decode record types.
*/
#define rtTopic(eType)       ((eType) & 0xF0)
#define rtIsDelete(eType)    (((eType) & 0x0F)==SORTED_DELETE)
#define rtIsSeparator(eType) (((eType) & 0x0F)==SORTED_SEPARATOR)
#define rtIsWrite(eType)     (((eType) & 0x0F)==SORTED_WRITE)




/*
** The following macros are used to access a page footer.
*/
#define SEGMENT_NRECORD_OFFSET(pgsz)        ((pgsz) - 2)
#define SEGMENT_FLAGS_OFFSET(pgsz)          ((pgsz) - 2 - 2)
#define SEGMENT_POINTER_OFFSET(pgsz)        ((pgsz) - 2 - 2 - 4)
#define SEGMENT_CELLPTR_OFFSET(pgsz, iCell) ((pgsz) - 2 - 2 - 4 - 2 - (iCell)*2)

    mcursorFreeComponents(pCsr);

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

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

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

  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(pDb, &pLevel->lhs, &pCsr->pBtCsr);
    assert( (rc==LSM_OK)==(pCsr->pBtCsr!=0) );
  }else{
    int i;
    int nAdd = pLevel->nRight + (eMode==MULTICURSOR_ADDLEVEL_ALL);
    int nByte;
    SegmentPtr *aNew;

    /* Grow the pCsr->aPtr array */
    nByte = sizeof(SegmentPtr) * (pCsr->nPtr + nAdd);
    aNew = (SegmentPtr *)lsmRealloc(pDb->pEnv, pCsr->aPtr, nByte);
    if( aNew==0 ) return LSM_NOMEM_BKPT;
    memset(&aNew[pCsr->nPtr], 0, nAdd * sizeof(SegmentPtr));
    pCsr->aPtr = aNew;

    /* If this is ALL, add the left-hand-side segment */
    if( eMode==MULTICURSOR_ADDLEVEL_ALL ){
      aNew[pCsr->nPtr].pSeg = &pLevel->lhs;
      aNew[pCsr->nPtr].pLevel = pLevel;
      if( pLevel->nRight && pLevel->pSplitKey==0 ){
        lsmSortedSplitkey(pDb, pLevel, &rc);
      }
      pCsr->nPtr++;
    }

    /* Add the right-hand-side segments */
    for(i=0; i<pLevel->nRight; i++){
      aNew[pCsr->nPtr].pSeg = &pLevel->aRhs[i];
      aNew[pCsr->nPtr].pLevel = pLevel;
      pCsr->nPtr++;
    }
  }

  return rc;
}

#define TREE_NONE 0
#define TREE_OLD  1
#define TREE_BOTH 2

/*
** Parameter eTree must be set to TREE_NONE, OLD or BOTH.
*/
static int multiCursorNew(
  lsm_db *pDb,                    /* Database handle */
  Snapshot *pSnap,                /* Snapshot to use for this cursor */
  int eTree,                      /* One of the TREE_XXX values above */
  int bUserOnly,                  /* If true, ignore all system data */
  MultiCursor **ppCsr             /* OUT: Allocated cursor */
){
  int rc = LSM_OK;                /* Return Code */
  MultiCursor *pCsr = *ppCsr;     /* Allocated multi-cursor */

  assert( eTree==TREE_NONE || eTree==TREE_OLD || eTree==TREE_BOTH );

  if( pCsr==0 ){
    pCsr = (MultiCursor *)lsmMallocZeroRc(pDb->pEnv, sizeof(MultiCursor), &rc);
    if( pCsr ){
      pCsr->pNext = pDb->pCsr;
      pDb->pCsr = pCsr;
      if( bUserOnly ) pCsr->flags |= CURSOR_IGNORE_SYSTEM;
      pCsr->pDb = pDb;
    }
  }

  /* Add a tree cursor on the 'old' tree, if required. */
  if( rc==LSM_OK 
   && eTree!=TREE_NONE 
   && lsmTreeHasOld(pDb)
   && pDb->treehdr.iOldLog!=pSnap->iLogOff
  ){
    rc = lsmTreeCursorNew(pDb, 1, &pCsr->apTreeCsr[1]);
  }

  /* Add a tree cursor on the 'current' tree, if required. */
  if( rc==LSM_OK && eTree==TREE_BOTH ){
    rc = lsmTreeCursorNew(pDb, 0, &pCsr->apTreeCsr[0]);
  }



















  if( rc!=LSM_OK ){

    lsmMCursorClose(pCsr);


    pCsr = 0;


  }










































  *ppCsr = pCsr;


  return rc;
}


void lsmSortedRemap(lsm_db *pDb){
  MultiCursor *pCsr;
  for(pCsr=pDb->pCsr; pCsr; pCsr=pCsr->pNext){
    int iPtr;
    if( pCsr->pBtCsr ){
      btreeCursorLoadKey(pCsr->pBtCsr);

*/
static void multiCursorIgnoreDelete(MultiCursor *pCsr){
  if( pCsr ) pCsr->flags |= CURSOR_IGNORE_DELETE;
}

/*
** If the free-block list is not empty, then have this cursor visit a key
** with (a) the system bit set, and (b) the key "F" and (c) a value blob
** containing the entire serialized free-block list.
*/
static void multiCursorVisitFreelist(MultiCursor *pCsr, int *pnOvfl){
  assert( pCsr );
  pCsr->pnOvfl = pnOvfl;
  pCsr->flags |= CURSOR_NEW_SYSTEM;
}

/*
** Allocate a new cursor to read the database (the in-memory tree and all
** levels). If successful, set *ppCsr to point to the new cursor object
** and return SQLITE4_OK. Otherwise, set *ppCsr to NULL and return an
** lsm error code.
**
** If parameter bSystem is true, this is a system cursor. In that case
** the behaviour of this function is modified as follows:
**
**   * the worker snapshot is used instead of the client snapshot, and
**   * the in-memory tree is ignored.
*/
static int multiCursorAllocate(
  lsm_db *pDb,                    /* Database handle */
  int bSystem,                    /* True for a system cursor */
  MultiCursor **ppCsr             /* OUT: Allocated cursor */
){
  int rc = LSM_OK;                /* Return Code */
  MultiCursor *pCsr = *ppCsr;     /* Allocated multi-cursor */
  Level *p;                       /* Level iterator */
  Snapshot *pSnap;                /* Snapshot to use for cursor */

  pSnap = (bSystem ? pDb->pWorker : pDb->pClient);
  assert( pSnap );

  rc = multiCursorNew(pDb, pSnap, 
      (bSystem ? TREE_NONE : TREE_BOTH), !bSystem, &pCsr
  );

  multiCursorIgnoreDelete(pCsr);
  for(p=lsmDbSnapshotLevel(pSnap); p && rc==LSM_OK; p=p->pNext){
    rc = multiCursorAddLevel(pCsr, p, MULTICURSOR_ADDLEVEL_ALL);
  }

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

/*
** Allocate and return a new database cursor.
*/
int lsmMCursorNew(
  lsm_db *pDb,                    /* Database handle */
  MultiCursor **ppCsr             /* OUT: Allocated cursor */
){
  MultiCursor *pCsr = 0;
  int rc;

  rc = multiCursorAllocate(pDb, 0, &pCsr);






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

static int multiCursorGetVal(
  MultiCursor *pCsr, 

}

int lsmSortedLoadFreelist(
  lsm_db *pDb,                    /* Database handle (must be worker) */
  void **ppVal,                   /* OUT: Blob containing LSM free-list */
  int *pnVal                      /* OUT: Size of *ppVal blob in bytes */
){
  MultiCursor *pCsr = 0;          /* Cursor used to retreive free-list */
  int rc;                         /* Return Code */

  assert( pDb->pWorker );
  assert( *ppVal==0 && *pnVal==0 );



  rc = multiCursorAllocate(pDb, 1, &pCsr);



  if( rc==LSM_OK ){
    rc = lsmMCursorLast(pCsr);
    if( rc==LSM_OK 
     && pCsr->eType==SORTED_SYSTEM_WRITE 
     && pCsr->key.nData==8 
     && 0==memcmp(pCsr->key.pData, "FREELIST", 8)
    ){

int mcursorSave(MultiCursor *pCsr){
  int rc = LSM_OK;
  if( pCsr->aTree ){
    int iTree = pCsr->aTree[1];
    if( iTree==CURSOR_DATA_TREE0 || iTree==CURSOR_DATA_TREE1 ){
      multiCursorCacheKey(pCsr, &rc);
    }
    mcursorFreeComponents(pCsr);
  }

  return rc;
}

int mcursorRestore(lsm_db *pDb, MultiCursor *pCsr){
  int rc;
  rc = multiCursorAllocate(pDb, 0, &pCsr);
  if( rc==LSM_OK && pCsr->key.pData ){
    rc = lsmMCursorSeek(pCsr, pCsr->key.pData, pCsr->key.nData, +1);
  }
  return rc;
}

int lsmSaveCursors(lsm_db *pDb){

  int nWrite = 0;                 /* Number of database pages written */

  assert( pnOvfl );

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





  /* Create a cursor to gather the data required by the new segment. The new
  ** segment contains everything in the tree and pointers to the next segment
  ** in the database (if any).  */
  if( rc==LSM_OK ){
    rc = multiCursorNew(pDb, pDb->pWorker, eTree, 0, &pCsr);
    if( rc==LSM_OK ){
      pNew->pNext = pNext;
      lsmDbSnapshotSetLevel(pDb->pWorker, pNew);
    }
    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);
      }
    }

    if( rc==LSM_OK ){
      multiCursorVisitFreelist(pCsr, pnOvfl);
      multiCursorReadSeparators(pCsr);
    }
  }

  if( rc!=LSM_OK ){
    lsmMCursorClose(pCsr);
  }else{
    Merge merge;                  /* Merge object used to create new level */

    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( rc==LSM_OK ){
    sortedInvokeWorkHook(pDb);
  }

  if( pnWrite ) *pnWrite = nWrite;
  pDb->pWorker->nWrite += nWrite;
#if 0
  lsmSortedDumpStructure(pDb, pDb->pWorker, 0, 0, "new-toplevel");
#endif



  return rc;
}

/*
** The nMerge levels in the LSM beginning with pLevel consist of a
** left-hand-side segment only. Replace these levels with a single new
** level consisting of a new empty segment on the left-hand-side and the

  lsm_db *pDb,                    /* Db connection to do merge work */
  Level *pLevel,                  /* Level to work on merging */
  MergeWorker *pMW                /* Object to initialize */
){
  int rc = LSM_OK;                /* Return code */
  Merge *pMerge = pLevel->pMerge; /* Persistent part of merge state */
  MultiCursor *pCsr = 0;          /* Cursor opened for pMW */


  assert( pDb->pWorker );
  assert( pLevel->pMerge );
  assert( pLevel->nRight>0 );

  memset(pMW, 0, sizeof(MergeWorker));
  pMW->pDb = pDb;
  pMW->pLevel = pLevel;
  pMW->aGobble = lsmMallocZeroRc(pDb->pEnv, sizeof(Pgno) * pLevel->nRight, &rc);

  /* Create a multi-cursor to read the data to write to the new
  ** segment. The new segment contains:
  **
  **   1. Records from LHS of each of the nMerge levels being merged.
  **   2. Separators from either the last level being merged, or the
  **      separators attached to the LHS of the following level, or neither.
  **
  ** If the new level is the lowest (oldest) in the db, discard any
  ** delete keys. Key annihilation.
  */
  if( rc==LSM_OK ){
    rc = multiCursorNew(pDb, pDb->pWorker, TREE_NONE, 0, &pCsr);
  }
  if( rc==LSM_OK ){
    rc = multiCursorAddLevel(pCsr, pLevel, MULTICURSOR_ADDLEVEL_RHS);
  }
  if( rc==LSM_OK && pLevel->pNext ){
    if( pMerge->nInput > pLevel->nRight ){

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

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

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

  /* Position the cursor. */