/*

** 2011-08-14

**

** The author disclaims copyright to this source code. In place of

** a legal notice, here is a blessing:

**

** May you do good and not evil.

** May you find forgiveness for yourself and forgive others.

** May you share freely, never taking more than you give.

**

*************************************************************************

**

** PAGE FORMAT:

**

** The maximum page size is 65536 bytes.

**

** Since all records are equal to or larger than 2 bytes in size, and

** some space within the page is consumed by the page footer, there must

** be less than 2^15 records on each page.

**

** Each page ends with a footer that describes the pages contents. This

** footer serves as similar purpose to the page header in an SQLite database.

** A footer is used instead of a header because it makes it easier to

** populate a new page based on a sorted list of key/value pairs.

**

** The footer consists of the following values (starting at the end of

** the page and continuing backwards towards the start). All values are

** stored as unsigned big-endian integers.

**

** * Number of records on page (2 bytes).

** * Flags field (2 bytes).

** * Left-hand pointer value (8 bytes).

** * The starting offset of each record (2 bytes per record).

**

** Records may span pages. Unless it happens to be an exact fit, the part

** of the final record that starts on page X that does not fit on page X

** is stored at the start of page (X+1). This means there may be pages where

** (N==0). And on most pages the first record that starts on the page will

** not start at byte offset 0. For example:

**

** aaaaa bbbbb ccc <footer> cc eeeee fffff g <footer> gggg....

**

** RECORD FORMAT:

**

** The first byte of the record is a flags byte. It is a combination

** of the following flags (defined in lsmInt.h):

**

** LSM_START_DELETE

** LSM_END_DELETE

** LSM_POINT_DELETE

** LSM_INSERT

** LSM_SEPARATOR

** LSM_SYSTEMKEY

**

** Immediately following the type byte is a pointer to the smallest key

** in the next file that is larger than the key in the current record. The

** pointer is encoded as a varint. When added to the 32-bit page number

** stored in the footer, it is the page number of the page that contains the

** smallest key in the next sorted file that is larger than this key.

**

** Next is the number of bytes in the key, encoded as a varint.

**

** If the LSM_INSERT flag is set, the number of bytes in the value, as

** a varint, is next.

**

** Finally, the blob of data containing the key, and for LSM_INSERT

** records, the value as well.

*/

#ifndef _LSM_INT_H

# include "lsmInt.h"

#endif

#define LSM_LOG_STRUCTURE 0

#define LSM_LOG_DATA 0

/*

** Macros to help decode record types.

*/

#define rtTopic(eType) ((eType) & LSM_SYSTEMKEY)

#define rtIsDelete(eType) (((eType) & 0x0F)==LSM_POINT_DELETE)

#define rtIsSeparator(eType) (((eType) & LSM_SEPARATOR)!=0)

#define rtIsWrite(eType) (((eType) & LSM_INSERT)!=0)

#define rtIsSystem(eType) (((eType) & LSM_SYSTEMKEY)!=0)

/*

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

#define SEGMENT_CELLPTR_OFFSET(pgsz, iCell) ((pgsz) - 2 - 2 - 8 - 2 - (iCell)*2)

#define SEGMENT_EOF(pgsz, nEntry) SEGMENT_CELLPTR_OFFSET(pgsz, nEntry-1)

#define SEGMENT_BTREE_FLAG 0x0001

#define PGFTR_SKIP_NEXT_FLAG 0x0002

#define PGFTR_SKIP_THIS_FLAG 0x0004

#ifndef LSM_SEGMENTPTR_FREE_THRESHOLD

# define LSM_SEGMENTPTR_FREE_THRESHOLD 1024

#endif

typedef struct SegmentPtr SegmentPtr;

typedef struct LsmBlob LsmBlob;

struct LsmBlob {

lsm_env *pEnv;

void *pData;

int nData;

int nAlloc;

};

/*

** A SegmentPtr object may be used for one of two purposes:

**

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

Level *pLevel; /* Level object segment is part of */

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

