Use the pointer-map pages to make the incremental blob API more efficient. (CVS 3896)

FossilOrigin-Name: 93a3bf71d576096f4b5a3db256ca6f9b5521d137

5 files changed, 297 insertions, 117 deletions

diff --git a/manifest b/manifest
index 3b5f8ab4f..76faeb3de 100644
--- a/manifest
+++ b/manifest
@@ -1,5 +1,5 @@
-C Fix\ssome\scompiler\swarnings.\s\sAdd\sthe\s(untested)\szeroblob()\sSQL\sfunction.\s(CVS\s3895)
-D 2007-05-02T02:08:29
+C Use\sthe\spointer-map\spages\sto\smake\sthe\sincremental\sblob\sAPI\smore\sefficient.\s(CVS\s3896)
+D 2007-05-02T13:16:30
 F Makefile.in 8cab54f7c9f5af8f22fd97ddf1ecfd1e1860de62
 F Makefile.linux-gcc 2d8574d1ba75f129aba2019f0b959db380a90935
 F README 9c4e2d6706bdcc3efdd773ce752a8cdab4f90028
@@ -59,7 +59,7 @@ F src/alter.c 2c79ec40f65e33deaf90ca493422c74586e481a3
 F src/analyze.c 4bbf5ddf9680587c6d4917e02e378b6037be3651
 F src/attach.c a16ada4a4654a0d126b8223ec9494ebb81bc5c3c
 F src/auth.c 902f4722661c796b97f007d9606bd7529c02597f
-F src/btree.c 4e0735d1826a8cefb5ee25aa9615d43860881f10
+F src/btree.c 0b2c181ea3ee23b5daef6f89d07a8a60d0f6370f
 F src/btree.h b2ef1ccc337fd37c58c8c17189a237aea341fb48
 F src/build.c 02e01ec7907c7d947ab3041fda0e81eaed05db42
 F src/callback.c 6414ed32d55859d0f65067aa5b88d2da27b3af9e
@@ -101,7 +101,7 @@ F src/sqlite.h.in 1e053c58fd4df28c38ffdca2443b16d5f76f6f1e
 F src/sqlite3ext.h 7d0d363ea7327e817ef0dfe1b7eee1f171b72890
 F src/sqliteInt.h 0b14d0eae083aafca0562d2261a404e5e5abc5f0
 F src/table.c 6d0da66dde26ee75614ed8f584a1996467088d06
-F src/tclsqlite.c 82f7be1e8015ef224e2a9410a8f98dd6f61d64e9
+F src/tclsqlite.c 23082fa8affdf3ae73937ca0755754fc562674bc
 F src/test1.c bf70db366aa28b813810f63fc48fec424034502d
 F src/test2.c 24458b17ab2f3c90cbc1c8446bd7ffe69be62f88
 F src/test3.c 946ea9d1a8c928656e3c70f0a2fcb8e733a15e86
@@ -241,7 +241,7 @@ F test/fts2n.test a70357e72742681eaebfdbe9007b87ff3b771638
 F test/func.test 8a3bc8e8365dc0053c826923c0f738645f50f2f5
 F test/hook.test 7e7645fd9a033f79cce8fdff151e32715e7ec50a
 F test/in.test 369cb2aa1eab02296b4ec470732fe8c131260b1d
-F test/incrblob.test 86708ae039f564535e4200ac0e61dab5a6d6f18e
+F test/incrblob.test 09db22f90137dc4f449cf6c7f8e554156fb68fd2
 F test/incrvacuum.test a4c9022d7b26b10495616cc5a255f11afb683be8
 F test/incrvacuum_ioerr.test 0ebc382bcc2036ec58cf49cc5ffada45f75d907b
 F test/index.test e65df12bed94b2903ee89987115e1578687e9266
@@ -470,7 +470,7 @@ F www/tclsqlite.tcl bb0d1357328a42b1993d78573e587c6dcbc964b9
 F www/vdbe.tcl 87a31ace769f20d3627a64fa1fade7fed47b90d0
 F www/version3.tcl 890248cf7b70e60c383b0e84d77d5132b3ead42b
 F www/whentouse.tcl fc46eae081251c3c181bd79c5faef8195d7991a5
