The OP_Column opcode caches large column values coming from overflow pages.

FossilOrigin-Name: ab1edcc7fedcf27922d5db4bc1bc673b1495ca9c66eb6debdda7b7776c068888

5 files changed, 120 insertions, 13 deletions

diff --git a/src/btree.c b/src/btree.c
index 8817efc71..c23f86e1d 100644
--- a/src/btree.c
+++ b/src/btree.c
@@ -4826,7 +4826,6 @@ void sqlite3BtreeCursorUnpin(BtCursor *pCur){
   pCur->curFlags &= ~BTCF_Pinned;
 }
 
-#ifdef SQLITE_ENABLE_OFFSET_SQL_FUNC
 /*
 ** Return the offset into the database file for the start of the
 ** payload to which the cursor is pointing.
@@ -4838,7 +4837,6 @@ i64 sqlite3BtreeOffset(BtCursor *pCur){
   return (i64)pCur->pBt->pageSize*((i64)pCur->pPage->pgno - 1) +
          (i64)(pCur->info.pPayload - pCur->pPage->aData);
 }
-#endif /* SQLITE_ENABLE_OFFSET_SQL_FUNC */
 
 /*
 ** Return the number of bytes of payload for the entry that pCur is
diff --git a/src/btree.h b/src/btree.h
index b9078f901..b45ace7e1 100644
--- a/src/btree.h
+++ b/src/btree.h
@@ -321,9 +321,7 @@ int sqlite3BtreePrevious(BtCursor*, int flags);
 i64 sqlite3BtreeIntegerKey(BtCursor*);
 void sqlite3BtreeCursorPin(BtCursor*);
 void sqlite3BtreeCursorUnpin(BtCursor*);
-#ifdef SQLITE_ENABLE_OFFSET_SQL_FUNC
 i64 sqlite3BtreeOffset(BtCursor*);
-#endif
 int sqlite3BtreePayload(BtCursor*, u32 offset, u32 amt, void*);
 const void *sqlite3BtreePayloadFetch(BtCursor*, u32 *pAmt);
 u32 sqlite3BtreePayloadSize(BtCursor*);
diff --git a/src/vdbe.c b/src/vdbe.c
index 075a63211..f479440da 100644
--- a/src/vdbe.c
+++ b/src/vdbe.c
@@ -695,6 +695,83 @@ static u64 filterHash(const Mem *aMem, const Op *pOp){
   return h;
 }
 
+
+/*
+** For OP_Column, factor out the case where content is loaded from
+** overflow pages, so that the code to implement this case is separate
+** the common case where all content fits on the page.  Factoring out
+** the code reduces register pressure and helps the common case
+** to run faster.
+*/
+static SQLITE_NOINLINE int vdbeColumnFromOverflow(
+  VdbeCursor *pC,       /* The BTree cursor from which we are reading */
+  int iCol,             /* The column to read */
+  int t,                /* The serial-type code for the column value */
+  i64 iOffset,          /* Offset to the start of the content value */
+  Mem *pDest            /* Store the value into this register. */
+){
+  int rc;
+  sqlite3 *db = pDest->db;
+  int encoding = pDest->enc;
+  int len = sqlite3VdbeSerialTypeLen(t);
+  assert( pC->eCurType==CURTYPE_BTREE );
+  if( len>db->aLimit[SQLITE_LIMIT_LENGTH] ) return SQLITE_TOOBIG;
+  if( len > 4000 ){
+    /* Cache large column values that are on overflow pages using
+    ** an RCStr (reference counted string) so that if they are reloaded,
+    ** that do not have to be copied a second time.  The overhead of
+    ** creating and managing the cache is such that this is only
+    ** profitable for larger TEXT and BLOB values.
+    */
+    VdbeTxtBlbCache *pCache;
+    char *pBuf;
+    if( pC->colCache==0 ){
+      pC->pCache = sqlite3DbMallocZero(db, sizeof(VdbeTxtBlbCache) );
+      if( pC->pCache==0 ) return SQLITE_NOMEM;
+      pC->colCache = 1;
+    }
+    pCache = pC->pCache;
+    if( pCache->pCValue==0
+     || pCache->iCol!=iCol
+     || pCache->iOffset!=sqlite3BtreeOffset(pC->uc.pCursor)
+    ){
+      if( pCache->pCValue ) sqlite3RCStrUnref(pCache->pCValue);
+      pBuf = pCache->pCValue = sqlite3RCStrNew( len+3 );
+      if( pBuf==0 ) return SQLITE_NOMEM;
+      rc = sqlite3BtreePayload(pC->uc.pCursor, iOffset, len, pBuf);
+      if( rc ) return rc;
+      pBuf[len] = 0;
+      pBuf[len+1] = 0;
+      pBuf[len+2] = 0;
+      pCache->iCol = iCol;
+      pCache->iOffset = sqlite3BtreeOffset(pC->uc.pCursor);
+    }else{
+      pBuf = pCache->pCValue;
+    }
+    assert( t>=12 );
+    sqlite3RCStrRef(pBuf);
+    if( t&1 ){
+      rc = sqlite3VdbeMemSetStr(pDest, pBuf, len, encoding,
+                                (void(*)(void*))sqlite3RCStrUnref);
+      pDest->flags |= MEM_Term;
+    }else{
+      rc = sqlite3VdbeMemSetStr(pDest, pBuf, len, 0,
+                                (void(*)(void*))sqlite3RCStrUnref);
+    }
+  }else{
+    rc = sqlite3VdbeMemFromBtree(pC->uc.pCursor, iOffset, len, pDest);
+    if( rc ) return rc;
+    sqlite3VdbeSerialGet((const u8*)pDest->z, t, pDest);
+    if( (t&1)!=0 && encoding==SQLITE_UTF8 ){
+      pDest->z[len] = 0;
+      pDest->flags |= MEM_Term;
+    }
+  }
+  pDest->flags &= ~MEM_Ephem;
+  return rc;
+}
+
+
 /*
 ** Return the symbolic name for the data type of a pMem
 */