LsmPgno iPtr; /* Base cascade pointer */

/* Current cell. See segmentPtrLoadCell() */

int iCell; /* Current record within page pPg */

int eType; /* Type of current record */

LsmPgno iPgPtr; /* Cascade pointer offset */

void *pKey; int nKey; /* Key associated with current record */

void *pVal; int nVal; /* Current record value (eType==WRITE only) */

/* Blobs used to allocate buffers for pKey and pVal as required */

LsmBlob blob1;

LsmBlob blob2;

};

/*

** Used to iterate through the keys stored in a b-tree hierarchy from start

** to finish. Only First() and Next() operations are required.

**

** btreeCursorNew()

** btreeCursorFirst()

** btreeCursorNext()

** btreeCursorFree()

** btreeCursorPosition()

** btreeCursorRestore()

*/

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;

int nKey;

int eType;

LsmPgno iPtr;

/* Storage for key, if not local */

LsmBlob blob;

};

/*

** A cursor used for merged searches or iterations through up to one

** Tree structure and any number of sorted files.

**

** lsmMCursorNew()

** lsmMCursorSeek()

** lsmMCursorNext()

** lsmMCursorPrev()

** lsmMCursorFirst()

** lsmMCursorLast()

** lsmMCursorKey()

** lsmMCursorValue()

** lsmMCursorValid()

**

** iFree:

** This variable is only used by cursors providing input data for a

** new top-level segment. Such cursors only ever iterate forwards, not

** backwards.

*/

struct MultiCursor {

lsm_db *pDb; /* Connection that owns this cursor */

MultiCursor *pNext; /* Next cursor owned by connection pDb */

int flags; /* Mask of CURSOR_XXX flags */

int eType; /* Cache of current key type */

LsmBlob key; /* Cache of current key (or NULL) */

LsmBlob val; /* Cache of current value */

/* All the component cursors: */

TreeCursor *apTreeCsr[2]; /* Up to two tree cursors */

int iFree; /* Next element of free-list (-ve for eof) */

SegmentPtr *aPtr; /* Array of segment pointers */

int nPtr; /* Size of array aPtr[] */

BtreeCursor *pBtCsr; /* b-tree cursor (db writes only) */

/* Comparison results */

int nTree; /* Size of aTree[] array */

int *aTree; /* Array of comparison results */

/* Used by cursors flushing the in-memory tree only */

void *pSystemVal; /* Pointer to buffer to free */

/* Used by worker cursors only */

LsmPgno *pPrevMergePtr;

};

/*

** The following constants are used to assign integers to each component

** cursor of a multi-cursor.

*/

#define CURSOR_DATA_TREE0 0 /* Current tree cursor (apTreeCsr[0]) */

#define CURSOR_DATA_TREE1 1 /* The "old" tree, if any (apTreeCsr[1]) */

#define CURSOR_DATA_SYSTEM 2 /* Free-list entries (new-toplevel only) */

#define CURSOR_DATA_SEGMENT 3 /* First segment pointer (aPtr[0]) */

/*

** CURSOR_IGNORE_DELETE

** If set, this cursor will not visit SORTED_DELETE keys.

**

** CURSOR_FLUSH_FREELIST

** This cursor is being used to create a new toplevel. It should also

** iterate through the contents of the in-memory free block list.

**

** CURSOR_IGNORE_SYSTEM

** If set, this cursor ignores system keys.

**

** CURSOR_NEXT_OK

** Set if it is Ok to call lsm_csr_next().

**

** CURSOR_PREV_OK

** Set if it is Ok to call lsm_csr_prev().

**

** CURSOR_READ_SEPARATORS

** Set if this cursor should visit the separator keys in segment

** aPtr[nPtr-1].

**

** CURSOR_SEEK_EQ

** Cursor has undergone a successful lsm_csr_seek(LSM_SEEK_EQ) operation.

** The key and value are stored in MultiCursor.key and MultiCursor.val

** respectively.

*/

#define CURSOR_IGNORE_DELETE 0x00000001

