Performance improvements for the RowSet object when it undergoes many

cycles between RowSetInsert and RowSetTest.

FossilOrigin-Name: 49d20ede5f4c0895a165126d5cf7c95a0510ba35

manifest

@@ -1,5 +1,5 @@
-C Add\sthe\s".trace"\soption\sto\sthe\scommand-line\sshell.
+C Performance\simprovements\sfor\sthe\sRowSet\sobject\swhen\sit\sundergoes\smany\ncycles\sbetween\sRowSetInsert\sand\sRowSetTest.
-D 2012-04-04T16:56:23.330
+D 2012-04-05T01:37:32.902
 F Makefile.arm-wince-mingw32ce-gcc d6df77f1f48d690bd73162294bbba7f59507c72f
 F Makefile.in 2f37e468503dbe79d35c9f6dffcf3fae1ae9ec20
 F Makefile.linux-gcc 91d710bdc4998cb015f39edf3cb314ec4f4d7e23
@@ -180,7 +180,7 @@ F src/prepare.c ec4989f7f480544bdc4192fe663470d2a2d7d61e
 F src/printf.c 7ffb4ebb8b341f67e049695ba031da717b3d2699
 F src/random.c cd4a67b3953b88019f8cd4ccd81394a8ddfaba50
 F src/resolve.c 3d3e80a98f203ac6b9329e9621e29eda85ddfd40
-F src/rowset.c 69afa95a97c524ba6faf3805e717b5b7ae85a697
+F src/rowset.c f6a49f3e9579428024662f6e2931832511f831a1
 F src/select.c 5e0e481c7d215d3c7ca8ccae1e688aa30163c6c1
 F src/shell.c ce4d41582182b8fad3be364e2fa295b70bc342ab
 F src/sqlite.h.in 11a883919b0baf4ffaa7550cfeef99be613ec2bf
@@ -1000,7 +1000,7 @@ F tool/tostr.awk e75472c2f98dd76e06b8c9c1367f4ab07e122d06
 F tool/vdbe-compress.tcl d70ea6d8a19e3571d7ab8c9b75cba86d1173ff0f
 F tool/warnings-clang.sh 9f406d66e750e8ac031c63a9ef3248aaa347ef2a
 F tool/warnings.sh fbc018d67fd7395f440c28f33ef0f94420226381
-P 9e1e2fe2950bb96784413eae934314d95bce08e7
+P b9ac3d7e340eb616fd23cc7dbdef6fdd66a79fe4
-R 7ce64940a4af6b82ea18cf10b0da61e8
+R a1d15ac4a2f200ebad8c15bdd80077ea
 U drh
-Z 783b4beff01e2d5037d4d8570a1550e1
+Z 3eabbe65a412e07d9d0e849fd433908a

manifest.uuid

@@ -1 +1 @@
-b9ac3d7e340eb616fd23cc7dbdef6fdd66a79fe4
+49d20ede5f4c0895a165126d5cf7c95a0510ba35

src/rowset.c

