/*

** 2004 April 6

**

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

**

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

** $Id: btree.cpp 1282 2008-11-13 09:31:33Z LarsPson $

**

** This file implements a external (disk-based) database using BTrees.

** See the header comment on "btreeInt.h" for additional information.

** Including a description of file format and an overview of operation.

*/

#include "btreeInt.h"

/*

** The header string that appears at the beginning of every

** SQLite database.

*/

static const char zMagicHeader[] = SQLITE_FILE_HEADER;

/*

** Set this global variable to 1 to enable tracing using the TRACE

** macro.

*/

#if SQLITE_TEST

int sqlite3_btree_trace=0;  /* True to enable tracing */

#endif

#ifndef SQLITE_OMIT_SHARED_CACHE

/*

** A flag to indicate whether or not shared cache is enabled.  Also,

** a list of BtShared objects that are eligible for participation

** in shared cache.  The variables have file scope during normal builds,

** but the test harness needs to access these variables so we make them

** global for test builds.

*/

#ifdef SQLITE_TEST

BtShared *sqlite3SharedCacheList = 0;

int sqlite3SharedCacheEnabled = 0;

#else

static BtShared *sqlite3SharedCacheList = 0;

static int sqlite3SharedCacheEnabled = 0;

#endif

#endif /* SQLITE_OMIT_SHARED_CACHE */

#ifndef SQLITE_OMIT_SHARED_CACHE

/*

** Enable or disable the shared pager and schema features.

**

** This routine has no effect on existing database connections.

** The shared cache setting effects only future calls to

** sqlite3_open(), sqlite3_open16(), or sqlite3_open_v2().

*/

EXPORT_C int sqlite3_enable_shared_cache(int enable){

  sqlite3SharedCacheEnabled = enable;

  return SQLITE_OK;

}

#endif

/*

** Forward declaration

*/

static int checkReadLocks(Btree*,Pgno,BtCursor*);

#ifdef SQLITE_OMIT_SHARED_CACHE

  /*

  ** The functions queryTableLock(), lockTable() and unlockAllTables()

  ** manipulate entries in the BtShared.pLock linked list used to store

  ** shared-cache table level locks. If the library is compiled with the

  ** shared-cache feature disabled, then there is only ever one user

  ** of each BtShared structure and so this locking is not necessary. 

  ** So define the lock related functions as no-ops.

  */

  #define queryTableLock(a,b,c) SQLITE_OK

  #define lockTable(a,b,c) SQLITE_OK

  #define unlockAllTables(a)

#endif

#ifndef SQLITE_OMIT_SHARED_CACHE

/*

** Query to see if btree handle p may obtain a lock of type eLock 

** (READ_LOCK or WRITE_LOCK) on the table with root-page iTab. Return

** SQLITE_OK if the lock may be obtained (by calling lockTable()), or

** SQLITE_LOCKED if not.

*/

static int queryTableLock(Btree *p, Pgno iTab, u8 eLock){

  BtShared *pBt = p->pBt;

  BtLock *pIter;

  assert( sqlite3BtreeHoldsMutex(p) );

  /* This is a no-op if the shared-cache is not enabled */

  if( !p->sharable ){

    return SQLITE_OK;

  }

  /* This (along with lockTable()) is where the ReadUncommitted flag is

  ** dealt with. If the caller is querying for a read-lock and the flag is

  ** set, it is unconditionally granted - even if there are write-locks

  ** on the table. If a write-lock is requested, the ReadUncommitted flag

  ** is not considered.

  **

  ** In function lockTable(), if a read-lock is demanded and the 

  ** ReadUncommitted flag is set, no entry is added to the locks list 

  ** (BtShared.pLock).

  **

  ** To summarize: If the ReadUncommitted flag is set, then read cursors do

  ** not create or respect table locks. The locking procedure for a 

  ** write-cursor does not change.

  */

  if( 

    !p->db || 

    0==(p->db->flags&SQLITE_ReadUncommitted) || 

    eLock==WRITE_LOCK ||

    iTab==MASTER_ROOT

  ){

    for(pIter=pBt->pLock; pIter; pIter=pIter->pNext){

      if( pIter->pBtree!=p && pIter->iTable==iTab && 

          (pIter->eLock!=eLock || eLock!=READ_LOCK) ){

        return SQLITE_LOCKED;

      }

    }

  }

  return SQLITE_OK;

}

#endif /* !SQLITE_OMIT_SHARED_CACHE */

#ifndef SQLITE_OMIT_SHARED_CACHE

/*

** Add a lock on the table with root-page iTable to the shared-btree used

** by Btree handle p. Parameter eLock must be either READ_LOCK or 

** WRITE_LOCK.

**

** SQLITE_OK is returned if the lock is added successfully. SQLITE_BUSY and

** SQLITE_NOMEM may also be returned.

*/