#define CURSOR_FLUSH_FREELIST 0x00000002

#define CURSOR_IGNORE_SYSTEM 0x00000010

#define CURSOR_NEXT_OK 0x00000020

#define CURSOR_PREV_OK 0x00000040

#define CURSOR_READ_SEPARATORS 0x00000080

#define CURSOR_SEEK_EQ 0x00000100

typedef struct MergeWorker MergeWorker;

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

LsmPgno *aGobble; /* Gobble point for each input segment */

LsmPgno iIndirect;

struct SavedPgno {

LsmPgno 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;

}

/*UNUSED static u8 *fsPageDataPtr(Page *pPg){

return ((struct FilePage *)(pPg))->aData;

}*/

/*

** Write nVal as a 16-bit unsigned big-endian integer into buffer aOut.

*/

void lsmPutU16(u8 *aOut, u16 nVal){

aOut[0] = (u8)((nVal>>8) & 0xFF);

aOut[1] = (u8)(nVal & 0xFF);

}

void lsmPutU32(u8 *aOut, u32 nVal){

aOut[0] = (u8)((nVal>>24) & 0xFF);

aOut[1] = (u8)((nVal>>16) & 0xFF);

aOut[2] = (u8)((nVal>> 8) & 0xFF);

aOut[3] = (u8)((nVal ) & 0xFF);

}

int lsmGetU16(u8 *aOut){

return (aOut[0] << 8) + aOut[1];

}

u32 lsmGetU32(u8 *aOut){

return ((u32)aOut[0] << 24)

+ ((u32)aOut[1] << 16)

+ ((u32)aOut[2] << 8)

+ ((u32)aOut[3]);

}

u64 lsmGetU64(u8 *aOut){

return ((u64)aOut[0] << 56)

+ ((u64)aOut[1] << 48)

+ ((u64)aOut[2] << 40)

+ ((u64)aOut[3] << 32)

+ ((u64)aOut[4] << 24)

+ ((u32)aOut[5] << 16)

+ ((u32)aOut[6] << 8)

+ ((u32)aOut[7]);

}

void lsmPutU64(u8 *aOut, u64 nVal){

aOut[0] = (u8)((nVal>>56) & 0xFF);

aOut[1] = (u8)((nVal>>48) & 0xFF);

aOut[2] = (u8)((nVal>>40) & 0xFF);

aOut[3] = (u8)((nVal>>32) & 0xFF);

aOut[4] = (u8)((nVal>>24) & 0xFF);

aOut[5] = (u8)((nVal>>16) & 0xFF);

aOut[6] = (u8)((nVal>> 8) & 0xFF);

aOut[7] = (u8)((nVal ) & 0xFF);

}

static int sortedBlobGrow(lsm_env *pEnv, LsmBlob *pBlob, int nData){

assert( pBlob->pEnv==pEnv || (pBlob->pEnv==0 && pBlob->pData==0) );

if( pBlob->nAlloc<nData ){

pBlob->pData = lsmReallocOrFree(pEnv, pBlob->pData, nData);

if( !pBlob->pData ) return LSM_NOMEM_BKPT;

pBlob->nAlloc = nData;

pBlob->pEnv = pEnv;

}

return LSM_OK;

}

static int sortedBlobSet(lsm_env *pEnv, LsmBlob *pBlob, void *pData, int nData){

if( sortedBlobGrow(pEnv, pBlob, nData) ) return LSM_NOMEM;

memcpy(pBlob->pData, pData, nData);

pBlob->nData = nData;

return LSM_OK;

}

#if 0

static int sortedBlobCopy(LsmBlob *pDest, LsmBlob *pSrc){

return sortedBlobSet(pDest, pSrc->pData, pSrc->nData);

}

#endif

static void sortedBlobFree(LsmBlob *pBlob){

assert( pBlob->pEnv || pBlob->pData==0 );

if( pBlob->pData ) lsmFree(pBlob->pEnv, pBlob->pData);

memset(pBlob, 0, sizeof(LsmBlob));

}

