4 files changed, 260 insertions, 67 deletions

diff --git a/src/rowhash.c b/src/rowhash.c
index 28c344707..50559e6d8 100644
--- a/src/rowhash.c
+++ b/src/rowhash.c
@@ -31,7 +31,7 @@
 ** The caller is responsible for insuring that there are no duplicate
 ** INSERTs.
 **
-** $Id: rowhash.c,v 1.4 2009/04/21 18:20:45 danielk1977 Exp $
+** $Id: rowhash.c,v 1.5 2009/04/22 00:47:01 drh Exp $
 */
 #include "sqliteInt.h"
 
@@ -135,9 +135,9 @@ struct RowHashBlock {
 ** around and used as opaque handles by code in other modules.
 */
 struct RowHash {
-  int nUsed;              /* Number of used entries in first RowHashBlock */
-  int nEntry;             /* Number of used entries over all RowHashBlocks */
+  unsigned int nEntry;    /* Number of used entries over all RowHashBlocks */
   int iBatch;             /* The current insert batch number */
+  u16 nUsed;              /* Number of used entries in first RowHashBlock */
   u8 nHeight;             /* Height of tree of hash pages */
   u8 nLinearLimit;        /* Linear search limit (used if pHash==0) */
   int nBucket;            /* Number of buckets in hash table */
diff --git a/src/rowset.c b/src/rowset.c
index e8f7a52e4..4e27676cb 100644
--- a/src/rowset.c
+++ b/src/rowset.c
@@ -10,47 +10,89 @@
 **
 *************************************************************************
 **
-** This module implements an object we call a "Row Set".
+** This module implements an object we call a "RowSet".
 **
-** The RowSet object is a bag of rowids.  Rowids
-** are inserted into the bag in an arbitrary order.  Then they are
-** pulled from the bag in sorted order.  Rowids only appear in the
-** bag once.  If the same rowid is inserted multiple times, the
-** second and subsequent inserts make no difference on the output.
+** The RowSet object is a collection of rowids.  Rowids
+** are inserted into the RowSet in an arbitrary order.  Inserts
+** can be intermixed with tests to see if a given rowid has been
+** previously inserted into the RowSet.
 **
-** This implementation accumulates rowids in a linked list.  For
-** output, it first sorts the linked list (removing duplicates during
-** the sort) then returns elements one by one by walking the list.
+** After all inserts are finished, it is possible to extract the
+** elements of the RowSet in sorted order.  Once this extraction
+** process has started, no new elements may be inserted.
 **
-** Big chunks of rowid/next-ptr pairs are allocated at a time, to
-** reduce the malloc overhead.
+** Hence, the primitive operations for a RowSet are:
 **
-** $Id: rowset.c,v 1.4 2009/04/01 19:35:55 drh Exp $
+**    CREATE
+**    INSERT
+**    TEST
+**    SMALLEST
+**    DESTROY
+**
+** The CREATE and DESTROY primitives are the constructor and destructor,
+** obviously.  The INSERT primitive adds a new element to the RowSet.
+** TEST checks to see if an element is already in the RowSet.  SMALLEST
+** extracts the least value from the RowSet.
+**
+** The INSERT primitive might allocate additional memory.  Memory is
+** allocated in chunks so most INSERTs do no allocation.  There is an 
+** upper bound on the size of allocated memory.  No memory is freed
+** until DESTROY.
+**
+** The TEST primitive includes a "batch" number.  The TEST primitive
+** will only see elements that were inserted before the last change
+** in the batch number.  In other words, if an INSERT occurs between
+** two TESTs where the TESTs have the same batch nubmer, then the
+** value added by the INSERT will not be visible to the second TEST.
+** The initial batch number is zero, so if the very first TEST contains
+** a non-zero batch number, it will see all prior INSERTs.
+**
+** No INSERTs may occurs after a SMALLEST.  An assertion will fail if
+** that is attempted.
+**
+** The cost of an INSERT is roughly constant.  (Sometime new memory
+** has to be allocated on an INSERT.)  The cost of a TEST with a new
+** batch number is O(NlogN) where N is the number of elements in the RowSet.
+** The cost of a TEST using the same batch number is O(logN).  The cost
+** of the first SMALLEST is O(NlogN).  Second and subsequent SMALLEST
+** primitives are constant time.  The cost of DESTROY is O(N).
+**
+** There is an added cost of O(N) when switching between TEST and
+** SMALLEST primitives.
+**
+** $Id: rowset.c,v 1.5 2009/04/22 00:47:01 drh Exp $
 */
 #include "sqliteInt.h"
 
+
+/*
+** Target size for allocation chunks.
+*/
+#define ROWSET_ALLOCATION_SIZE 1024
+
 /*
 ** The number of rowset entries per allocation chunk.
 */
-#define ROWSET_ENTRY_PER_CHUNK  63
+#define ROWSET_ENTRY_PER_CHUNK  \
+                       ((ROWSET_ALLOCATION_SIZE-8)/sizeof(struct RowSetEntry))
 
 /*
-** Each entry in a RowSet is an instance of the following
-** structure:
+** Each entry in a RowSet is an instance of the following object.
 */
 struct RowSetEntry {            
   i64 v;                        /* ROWID value for this entry */
-  struct RowSetEntry *pNext;    /* Next entry on a list of all entries */
+  struct RowSetEntry *pRight;   /* Right subtree (larger entries) or list */
+  struct RowSetEntry *pLeft;    /* Left subtree (smaller entries) */
 };
 
 /*
-** Index entries are allocated in large chunks (instances of the
+** RowSetEntry objects are allocated in large chunks (instances of the
 ** following structure) to reduce memory allocation overhead.  The
 ** chunks are kept on a linked list so that they can be deallocated
 ** when the RowSet is destroyed.
 */
 struct RowSetChunk {
-  struct RowSetChunk *pNext;             /* Next chunk on list of them all */
+  struct RowSetChunk *pNextChunk;        /* Next chunk on list of them all */
   struct RowSetEntry aEntry[ROWSET_ENTRY_PER_CHUNK]; /* Allocated entries */
 };
 
@@ -62,11 +104,13 @@ struct RowSetChunk {
 struct RowSet {
   struct RowSetChunk *pChunk;    /* List of all chunk allocations */
   sqlite3 *db;                   /* The database connection */
-  struct RowSetEntry *pEntry;    /* List of entries in the 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 */
   u16 nFresh;                    /* Number of objects on pFresh */
-  u8 isSorted;                   /* True if content is sorted */
+  u8 isSorted;                   /* True if pEntry is sorted */
+  u8 iBatch;                     /* Current insert batch */
 };
 
 /*
@@ -89,25 +133,30 @@ RowSet *sqlite3RowSetInit(sqlite3 *db, void *pSpace, unsigned int N){
   p->db = db;
   p->pEntry = 0;
   p->pLast = 0;
+  p->pTree = 0;
   p->pFresh = (struct RowSetEntry*)&p[1];
   p->nFresh = (u16)((N - sizeof(*p))/sizeof(struct RowSetEntry));
   p->isSorted = 1;
+  p->iBatch = 0;
   return p;
 }
 
 /*
-** Deallocate all chunks from a RowSet.
+** Deallocate all chunks from a RowSet.  This frees all memory that
+** the RowSet has allocated over its lifetime.  This routine is
+** the destructor for the RowSet.
 */
 void sqlite3RowSetClear(RowSet *p){
   struct RowSetChunk *pChunk, *pNextChunk;
   for(pChunk=p->pChunk; pChunk; pChunk = pNextChunk){
-    pNextChunk = pChunk->pNext;
+    pNextChunk = pChunk->pNextChunk;
     sqlite3DbFree(p->db, pChunk);
   }
   p->pChunk = 0;
   p->nFresh = 0;
   p->pEntry = 0;
   p->pLast = 0;
+  p->pTree = 0;
   p->isSorted = 1;
 }
 
@@ -118,8 +167,8 @@ void sqlite3RowSetClear(RowSet *p){
 ** memory allocation fails.
 */
 void sqlite3RowSetInsert(RowSet *p, i64 rowid){
-  struct RowSetEntry *pEntry;
-  struct RowSetEntry *pLast;
+  struct RowSetEntry *pEntry;  /* The new entry */
+  struct RowSetEntry *pLast;   /* The last prior entry */
   assert( p!=0 );
   if( p->nFresh==0 ){
     struct RowSetChunk *pNew;
@@ -127,7 +176,7 @@ void sqlite3RowSetInsert(RowSet *p, i64 rowid){
     if( pNew==0 ){
       return;
     }
-    pNew->pNext = p->pChunk;
+    pNew->pNextChunk = p->pChunk;
     p->pChunk = pNew;
     p->pFresh = pNew->aEntry;
     p->nFresh = ROWSET_ENTRY_PER_CHUNK;
@@ -135,26 +184,27 @@ void sqlite3RowSetInsert(RowSet *p, i64 rowid){
   pEntry = p->pFresh++;
   p->nFresh--;
   pEntry->v = rowid;
-  pEntry->pNext = 0;
+  pEntry->pRight = 0;
   pLast = p->pLast;
   if( pLast ){
     if( p->isSorted && rowid<=pLast->v ){
       p->isSorted = 0;
     }
-    pLast->pNext = pEntry;
+    pLast->pRight = pEntry;
   }else{
-    assert( p->pEntry==0 );
+    assert( p->pEntry==0 ); /* Fires if INSERT after SMALLEST */
     p->pEntry = pEntry;
   }
   p->pLast = pEntry;
 }
 
 /*
-** Merge two lists of RowSet entries.  Remove duplicates.
+** Merge two lists of RowSetEntry objects.  Remove duplicates.
 **
-** The input lists are assumed to be in sorted order.
+** The input lists are connected via pRight pointers and are 
+** assumed to each already be in sorted order.
 */
-static struct RowSetEntry *boolidxMerge(
+static struct RowSetEntry *rowSetMerge(
   struct RowSetEntry *pA,    /* First sorted list to be merged */
   struct RowSetEntry *pB     /* Second sorted list to be merged */
 ){
@@ -163,34 +213,34 @@ static struct RowSetEntry *boolidxMerge(
 
   pTail = &head;
   while( pA && pB ){
-    assert( pA->pNext==0 || pA->v<=pA->pNext->v );
-    assert( pB->pNext==0 || pB->v<=pB->pNext->v );
+    assert( pA->pRight==0 || pA->v<=pA->pRight->v );
+    assert( pB->pRight==0 || pB->v<=pB->pRight->v );
     if( pA->v<pB->v ){
-      pTail->pNext = pA;
-      pA = pA->pNext;
-      pTail = pTail->pNext;
+      pTail->pRight = pA;
+      pA = pA->pRight;
+      pTail = pTail->pRight;
     }else if( pB->v<pA->v ){
-      pTail->pNext = pB;
-      pB = pB->pNext;
-      pTail = pTail->pNext;
+      pTail->pRight = pB;
+      pB = pB->pRight;
+      pTail = pTail->pRight;
     }else{
-      pA = pA->pNext;
+      pA = pA->pRight;
     }
   }
   if( pA ){
-    assert( pA->pNext==0 || pA->v<=pA->pNext->v );
-    pTail->pNext = pA;
+    assert( pA->pRight==0 || pA->v<=pA->pRight->v );
+    pTail->pRight = pA;
   }else{
-    assert( pB==0 || pB->pNext==0 || pB->v<=pB->pNext->v );
-    pTail->pNext = pB;
+    assert( pB==0 || pB->pRight==0 || pB->v<=pB->pRight->v );
+    pTail->pRight = pB;
   }
-  return head.pNext;
+  return head.pRight;
 }
 
 /*
-** Sort all elements of the RowSet into ascending order.
+** Sort all elements on the pEntry list of the RowSet into ascending order.
 */ 
-static void sqlite3RowSetSort(RowSet *p){
+static void rowSetSort(RowSet *p){
   unsigned int i;
   struct RowSetEntry *pEntry;
   struct RowSetEntry *aBucket[40];
@@ -199,35 +249,147 @@ static void sqlite3RowSetSort(RowSet *p){
   memset(aBucket, 0, sizeof(aBucket));
   while( p->pEntry ){
     pEntry = p->pEntry;
-    p->pEntry = pEntry->pNext;
-    pEntry->pNext = 0;
+    p->pEntry = pEntry->pRight;
+    pEntry->pRight = 0;
     for(i=0; aBucket[i]; i++){
-      pEntry = boolidxMerge(aBucket[i],pEntry);
+      pEntry = rowSetMerge(aBucket[i], pEntry);
       aBucket[i] = 0;
     }
     aBucket[i] = pEntry;
   }
   pEntry = 0;
   for(i=0; i<sizeof(aBucket)/sizeof(aBucket[0]); i++){
-    pEntry = boolidxMerge(pEntry,aBucket[i]);
+    pEntry = rowSetMerge(pEntry, aBucket[i]);
   }
   p->pEntry = pEntry;
   p->pLast = 0;
   p->isSorted = 1;
 }
 
+
+/*
+** The input, pIn, is a binary tree (or subtree) of RowSetEntry objects.
+** Convert this tree into a linked list connected by the pRight pointers
+** and return pointers to the first and last elements of the new list.
+*/
+static void rowSetTreeToList(
+  struct RowSetEntry *pIn,         /* Root of the input tree */
+  struct RowSetEntry **ppFirst,    /* Write head of the output list here */
+  struct RowSetEntry **ppLast      /* Write tail of the output list here */
+){
+  if( pIn==0 ){
+    *ppFirst = *ppLast = 0;
+  }
+  if( pIn->pLeft ){
+    struct RowSetEntry *p;
+    rowSetTreeToList(pIn->pLeft, ppFirst, &p);
+    p->pRight = pIn;
+  }else{
+    *ppFirst = pIn;
+  }
+  if( pIn->pRight ){
+    rowSetTreeToList(pIn->pRight, &pIn->pRight, ppLast);
+  }else{
+    *ppLast = pIn;
+  }
+  assert( (*ppLast)->pRight==0 );
+}
+
+
+/*
+** Convert a sorted list of elements (connected by pRight) into a binary
+** tree with depth of iDepth.  A depth of 1 means the tree contains a single
+** node taken from the head of *ppList.  A depth of 2 means a tree with
+** three nodes.  And so forth.
+**
+** Use as many entries from the input list as required and update the
+** *ppList to point to the unused elements of the list.  If the input
+** list contains too few elements, then construct an incomplete tree
+** and leave *ppList set to NULL.
+**
+** Return a pointer to the root of the constructed binary tree.
+*/
+static struct RowSetEntry *rowSetNDeepTree(
+  struct RowSetEntry **ppList,
+  int iDepth
+){
+  struct RowSetEntry *p;         /* Root of the new tree */
+  struct RowSetEntry *pLeft;     /* Left subtree */
+  if( *ppList==0 ){
+    return 0;
+  }
+  if( iDepth==1 ){
+    p = *ppList;
+    *ppList = p->pRight;
+    p->pLeft = p->pRight = 0;
+    return p;
+  }
+  pLeft = rowSetNDeepTree(ppList, iDepth-1);
+  p = *ppList;
+  if( p==0 ){
+    return pLeft;
+  }
+  p->pLeft = pLeft;
+  *ppList = p->pRight;
+  p->pRight = rowSetNDeepTree(ppList, iDepth-1);
+  return p;
+}
+
+/*
+** Convert a sorted list of elements into a binary tree. Make the tree
+** as deep as it needs to be in order to contain the entire list.
+*/
+static struct RowSetEntry *rowSetListToTree(struct RowSetEntry *pList){
+  int iDepth;           /* Depth of the tree so far */
+  struct RowSetEntry *p;       /* Current tree root */
+  struct RowSetEntry *pLeft;   /* Left subtree */
+
+  if( pList==0 ){
+    return 0;
+  }
+  p = pList;
+  pList = p->pRight;
+  p->pLeft = p->pRight = 0;
+  for(iDepth=1; pList; iDepth++){
+    pLeft = p;
+    p = pList;
+    pList = p->pRight;
+    p->pLeft = pLeft;
+    p->pRight = rowSetNDeepTree(&pList, iDepth);
+  }
+  return p;
+}
+
+/*
+** Convert the list in p->pEntry into a sorted list if it is not
+** sorted already.  If there is a binary tree on p->pTree, then
+** convert it into a list too and merge it into the p->pEntry list.
+*/
+static void rowSetToList(RowSet *p){
+  if( !p->isSorted ){
+    rowSetSort(p);
+  }
+  if( p->pTree ){
+    struct RowSetEntry *pHead, *pTail;
+    rowSetTreeToList(p->pTree, &pHead, &pTail);
+    p->pTree = 0;
+    p->pEntry = rowSetMerge(p->pEntry, pHead);
+  }
+}
+
 /*
-** Extract the next (smallest) element from the RowSet.
+** Extract the smallest element from the RowSet.
 ** Write the element into *pRowid.  Return 1 on success.  Return
 ** 0 if the RowSet is already empty.
+**
+** After this routine has been called, the sqlite3RowSetInsert()
+** routine may not be called again.  
 */
 int sqlite3RowSetNext(RowSet *p, i64 *pRowid){
-  if( !p->isSorted ){
-    sqlite3RowSetSort(p);
-  }
+  rowSetToList(p);
   if( p->pEntry ){
     *pRowid = p->pEntry->v;
-    p->pEntry = p->pEntry->pNext;
+    p->pEntry = p->pEntry->pRight;
     if( p->pEntry==0 ){
       sqlite3RowSetClear(p);
     }
@@ -236,3 +398,31 @@ int sqlite3RowSetNext(RowSet *p, i64 *pRowid){
     return 0;
   }
 }
+
+/*
+** 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.
+*/
+int sqlite3RowSetTest(RowSet *pRowSet, u8 iBatch, sqlite3_int64 iRowid){
+  struct RowSetEntry *p;
+  if( iBatch!=pRowSet->iBatch ){
+    if( pRowSet->pEntry ){
+      rowSetToList(pRowSet);
+      pRowSet->pTree = rowSetListToTree(pRowSet->pEntry);
+      pRowSet->pEntry = 0;
+      pRowSet->pLast = 0;
+    }
+    pRowSet->iBatch = iBatch;
+  }
+  p = pRowSet->pTree;
+  while( p ){
+    if( p->v<iRowid ){
+      p = p->pRight;
+    }else if( p->v>iRowid ){
+      p = p->pLeft;
+    }else{
+      return 1;
+    }
+  }
+  return 0;
+}
diff --git a/src/sqliteInt.h b/src/sqliteInt.h
index fe2b89333..2e51888e0 100644
--- a/src/sqliteInt.h
+++ b/src/sqliteInt.h
@@ -11,7 +11,7 @@
 *************************************************************************
 ** Internal interface definitions for SQLite.
 **
-** @(#) $Id: sqliteInt.h,v 1.856 2009/04/21 16:15:15 drh Exp $
+** @(#) $Id: sqliteInt.h,v 1.857 2009/04/22 00:47:01 drh Exp $
 */
 #ifndef _SQLITEINT_H_
 #define _SQLITEINT_H_
@@ -2400,6 +2400,7 @@ int sqlite3BitvecBuiltinTest(int,int*);
 RowSet *sqlite3RowSetInit(sqlite3*, void*, unsigned int);
 void sqlite3RowSetClear(RowSet*);
 void sqlite3RowSetInsert(RowSet*, i64);
+int sqlite3RowSetTest(RowSet*, u8 iBatch, i64);
 int sqlite3RowSetNext(RowSet*, i64*);
 
 int sqlite3RowhashInsert(sqlite3*, RowHash **pp, i64 iVal);
diff --git a/src/vdbe.c b/src/vdbe.c
index 6f22d9148..8cc33d0be 100644
--- a/src/vdbe.c
+++ b/src/vdbe.c
@@ -43,7 +43,7 @@
 ** in this file for details.  If in doubt, do not deviate from existing
 ** commenting and indentation practices when changing or adding code.
 **
-** $Id: vdbe.c,v 1.835 2009/04/21 16:15:15 drh Exp $
+** $Id: vdbe.c,v 1.836 2009/04/22 00:47:01 drh Exp $
 */
 #include "sqliteInt.h"
 #include "vdbeInt.h"
@@ -428,6 +428,8 @@ static void memTracePrint(FILE *out, Mem *p){
     fprintf(out, " i:%lld", p->u.i);
   }else if( p->flags & MEM_Real ){
     fprintf(out, " r:%g", p->r);
+  }else if( p->flags & MEM_RowSet ){
+    fprintf(out, " (rowset)");
   }else{
     char zBuf[200];
     sqlite3VdbeMemPrettyPrint(p, zBuf);
@@ -4631,23 +4633,23 @@ case OP_RowHash: {                     /* jump, in1, in3 */
   ** delete it now and initialize P1 with an empty row-hash (a null pointer
   ** is an acceptable representation of an empty row-hash).
   */
-  if( (pIn1->flags & MEM_RowHash)==0 ){
-    sqlite3VdbeMemReleaseExternal(pIn1);
-    pIn1->u.pRowHash = 0;
-    pIn1->flags = MEM_RowHash;
+  if( (pIn1->flags & MEM_RowSet)==0 ){
+    sqlite3VdbeMemSetRowSet(pIn1);
+    if( (pIn1->flags & MEM_RowSet)==0 ) goto no_mem;
   }
 
   assert( pOp->p4type==P4_INT32 );
   if( iSet ){
     int exists;
-    rc = sqlite3RowhashTest(pIn1->u.pRowHash, pOp->p4.i, pIn3->u.i, &exists);
+    exists = sqlite3RowSetTest(pIn1->u.pRowSet, iSet>=0 ? iSet & 0xf : 0xff,
+                               pIn3->u.i);
     if( exists ){
       pc = pOp->p2 - 1;
       break;
     }
   }
   if( iSet>=0 ){
-    rc = sqlite3RowhashInsert(db, &pIn1->u.pRowHash, pIn3->u.i);
+    sqlite3RowSetInsert(pIn1->u.pRowSet, pIn3->u.i);
   }
   break;
 }