static int lockTable(Btree *p, Pgno iTable, u8 eLock){

  BtShared *pBt = p->pBt;

  BtLock *pLock = 0;

  BtLock *pIter;

  assert( sqlite3BtreeHoldsMutex(p) );

  /* This is a no-op if the shared-cache is not enabled */

  if( !p->sharable ){

    return SQLITE_OK;

  }

  assert( SQLITE_OK==queryTableLock(p, iTable, eLock) );

  /* If the read-uncommitted flag is set and a read-lock is requested,

  ** return early without adding an entry to the BtShared.pLock list. See

  ** comment in function queryTableLock() for more info on handling 

  ** the ReadUncommitted flag.

  */

  if( 

    (p->db) && 

    (p->db->flags&SQLITE_ReadUncommitted) && 

    (eLock==READ_LOCK) &&

    iTable!=MASTER_ROOT

  ){

    return SQLITE_OK;

  }

  /* First search the list for an existing lock on this table. */

  for(pIter=pBt->pLock; pIter; pIter=pIter->pNext){

    if( pIter->iTable==iTable && pIter->pBtree==p ){

      pLock = pIter;

      break;

    }

  }

  /* If the above search did not find a BtLock struct associating Btree p

  ** with table iTable, allocate one and link it into the list.

  */

  if( !pLock ){

    pLock = (BtLock *)sqlite3MallocZero(sizeof(BtLock));

    if( !pLock ){

      return SQLITE_NOMEM;

    }

    pLock->iTable = iTable;

    pLock->pBtree = p;

    pLock->pNext = pBt->pLock;

    pBt->pLock = pLock;

  }

  /* Set the BtLock.eLock variable to the maximum of the current lock

  ** and the requested lock. This means if a write-lock was already held

  ** and a read-lock requested, we don't incorrectly downgrade the lock.

  */

  assert( WRITE_LOCK>READ_LOCK );

  if( eLock>pLock->eLock ){

    pLock->eLock = eLock;

  }

  return SQLITE_OK;

}

#endif /* !SQLITE_OMIT_SHARED_CACHE */

#ifndef SQLITE_OMIT_SHARED_CACHE

/*

** Release all the table locks (locks obtained via calls to the lockTable()

** procedure) held by Btree handle p.

*/

static void unlockAllTables(Btree *p){

  BtLock **ppIter = &p->pBt->pLock;

  assert( sqlite3BtreeHoldsMutex(p) );

  assert( p->sharable || 0==*ppIter );

  while( *ppIter ){

    BtLock *pLock = *ppIter;

    if( pLock->pBtree==p ){

      *ppIter = pLock->pNext;

      sqlite3_free(pLock);

    }else{

      ppIter = &pLock->pNext;

    }

  }

}

#endif /* SQLITE_OMIT_SHARED_CACHE */

static void releasePage(MemPage *pPage);  /* Forward reference */

/*

** Verify that the cursor holds a mutex on the BtShared

*/

#ifndef NDEBUG

static int cursorHoldsMutex(BtCursor *p){

  return sqlite3_mutex_held(p->pBt->mutex);

}

#endif

#ifndef SQLITE_OMIT_INCRBLOB

/*

** Invalidate the overflow page-list cache for cursor pCur, if any.

*/

static void invalidateOverflowCache(BtCursor *pCur){

  assert( cursorHoldsMutex(pCur) );

  sqlite3_free(pCur->aOverflow);

  pCur->aOverflow = 0;

}

/*

** Invalidate the overflow page-list cache for all cursors opened

** on the shared btree structure pBt.

*/

static void invalidateAllOverflowCache(BtShared *pBt){

  BtCursor *p;

  assert( sqlite3_mutex_held(pBt->mutex) );

  for(p=pBt->pCursor; p; p=p->pNext){

    invalidateOverflowCache(p);

  }

}

#else

  #define invalidateOverflowCache(x)

  #define invalidateAllOverflowCache(x)

#endif

/*

** Save the current cursor position in the variables BtCursor.nKey 

** and BtCursor.pKey. The cursor's state is set to CURSOR_REQUIRESEEK.

*/