static int sortedReadData(

Segment *pSeg,

Page *pPg,

int iOff,

int nByte,

void **ppData,

LsmBlob *pBlob

){

int rc = LSM_OK;

int iEnd;

int nData;

int nCell;

u8 *aData;

aData = fsPageData(pPg, &nData);

nCell = lsmGetU16(&aData[SEGMENT_NRECORD_OFFSET(nData)]);

iEnd = SEGMENT_EOF(nData, nCell);

assert( iEnd>0 && iEnd<nData );

if( iOff+nByte<=iEnd ){

*ppData = (void *)&aData[iOff];

}else{

int nRem = nByte;

int i = iOff;

u8 *aDest;

/* Make sure the blob is big enough to store the value being loaded. */

rc = sortedBlobGrow(lsmPageEnv(pPg), pBlob, nByte);

if( rc!=LSM_OK ) return rc;

pBlob->nData = nByte;

aDest = (u8 *)pBlob->pData;

*ppData = pBlob->pData;

/* Increment the pointer pages ref-count. */

lsmFsPageRef(pPg);

while( rc==LSM_OK ){

Page *pNext;

int flags;

/* Copy data from pPg into the output buffer. */

int nCopy = LSM_MIN(nRem, iEnd-i);

if( nCopy>0 ){

memcpy(&aDest[nByte-nRem], &aData[i], nCopy);

nRem -= nCopy;

i += nCopy;

assert( nRem==0 || i==iEnd );

}

assert( nRem>=0 );

if( nRem==0 ) break;

i -= iEnd;

/* Grab the next page in the segment */

do {

rc = lsmFsDbPageNext(pSeg, pPg, 1, &pNext);

if( rc==LSM_OK && pNext==0 ){

rc = LSM_CORRUPT_BKPT;

}

if( rc ) break;

lsmFsPageRelease(pPg);

pPg = pNext;

aData = fsPageData(pPg, &nData);

flags = lsmGetU16(&aData[SEGMENT_FLAGS_OFFSET(nData)]);

}while( flags&SEGMENT_BTREE_FLAG );

iEnd = SEGMENT_EOF(nData, lsmGetU16(&aData[nData-2]));

assert( iEnd>0 && iEnd<nData );

}

lsmFsPageRelease(pPg);

}

return rc;

}

static int pageGetNRec(u8 *aData, int nData){

return (int)lsmGetU16(&aData[SEGMENT_NRECORD_OFFSET(nData)]);

}

static LsmPgno pageGetPtr(u8 *aData, int nData){

return (LsmPgno)lsmGetU64(&aData[SEGMENT_POINTER_OFFSET(nData)]);

}

static int pageGetFlags(u8 *aData, int nData){

return (int)lsmGetU16(&aData[SEGMENT_FLAGS_OFFSET(nData)]);

}

static u8 *pageGetCell(u8 *aData, int nData, int iCell){

return &aData[lsmGetU16(&aData[SEGMENT_CELLPTR_OFFSET(nData, iCell)])];

}

/*

** Return the number of cells on page pPg.

*/

static int pageObjGetNRec(Page *pPg){

int nData;

u8 *aData = lsmFsPageData(pPg, &nData);

return pageGetNRec(aData, nData);

}

/*

** Return the decoded (possibly relative) pointer value stored in cell

** iCell from page aData/nData.

*/

static LsmPgno pageGetRecordPtr(u8 *aData, int nData, int iCell){

LsmPgno iRet; /* Return value */

u8 *aCell; /* Pointer to cell iCell */

assert( iCell<pageGetNRec(aData, nData) && iCell>=0 );

aCell = pageGetCell(aData, nData, iCell);

lsmVarintGet64(&aCell[1], &iRet);

return iRet;

}