-P 7a01836dde45098796693bc6cb6045c4059adf1a
-R 05e0b3dbc6feb6e7c7aca220b80c8909
-U drh
-Z cffc5ba75c4a156de15690c278520f4f
+P 6f4f8ba7ec15f214f36fa78e593dd4522ab717f5
+R 89ce3536d9d902e9d75a561e20ef3775
+U danielk1977
+Z 5170bc200f54fb706f822cc7c0bea383
diff --git a/manifest.uuid b/manifest.uuid
index 4319a33ae..2903a5e59 100644
--- a/manifest.uuid
+++ b/manifest.uuid
@@ -1 +1 @@
-6f4f8ba7ec15f214f36fa78e593dd4522ab717f5
\ No newline at end of file
+93a3bf71d576096f4b5a3db256ca6f9b5521d137
\ No newline at end of file
diff --git a/src/btree.c b/src/btree.c
index 2ad735805..484b52879 100644
--- a/src/btree.c
+++ b/src/btree.c
@@ -9,7 +9,7 @@
 **    May you share freely, never taking more than you give.
 **
 *************************************************************************
-** $Id: btree.c,v 1.364 2007/05/02 01:34:31 drh Exp $
+** $Id: btree.c,v 1.365 2007/05/02 13:16:30 danielk1977 Exp $
 **
 ** This file implements a external (disk-based) database using BTrees.
 ** For a detailed discussion of BTrees, refer to
@@ -3025,6 +3025,91 @@ int sqlite3BtreeDataSize(BtCursor *pCur, u32 *pSize){
 }
 
 /*
+** Given the page number of an overflow page in the database (parameter
+** ovfl), this function finds the page number of the next page in the 
+** linked list of overflow pages. If possible, it uses the auto-vacuum
+** pointer-map data instead of reading the content of page ovfl to do so. 
+**
+** If an error occurs an SQLite error code is returned. Otherwise:
+**
+** Unless pPgnoNext is NULL, the page number of the next overflow 
+** page in the linked list is written to *pPgnoNext. If page ovfl
+** is the last page in it's linked list, *pPgnoNext is set to zero. 
+**
+** If ppPage is not NULL, *ppPage is set to the MemPage* handle
+** for page ovfl. The underlying pager page may have been requested
+** with the noContent flag set, so the page data accessable via
+** this handle may not be trusted.
+*/
+static int getOverflowPage(
+  BtShared *pBt, 
+  Pgno ovfl,                   /* Overflow page */
+  MemPage **ppPage,            /* OUT: MemPage handle */
+  Pgno *pPgnoNext              /* OUT: Next overflow page number */
+){
+  Pgno next = 0;
+  int rc;
+
+  /* One of these must not be NULL. Otherwise, why call this function? */
+  assert(ppPage || pPgnoNext);
+
+  /* If pPgnoNext is NULL, then this function is being called to obtain
+  ** a MemPage* reference only. No page-data is required in this case.
+  */
+  if( !pPgnoNext ){
+    return getPage(pBt, ovfl, ppPage, 1);
+  }
+
+#ifndef SQLITE_OMIT_AUTOVACUUM
+  /* Try to find the next page in the overflow list using the
+  ** autovacuum pointer-map pages. Guess that the next page in 
+  ** the overflow list is page number (ovfl+1). If that guess turns 
+  ** out to be wrong, fall back to loading the data of page 
+  ** number ovfl to determine the next page number.
+  */
+  if( pBt->autoVacuum ){
+    Pgno pgno;
+    Pgno iGuess = ovfl+1;
+    u8 eType;
+
+    while( PTRMAP_ISPAGE(pBt, iGuess) || iGuess==PENDING_BYTE_PAGE(pBt) ){
+      iGuess++;
+    }
+
+    if( iGuess<sqlite3PagerPagecount(pBt->pPager) ){
+      rc = ptrmapGet(pBt, iGuess, &eType, &pgno);
+      if( rc!=SQLITE_OK ){
+        return rc;
+      }
+      if( eType==PTRMAP_OVERFLOW2 && pgno==ovfl ){
+        next = iGuess;
+      }
+    }
+  }
+#endif
+
+  if( next==0 || ppPage ){
+    MemPage *pPage = 0;
+
+    rc = getPage(pBt, ovfl, &pPage, next!=0);
+    assert(rc==SQLITE_OK || pPage==0);
+    if( next==0 && rc==SQLITE_OK ){
+      next = get4byte(pPage->aData);
+    }
+
+    if( ppPage ){
+      *ppPage = pPage;
+    }else{
+      releasePage(pPage);
+    }
+  }
+  *pPgnoNext = next;
+
+  return rc;
+}
+
+
+/*
 ** Read payload information from the entry that the pCur cursor is
 ** pointing to.  Begin reading the payload at "offset" and read
 ** a total of "amt" bytes.  Put the result in zBuf.
@@ -3086,15 +3171,28 @@ static int getPayload(
   if( amt>0 ){
     nextPage = get4byte(&aPayload[pCur->info.nLocal]);
     while( amt>0 && nextPage ){
-      DbPage *pDbPage;
-      rc = sqlite3PagerGet(pBt->pPager, nextPage, &pDbPage);
-      if( rc!=0 ){
-        return rc;
-      }
-      aPayload = sqlite3PagerGetData(pDbPage);
-      nextPage = get4byte(aPayload);
-      if( offset<ovflSize ){
+      if( offset>=ovflSize ){
+        /* The only reason to read this page is to obtain the page
+        ** number for the next page in the overflow chain. So try
+        ** the getOverflowPage() shortcut.
+        */
+        rc = getOverflowPage(pBt, nextPage, 0, &nextPage);
+        if( rc!=SQLITE_OK ){
+          return rc;
+        }
+        offset -= ovflSize;
+      }else{
+        /* Need to read this page properly, to obtain data to copy into
+        ** the caller's buffer.
+        */
+        DbPage *pDbPage;
         int a = amt;
+        rc = sqlite3PagerGet(pBt->pPager, nextPage, &pDbPage);
+        if( rc!=0 ){
+          return rc;
+        }
+        aPayload = sqlite3PagerGetData(pDbPage);
+        nextPage = get4byte(aPayload);
         if( a + offset > ovflSize ){
           a = ovflSize - offset;
         }
@@ -3102,10 +3200,8 @@ static int getPayload(
         offset = 0;
         amt -= a;
         pBuf += a;
-      }else{
-        offset -= ovflSize;
+        sqlite3PagerUnref(pDbPage);
       }
-      sqlite3PagerUnref(pDbPage);
     }
   }
 