@@ -76,6 +76,11 @@
 
 /*
 ** Each entry in a RowSet is an instance of the following object.
+**
+** This same object is reused to store a linked list of trees of RowSetEntry
+** objects.  In that alternative use, pRight points to the next entry
+** in the list, pLeft points to the tree, and v is unused.  The
+** RowSet.pForest value points to the head of this forest list.
 */
 struct RowSetEntry {            
   i64 v;                        /* ROWID value for this entry */
@@ -105,12 +110,18 @@ struct RowSet {
   struct RowSetEntry *pEntry;    /* List of entries using pRight */
   struct RowSetEntry *pLast;     /* Last entry on the pEntry list */
   struct RowSetEntry *pFresh;    /* Source of new entry objects */
-  struct RowSetEntry *pTree;     /* Binary tree of entries */
+  struct RowSetEntry *pForest;   /* List of binary trees of entries */
   u16 nFresh;                    /* Number of objects on pFresh */
-  u8 isSorted;                   /* True if pEntry is sorted */
+  u8 rsFlags;                    /* Various flags */
   u8 iBatch;                     /* Current insert batch */
 };
 
+/*
+** Allowed values for RowSet.rsFlags
+*/
+#define ROWSET_SORTED  0x01   /* True if RowSet.pEntry is sorted */
+#define ROWSET_NEXT    0x02   /* True if sqlite3RowSetNext() has been called */
+
 /*
 ** Turn bulk memory into a RowSet object.  N bytes of memory
 ** are available at pSpace.  The db pointer is used as a memory context
@@ -131,10 +142,10 @@ RowSet *sqlite3RowSetInit(sqlite3 *db, void *pSpace, unsigned int N){
   p->db = db;
   p->pEntry = 0;
   p->pLast = 0;
-  p->pTree = 0;
+  p->pForest = 0;
   p->pFresh = (struct RowSetEntry*)(ROUND8(sizeof(*p)) + (char*)p);
   p->nFresh = (u16)((N - ROUND8(sizeof(*p)))/sizeof(struct RowSetEntry));
-  p->isSorted = 1;
+  p->rsFlags = ROWSET_SORTED;
   p->iBatch = 0;
   return p;
 }
@@ -154,8 +165,33 @@ void sqlite3RowSetClear(RowSet *p){
   p->nFresh = 0;
   p->pEntry = 0;
   p->pLast = 0;
-  p->pTree = 0;
+  p->pForest = 0;
-  p->isSorted = 1;
+  p->rsFlags = ROWSET_SORTED;
+}
+
+/*
+** Allocate a new RowSetEntry object that is associated with the
+** given RowSet.  Return a pointer to the new and completely uninitialized
+** objected.
+**
+** In an OOM situation, the RowSet.db->mallocFailed flag is set and this
+** routine returns NULL.
+*/
+static struct RowSetEntry *rowSetEntryAlloc(RowSet *p){
+  assert( p!=0 );
+  if( p->nFresh==0 ){
+    struct RowSetChunk *pNew;
+    pNew = sqlite3DbMallocRaw(p->db, sizeof(*pNew));
+    if( pNew==0 ){
+      return 0;
+    }
+    pNew->pNextChunk = p->pChunk;
+    p->pChunk = pNew;
+    p->pFresh = pNew->aEntry;
+    p->nFresh = ROWSET_ENTRY_PER_CHUNK;
+  }
+  p->nFresh--;
+  return p->pFresh++;
 }
 
 /*
@@ -167,30 +203,21 @@ void sqlite3RowSetClear(RowSet *p){
 void sqlite3RowSetInsert(RowSet *p, i64 rowid){
   struct RowSetEntry *pEntry;  /* The new entry */
   struct RowSetEntry *pLast;   /* The last prior entry */
-  assert( p!=0 );
+
-  if( p->nFresh==0 ){
+  /* This routine is never called after sqlite3RowSetNext() */
-    struct RowSetChunk *pNew;
+  assert( p!=0 && (p->rsFlags & ROWSET_NEXT)==0 );
-    pNew = sqlite3DbMallocRaw(p->db, sizeof(*pNew));
+
-    if( pNew==0 ){
+  pEntry = rowSetEntryAlloc(p);
-      return;
+  if( pEntry==0 ) return;
-    }
-    pNew->pNextChunk = p->pChunk;
-    p->pChunk = pNew;
-    p->pFresh = pNew->aEntry;
-    p->nFresh = ROWSET_ENTRY_PER_CHUNK;
-  }
-  pEntry = p->pFresh++;
-  p->nFresh--;
   pEntry->v = rowid;
   pEntry->pRight = 0;
   pLast = p->pLast;
   if( pLast ){
-    if( p->isSorted && rowid<=pLast->v ){
+    if( (p->rsFlags & ROWSET_SORTED)!=0 && rowid<=pLast->v ){
-      p->isSorted = 0;
+      p->rsFlags &= ~ROWSET_SORTED;
     }
     pLast->pRight = pEntry;
   }else{
-    assert( p->pEntry==0 ); /* Fires if INSERT after SMALLEST */
     p->pEntry = pEntry;
   }
   p->pLast = pEntry;
@@ -202,7 +229,7 @@ void sqlite3RowSetInsert(RowSet *p, i64 rowid){
 ** The input lists are connected via pRight pointers and are 
 ** assumed to each already be in sorted order.
 */
-static struct RowSetEntry *rowSetMerge(
+static struct RowSetEntry *rowSetEntryMerge(
   struct RowSetEntry *pA,    /* First sorted list to be merged */
   struct RowSetEntry *pB     /* Second sorted list to be merged */
 ){
@@ -236,32 +263,29 @@ static struct RowSetEntry *rowSetMerge(
 }
 
 /*
-** Sort all elements on the pEntry list of the RowSet into ascending order.
+** Sort all elements on the list of RowSetEntry objects into order of
+** increasing v.
 */ 
-static void rowSetSort(RowSet *p){
+static struct RowSetEntry *rowSetEntrySort(struct RowSetEntry *pIn){
   unsigned int i;
-  struct RowSetEntry *pEntry;
+  struct RowSetEntry *pNext, *aBucket[40];
-  struct RowSetEntry *aBucket[40];
 
-  assert( p->isSorted==0 );
   memset(aBucket, 0, sizeof(aBucket));
-  while( p->pEntry ){
+  while( pIn ){
-    pEntry = p->pEntry;
+    pNext = pIn->pRight;
-    p->pEntry = pEntry->pRight;
+    pIn->pRight = 0;
-    pEntry->pRight = 0;
     for(i=0; aBucket[i]; i++){
-      pEntry = rowSetMerge(aBucket[i], pEntry);
+      pIn = rowSetEntryMerge(aBucket[i], pIn);
       aBucket[i] = 0;
     }
-    aBucket[i] = pEntry;
+    aBucket[i] = pIn;
+    pIn = pNext;
   }
-  pEntry = 0;
+  pIn = 0;
   for(i=0; i<sizeof(aBucket)/sizeof(aBucket[0]); i++){
-    pEntry = rowSetMerge(pEntry, aBucket[i]);
+    pIn = rowSetEntryMerge(pIn, aBucket[i]);
   }
-  p->pEntry = pEntry;
+  return pIn;
-  p->pLast = 0;
-  p->isSorted = 1;
 }
 
 
@@ -355,20 +379,37 @@ static struct RowSetEntry *rowSetListToTree(struct RowSetEntry *pList){
 }
 
 /*
-** Convert the list in p->pEntry into a sorted list if it is not
+** Take all the entries on p->pEntry and on the trees in p->pForest and
-** sorted already.  If there is a binary tree on p->pTree, then
+** sort them all together into one big ordered list on p->pEntry.
-** convert it into a list too and merge it into the p->pEntry list.
+**
+** This routine should only be called once in the life of a RowSet.
 */
 static void rowSetToList(RowSet *p){
-  if( !p->isSorted ){
+
-    rowSetSort(p);
+  /* This routine is called only once */
+  assert( p!=0 && (p->rsFlags & ROWSET_NEXT)==0 );
+
+  if( (p->rsFlags & ROWSET_SORTED)==0 ){
+    p->pEntry = rowSetEntrySort(p->pEntry);
   }
-  if( p->pTree ){
+
+  /* While this module could theoretically support it, sqlite3RowSetNext()
+  ** is never called after sqlite3RowSetText() for the same RowSet.  So
+  ** there is never a forest to deal with.  Should this change, simply
+  ** remove the assert() and the #if 0. */
+  assert( p->pForest==0 );
+#if 0
+  while( p->pForest ){
+    struct RowSetEntry *pTree = p->pForest->pLeft;
+    if( pTree ){
       struct RowSetEntry *pHead, *pTail;
-    rowSetTreeToList(p->pTree, &pHead, &pTail);
+      rowSetTreeToList(pTree, &pHead, &pTail);
-    p->pTree = 0;
+      p->pEntry = rowSetEntryMerge(p->pEntry, pHead);
-    p->pEntry = rowSetMerge(p->pEntry, pHead);
     }
+    p->pForest = p->pForest->pRight;
+  }
+#endif
+  p->rsFlags |= ROWSET_NEXT;  /* Verify this routine is never called again */
 }
 
 /*
@@ -380,7 +421,12 @@ static void rowSetToList(RowSet *p){
 ** routine may not be called again.  
 */
 int sqlite3RowSetNext(RowSet *p, i64 *pRowid){
-  rowSetToList(p);
+  assert( p!=0 );
+
+  /* Merge the forest into a single sorted list on first call */
+  if( (p->rsFlags & ROWSET_NEXT)==0 ) rowSetToList(p);
+
+  /* Return the next entry on the list */
   if( p->pEntry ){
     *pRowid = p->pEntry->v;
     p->pEntry = p->pEntry->pRight;
@@ -396,19 +442,58 @@ int sqlite3RowSetNext(RowSet *p, i64 *pRowid){
 /*
 ** Check to see if element iRowid was inserted into the the rowset as
 ** part of any insert batch prior to iBatch.  Return 1 or 0.
+**
+** If this is the first test of a new batch and if there exist entires
+** on pRowSet->pEntry, then sort those entires into the forest at
+** pRowSet->pForest so that they can be tested.
 */
 int sqlite3RowSetTest(RowSet *pRowSet, u8 iBatch, sqlite3_int64 iRowid){
-  struct RowSetEntry *p;
+  struct RowSetEntry *p, *pTree;
+
+  /* This routine is never called after sqlite3RowSetNext() */
+  assert( pRowSet!=0 && (pRowSet->rsFlags & ROWSET_NEXT)==0 );
+
+  /* Sort entries into the forest on the first test of a new batch 
+  */
   if( iBatch!=pRowSet->iBatch ){
-    if( pRowSet->pEntry ){
+    p = pRowSet->pEntry;
-      rowSetToList(pRowSet);
+    if( p ){
-      pRowSet->pTree = rowSetListToTree(pRowSet->pEntry);
+      struct RowSetEntry **ppPrevTree = &pRowSet->pForest;
+      if( (pRowSet->rsFlags & ROWSET_SORTED)==0 ){
+        p = rowSetEntrySort(p);
+      }
+      for(pTree = pRowSet->pForest; pTree; pTree=pTree->pRight){
+        ppPrevTree = &pTree->pRight;
+        if( pTree->pLeft==0 ){
+          pTree->pLeft = rowSetListToTree(p);
+          break;
+        }else{
+          struct RowSetEntry *pAux, *pTail;
+          rowSetTreeToList(pTree->pLeft, &pAux, &pTail);
+          pTree->pLeft = 0;
+          p = rowSetEntryMerge(pAux, p);
+        }
+      }
+      if( pTree==0 ){
+        *ppPrevTree = pTree = rowSetEntryAlloc(pRowSet);
+        if( pTree ){
+          pTree->v = 0;
+          pTree->pRight = 0;
+          pTree->pLeft = rowSetListToTree(p);
+        }
+      }
       pRowSet->pEntry = 0;
       pRowSet->pLast = 0;
+      pRowSet->rsFlags |= ROWSET_SORTED;
     }
     pRowSet->iBatch = iBatch;
   }
-  p = pRowSet->pTree;
+
+  /* Test to see if the iRowid value appears anywhere in the forest.
+  ** Return 1 if it does and 0 if not.
+  */
+  for(pTree = pRowSet->pForest; pTree; pTree=pTree->pRight){
+    p = pTree->pLeft;
     while( p ){
       if( p->v<iRowid ){
         p = p->pRight;
@@ -418,5 +503,6 @@ int sqlite3RowSetTest(RowSet *pRowSet, u8 iBatch, sqlite3_int64 iRowid){
         return 1;
       }
     }
+  }
   return 0;
 }