static u8 *pageGetKey(

Segment *pSeg, /* Segment pPg belongs to */

Page *pPg, /* Page to read from */

int iCell, /* Index of cell on page to read */

int *piTopic, /* OUT: Topic associated with this key */

int *pnKey, /* OUT: Size of key in bytes */

LsmBlob *pBlob /* If required, use this for dynamic memory */

){

u8 *pKey;

i64 nDummy;

int eType;

u8 *aData;

int nData;

aData = fsPageData(pPg, &nData);

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

assert( iCell<pageGetNRec(aData, nData) );

pKey = pageGetCell(aData, nData, iCell);

eType = *pKey++;

pKey += lsmVarintGet64(pKey, &nDummy);

pKey += lsmVarintGet32(pKey, pnKey);

if( rtIsWrite(eType) ){

pKey += lsmVarintGet64(pKey, &nDummy);

}

*piTopic = rtTopic(eType);

sortedReadData(pSeg, pPg, pKey-aData, *pnKey, (void **)&pKey, pBlob);

return pKey;

}

static int pageGetKeyCopy(

lsm_env *pEnv, /* Environment handle */

Segment *pSeg, /* Segment pPg belongs to */

Page *pPg, /* Page to read from */

int iCell, /* Index of cell on page to read */

int *piTopic, /* OUT: Topic associated with this key */

LsmBlob *pBlob /* If required, use this for dynamic memory */

){

int rc = LSM_OK;

int nKey;

u8 *aKey;

aKey = pageGetKey(pSeg, pPg, iCell, piTopic, &nKey, pBlob);

assert( (void *)aKey!=pBlob->pData || nKey==pBlob->nData );

if( (void *)aKey!=pBlob->pData ){

rc = sortedBlobSet(pEnv, pBlob, aKey, nKey);

}

return rc;

}

static LsmPgno pageGetBtreeRef(Page *pPg, int iKey){

LsmPgno iRef;

u8 *aData;

int nData;

u8 *aCell;

aData = fsPageData(pPg, &nData);

aCell = pageGetCell(aData, nData, iKey);

assert( aCell[0]==0 );

aCell++;

aCell += lsmVarintGet64(aCell, &iRef);

lsmVarintGet64(aCell, &iRef);

assert( iRef>0 );

return iRef;

}

#define GETVARINT64(a, i) (((i)=((u8*)(a))[0])<=240?1:lsmVarintGet64((a), &(i)))

#define GETVARINT32(a, i) (((i)=((u8*)(a))[0])<=240?1:lsmVarintGet32((a), &(i)))

static int pageGetBtreeKey(

Segment *pSeg, /* Segment page pPg belongs to */

Page *pPg,

int iKey,

LsmPgno *piPtr,

int *piTopic,

void **ppKey,

int *pnKey,

LsmBlob *pBlob

){

u8 *aData;

int nData;

u8 *aCell;

int eType;

aData = fsPageData(pPg, &nData);

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

assert( iKey>=0 && iKey<pageGetNRec(aData, nData) );

aCell = pageGetCell(aData, nData, iKey);

eType = *aCell++;

aCell += GETVARINT64(aCell, *piPtr);

if( eType==0 ){

int rc;

LsmPgno iRef; /* Page number of referenced page */

Page *pRef;

aCell += GETVARINT64(aCell, iRef);

rc = lsmFsDbPageGet(lsmPageFS(pPg), pSeg, iRef, &pRef);

if( rc!=LSM_OK ) return rc;

pageGetKeyCopy(lsmPageEnv(pPg), pSeg, pRef, 0, &eType, pBlob);

lsmFsPageRelease(pRef);

*ppKey = pBlob->pData;

*pnKey = pBlob->nData;

}else{

aCell += GETVARINT32(aCell, *pnKey);

*ppKey = aCell;

}

if( piTopic ) *piTopic = rtTopic(eType);

return LSM_OK;

}