static int saveCursorPosition(BtCursor *pCur){

  int rc;

  assert( CURSOR_VALID==pCur->eState );

  assert( 0==pCur->pKey );

  assert( cursorHoldsMutex(pCur) );

  rc = sqlite3BtreeKeySize(pCur, &pCur->nKey);

  /* If this is an intKey table, then the above call to BtreeKeySize()

  ** stores the integer key in pCur->nKey. In this case this value is

  ** all that is required. Otherwise, if pCur is not open on an intKey

  ** table, then malloc space for and store the pCur->nKey bytes of key 

  ** data.

  */

  if( rc==SQLITE_OK && 0==pCur->pPage->intKey){

    void *pKey = sqlite3_malloc(pCur->nKey);

    if( pKey ){

      rc = sqlite3BtreeKey(pCur, 0, pCur->nKey, pKey);

      if( rc==SQLITE_OK ){

        pCur->pKey = pKey;

      }else{

        sqlite3_free(pKey);

      }

    }else{

      rc = SQLITE_NOMEM;

    }

  }

  assert( !pCur->pPage->intKey || !pCur->pKey );

  if( rc==SQLITE_OK ){

    releasePage(pCur->pPage);

    pCur->pPage = 0;

    pCur->eState = CURSOR_REQUIRESEEK;

  }

  invalidateOverflowCache(pCur);

  return rc;

}

/*

** Save the positions of all cursors except pExcept open on the table 

** with root-page iRoot. Usually, this is called just before cursor

** pExcept is used to modify the table (BtreeDelete() or BtreeInsert()).

*/

static int saveAllCursors(BtShared *pBt, Pgno iRoot, BtCursor *pExcept){

  BtCursor *p;

  assert( sqlite3_mutex_held(pBt->mutex) );

  assert( pExcept==0 || pExcept->pBt==pBt );

  for(p=pBt->pCursor; p; p=p->pNext){

    if( p!=pExcept && (0==iRoot || p->pgnoRoot==iRoot) && 

        p->eState==CURSOR_VALID ){

      int rc = saveCursorPosition(p);

      if( SQLITE_OK!=rc ){

        return rc;

      }

    }

  }

  return SQLITE_OK;

}

/*

** Clear the current cursor position.

*/

static void clearCursorPosition(BtCursor *pCur){

  assert( cursorHoldsMutex(pCur) );

  sqlite3_free(pCur->pKey);

  pCur->pKey = 0;

  pCur->eState = CURSOR_INVALID;

}

/*

** Restore the cursor to the position it was in (or as close to as possible)

** when saveCursorPosition() was called. Note that this call deletes the 

** saved position info stored by saveCursorPosition(), so there can be

** at most one effective restoreOrClearCursorPosition() call after each 

** saveCursorPosition().

**

** If the second argument argument - doSeek - is false, then instead of 

** returning the cursor to its saved position, any saved position is deleted

** and the cursor state set to CURSOR_INVALID.

*/

int sqlite3BtreeRestoreOrClearCursorPosition(BtCursor *pCur){

  int rc;

  assert( cursorHoldsMutex(pCur) );

  assert( pCur->eState>=CURSOR_REQUIRESEEK );

  if( pCur->eState==CURSOR_FAULT ){

    return pCur->skip;

  }

#ifndef SQLITE_OMIT_INCRBLOB

  if( pCur->isIncrblobHandle ){

    return SQLITE_ABORT;

  }

#endif

  pCur->eState = CURSOR_INVALID;

  rc = sqlite3BtreeMoveto(pCur, pCur->pKey, pCur->nKey, 0, &pCur->skip);

  if( rc==SQLITE_OK ){

    sqlite3_free(pCur->pKey);

    pCur->pKey = 0;

    assert( pCur->eState==CURSOR_VALID || pCur->eState==CURSOR_INVALID );

  }

  return rc;

}

#define restoreOrClearCursorPosition(p) \

  (p->eState>=CURSOR_REQUIRESEEK ? \

         sqlite3BtreeRestoreOrClearCursorPosition(p) : \

         SQLITE_OK)

#ifndef SQLITE_OMIT_AUTOVACUUM

/*

** Given a page number of a regular database page, return the page

** number for the pointer-map page that contains the entry for the

** input page number.

*/

static Pgno ptrmapPageno(BtShared *pBt, Pgno pgno){

  int nPagesPerMapPage, iPtrMap, ret;

  assert( sqlite3_mutex_held(pBt->mutex) );

  nPagesPerMapPage = (pBt->usableSize/5)+1;

  iPtrMap = (pgno-2)/nPagesPerMapPage;

  ret = (iPtrMap*nPagesPerMapPage) + 2; 

  if( ret==PENDING_BYTE_PAGE(pBt) ){

    ret++;

  }

  return ret;

}

/*

** Write an entry into the pointer map.

**

** This routine updates the pointer map entry for page number 'key'

** so that it maps to type 'eType' and parent page number 'pgno'.

** An error code is returned if something goes wrong, otherwise SQLITE_OK.

*/

