** http://www.hwaci.com/drh/
**
*************************************************************************
-** $Id: btree.c,v 1.11 2001/06/08 00:21:52 drh Exp $
+** $Id: btree.c,v 1.12 2001/06/10 19:56:59 drh Exp $
**
** This file implements a external (disk-based) database using BTrees.
** For a detailed discussion of BTrees, refer to
/*
** Primitive data types. u32 must be 4 bytes and u16 must be 2 bytes.
+** The uptr type must be big enough to hold a pointer.
** Change these typedefs when porting to new architectures.
*/
+typedef unsigned int uptr;
typedef unsigned int u32;
typedef unsigned short int u16;
typedef unsigned char u8;
** structure.
**
** PageHdr.firstFree is 0 if there is no free space on this page.
-** Otherwise, PageHdr.firstFree is the index in MemPage.aDisk[] of a
+** Otherwise, PageHdr.firstFree is the index in MemPage.u.aDisk[] of a
** FreeBlk structure that describes the first block of free space.
** All free space is defined by a linked list of FreeBlk structures.
**
** Data is stored in a linked list of Cell structures. PageHdr.firstCell
-** is the index into MemPage.aDisk[] of the first cell on the page. The
+** is the index into MemPage.u.aDisk[] of the first cell on the page. The
** Cells are kept in sorted order.
**
** A Cell contains all information about a database entry and a pointer
*/
struct PageHdr {
Pgno rightChild; /* Child page that comes after all cells on this page */
- u16 firstCell; /* Index in MemPage.aDisk[] of the first cell */
- u16 firstFree; /* Index in MemPage.aDisk[] of the first free block */
+ u16 firstCell; /* Index in MemPage.u.aDisk[] of the first cell */
+ u16 firstFree; /* Index in MemPage.u.aDisk[] of the first free block */
};
/*
struct CellHdr {
Pgno leftChild; /* Child page that comes before this cell */
u16 nKey; /* Number of bytes in the key */
- u16 iNext; /* Index in MemPage.aDisk[] of next cell in sorted order */
+ u16 iNext; /* Index in MemPage.u.aDisk[] of next cell in sorted order */
u32 nData; /* Number of bytes of data */
}
*/
struct FreeBlk {
u16 iSize; /* Number of bytes in this block of free space */
- u16 iNext; /* Index in MemPage.aDisk[] of the next free block */
+ u16 iNext; /* Index in MemPage.u.aDisk[] of the next free block */
};
/*
-** Number of bytes on a single overflow page.
+** The number of bytes of payload that will fit on a single overflow page.
*/
#define OVERFLOW_SIZE (SQLITE_PAGE_SIZE-sizeof(Pgno))
** pages.
*/
struct OverflowPage {
- Pgno next;
+ Pgno iNext;
char aPayload[OVERFLOW_SIZE];
};
/*
** For every page in the database file, an instance of the following structure
-** is stored in memory. The aDisk[] array contains the raw bits read from
+** is stored in memory. The u.aDisk[] array contains the raw bits read from
** the disk. The rest is auxiliary information that held in memory only. The
** auxiliary info is only valid for regular database pages - it is not
** used for overflow pages and pages on the freelist.
**
** Of particular interest in the auxiliary info is the apCell[] entry. Each
-** apCell[] entry is a pointer to a Cell structure in aDisk[]. The cells are
+** apCell[] entry is a pointer to a Cell structure in u.aDisk[]. The cells are
** put in this array so that they can be accessed in constant time, rather
** than in linear time which would be needed if we had to walk the linked
** list on every access.
**
+** Note that apCell[] contains enough space to hold up to two more Cells
+** than can possibly fit on one page. In the steady state, every apCell[]
+** points to memory inside u.aDisk[]. But in the middle of an insert
+** operation, some apCell[] entries may temporarily point to data space
+** outside of u.aDisk[]. This is a transient situation that is quickly
+** resolved. But while it is happening, it is possible for a database
+** page to hold as many as two more cells than it might otherwise hold.
+** The extra too entries in apCell[] are an allowance for this situation.
+**
** The pParent field points back to the parent page. This allows us to
** walk up the BTree from any leaf to the root. Care must be taken to
** unref() the parent page pointer when this page is no longer referenced.
** The pageDestructor() routine handles that chore.
*/
struct MemPage {
- char aDisk[SQLITE_PAGE_SIZE]; /* Page data stored on disk */
+ union {
+ char aDisk[SQLITE_PAGE_SIZE]; /* Page data stored on disk */
+ PageHdr hdr; /* Overlay page header */
+ } u;
int isInit; /* True if auxiliary data is initialized */
MemPage *pParent; /* The parent of this page. NULL for root */
- int nFree; /* Number of free bytes in aDisk[] */
+ int nFree; /* Number of free bytes in u.aDisk[] */
int nCell; /* Number of entries on this page */
- Cell *apCell[MX_CELL+1]; /* All data entires in sorted order */
+ int isOverfull; /* Some apCell[] points outside u.aDisk[] */
+ Cell *apCell[MX_CELL+2]; /* All data entires in sorted order */
}
/*
*/
struct BtCursor {
Btree *pBt; /* The Btree to which this cursor belongs */
- BtCursor *pPrev, *pNext; /* List of all cursors */
+ BtCursor *pNext, *pPrev; /* Forms a linked list of all cursors */
Pgno pgnoRoot; /* The root page of this tree */
MemPage *pPage; /* Page that contains the entry */
u16 idx; /* Index of the entry in pPage->apCell[] */
** into one big FreeBlk at the end of the page.
*/
static void defragmentPage(MemPage *pPage){
- int pc;
- int i, n;
+ int pc, i, n;
FreeBlk *pFBlk;
char newPage[SQLITE_PAGE_SIZE];
pc = sizeof(PageHdr);
- ((PageHdr*)pPage)->firstCell = pc;
- memcpy(newPage, pPage->aDisk, pc);
+ pPage->u.hdr.firstCell = pc;
+ memcpy(newPage, pPage->u.aDisk, pc);
for(i=0; i<pPage->nCell; i++){
Cell *pCell = &pPage->apCell[i];
n = cellSize(pCell);
- pCell->h.iNext = i<pPage->nCell ? pc + n : 0;
+ pCell->h.iNext = i<pPage->nCell-1 ? pc + n : 0;
memcpy(&newPage[pc], pCell, n);
- pPage->apCell[i] = (Cell*)&pPage->aDisk[pc];
+ pPage->apCell[i] = (Cell*)&pPage->u.aDisk[pc];
pc += n;
}
assert( pPage->nFree==SQLITE_PAGE_SIZE-pc );
- memcpy(pPage->aDisk, newPage, pc);
- pFBlk = &pPage->aDisk[pc];
+ memcpy(pPage->u.aDisk, newPage, pc);
+ pFBlk = &pPage->u.aDisk[pc];
pFBlk->iSize = SQLITE_PAGE_SIZE - pc;
pFBlk->iNext = 0;
- ((PageHdr*)pPage)->firstFree = pc;
+ pPage->u.hdr.firstFree = pc;
memset(&pFBlk[1], 0, SQLITE_PAGE_SIZE - pc - sizeof(FreeBlk));
}
** Allocate nByte bytes of space on a page. nByte must be a
** multiple of 4.
**
-** Return the index into pPage->aDisk[] of the first byte of
+** Return the index into pPage->u.aDisk[] of the first byte of
** the new allocation. Or return 0 if there is not enough free
** space on the page to satisfy the allocation request.
**
int start;
assert( nByte==ROUNDUP(nByte) );
- if( pPage->nFree<nByte ) return 0;
- pIdx = &((PageHdr*)pPage)->firstFree;
- p = (FreeBlk*)&pPage->aDisk[*pIdx];
+ if( pPage->nFree<nByte || pPage->isOverfull ) return 0;
+ pIdx = &pPage->u.hdr.firstFree;
+ p = (FreeBlk*)&pPage->u.aDisk[*pIdx];
while( p->iSize<nByte ){
if( p->iNext==0 ){
defragmentPage(pPage);
- pIdx = &((PageHdr*)pPage)->firstFree;
+ pIdx = &pPage->u.hdr.firstFree;
}else{
pIdx = &p->iNext;
}
- p = (FreeBlk*)&pPage->aDisk[*pIdx];
+ p = (FreeBlk*)&pPage->u.aDisk[*pIdx];
}
if( p->iSize==nByte ){
start = *pIdx;
*pIdx = p->iNext;
}else{
start = *pIdx;
- FreeBlk *pNew = (FreeBlk*)&pPage->aDisk[start + nByte];
+ FreeBlk *pNew = (FreeBlk*)&pPage->u.aDisk[start + nByte];
pNew->iNext = p->iNext;
pNew->iSize = p->iSize - nByte;
*pIdx = start + nByte;
}
/*
-** Return a section of the MemPage.aDisk[] to the freelist.
-** The first byte of the new free block is pPage->aDisk[start]
-** and the size of the block is "size" bytes.
+** Return a section of the MemPage.u.aDisk[] to the freelist.
+** The first byte of the new free block is pPage->u.aDisk[start]
+** and the size of the block is "size" bytes. Size must be
+** a multiple of 4.
**
** Most of the effort here is involved in coalesing adjacent
** free blocks into a single big free block.
assert( size == ROUNDUP(size) );
assert( start == ROUNDUP(start) );
- pIdx = &((PageHdr*)pPage)->firstFree;
+ pIdx = &pPage->u.hdr.firstFree;
idx = *pIdx;
while( idx!=0 && idx<start ){
- pFBlk = (FreeBlk*)&pPage->aDisk[idx];
+ pFBlk = (FreeBlk*)&pPage->u.aDisk[idx];
if( idx + pFBlk->iSize == start ){
pFBlk->iSize += size;
if( idx + pFBlk->iSize == pFBlk->iNext ){
- pNext = (FreeBlk*)&pPage->aDisk[pFblk->iNext];
+ pNext = (FreeBlk*)&pPage->u.aDisk[pFblk->iNext];
pFBlk->iSize += pNext->iSize;
pFBlk->iNext = pNext->iNext;
}
pIdx = &pFBlk->iNext;
idx = *pIdx;
}
- pNew = (FreeBlk*)&pPage->aDisk[start];
+ pNew = (FreeBlk*)&pPage->u.aDisk[start];
if( idx != end ){
pNew->iSize = size;
pNew->iNext = idx;
}else{
- pNext = (FreeBlk*)&pPage->aDisk[idx];
+ pNext = (FreeBlk*)&pPage->u.aDisk[idx];
pNew->iSize = size + pNext->iSize;
pNew->iNext = pNext->iNext;
}
** we failed to detect any corruption.
*/
static int initPage(MemPage *pPage, Pgno pgnoThis, MemPage *pParent){
- int idx; /* An index into pPage->aDisk[] */
- Cell *pCell; /* A pointer to a Cell in pPage->aDisk[] */
- FreeBlk *pFBlk; /* A pointer to a free block in pPage->aDisk[] */
+ int idx; /* An index into pPage->u.aDisk[] */
+ Cell *pCell; /* A pointer to a Cell in pPage->u.aDisk[] */
+ FreeBlk *pFBlk; /* A pointer to a free block in pPage->u.aDisk[] */
int sz; /* The size of a Cell in bytes */
int freeSpace; /* Amount of free space on the page */
pPage->isInit = 1;
pPage->nCell = 0;
freeSpace = SQLITE_PAGE_SIZE - sizeof(PageHdr);
- idx = ((PageHdr*)pPage)->firstCell;
+ idx = pPage->u.hdr.firstCell;
while( idx!=0 ){
if( idx>SQLITE_PAGE_SIZE-MN_CELL_SIZE ) goto page_format_error;
if( idx<sizeof(PageHdr) ) goto page_format_error;
- pCell = (Cell*)&pPage->aDisk[idx];
+ pCell = (Cell*)&pPage->u.aDisk[idx];
sz = cellSize(pCell);
if( idx+sz > SQLITE_PAGE_SIZE ) goto page_format_error;
freeSpace -= sz;
idx = pCell->h.iNext;
}
pPage->nFree = 0;
- idx = ((PageHdr*)pPage)->firstFree;
+ idx = pPage->u.hdr.firstFree;
while( idx!=0 ){
if( idx>SQLITE_PAGE_SIZE-sizeof(FreeBlk) ) goto page_format_error;
if( idx<sizeof(PageHdr) ) goto page_format_error;
- pFBlk = (FreeBlk*)&pPage->aDisk[idx];
+ pFBlk = (FreeBlk*)&pPage->u.aDisk[idx];
pPage->nFree += pFBlk->iSize;
if( pFBlk->iNext <= idx ) goto page_format_error;
idx = pFBlk->iNext;
return SQLITE_CORRUPT;
}
-/*
-** Recompute the MemPage.apCell[], MemPage.nCell, and MemPage.nFree parameters
-** for a cell after the MemPage.aDisk[] content has be changed significantly.
-**
-** The computation here is similar to initPage() except that in this case
-** the MemPage.aDisk[] field has been set up internally (instead of
-** having been read from disk) so we do not need to do as much error
-** checking.
-*/
-static void reinitPage(MemPage *pPage){
- Cell *pCell;
-
- pPage->nCell = 0;
- idx = ((PageHdr*)pPage)->firstCell;
- while( idx!=0 ){
- pCell = (Cell*)&pPage->aDisk[idx];
- sz = cellSize(pCell);
- pPage->apCell[pPage->nCell++] = pCell;
- idx = pCell->h.iNext;
- }
- pPage->nFree = 0;
- idx = ((PageHdr*)pPage)->firstFree;
- while( idx!=0 ){
- pFBlk = (FreeBlk*)&pPage->aDisk[idx];
- pPage->nFree += pFBlk->iSize;
- idx = pFBlk->iNext;
- }
- return SQLITE_OK;
-}
-
/*
** Set up a raw page so that it looks like a database page holding
** no entries.
PageHdr *pHdr;
FreeBlk *pFBlk;
memset(pPage, 0, SQLITE_PAGE_SIZE);
- pHdr = (PageHdr*)pPage;
+ pHdr = &pPage->u.hdr;
pHdr->firstCell = 0;
pHdr->firstFree = sizeof(*pHdr);
pFBlk = (FreeBlk*)&pHdr[1];
if( rc!=SQLITE_OK ){
goto create_cursor_exception;
}
- pCur->pPrev = 0;
+ pCur->pBt = pBt;
+ pCur->idx = 0;
pCur->pNext = pBt->pCursor;
if( pCur->pNext ){
pCur->pNext->pPrev = pCur;
}
+ pCur->pPrev = 0;
pBt->pCursor = pCur;
- pCur->pBt = pBt;
- pCur->idx = 0;
*ppCur = pCur;
return SQLITE_OK;
** Make a temporary cursor by filling in the fields of pTempCur.
** The temporary cursor is not on the cursor list for the Btree.
*/
-static void CreateTemporaryCursor(BtCursor *pCur, BtCursor *pTempCur){
+static void getTempCursor(BtCursor *pCur, BtCursor *pTempCur){
memcpy(pTempCur, pCur, sizeof(*pCur));
pTempCur->pNext = 0;
pTempCur->pPrev = 0;
** Delete a temporary cursor such as was made by the CreateTemporaryCursor()
** function above.
*/
-static void DestroyTemporaryCursor(BeCursor *pCur){
+static void releaseTempCursor(BtCursor *pCur){
sqlitepager_unref(pCur->pPage);
}
if( rc!=0 ){
return rc;
}
- nextPage = pOvfl->next;
+ nextPage = pOvfl->iNext;
if( offset<OVERFLOW_SIZE ){
int a = amt;
if( a + offset > OVERFLOW_SIZE ){
if( rc ){
return rc;
}
- nextPage = pOvfl->next;
+ nextPage = pOvfl->iNext;
n = nKey;
if( n>OVERFLOW_SIZE ){
n = OVERFLOW_SIZE;
}
assert( lwr==upr+1 );
if( lwr>=pPage->nCell ){
- chldPg = ((PageHdr*)pPage)->rightChild;
+ chldPg = pPage->u.hdr.rightChild;
}else{
chldPg = pPage->apCell[lwr]->h.leftChild;
}
}
pCur->idx++;
if( pCur->idx>=pCur->pPage->nCell ){
- if( ((PageHdr*)pPage)->rightChild ){
- rc = moveToChild(pCur, ((PageHdr*)pPage)->rightChild);
+ if( pPage->u.hdr.rightChild ){
+ rc = moveToChild(pCur, pPage->u.hdr.rightChild);
if( rc ) return rc;
rc = moveToLeftmost(pCur);
if( rc ) return rc;
sqlitepager_unref(pOvfl);
return rc;
}
- pPage1->freeList = pOvfl->next;
+ pPage1->freeList = pOvfl->iNext;
*ppPage = (MemPage*)pOvfl;
}else{
*pPgno = sqlitepager_pagecount(pBt->pPager);
if( needOvflUnref ) sqlitepager_unref(pOvfl);
return rc;
}
- pOvfl->next = pPage1->freeList;
+ pOvfl->iNext = pPage1->freeList;
pPage1->freeList = pgno;
memset(pOvfl->aPayload, 0, OVERFLOW_SIZE);
pPage->isInit = 0;
while( ovfl ){
rc = sqlitepager_get(pPager, ovfl, &pOvfl);
if( rc ) return rc;
- nextOvfl = pOvfl->next;
+ nextOvfl = pOvfl->iNext;
rc = freePage(pBt, pOvfl, ovfl);
if( rc ) return rc;
ovfl = nextOvfl;
}
spaceLeft = OVERFLOW_SIZE;
pSpace = pOvfl->aPayload;
- pNextPg = &pOvfl->next;
+ pNextPg = &pOvfl->iNext;
}
n = nPayload;
if( n>spaceLeft ) n = spaceLeft;
for(i=0; i<pPage->nCell; i++){
reparentPage(pPager, pPage->apCell[i]->leftChild, pPage);
}
- reparentPage(pPager, ((PageHdr*)pPage)->rightChild, pPage);
+ reparentPage(pPager, pPage->u.hdr.rightChild, pPage);
+}
+
+/*
+** Remove the i-th cell from pPage. This routine effects pPage only.
+** The cell content is not freed or deallocated. It is assumed that
+** the cell content has been copied someplace else. This routine just
+** removes the reference to the cell from pPage.
+**
+** "sz" must be the number of bytes in the cell.
+**
+** Do not bother maintaining the integrity of the linked list of Cells.
+** Only pPage->apCell[] is important. The relinkCellList() routine
+** will be called soon after this routine in order to rebuild the
+** linked list.
+*/
+static void dropCell(MemPage *pPage, int i, int sz){
+ int j;
+ assert( i>=0 && i<pPage->nCell );
+ assert( sz==cellSize(pPage->apCell[i]);
+ freeSpace(pPage, idx, sz);
+ for(j=i, j<pPage->nCell-2; j++){
+ pPage->apCell[j] = pPage->apCell[j+1];
+ }
+ pPage->nCell--;
+}
+
+/*
+** Insert a new cell on pPage at cell index "i". pCell points to the
+** content of the cell.
+**
+** If the cell content will fit on the page, then put it there. If it
+** will not fit, then just make pPage->apCell[i] point to the content
+** and set pPage->isOverfull.
+**
+** Do not bother maintaining the integrity of the linked list of Cells.
+** Only pPage->apCell[] is important. The relinkCellList() routine
+** will be called soon after this routine in order to rebuild the
+** linked list.
+*/
+static void insertCell(MemPage *pPage, int i, Cell *pCell, int sz){
+ int idx, j;
+ assert( i>=0 && i<=pPage->nCell );
+ assert( sz==cellSize(pCell) );
+ for(j=pPage->nCell; j>i; j--){
+ pPage->apCell[j] = pPage->apCell[j-1];
+ }
+ pPage->nCell++;
+ idx = allocateSpace(pPage, sz);
+ if( idx<=0 ){
+ pPage->isOverfull = 1;
+ pPage->apCell[i] = pCell;
+ }else{
+ memcpy(&pPage->u.aDisk[idx], pCell, sz);
+ pPage->apCell[i] = (Cell*)&pPage->u.aDisk[idx]);
+ }
+}
+
+/*
+** Rebuild the linked list of cells on a page so that the cells
+** occur in the order specified by pPage->apCell[]. Invoke this
+** routine once to repair damage after one or more invocations
+** of either insertCell() or dropCell().
+*/
+static void relinkCellList(MemPage *pPage){
+ int i;
+ u16 *pIdx;
+ pIdx = &pPage->u.hdr.firstCell;
+ for(i=0; i<pPage->nCell; i++){
+ int idx = ((uptr)pPage->apCell[i]) - (uptr)pPage;
+ *pIdx = idx;
+ pIdx = &pPage->apCell[i]->h.iNext;
+ }
+ *pIdx = 0;
+}
+
+/*
+** Make a copy of the contents of pFrom into pTo. The pFrom->apCell[]
+** pointers that point intto pFrom->u.aDisk[] must be adjusted to point
+** intto pTo->u.aDisk[] instead. But some pFrom->apCell[] entries might
+** not point to pFrom->u.aDisk[]. Those are unchanged.
+*/
+static void copyPage(MemPage *pTo, MemPage *pFrom){
+ uptr from, to;
+ int i;
+ memcpy(pTo->u.aDisk, pFrom->u.aDisk, SQLITE_PAGE_SIZE);
+ pTo->pParent = pFrom->pParent;
+ pTo->isInit = 1;
+ pTo->nCell = pFrom->nCell;
+ pTo->nFree = pFrom->nFree;
+ pTo->isOverfull = pFrom->isOverfull;
+ to = (unsigned int)pTo;
+ from = (unsigned int)pFrom;
+ for(i=0; i<pTo->nCell; i++){
+ uptr addr = (uptr)(pFrom->apCell[i]);
+ if( addr>from && addr<from+SQLITE_PAGE_SIZE ){
+ *((uptr*)&pTo->apCell[i]) = addr + to - from;
+ }
+ }
}
/*
** This routine redistributes Cells on pPage and up to two siblings
** of pPage so that all pages have about the same amount of free space.
-** Usually one siblings on either side of pPage are used in the repack,
+** Usually one sibling on either side of pPage is used in the balancing,
** though both siblings might come from one side if pPage is the first
** or last child of its parent. If pPage has fewer than two siblings
** (something which can only happen if pPage is the root page or a
-** child of root) then all siblings participate in the repack.
+** child of root) then all available siblings participate in the balancing.
**
** The number of siblings of pPage might be increased or decreased by
** one in order to keep all pages between 2/3 and completely full. If
** or decreased by one, as necessary, to keep the root page from being
** overfull or empty.
**
+** This routine calls relinkCellList() on its input page regardless of
+** whether or not it does any real balancing. Client routines will typically
+** invoke insertCell() or dropCell() before calling this routine, so we
+** need to call relinkCellList() to clean up the mess that those other
+** routines left behind.
+**
+** pCur is left pointing to the same cell as when this routine was called
+** event if that cell gets moved to a different page. pCur may be NULL.
+**
** Note that when this routine is called, some of the Cells on pPage
-** might not actually be stored in pPage->aDisk[]. This can happen
+** might not actually be stored in pPage->u.aDisk[]. This can happen
** if the page is overfull. Part of the job of this routine is to
-** make sure all Cells for pPage once again fit in pPage->aDisk[].
+** make sure all Cells for pPage once again fit in pPage->u.aDisk[].
+**
+** If this routine fails for any reason, it means the database may have
+** been left in a corrupted state and should be rolled back.
*/
-static int repack(Btree *pBt, MemPage *pPage){
+static int balance(Btree *pBt, MemPage *pPage, BtCursor *pCur){
MemPage *pParent; /* The parent of pPage */
- MemPage *apOld[3]; /* pPage and up to two siblings before repack */
+ MemPage *apOld[3]; /* pPage and up to two siblings */
Pgno pgnoOld[3]; /* Page numbers for each page in apOld[] */
- MemPage *apNew[4]; /* pPage and up to 3 siblings after repack */
+ MemPage *apNew[4]; /* pPage and up to 3 siblings after balancing */
+ Pgno pgnoNew[4]; /* Page numbers for each page in apNew[] */
int idxDiv[3]; /* Indices of divider cells in pParent */
Cell *apDiv[3]; /* Divider cells in pParent */
int nCell; /* Number of cells in apCell[] */
int nNew; /* Number of pages in apNew[] */
int perPage; /* Approximate number of bytes per page */
int nDiv; /* Number of cells in apDiv[] */
- Cell *apCell[MX_CELL*3+5]; /* All cells from pages being repacked */
- int unrefPage = 0; /* If true, then unref pPage when done */
-
- /*
- ** Early out if no repacking is needed.
+ int i, j, k; /* Loop counters */
+ int idx; /* Index of pPage in pParent->apCell[] */
+ int nxDiv; /* Next divider slot in pParent->apCell[] */
+ int rc; /* The return code */
+ int iCur; /* apCell[iCur] is the cell of the cursor */
+ int usedPerPage; /* Memory needed for each page */
+ int freePerPage; /* Average free space per page */
+ Cell *apCell[MX_CELL*3+5]; /* All cells from pages being balanceed */
+ int szCell[MX_CELL*3+5]; /* Local size of all cells */
+ Cell aTemp[2]; /* Temporary holding area for apDiv[] */
+ MemPage aOld[3]; /* Temporary copies of pPage and its siblings */
+
+ /*
+ ** Return without doing any work if pPage is neither overfull nor
+ ** underfull.
*/
- if( pPage->nFree>=0 && pPage->nFree<SQLITE_PAGE_SIZE/2 ){
+ if( !pPage->isOverfull && pPage->nFree<SQLITE_PAGE_SIZE/2 ){
+ relinkCellList(pPage);
return SQLITE_OK;
}
/*
- ** Find the parent of the page to be repacked.
- */
- pParent = pPage->pParent;
-
- /*
- ** If there is no parent, it means the page is the root page.
+ ** Find the parent of the page to be balanceed.
+ ** If there is no parent, it means this page is the root page and
** special rules apply.
*/
+ pParent = pPage->pParent;
if( pParent==0 ){
Pgno pgnoChild;
Page *pChild;
if( pPage->nCell==0 ){
- if( ((PageHdr*)pPage)->rightChild ){
- /* The root page is under full. Copy the one child page
+ if( pPage->u.hdr.rightChild ){
+ /*
+ ** The root page is empty. Copy the one child page
** into the root page and return. This reduces the depth
** of the BTree by one.
*/
- pgnoChild = ((PageHdr*)pPage->rightChild;
+ rc = sqlitepager_write(pPage);
+ if( rc ) return rc;
+ pgnoChild = pPage->u.hdr.rightChild;
rc = sqlitepager_get(pBt, pgnoChild, &pChild);
if( rc ) return rc;
memcpy(pPage, pChild, SQLITE_PAGE_SIZE);
}
return SQLITE_OK;
}
- if( pPage->nFree>=0 ){
+ if( !pPage->isOverfull ){
/* It is OK for the root page to be less than half full.
*/
+ relinkCellList(pPage);
return SQLITE_OK;
}
- /* If we get to here, it means the root page is over full.
+ /*
+ ** If we get to here, it means the root page is overfull.
** When this happens, Create a new child page and copy the
** contents of the root into the child. Then make the root
- ** page and empty page with rightChild pointing to the new
+ ** page an empty page with rightChild pointing to the new
** child. Then fall thru to the code below which will cause
** the overfull child page to be split.
*/
+ rc = sqlitepager_write(pPage);
+ if( rc ) return rc;
rc = allocatePage(pBt, &pChild, &pgnoChild);
if( rc ) return rc;
- memcpy(pChild, pPage, SQLITE_PAGE_SIZE);
- for(i=0; i<pPage->nCell; i++){
- if( pPage->apCell[i]>(Cell*)pPage && pPage->apCell[i]<(Cell*)&pPage[1] ){
- int offset = (int)pPage->apCell[i] - (int)pPage;
- pChild->apCell[i] = (Cell*)((int)pChild + offset);
- }else{
- pChild->apCell[i] = pPage->apCell[i];
- }
+ copyPage(pChild, pPage);
+ pChild->pParent = pPage;
+ pChild->isOverfull = 1;
+ if( pCur ){
+ sqlitepager_ref(pChild);
+ sqlitepager_unref(pCur->pPage);
+ pCur->pPage = pChild;
}
- pChild->isInit = 1;
- pChild->nCell = pPage->nCell;
- pChild->nFree = pPage->nFree;
- /* reparentChildPages(pBt->pPager, pChild); */
zeroPage(pPage);
- ((PageHdr*)pPage)->rightChild = pgnoChild;
+ pPage->u.hdr.rightChild = pgnoChild;
pParent = pPage;
pPage = pChild;
- unrefPage = 1;
+ }else{
+ rc = sqlitepager_write(pPage);
+ if( rc ) return rc;
}
-
+
/*
- ** Find the Cell in the parent page that refers to the page
- ** to be repacked.
+ ** Find the Cell in the parent page whose h.leftChild points back
+ ** to pPage. The "idx" variable is the index of that cell. If pPage
+ ** is the rightmost child of pParent then set idx to pParent->nCell
*/
idx = -1;
pgno = sqlitepager_pagenumber(pPage);
break;
}
}
- if( idx<0 && ((PageHdr*)pPage)->rightChild==pgno ){
+ if( idx<0 && pPage->u.hdr.rightChild==pgno ){
idx = pPage->nCell;
}
if( idx<0 ){
- rc = SQLITE_CORRUPT;
- goto end_of_repack;
+ return SQLITE_CORRUPT;
}
/*
- ** Get sibling pages and their dividers
+ ** Initialize variables so that it will be safe to jump
+ ** directory to balance_cleanup at any moment.
*/
- if( idx==pPage->nCell ){
- idx -= 2;
+ nOld = nNew = 0;
+ sqlitepager_ref(pParent);
+
+ /*
+ ** Find sibling pages to pPage and the Cells in pParent that divide
+ ** the siblings. An attempt is made to find one sibling on either
+ ** side of pPage. Both siblings are taken from one side, however, if
+ ** pPage is either the first or last child of its parent. If pParent
+ ** has 3 or fewer children then all children of pParent are taken.
+ */
+ if( idx==pParent->nCell ){
+ nxDiv = idx - 2;
}else{
- idx--;
+ nxDiv = idx - 1;
}
- if( idx<0 ) idx = 0;
+ if( nxDiv<0 ) nxDiv = 0;
nDiv = 0;
- nOld = 0;
- for(i=0; i<3; i++){
- if( i+idx<pParent->nCell ){
- idxDiv[i] = i+idx;
- apDiv[i] = pParent->apCell[i+idx];
+ for(i=0, k=nxDiv; i<3; i++, k++){
+ if( k<pParent->nCell ){
+ idxDiv[i] = k;
+ apDiv[i] = pParent->apCell[k];
nDiv++;
pgnoOld[i] = apDiv[i]->h.leftChild;
- rc = sqlitepager_get(pBt, pgnoOld[i], &apOld[i]);
- if( rc ) goto end_of_repack;
- nOld++;
- }
- if( i+idx==pParent->nCell ){
+ }else if( k==pParent->nCell ){
pgnoOld[i] = pParent->rightChild;
- rc = sqlitepager_get(pBt, pgnoOld[i], &apOld[i]);
- if( rc ) goto end_of_repack;
- nOld++;
+ }else{
+ break;
+ }
+ rc = sqlitepager_get(pBt, pgnoOld[i], &apOld[i]);
+ if( rc ) goto balance_cleanup;
+ nOld++;
+ }
+
+ /*
+ ** Set iCur to be the index in apCell[] of the cell that the cursor
+ ** is pointing to. We will need this later on in order to keep the
+ ** cursor pointing at the same cell.
+ */
+ if( pCur ){
+ iCur = pCur->idx;
+ for(i=0; idxDiv[i]<idx; i++){
+ iCur += apOld[i]->nCell + 1;
}
+ sqlitepager_unref(pCur->pPage);
+ pCur->pPage = 0;
}
/*
- ** Get all cells
+ ** Make copies of the content of pPage and its siblings into aOld[].
+ ** The rest of this function will use data from the copies rather
+ ** that the original pages since the original pages will be in the
+ ** process of being overwritten.
+ */
+ for(i=0; i<nOld; i++){
+ copyPage(&aOld[i], apOld[i]);
+ rc = freePage(pBt, apOld[i], pgnoOld[i]);
+ if( rc ) goto balance_cleanup;
+ apOld[i] = &aOld[i];
+ }
+
+ /*
+ ** Load pointers to all cells on sibling pages and the divider cells
+ ** into the local apCell[] array. Make copies of the divider cells
+ ** into aTemp[] and remove the the divider Cells from pParent.
*/
nCell = 0;
for(i=0; i<nOld; i++){
MemPage *pOld = apOld[i];
for(j=0; j<pOld->nCell; j++){
- apCell[nCell++] = pOld->apCell[j];
+ apCell[nCell] = pOld->apCell[j];
+ szCell[nCell] = cellSize(apCell[nCell]);
+ nCell++;
}
if( i<nOld-1 ){
- apCell[nCell++] = apDiv[i];
+ szCell[nCell] = cellSize(apDiv[i]);
+ memcpy(aTemp[i], apDiv[i], szCell[nCell]);
+ apCell[nCell] = &aTemp[i];
+ dropCell(pParent, nxDiv, szCell[nCell]);
+ assert( apCell[nCell]->h.leftChild==pgnoOld[i] );
+ apCell[nCell]->h.leftChild = pOld->u.hdr.rightChild;
+ nCell++;
}
}
/*
- ** Estimate the number of pages needed
+ ** Estimate the number of pages needed. Record this number in "k"
+ ** for now. It will get transferred to nNew as we allocate the
+ ** new pages.
*/
totalSize = 0;
for(i=0; i<nCell; i++){
- totalSize += cellSize(apCell[i]);
+ totalSize += szCell[i];
}
- nNew = (totalSize + (SQLITE_PAGE_SIZE - sizeof(PageHdr) - 1)) /
+ k = (totalSize + (SQLITE_PAGE_SIZE - sizeof(PageHdr) - 1)) /
(SQLITE_PAGE_SIZE - sizeof(PageHdr));
- perPage = totalSize/nNew;
+ usedPerPage = (totalSize+k-1)/k;
+ freePerPage = SQLITE_PAGE_SIZE - usedPerPage;
/*
** Allocate new pages
*/
- for(i=0; i<nNew; i++){
+ for(i=0; i<k; i++){
rc = allocatePage(pBt, &apNew[i], &pgnoNew[i]);
- if( rc ) goto end_of_repack;
+ if( rc ) goto balance_cleanup;
+ nNew++;
zeroPage(apNew[i]);
}
/*
- ** Copy data into the new pages
+ ** Evenly distribute the data in apCell[] across the new pages.
+ ** Insert divider cells into pParent as necessary.
*/
+ j = 0;
for(i=0; i<nNew; i++){
+ MemPage *pNew = apNew[i];
+ while( j<nCell && pNew->nFree<freePerPage && szCell[j]<=pNew->nFree ){
+ if( pCur && iCur==j ){ pCur->pPage = pNew; pCur->idx = pNew->nCell; }
+ insertCell(pNew, pNew->nCell, apCell[j], szCell[j]);
+ j++;
+ }
+ assert( !pNew->isOverfull );
+ relinkCellList(pNew);
+ if( i<nNew-1 && j<nCell ){
+ pNew->u.hdr.rightChild = apCell[j]->h.leftChild;
+ apCell[j]->h.leftChild = pgnoNew[i];
+ if( pCur && iCur==j ){ pCur->pPage = pParent; pCur->idx = nxDiv; }
+ insertCell(pParent, nxDiv, apCell[j], szCell[j]);
+ j++;
+ nxDiv++;
+ }
}
-
- /*
- ** Reparent children of all cells
- */
-
- /*
- ** Release the old pages
- */
- for(i=0; i<nOld; i++){
- releasePage(pBt, apOld[i], 0);
- }
-
- /*
- ** Repack the parent page, if necessary
- */
- if( needToRepackParent ){
- return repack(pParent);
- }
- rc = SQLITE_OK;
-
-end_of_repack:
- if( unrefPage ){
- sqlitepager_unref(pPage);
- }
- return rc;
-}
-
-/*
-** Attempt to move N or more bytes out of the page that the cursor
-** points to into the left sibling page. (The left sibling page
-** contains cells that are less than the cells on this page.) The
-** entry that the cursor is pointing to cannot be moved. Return
-** TRUE if successful and FALSE if not.
-**
-** Reasons for not being successful include:
-**
-** (1) there is no left sibling,
-** (2) we could only move N-1 bytes or less,
-** (3) some kind of file I/O error occurred
-**
-** Note that a partial rotation may have occurred even if this routine
-** returns FALSE. Failure means we could not rotation a full N bytes.
-** If it is possible to rotation some smaller number M, then the
-** rotation occurs but we still return false.
-**
-** Example: Consider a segment of the Btree that looks like the
-** figure below prior to rotation. The cursor is pointing to the
-** entry *. The sort order of the entries is A B C D E * F Y.
-**
-**
-** -------------------------
-** ... | C | Y | ...
-** -------------------------
-** / \
-** --------- -----------------
-** | A | B | | D | E | * | F |
-** --------- -----------------
-**
-** After rotation of two cells (D and E), the same Btree segment
-** looks like this:
-**
-** -------------------------
-** ... | E | Y | ...
-** -------------------------
-** / \
-** ----------------- ---------
-** | A | B | C | D | | * | F |
-** ----------------- ---------
-**
-** The size of this rotation is the size by which the page containing
-** the cursor was reduced. In this case, the size of D and E.
-**
-*/
-static int rotateLeft(BtCursor *pCur, int N){
- return 0;
-}
-
-/*
-** This routine is the same as rotateLeft() except that it move data
-** to the right instead of to the left. See comments on the rotateLeft()
-** routine for additional information.
-*/
-static int rotateRight(BtCursor *pCur, int N){
- return 0;
-}
-
-/*
-** Append a cell onto the end of a page.
-**
-** The child page of the cell is reparented if pPager!=NULL.
-*/
-static void appendCell(
- Pager *pPager, /* The page cache. Needed for reparenting */
- Cell *pSrc, /* The Cell to be copied onto a new page */
- MemPage *pPage /* The page into which the cell is copied */
-){
- int pc;
- int sz;
- Cell *pDest;
-
- sz = cellSize(pSrc);
- pc = allocateSpace(pPage, sz);
- assert( pc>0 ){
- pDest = pPage->apCell[pPage->nCell] = &pPage->aDisk[pc];
- memcpy(pDest, pSrc, sz);
- pDest->h.iNext = 0;
- if( pPage->nCell>0 ){
- pPage->apCell[pPage->nCell-1]->h.iNext = pc;
+ apNew[nNew-1]->u.hdr.rightChild = apOld[nOld-1]->u.hdr.rightChild;
+ if( nxDiv==pParent->nCell ){
+ pParent->u.hdr.rightChild = pgnoNew[nNew-1];
}else{
- ((PageHdr*)pPage)->firstCell = pc;
+ pParent->apCell[nxDiv]->h.leftChild = pgnoNew[nNew-1];
}
- if( pPager && pDest->h.leftChild ){
- reparentPage(pPager, pDest->h.leftChild, pPage);
+ if( pCur ){
+ assert( pCur->pPage!=0 );
+ sqlitepager_ref(pCur->pPage);
}
-}
-
-/*
-** Split a single database page into two roughly equal-sized pages.
-**
-** The input is an existing page and a new Cell. The Cell might contain
-** a valid Cell.h.leftChild field pointing to a child page.
-**
-** The output is the Cell that divides the two new pages. The content
-** of this divider Cell is written into *pCenter. pCenter->h.leftChild
-** holds the page number of the new page that was created to hold the
-** smaller of the cells from the divided page. The larger cells from
-** the divided page are written to a newly allocated page and *ppOut
-** is made to point to that page. Or if ppOut==NULL then the larger cells
-** remain on pIn.
-**
-** Upon return, pCur should be pointing to the same cell, even if that
-** cell has moved to a new page. The cell that pCur points to cannot
-** be the pCenter cell.
-*/
-static int split(
- BtCursor *pCur, /* A cursor pointing at a cell on the page to be split */
- Cell *pNewCell, /* A new cell to add to pIn before dividing it up */
- Cell *pCenter, /* Write the cell that divides the two pages here */
- MemPage **ppOut /* If not NULL, put larger cells in new page at *ppOut */
-){
- MemPage *pLeft, *pRight;
- Pgno pgnoLeft, pgnoRight;
- PageHdr *pHdr;
- int rc;
- Pager *pPager = pCur->pBt->pPager;
- MemPage tempPage;
- /* Allocate pages to hold cells after the split and make pRight and
- ** pLeft point to the newly allocated pages.
+ /*
+ ** Reparent children of all cells.
*/
- rc = allocatePage(pCur->pBt, &pLeft, &pgnoLeft);
- if( rc ) return rc;
- if( ppOut ){
- rc = allocatePage(pCur->pBt, &pRight, &pgnoRight);
- if( rc ){
- freePage(pCur->pBt, pLeft, pgnoLeft);
- return rc;
- }
- *ppOut = pRight;
- }else{
- *ppOut = tempPage;
+ for(i=0; i<nNew; i++){
+ reparentChildPages(pBt->pPager, apNew[i]);
}
+ reparentChildPages(pBt->pPager, pParent);
- /* Copy the smaller cells from the original page into the left page
- ** of the split.
- */
- zeroPage(pLeft);
- if( pCur->idx==0 && pCur->match>0 ){
- appendCell(pPager, pNewCell, pLeft);
- }
- do{
- assert( i<pPage->nCell );
- appendCell(pPager, pPage->apCell[i++], pLeft);
- if( pCur->idx==i && pCur->iMatch>0 ){
- appendCell(pPager, pNewCell, Left);
- }
- }while( pc < SQLITE_PAGE_SIZE/2 );
-
- /* Copy the middle entry into *pCenter
- */
- assert( i<pPage->nCell );
- memcpy(pCenter, pPage->aCell[i], cellSize(pPage->aCell[i]));
- i++;
- pHdr = (PageHdr*)pLeft;
- pHdr->rightChild = pCenter->h.leftChild;
- if( pHdr->rightChild ){
- reparentPage(pPager, pHdr->rightChild, pLeft);
- }
- pCenter->h.leftChild = pgnoLeft;
-
- /* Copy the larger cells from the original page into the right
- ** page of the split
+ /*
+ ** balance the parent page.
*/
- zeroPage(pRight);
- while( i<pPage->nCell ){
- appendCell(0, pPage->apCell[i++], pRight);
- }
+ rc = balance(pBt, pParent, 0);
- /* If ppOut==NULL then copy the temporary right page over top of
- ** the original input page.
+ /*
+ ** Cleanup before returning.
*/
- if( ppOut==0 ){
- pRight->pParent = pPage->pParent;
- pRight->isInit = 1;
- memcpy(pPage, pRight, sizeof(*pPage));
- }
- reparentChildPages(pPager, pPage);
-}
-
-/*
-** Unlink a cell from a database page. Add the space used by the cell
-** back to the freelist for the database page on which the cell used to
-** reside.
-**
-** This operation overwrites the cell header and content.
-*/
-static void unlinkCell(BtCursor *pCur){
- MemPage *pPage; /* Page containing cell to be unlinked */
- int idx; /* The index of the cell to be unlinked */
- Cell *pCell; /* Pointer to the cell to be unlinked */
- u16 *piCell; /* iNext pointer from prior cell */
- int iCell; /* Index in pPage->aDisk[] of cell to be unlinked */
- int i; /* Loop counter */
-
- pPage = pCur->pPage;
- sqlitepager_write(pPage);
- idx = pCur->idx;
- pCell = pPage->apCell[idx];
- if( idx==0 ){
- piCell = &pPage->pHdr->firstCell;
- }else{
- piCell = &pPage->apCell[idx-1]->h.iNext;
+balance_cleanup:
+ for(i=0; i<nOld; i++){
+ sqlitepager_unref(apOld[i]);
}
- iCell = *piCell;
- *piCell = pCell->h.iNext;
- freeSpace(pPage, iCell, cellSize(pCell));
- pPage->nCell--;
- for(i=idx; i<pPage->nCell; i++){
- pPage->apCell[i] = pPage->apCell[i+1];
+ for(i=0; i<nNew; i++){
+ sqlitepager_unref(apNew[i]);
}
-}
-
-/*
-** Add a Cell to a database page at the spot indicated by the cursor.
-**
-** With this routine, we know that the Cell pNewCell will fit into the
-** database page that pCur points to. The calling routine has made
-** sure it will fit. All this routine needs to do is add the Cell
-** to the page. The addToPage() routine should be used for cases
-** were it is not known if the new cell will fit.
-**
-** The new cell is added to the page either before or after the cell
-** to which the cursor is pointing. The new cell is added before
-** the cursor cell if pCur->iMatch>0 and the new cell is added after
-** the cursor cell if pCur->iMatch<0. pCur->iMatch should have been set
-** by a prior call to sqliteBtreeMoveto() where the key was the key
-** of the cell being inserted. If sqliteBtreeMoveto() ended up on a
-** cell that is larger than the key, then pCur->iMatch was set to a
-** positive number, hence we insert the new record before the pointer
-** if pCur->iMatch is positive. If sqliteBtreeMaveto() ended up on a
-** cell that is smaller than the key then pCur->iMatch was set to a
-** negative number, hence we insert the new record after then pointer
-** if pCur->iMatch is negative.
-*/
-static int insertCell(BtCursor *pCur, Cell *pNewCell){
- int sz;
- int idx;
- int i;
- Cell *pCell, *pIdx;
- MemPage *pPage;
-
- pPage = pCur->pPage;
- sz = cellSize(pNewCell);
- idx = allocateSpace(pPage, sz);
- assert( idx>0 && idx<=SQLITE_PAGE_SIZE - sz );
- pCell = (Cell*)&pPage->aDisk[idx];
- memcpy(pCell, pNewCell, sz);
- pIdx = pPage->aDisk[pCur->idx];
- if( pCur->iMatch<0 ){
- /* Insert the new cell after the cell pCur points to */
- pCell->h.iNext = pIdx->h.iNext;
- pIdx->h.iNext = idx;
- for(i=pPage->nCell-1; i>pCur->idx; i--){
- pPage->apCell[i+1] = pPage->apCell[i];
- }
- pPage->apCell[pCur->idx+1] = pCell;
+ if( pCur && pCur->pPage==0 ){
+ pCur->pPage = pParent;
+ pCur->idx = 0;
}else{
- /* Insert the new cell before the cell pCur points to */
- pCell->h.iNext = pPage->pHdr->firstCell;
- pPage->pHdr->firstCell = idx;
- for(i=pPage->nCell; i>0; i++){
- pPage->apCell[i] = pPage->apCell[i-1];
- }
- pPage->apCell[0] = pCell;
- }
- pPage->nCell++;
- if( pCell->h.leftChild ){
- MemPage *pChild = sqlitepager_lookup(pCur->pBt, pCell->h.leftChild);
- if( pChild && pChild->pParent ){
- sqlitepager_unref(pChild->pParent);
- pChild->pParent = pPage;
- sqlitepager_ref(pChild->pParent);
- }
- }
- return SQLITE_OK;
-}
-
-/*
-** Insert pNewCell into the database page that pCur is pointing to at
-** the place where pCur is pointing.
-**
-** This routine works just like insertCell() except that the cell
-** to be inserted need not fit on the page. If the new cell does
-** not fit, then the page sheds data to its siblings to try to get
-** down to a size where the new cell will fit. If that effort fails,
-** then the page is split.
-*/
-static int addToPage(BtCursor *pCur, Cell *pNewCell){
- Cell tempCell;
- Cell centerCell;
-
- for(;;){
- MemPage *pPage = pCur->pPage;
- rc = sqlitepager_write(pPage);
- if( rc ) return rc;
- int sz = cellSize(pNewCell);
- if( sz<=pPage->nFree ){
- insertCell(pCur, pNewCell);
- return SQLITE_OK;
- }
- if( pPage->pParent==0 ){
- MemPage *pRight;
- PageHdr *pHdr;
- FreeBlk *pFBlk;
- int pc;
- rc = split(pCur, pNewCell, ¢erCell, &pRight);
- if( rc ) return rc;
- pHdr = pPage->pHdr;
- pHdr->right = sqlitepager_pagenumber(pRight);
- sqlitepager_unref(pRight);
- pHdr->firstCell = pc = sizeof(*pHdr);
- sz = cellSize(¢erCell);
- memcpy(&pPage->aDisk[pc], ¢erCell, sz);
- pc += sz;
- pHdr->firstFree = pc;
- pFBlk = (FreeBlk*)&pPage->aDisk[pc];
- pFBlk->iSize = SQLITE_PAGE_SIZE - pc;
- pFBlk->iNext = 0;
- memset(&pFBlk[1], 0, pFBlk->iSize-sizeof(*pFBlk));
- return SQLITE_OK;
- }
- if( rotateLeft(pCur, sz - pPage->nFree)
- || rotateRight(pCur, sz - pPage->nFree) ){
- insertCell(pCur, pNewCell);
- return SQLITE_OK;
- }
- rc = split(pCur, pNewCell, ¢erCell, 0);
- if( rc ) return rc;
- moveToParent(pCur);
- tempCell = centerCell;
- pNewPage = &tempCell;
+ sqlitepager_unref(pParent);
}
- /* NOT REACHED */
+ return rc;
}
/*
** Insert a new record into the BTree. The key is given by (pKey,nKey)
** and the data is given by (pData,nData). The cursor is used only to
** define what database the record should be inserted into. The cursor
-** is NOT left pointing at the new record.
+** is left pointing at the new record.
*/
int sqliteBtreeInsert(
BtCursor *pCur, /* Insert data into the table of this cursor */
Cell newCell;
int rc;
int loc;
+ int szNew;
MemPage *pPage;
Btree *pBt = pCur->pBt;
if( !pCur->pBt->inTrans ){
return SQLITE_ERROR; /* Must start a transaction first */
}
- rc = sqliteBtreeMoveTo(pCur, pKey, nKey, &loc);
+ rc = sqliteBtreeMoveto(pCur, pKey, nKey, &loc);
if( rc ) return rc;
- rc = sqlitepager_write(pCur->pPage);
+ pPage = pCur->pPage;
+ rc = sqlitepager_write(pPage);
if( rc ) return rc;
rc = fillInCell(pBt, &newCell, pKey, nKey, pData, nData);
if( rc ) return rc;
+ szNew = cellSize(&newCell);
if( loc==0 ){
- newCell.h.leftChild = pCur->pPage->apCell[pCur->idx]->h.leftChild;
- rc = clearCell(pBt, pCur->pPage->apCell[pCur->idx]);
+ newCell.h.leftChild = pPage->apCell[pCur->idx]->h.leftChild;
+ rc = clearCell(pBt, pPage->apCell[pCur->idx]);
if( rc ) return rc;
- unlinkCell(pCur);
- }
- return addToPage(pCur, &newCell);
-}
-
-/*
-** Check the page at which the cursor points to see if it is less than
-** half full. If it is less than half full, then try to increase
-** its fill factor by grabbing cells from siblings or by merging
-** the page with siblings.
-*/
-static int refillPage(BtCursor *pCur){
- MemPage *pPage;
- BtCursor tempCur;
- int rc;
- Pager *pPager;
-
- pPage = pCur->pPage;
- if( pPage->nFree < SQLITE_PAGE_SIZE/2 ){
- return SQLITE_OK;
- }
- rc = sqlitepager_write(pPage);
- if( rc ) return rc;
- pPager = pCur->pBt->pPager;
-
- if( pPage->nCell==0 ){
- /* The page being refilled is the root of the BTree and it has
- ** no entries of its own. If there is a child page, then make the
- ** child become the new root.
- */
- MemPage *pChild;
- Pgno pgnoChild;
- assert( pPage->pParent==0 );
- assert( sqlitepager_pagenumber(pPage)==pCur->pgnoRoot );
- pgnoChild = ((PageHdr*)pPage)->rightChild;
- if( pgnoChild==0 ){
- return SQLITE_OK;
- }
- rc = sqlitepager_get(pPager, pgno, &pChild);
- if( rc ) return rc;
- memcpy(pPage, pChild, SQLITE_PAGE_SIZE);
- memset(&pPage->aDisk[SQLITE_PAGE_SIZE], 0, EXTRA_SIZE);
- freePage(pCur->pBt, pChild, pgnoChild);
- sqlitepager_unref(pChild);
- rc = initPage(pPage, pCur->pgnoRoot, 0);
- reparentChildPages(pPager, pPage);
- return SQLITE_OK;
+ dropCell(pPage, pCur->idx, cellSize(pPage->apCell[pCur->idx]));
+ }else if( loc>0 ){
+ assert( pPage->u.hdr.rightChild==0 ); /* Must be a leaf page */
+ pCur->idx++;
+ }else{
+ assert( pPage->u.hdr.rightChild==0 ); /* Must be a leaf page */
}
-
- /** merge with siblings **/
-
- /** borrow from siblings **/
-}
-
-/*
-** Replace the content of the cell that pCur is pointing to with the content
-** in pNewContent. The pCur cell is not unlinked or moved in the Btree,
-** its content is just replaced.
-**
-** If the size of pNewContent is greater than the current size of the
-** cursor cell then the page that cursor points to might have to split.
-*/
-static int ReplaceContentOfCell(BtCursor *pCur, Cell *pNewContent){
- Cell *pCell; /* The cell whose content will be changed */
- Pgno pgno; /* Temporary storage for a page number */
-
- pCell = pCur->pPage->apCell[pCur->idx];
- rc = clearCell(pCur->pBt, pCell);
- if( rc ) return rc;
- pgno = pNewCell->h.leftChild;
- pNewCell->h.leftChild = pCell->h.leftChild;
- unlinkCell(pCur);
- rc = addToPage(pCur, pNewCell);
- pNewCell->h.leftChild = pgno;
+ insertCell(pPage, pCur->idx, &newCell, cellSize(&newCell));
+ rc = balance(pCur->pBt, pPage, pCur);
return rc;
}
rc = sqlitepager_write(pPage);
if( rc ) return rc;
pCell = pPage->apCell[pCur->idx];
- if( pPage->pHdr->rightChild ){
- /* The entry to be deleted is not on a leaf page. Non-leaf entries
- ** cannot be deleted directly because they have to be present to
- ** hold pointers to subpages. So what we do is look at the next
- ** entry in sequence. The next entry is guaranteed to exist and
- ** be a leaf. We copy the payload from the next entry into this
- ** entry, then delete the next entry.
+ pgnoChild = pCell->h.leftChild;
+ clearCell(pCell);
+ dropCell(pPage, pCur->idx, cellSize(pCell));
+ if( pgnoChild ){
+ /*
+ ** If the entry we just deleted is not a leaf, then we've left a
+ ** whole in an internal page. We have to fill the whole by moving
+ ** in a page from a leaf. The next Cell after the one just deleted
+ ** is guaranteed to exist and to be a leaf so we can use it.
*/
- BtCursor origCur;
- CreateTemporaryCursor(pCur, &origCur);
- rc = sqliteBtreeNext(pCur, 0);
- if( rc==SQLITE_OK ){
- pPage = pCur->pPage;
- pCell = pPage->apCell[pCur->idx];
- rc = ReplaceContentOfCell(&origCur, pCell);
+ BtCursor leafCur;
+ Cell *pNext;
+ int szNext;
+ getTempCursor(pCur, &leafCur);
+ rc = sqliteBtreeNext(&leafCur, 0);
+ if( rc!=SQLITE_OK ){
+ return SQLITE_CORRUPT;
}
- DestroyTemporaryCursor(&origCur);
+ pNext = leafCur.pPage->apCell[leafCur.idx]
+ szNext = cellSize(pNext);
+ insertCell(pPage, pCur->idx, pNext, szNext);
+ rc = balance(pCur->pBt, pPage, pCur);
if( rc ) return rc;
- }
- rc = clearCell(pCell);
- if( rc ) return rc;
- unlinkCell(pCur->pBt, pCell);
- if( pCur->idx == 0 ){
pCur->bSkipNext = 1;
+ dropCell(leafCur.pPage, leafCur.idx, szNext);
+ rc = balance(pCur->pBt, leafCur.pPage, 0);
+ releaseTempCur(&leafCur);
}else{
- pCur->idx--;
+ rc = balance(pCur->pBt, pPage, pCur);
+ pCur->bSkipNext = 1;
}
- rc = refillPage(pCur);
return rc;
}
rc = sqlitepager_get(pBt->pPager, pgno, &pPage);
if( rc ) return rc;
- idx = ((PageHdr*)pPage)->firstCell;
+ idx = pPage->u.hdr.firstCell;
while( idx>0 ){
- pCell = (Cell*)&pPage->aDisk[idx];
+ pCell = (Cell*)&pPage->u.aDisk[idx];
idx = pCell->h.iNext;
if( pCell->h.leftChild ){
rc = clearDatabasePage(pBt, pCell->h.leftChild);
projects / thirdparty / sqlite.git / commitdiff