static int btreeCursorLoadKey(BtreeCursor *pCsr){

int rc = LSM_OK;

if( pCsr->iPg<0 ){

pCsr->pKey = 0;

pCsr->nKey = 0;

pCsr->eType = 0;

}else{

LsmPgno dummy;

int iPg = pCsr->iPg;

int iCell = pCsr->aPg[iPg].iCell;

while( iCell<0 && (--iPg)>=0 ){

iCell = pCsr->aPg[iPg].iCell-1;

}

if( iPg<0 || iCell<0 ) return LSM_CORRUPT_BKPT;

rc = pageGetBtreeKey(

pCsr->pSeg,

pCsr->aPg[iPg].pPage, iCell,

&dummy, &pCsr->eType, &pCsr->pKey, &pCsr->nKey, &pCsr->blob

);

pCsr->eType |= LSM_SEPARATOR;

}

return rc;

}

static LsmPgno btreeCursorPtr(u8 *aData, int nData, int iCell){

int nCell;

nCell = pageGetNRec(aData, nData);

if( iCell>=nCell ){

return pageGetPtr(aData, nData);

}

return pageGetRecordPtr(aData, nData, iCell);

}

static int btreeCursorNext(BtreeCursor *pCsr){

int rc = LSM_OK;

BtreePg *pPg = &pCsr->aPg[pCsr->iPg];

int nCell;

u8 *aData;

int nData;

assert( pCsr->iPg>=0 );

assert( pCsr->iPg==pCsr->nDepth-1 );

aData = fsPageData(pPg->pPage, &nData);

nCell = pageGetNRec(aData, nData);

assert( pPg->iCell<=nCell );

pPg->iCell++;

if( pPg->iCell==nCell ){

LsmPgno iLoad;

/* Up to parent. */

lsmFsPageRelease(pPg->pPage);

pPg->pPage = 0;

pCsr->iPg--;

while( pCsr->iPg>=0 ){

pPg = &pCsr->aPg[pCsr->iPg];

aData = fsPageData(pPg->pPage, &nData);

if( pPg->iCell<pageGetNRec(aData, nData) ) break;

lsmFsPageRelease(pPg->pPage);

pCsr->iPg--;

}

/* Read the key */

rc = btreeCursorLoadKey(pCsr);

/* Unless the cursor is at EOF, descend to cell -1 (yes, negative one) of

** the left-most most descendent. */

if( pCsr->iPg>=0 ){

pCsr->aPg[pCsr->iPg].iCell++;

iLoad = btreeCursorPtr(aData, nData, pPg->iCell);

do {

Page *pLoad;

pCsr->iPg++;

rc = lsmFsDbPageGet(pCsr->pFS, pCsr->pSeg, iLoad, &pLoad);

pCsr->aPg[pCsr->iPg].pPage = pLoad;

pCsr->aPg[pCsr->iPg].iCell = 0;

if( rc==LSM_OK ){

if( pCsr->iPg==(pCsr->nDepth-1) ) break;

aData = fsPageData(pLoad, &nData);

iLoad = btreeCursorPtr(aData, nData, 0);

}

}while( rc==LSM_OK && pCsr->iPg<(pCsr->nDepth-1) );

pCsr->aPg[pCsr->iPg].iCell = -1;

}

}else{

rc = btreeCursorLoadKey(pCsr);

}

if( rc==LSM_OK && pCsr->iPg>=0 ){

aData = fsPageData(pCsr->aPg[pCsr->iPg].pPage, &nData);

pCsr->iPtr = btreeCursorPtr(aData, nData, pCsr->aPg[pCsr->iPg].iCell+1);

}

return rc;

}

static void btreeCursorFree(BtreeCursor *pCsr){

if( pCsr ){

int i;

lsm_env *pEnv = lsmFsEnv(pCsr->pFS);

for(i=0; i<=pCsr->iPg; i++){

lsmFsPageRelease(pCsr->aPg[i].pPage);

}

sortedBlobFree(&pCsr->blob);

lsmFree(pEnv, pCsr->aPg);

lsmFree(pEnv, pCsr);

}

}