static int ptrmapPut(BtShared *pBt, Pgno key, u8 eType, Pgno parent){

  DbPage *pDbPage;  /* The pointer map page */

  u8 *pPtrmap;      /* The pointer map data */

  Pgno iPtrmap;     /* The pointer map page number */

  int offset;       /* Offset in pointer map page */

  int rc;

  assert( sqlite3_mutex_held(pBt->mutex) );

  /* The master-journal page number must never be used as a pointer map page */

  assert( 0==PTRMAP_ISPAGE(pBt, PENDING_BYTE_PAGE(pBt)) );

  assert( pBt->autoVacuum );

  if( key==0 ){

    return SQLITE_CORRUPT_BKPT;

  }

  iPtrmap = PTRMAP_PAGENO(pBt, key);

  rc = sqlite3PagerGet(pBt->pPager, iPtrmap, &pDbPage);

  if( rc!=SQLITE_OK ){

    return rc;

  }

  offset = PTRMAP_PTROFFSET(pBt, key);

  pPtrmap = (u8 *)sqlite3PagerGetData(pDbPage);

  if( eType!=pPtrmap[offset] || get4byte(&pPtrmap[offset+1])!=parent ){

    TRACE(("PTRMAP_UPDATE: %d->(%d,%d)\n", key, eType, parent));

    rc = sqlite3PagerWrite(pDbPage);

    if( rc==SQLITE_OK ){

      pPtrmap[offset] = eType;

      put4byte(&pPtrmap[offset+1], parent);

    }

  }

  sqlite3PagerUnref(pDbPage);

  return rc;

}

/*

** Read an entry from the pointer map.

**

** This routine retrieves the pointer map entry for page 'key', writing

** the type and parent page number to *pEType and *pPgno respectively.

** An error code is returned if something goes wrong, otherwise SQLITE_OK.

*/

static int ptrmapGet(BtShared *pBt, Pgno key, u8 *pEType, Pgno *pPgno){

  DbPage *pDbPage;   /* The pointer map page */

  int iPtrmap;       /* Pointer map page index */

  u8 *pPtrmap;       /* Pointer map page data */

  int offset;        /* Offset of entry in pointer map */

  int rc;

  assert( sqlite3_mutex_held(pBt->mutex) );

  iPtrmap = PTRMAP_PAGENO(pBt, key);

  rc = sqlite3PagerGet(pBt->pPager, iPtrmap, &pDbPage);

  if( rc!=0 ){

    return rc;

  }

  pPtrmap = (u8 *)sqlite3PagerGetData(pDbPage);

  offset = PTRMAP_PTROFFSET(pBt, key);

  assert( pEType!=0 );

  *pEType = pPtrmap[offset];

  if( pPgno ) *pPgno = get4byte(&pPtrmap[offset+1]);

  sqlite3PagerUnref(pDbPage);

  if( *pEType<1 || *pEType>5 ) return SQLITE_CORRUPT_BKPT;

  return SQLITE_OK;

}

#endif /* SQLITE_OMIT_AUTOVACUUM */

/*

** Given a btree page and a cell index (0 means the first cell on

** the page, 1 means the second cell, and so forth) return a pointer

** to the cell content.

**

** This routine works only for pages that do not contain overflow cells.

*/

#define findCell(pPage, iCell) \

  ((pPage)->aData + get2byte(&(pPage)->aData[(pPage)->cellOffset+2*(iCell)]))

#ifdef SQLITE_TEST

u8 *sqlite3BtreeFindCell(MemPage *pPage, int iCell){

  assert( iCell>=0 );

  assert( iCell<get2byte(&pPage->aData[pPage->hdrOffset+3]) );

  return findCell(pPage, iCell);

}

#endif

/*

** This a more complex version of sqlite3BtreeFindCell() that works for

** pages that do contain overflow cells.  See insert

*/

static u8 *findOverflowCell(MemPage *pPage, int iCell){

  int i;

  assert( sqlite3_mutex_held(pPage->pBt->mutex) );

  for(i=pPage->nOverflow-1; i>=0; i--){

    int k;

	MemPage::_OvflCell *pOvfl;

    pOvfl = &pPage->aOvfl[i];

    k = pOvfl->idx;

    if( k<=iCell ){

      if( k==iCell ){

        return pOvfl->pCell;

      }

      iCell--;

    }

  }

  return findCell(pPage, iCell);

}

/*

** Parse a cell content block and fill in the CellInfo structure.  There

** are two versions of this function.  sqlite3BtreeParseCell() takes a 

** cell index as the second argument and sqlite3BtreeParseCellPtr() 

** takes a pointer to the body of the cell as its second argument.

**

** Within this file, the parseCell() macro can be called instead of

** sqlite3BtreeParseCellPtr(). Using some compilers, this will be faster.

*/

void sqlite3BtreeParseCellPtr(

  MemPage *pPage,         /* Page containing the cell */