@@ -4047,60 +4143,6 @@ static int freePage(MemPage *pPage){
 }
 
 /*
-** Get a MemPage structure for overflow page number ovfl. If it
-** is not zero, the page number of the overflow page following the
-** one retrieved is written to *pPgnoNext.
-**
-** If it is possible to figure out the page-number of the next
-** overflow page without reading the data of page ovfl, then 
-** sqlite3PagerAcquire() is passed the "noContent" flag when
-** page ovfl is retrieved.
-*/
-static int getOverflowPage(
-  BtShared *pBt, 
-  Pgno ovfl, 
-  MemPage **ppPage, 
-  Pgno *pPgnoNext
-){
-  Pgno next = 0;
-  int rc;
-
-  if( !pPgnoNext ){
-    return getPage(pBt, ovfl, ppPage, 1);
-  }
-
-#ifndef SQLITE_OMIT_AUTOVACUUM
-  if( pBt->autoVacuum ){
-    Pgno pgno;
-    Pgno iGuess = ovfl+1;
-    u8 eType;
-
-    while( PTRMAP_ISPAGE(pBt, iGuess) || iGuess==PENDING_BYTE_PAGE(pBt) ){
-      iGuess++;
-    }
-
-    if( iGuess<sqlite3PagerPagecount(pBt->pPager) ){
-      rc = ptrmapGet(pBt, iGuess, &eType, &pgno);
-      if( rc!=SQLITE_OK ){
-        return rc;
-      }
-      if( eType==PTRMAP_OVERFLOW2 && pgno==ovfl ){
-        next = iGuess;
-      }
-    }
-  }
-#endif
-
-  rc = getPage(pBt, ovfl, ppPage, 0);
-  if( rc==SQLITE_OK ){
-    assert(next==0 || next==get4byte((*ppPage)->aData));
-    next = get4byte((*ppPage)->aData);
-  }
-  *pPgnoNext = next;
-  return rc;
-}
-
-/*
 ** Free any overflow pages associated with the given Cell.
 */
 static int clearCell(MemPage *pPage, unsigned char *pCell){
@@ -6825,49 +6867,108 @@ int sqlite3BtreeLockTable(Btree *p, int iTab, u8 isWriteLock){
 ** to change the length of the data stored.
 */
 int sqlite3BtreePutData(BtCursor *pCsr, u32 offset, u32 amt, const void *z){
-  /* TODO: The following is only a stop-gap implementation. It needs
-  ** to be made efficient using the optimistic overflow page trick. 
-  ** Similar changes need to be made to sqlite3BtreeData().
-  */
-  i64 iKey;
+  BtShared *pBt = pCsr->pBtree->pBt;
   int rc;
 
-  int nCopy;
-  u32 nData;
-  char *zData;
+  u32 iRem = amt;        /* Remaining bytes to write */
+  u8 *zRem = (u8 *)z;    /* Pointer to data not yet written */
+  u32 iOffset = offset;  /* Offset from traversal point to start of write */
 
-  rc = sqlite3BtreeKeySize(pCsr, &iKey);
-  if( rc!=SQLITE_OK ){
-    return rc;
-  }
+  Pgno iOvfl;            /* Page number for next overflow page */
+  int ovflSize;          /* Bytes of data per overflow page. */
 
-  rc = sqlite3BtreeDataSize(pCsr, &nData);
-  if( rc!=SQLITE_OK ){
-    return rc;
-  }
+  CellInfo *pInfo;
 
-  zData = sqliteMalloc(nData);
-  if( !zData ){
-    return SQLITE_NOMEM;
+  /* Check some preconditions: 
+  **   (a) a write-transaction is open, 
+  **   (b) the cursor is open for writing,
+  **   (c) there is no read-lock on the table being modified and
+  **   (d) the cursor points at a valid row of an intKey table.
+  */
+  if( pBt->inTransaction!=TRANS_WRITE ){
+    /* Must start a transaction before doing an insert */
+    return pBt->readOnly ? SQLITE_READONLY : SQLITE_ERROR;
+  }
+  assert( !pBt->readOnly );
+  if( !pCsr->wrFlag ){
+    return SQLITE_PERM;   /* Cursor not open for writing */
+  }
+  if( checkReadLocks(pCsr->pBtree, pCsr->pgnoRoot, pCsr) ){
+    return SQLITE_LOCKED; /* The table pCur points to has a read lock */
+  }
+  if( pCsr->eState==CURSOR_INVALID || !pCsr->pPage->intKey ){
+    return SQLITE_ERROR;
   }
 
-  rc = sqlite3BtreeData(pCsr, 0, nData, (void *)zData);
-  if( rc!=SQLITE_OK ){
-    sqliteFree(zData);
-    return rc;
+  /* Parse the cell-info. Check that the cell-data area is large
+  ** enough for the proposed write operation.
+  */
+  getCellInfo(pCsr);
+  pInfo = &pCsr->info;
+  if( pInfo->nData<(offset+amt) ){
+    return SQLITE_ERROR;
   }
 
-  nCopy = amt;
-  if( nCopy>(nData-offset) ){
-    nCopy = nData-offset;
+  if( pInfo->nLocal>iOffset ){
+    /* In this case data must be written to the b-tree page. */
+    int iWrite = pInfo->nLocal - offset;
+    if( iWrite>iRem ){
+      iWrite = iRem;
+    }
+    rc = sqlite3PagerWrite(pCsr->pPage->pDbPage);
+    if( rc!=SQLITE_OK ){
+      return rc;
+    }
+    memcpy(&pInfo->pCell[iOffset+pInfo->nHeader], zRem, iWrite);
+
+    zRem += iWrite;
+    iRem -= iWrite;
   }
-  if( nCopy>0 ){
-    memcpy(&zData[offset], z, amt);
-    rc = sqlite3BtreeInsert(pCsr, 0, iKey, zData, nData, 0, 0);
+  iOffset = ((iOffset<pInfo->nLocal)?0:(iOffset-pInfo->nLocal));
+
+  ovflSize = pBt->usableSize - 4;
+  assert(pInfo->iOverflow>0 || iRem==0);
+  iOvfl = get4byte(&pInfo->pCell[pInfo->iOverflow]);
+  while( iRem>0 ){
+    if( iOffset>ovflSize ){
+      /* The only reason to read this page is to obtain the page
+      ** number for the next page in the overflow chain. So try
+      ** the getOverflowPage() shortcut.  */
+      rc = getOverflowPage(pBt, iOvfl, 0, &iOvfl);
+      if( rc!=SQLITE_OK ){
+        return rc;
+      }
+      iOffset -= ovflSize;
+    }else{
+      int iWrite = ovflSize - iOffset;
+      DbPage *pOvfl;          /* The overflow page. */
+      u8 *aData;              /* Page data */
+
+      rc = sqlite3PagerGet(pBt->pPager, iOvfl, &pOvfl);
+      if( rc!=SQLITE_OK ){
+        return rc;
+      }
+      rc = sqlite3PagerWrite(pOvfl);
+      if( rc!=SQLITE_OK ){
+        sqlite3PagerUnref(pOvfl);
+        return rc;
+      }
+
+      aData = sqlite3PagerGetData(pOvfl);
+      iOvfl = get4byte(aData);
+      if( iWrite>iRem ){
+        iWrite = iRem;
+      }
+      memcpy(&aData[iOffset+4], zRem, iWrite);
+      sqlite3PagerUnref(pOvfl);
+
+      zRem += iWrite;
+      iRem -= iWrite;
+      iOffset = ((iOffset<ovflSize)?0:(iOffset-ovflSize));
+    }
   }
 
-  sqliteFree(zData);
-  return rc;
+  return SQLITE_OK;
 }
 #endif
 
diff --git a/src/tclsqlite.c b/src/tclsqlite.c
index 936406183..70481a75a 100644
--- a/src/tclsqlite.c
+++ b/src/tclsqlite.c
@@ -12,7 +12,7 @@
 ** A TCL Interface to SQLite.  Append this file to sqlite3.c and
 ** compile the whole thing to build a TCL-enabled version of SQLite.
 **
-** $Id: tclsqlite.c,v 1.180 2007/05/01 17:49:49 danielk1977 Exp $
+** $Id: tclsqlite.c,v 1.181 2007/05/02 13:16:31 danielk1977 Exp $
 */
 #include "tcl.h"
 #include <errno.h>
@@ -89,6 +89,8 @@ struct SqlPreparedStmt {
   char zSql[1];            /* Text of the SQL statement */
 };
 
+typedef struct IncrblobChannel IncrblobChannel;
+
 /*
 ** There is one instance of this structure for each SQLite database
 ** that has been opened by the SQLite TCL interface.
@@ -114,20 +116,56 @@ struct SqliteDb {
   SqlPreparedStmt *stmtLast; /* Last statement in the list */
   int maxStmt;               /* The next maximum number of stmtList */
   int nStmt;                 /* Number of statements in stmtList */
+  IncrblobChannel *pIncrblob;/* Linked list of open incrblob channels */
 };
 
-typedef struct IncrblobChannel IncrblobChannel;
 struct IncrblobChannel {
-  sqlite3_blob *pBlob;
-  int iSeek;              /* Current seek offset */
+  SqliteDb *pDb;            /* Associated database connection */
+  sqlite3_blob *pBlob;      /* sqlite3 blob handle */
+  int iSeek;                /* Current seek offset */
+
+  Tcl_Channel channel;      /* Channel identifier */
+  IncrblobChannel *pNext;   /* Linked list of all open incrblob channels */
+  IncrblobChannel *pPrev;   /* Linked list of all open incrblob channels */
 };
 
 /*
+** Close all incrblob channels opened using database connection pDb.
+** This is called when shutting down the database connection.
+*/
+static void closeIncrblobChannels(SqliteDb *pDb){
+  IncrblobChannel *p;
+  IncrblobChannel *pNext;
+
+  for(p=pDb->pIncrblob; p; p=pNext){
+    pNext = p->pNext;
+
+    /* Note: Calling unregister here call Tcl_Close on the incrblob channel, 
+    ** which deletes the IncrblobChannel structure at *p. So do not
+    ** call Tcl_Free() here.
+    */
+    Tcl_UnregisterChannel(pDb->interp, p->channel);
+  }
+}
+
+/*
 ** Close an incremental blob channel.
 */
 static int incrblobClose(ClientData instanceData, Tcl_Interp *interp){
   IncrblobChannel *p = (IncrblobChannel *)instanceData;
   sqlite3_blob_close(p->pBlob);
+
+  /* Remove the channel from the SqliteDb.pIncrblob list. */
+  if( p->pNext ){
+    p->pNext->pPrev = p->pPrev;
+  }
+  if( p->pPrev ){
+    p->pPrev->pNext = p->pNext;
+  }
+  if( p->pDb->pIncrblob==p ){
+    p->pDb->pIncrblob = p->pNext;
+  }
+
   Tcl_Free((char *)p);
   return TCL_OK;
 }
@@ -164,6 +202,9 @@ static int incrblobInput(
   return nRead;
 }
 
+/*
+** Write data to an incremental blob channel.
+*/
 static int incrblobOutput(
   ClientData instanceData, 
   CONST char *buf, 
@@ -263,7 +304,6 @@ static int createIncrblobChannel(
   IncrblobChannel *p;
   sqlite3_blob *pBlob;
   int rc;
-  Tcl_Channel channel;
   int flags = TCL_READABLE|TCL_WRITABLE;
 
   /* This variable is used to name the channels: "incrblob_[incr count]" */
@@ -281,10 +321,19 @@ static int createIncrblobChannel(
   p->pBlob = pBlob;
 
   sprintf(zChannel, "incrblob_%d", ++count);
-  channel = Tcl_CreateChannel(&IncrblobChannelType, zChannel, p, flags);
-  Tcl_RegisterChannel(interp, channel);
+  p->channel = Tcl_CreateChannel(&IncrblobChannelType, zChannel, p, flags);
+  Tcl_RegisterChannel(interp, p->channel);
+
+  /* Link the new channel into the SqliteDb.pIncrblob list. */
+  p->pNext = pDb->pIncrblob;
+  p->pPrev = 0;
+  if( p->pNext ){
+    p->pNext->pPrev = p;
+  }
+  pDb->pIncrblob = p;
+  p->pDb = pDb;
 
-  Tcl_SetResult(interp, (char *)Tcl_GetChannelName(channel), TCL_VOLATILE);
+  Tcl_SetResult(interp, (char *)Tcl_GetChannelName(p->channel), TCL_VOLATILE);
   return TCL_OK;
 }
 
@@ -363,6 +412,7 @@ static void flushStmtCache( SqliteDb *pDb ){
 static void DbDeleteCmd(void *db){
   SqliteDb *pDb = (SqliteDb*)db;
   flushStmtCache(pDb);
+  closeIncrblobChannels(pDb);
   sqlite3_close(pDb->db);
   while( pDb->pFunc ){
     SqlFunc *pFunc = pDb->pFunc;
diff --git a/test/incrblob.test b/test/incrblob.test
index 3470f03af..176e61e67 100644
--- a/test/incrblob.test
+++ b/test/incrblob.test
@@ -9,7 +9,7 @@
 #
 #***********************************************************************
 #
-# $Id: incrblob.test,v 1.1 2007/05/01 17:49:49 danielk1977 Exp $
+# $Id: incrblob.test,v 1.2 2007/05/02 13:16:31 danielk1977 Exp $
 
 set testdir [file dirname $argv0]
 source $testdir/tester.tcl
@@ -48,6 +48,35 @@ do_test incrblob-1.2.6 {
   }
 } {1234567890}
 
-finish_test
+#--------------------------------------------------------------------
+# Test cases incrblob-2.X check that it is possible to read and write
+# regions of a blob that lie on overflow pages.
+do_test incrblob-2.1 {
+  set ::str "[string repeat . 10000]"
+  execsql {
+    INSERT INTO blobs(rowid, k, v) VALUES(3, 'three', $::str);
+  }
+} {}
+
+do_test incrblob-2.2 {
+  set ::blob [db incrblob blobs v 3]
+  seek $::blob 8500
+  read $::blob 10
+} {..........}
+
+do_test incrblob-2.3 {
+  seek $::blob 8500
+  puts -nonewline $::blob 1234567890
+} {}
+
+do_test incrblob-2.4 {
+  seek $::blob 8496
+  read $::blob 10
+} {....123456}
+
+do_test incrblob-2.10 {
+  close $::blob
+} {}
 
+finish_test