static int btreeCursorFirst(BtreeCursor *pCsr){

int rc;

Page *pPg = 0;

FileSystem *pFS = pCsr->pFS;

LsmPgno iPg = pCsr->pSeg->iRoot;

do {

rc = lsmFsDbPageGet(pFS, pCsr->pSeg, iPg, &pPg);

assert( (rc==LSM_OK)==(pPg!=0) );

if( rc==LSM_OK ){

u8 *aData;

int nData;

int flags;

aData = fsPageData(pPg, &nData);

flags = pageGetFlags(aData, nData);

if( (flags & SEGMENT_BTREE_FLAG)==0 ) break;

if( (pCsr->nDepth % 8)==0 ){

int nNew = pCsr->nDepth + 8;

pCsr->aPg = (BtreePg *)lsmReallocOrFreeRc(

lsmFsEnv(pFS), pCsr->aPg, sizeof(BtreePg) * nNew, &rc

);

if( rc==LSM_OK ){

memset(&pCsr->aPg[pCsr->nDepth], 0, sizeof(BtreePg) * 8);

}

}

if( rc==LSM_OK ){

assert( pCsr->aPg[pCsr->nDepth].iCell==0 );

pCsr->aPg[pCsr->nDepth].pPage = pPg;

pCsr->nDepth++;

iPg = pageGetRecordPtr(aData, nData, 0);

}

}

}while( rc==LSM_OK );

lsmFsPageRelease(pPg);

pCsr->iPg = pCsr->nDepth-1;

if( rc==LSM_OK && pCsr->nDepth ){

pCsr->aPg[pCsr->iPg].iCell = -1;

rc = btreeCursorNext(pCsr);

}

return rc;

}

static void btreeCursorPosition(BtreeCursor *pCsr, MergeInput *p){

if( pCsr->iPg>=0 ){

p->iPg = lsmFsPageNumber(pCsr->aPg[pCsr->iPg].pPage);

p->iCell = ((pCsr->aPg[pCsr->iPg].iCell + 1) << 8) + pCsr->nDepth;

}else{

p->iPg = 0;

p->iCell = 0;

}

}

static void btreeCursorSplitkey(BtreeCursor *pCsr, MergeInput *p){

int iCell = pCsr->aPg[pCsr->iPg].iCell;

if( iCell>=0 ){

p->iCell = iCell;

p->iPg = lsmFsPageNumber(pCsr->aPg[pCsr->iPg].pPage);

}else{

int i;

for(i=pCsr->iPg-1; i>=0; i--){

if( pCsr->aPg[i].iCell>0 ) break;

}

assert( i>=0 );

p->iCell = pCsr->aPg[i].iCell-1;

p->iPg = lsmFsPageNumber(pCsr->aPg[i].pPage);

}

}

static int sortedKeyCompare(

int (*xCmp)(void *, int, void *, int),

int iLhsTopic, void *pLhsKey, int nLhsKey,

int iRhsTopic, void *pRhsKey, int nRhsKey

){

int res = iLhsTopic - iRhsTopic;

if( res==0 ){

res = xCmp(pLhsKey, nLhsKey, pRhsKey, nRhsKey);

}

return res;

}