@@ -2902,6 +2979,10 @@ op_column_restart:
       ** dynamically allocated. */
       pC->aRow = 0;
       pC->szRow = 0;
+      if( pC->colCache && pC->pCache && pC->pCache->pCValue ){
+        sqlite3RCStrUnref(pC->pCache->pCValue);
+        pC->pCache->pCValue = 0;
+      }
 
       /* Make sure a corrupt database has not given us an oversize header.
       ** Do this now to avoid an oversize memory allocation.
@@ -3061,6 +3142,7 @@ op_column_restart:
   }else{
     u8 p5;
     pDest->enc = encoding;
+    assert( pDest->db==db );
     /* This branch happens only when content is on overflow pages */
     if( ((p5 = (pOp->p5 & OPFLAG_BYTELENARG))!=0
           && (p5==OPFLAG_TYPEOFARG
@@ -3084,14 +3166,12 @@ op_column_restart:
       */
       sqlite3VdbeSerialGet((u8*)sqlite3CtypeMap, t, pDest);
     }else{
-      if( len>db->aLimit[SQLITE_LIMIT_LENGTH] ) goto too_big;
-      rc = sqlite3VdbeMemFromBtree(pC->uc.pCursor, aOffset[p2], len, pDest);
-      if( rc!=SQLITE_OK ) goto abort_due_to_error;
-      sqlite3VdbeSerialGet((const u8*)pDest->z, t, pDest);
-      if( (t&1)!=0 && encoding==SQLITE_UTF8 ){
-        pDest->flags |= MEM_Term;
+      rc = vdbeColumnFromOverflow(pC, p2, t, aOffset[p2], pDest);
+      if( rc ){
+        if( rc==SQLITE_NOMEM ) goto no_mem;
+        if( rc==SQLITE_TOOBIG ) goto too_big;
+        goto abort_due_to_error;
       }
-      pDest->flags &= ~MEM_Ephem;
     }
   }
 
diff --git a/src/vdbeInt.h b/src/vdbeInt.h
index 3a5b961a2..f2ca65675 100644
--- a/src/vdbeInt.h
+++ b/src/vdbeInt.h
@@ -56,6 +56,9 @@ typedef struct VdbeSorter VdbeSorter;
 /* Elements of the linked list at Vdbe.pAuxData */
 typedef struct AuxData AuxData;
 
+/* A cache of large TEXT or BLOB values in a VdbeCursor */
+typedef struct VdbeTxtBlbCache VdbeTxtBlbCache;
+
 /* Types of VDBE cursors */
 #define CURTYPE_BTREE       0
 #define CURTYPE_SORTER      1
@@ -87,6 +90,7 @@ struct VdbeCursor {
   Bool useRandomRowid:1;  /* Generate new record numbers semi-randomly */
   Bool isOrdered:1;       /* True if the table is not BTREE_UNORDERED */
   Bool noReuse:1;         /* OpenEphemeral may not reuse this cursor */
+  Bool colCache:1;        /* pCache pointer is initialized and non-NULL */
   u16 seekHit;            /* See the OP_SeekHit and OP_IfNoHope opcodes */
   union {                 /* pBtx for isEphermeral.  pAltMap otherwise */
     Btree *pBtx;            /* Separate file holding temporary table */
@@ -127,6 +131,7 @@ struct VdbeCursor {
 #ifdef SQLITE_ENABLE_COLUMN_USED_MASK
   u64 maskUsed;           /* Mask of columns used by this cursor */
 #endif
+  VdbeTxtBlbCache *pCache; /* Cache of large TEXT or BLOB values */
 
   /* 2*nField extra array elements allocated for aType[], beyond the one
   ** static element declared in the structure.  nField total array slots for
@@ -139,13 +144,24 @@ struct VdbeCursor {
 #define IsNullCursor(P) \
   ((P)->eCurType==CURTYPE_PSEUDO && (P)->nullRow && (P)->seekResult==0)
 
-
 /*
 ** A value for VdbeCursor.cacheStatus that means the cache is always invalid.
 */
 #define CACHE_STALE 0
 
 /*
+** Large TEXT or BLOB values can be slow to load, so we want to avoid
+** loading them more than once.  For that reason, large TEXT and BLOB values
+** can be stored in a cache defined by this object, and attached to the
+** VdbeCursor using the pCache field.
+*/
+struct VdbeTxtBlbCache {
+  char *pCValue;        /* A RCStr buffer to hold the value */
+  i64 iOffset;          /* File offset of the row being cached */
+  int iCol;             /* Column for which the cache is valid */
+};
+
+/*
 ** When a sub-program is executed (OP_Program), a structure of this type
 ** is allocated to store the current value of the program counter, as
 ** well as the current memory cell array and various other frame specific
diff --git a/src/vdbeaux.c b/src/vdbeaux.c
index a0eff155d..52d3c9e06 100644
--- a/src/vdbeaux.c
+++ b/src/vdbeaux.c
@@ -1498,7 +1498,6 @@ void sqlite3VdbeReleaseRegisters(
 }
 #endif /* SQLITE_DEBUG */
 
-
 /*
 ** Change the value of the P4 operand for a specific instruction.
 ** This routine is useful when a large program is loaded from a
@@ -2723,7 +2722,23 @@ void sqlite3VdbeMakeReady(
 void sqlite3VdbeFreeCursor(Vdbe *p, VdbeCursor *pCx){
   if( pCx ) sqlite3VdbeFreeCursorNN(p,pCx);
 }
+static SQLITE_NOINLINE void freeCursorWithCache(Vdbe *p, VdbeCursor *pCx){
+  VdbeTxtBlbCache *pCache = pCx->pCache;
+  assert( pCx->colCache );
+  pCx->colCache = 0;
+  pCx->pCache = 0;
+  if( pCache->pCValue ){
+    sqlite3RCStrUnref(pCache->pCValue);
+    pCache->pCValue = 0;
+  }
+  sqlite3DbFree(p->db, pCache);
+  sqlite3VdbeFreeCursorNN(p, pCx);
+}
 void sqlite3VdbeFreeCursorNN(Vdbe *p, VdbeCursor *pCx){
+  if( pCx->colCache ){
+    freeCursorWithCache(p, pCx);
+    return;
+  }
   switch( pCx->eCurType ){
     case CURTYPE_SORTER: {
       sqlite3VdbeSorterClose(p->db, pCx);