static int btreeCursorRestore(

BtreeCursor *pCsr,

int (*xCmp)(void *, int, void *, int),

MergeInput *p

){

int rc = LSM_OK;

if( p->iPg ){

lsm_env *pEnv = lsmFsEnv(pCsr->pFS);

int iCell; /* Current cell number on leaf page */

LsmPgno iLeaf; /* Page number of current leaf page */

int nDepth; /* Depth of b-tree structure */

Segment *pSeg = pCsr->pSeg;

/* Decode the MergeInput structure */

iLeaf = p->iPg;

nDepth = (p->iCell & 0x00FF);

iCell = (p->iCell >> 8) - 1;

/* Allocate the BtreeCursor.aPg[] array */

assert( pCsr->aPg==0 );

pCsr->aPg = (BtreePg *)lsmMallocZeroRc(pEnv, sizeof(BtreePg) * nDepth, &rc);

/* Populate the last entry of the aPg[] array */

if( rc==LSM_OK ){

Page **pp = &pCsr->aPg[nDepth-1].pPage;

pCsr->iPg = nDepth-1;

pCsr->nDepth = nDepth;

pCsr->aPg[pCsr->iPg].iCell = iCell;

rc = lsmFsDbPageGet(pCsr->pFS, pSeg, iLeaf, pp);

}

/* Populate any other aPg[] array entries */

if( rc==LSM_OK && nDepth>1 ){

LsmBlob blob = {0,0,0};

void *pSeek;

int nSeek;

int iTopicSeek;

int iPg = 0;

LsmPgno iLoad = pSeg->iRoot;

Page *pPg = pCsr->aPg[nDepth-1].pPage;

if( pageObjGetNRec(pPg)==0 ){

/* This can happen when pPg is the right-most leaf in the b-tree.

** In this case, set the iTopicSeek/pSeek/nSeek key to a value

** greater than any real key. */

assert( iCell==-1 );

iTopicSeek = 1000;

pSeek = 0;

nSeek = 0;

}else{

LsmPgno dummy;

rc = pageGetBtreeKey(pSeg, pPg,

0, &dummy, &iTopicSeek, &pSeek, &nSeek, &pCsr->blob

);

}

do {

Page *pPg2;

rc = lsmFsDbPageGet(pCsr->pFS, pSeg, iLoad, &pPg2);

assert( rc==LSM_OK || pPg2==0 );

if( rc==LSM_OK ){

u8 *aData; /* Buffer containing page data */

int nData; /* Size of aData[] in bytes */

int iMin;

int iMax;

int iCell2;

aData = fsPageData(pPg2, &nData);

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

iLoad = pageGetPtr(aData, nData);

iCell2 = pageGetNRec(aData, nData);

iMax = iCell2-1;

iMin = 0;

while( iMax>=iMin ){

int iTry = (iMin+iMax)/2;

void *pKey; int nKey; /* Key for cell iTry */

int iTopic; /* Topic for key pKeyT/nKeyT */

LsmPgno iPtr; /* Pointer for cell iTry */

int res; /* (pSeek - pKeyT) */

rc = pageGetBtreeKey(

pSeg, pPg2, iTry, &iPtr, &iTopic, &pKey, &nKey, &blob

);

if( rc!=LSM_OK ) break;

res = sortedKeyCompare(

xCmp, iTopicSeek, pSeek, nSeek, iTopic, pKey, nKey

);

assert( res!=0 );

if( res<0 ){

iLoad = iPtr;

iCell2 = iTry;

iMax = iTry-1;

}else{

iMin = iTry+1;

}

}

pCsr->aPg[iPg].pPage = pPg2;

pCsr->aPg[iPg].iCell = iCell2;

iPg++;

assert( iPg!=nDepth-1

|| lsmFsRedirectPage(pCsr->pFS, pSeg->pRedirect, iLoad)==iLeaf

);

}

}while( rc==LSM_OK && iPg<(nDepth-1) );

sortedBlobFree(&blob);

}

/* Load the current key and pointer */

if( rc==LSM_OK ){

BtreePg *pBtreePg;

u8 *aData;

int nData;

pBtreePg = &pCsr->aPg[pCsr->iPg];

aData = fsPageData(pBtreePg->pPage, &nData);

pCsr->iPtr = btreeCursorPtr(aData, nData, pBtreePg->iCell+1);

if( pBtreePg->iCell<0 ){

LsmPgno dummy;

int i;

for(i=pCsr->iPg-1; i>=0; i--){

if( pCsr->aPg[i].iCell>0 ) break;

}

assert( i>=0 );

rc = pageGetBtreeKey(pSeg,

pCsr->aPg[i].pPage, pCsr->aPg[i].iCell-1,

&dummy, &pCsr->eType, &pCsr->pKey, &pCsr->nKey, &pCsr->blob

);

pCsr->eType |= LSM_SEPARATOR;

}else{

rc = btreeCursorLoadKey(pCsr);

}

}

}

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;

}

static void segmentPtrSetPage(SegmentPtr *pPtr, Page *pNext){

lsmFsPageRelease(pPtr->pPg);

if( pNext ){