The compound-select merge optimization passes quick.test with no errors. (CVS 5299)

FossilOrigin-Name: 8bbfa97837a74ef0514e0c92ea2a6576f02cc361

5 files changed, 294 insertions, 134 deletions

diff --git a/src/select.c b/src/select.c
index b54230e38..7045c32b2 100644
--- a/src/select.c
+++ b/src/select.c
@@ -12,7 +12,7 @@
 ** This file contains C code routines that are called by the parser
 ** to handle SELECT statements in SQLite.
 **
-** $Id: select.c,v 1.435 2008/06/24 12:46:31 drh Exp $
+** $Id: select.c,v 1.436 2008/06/25 00:12:41 drh Exp $
 */
 #include "sqliteInt.h"
 
@@ -430,7 +430,7 @@ static void pushOntoSorter(
   if( pSelect->iLimit ){
     int addr1, addr2;
     int iLimit;
-    if( pSelect->pOffset ){
+    if( pSelect->iOffset ){
       iLimit = pSelect->iOffset+1;
     }else{
       iLimit = pSelect->iLimit;
@@ -561,8 +561,9 @@ static void selectInnerLoop(
     nResultCol = pEList->nExpr;
   }
   if( pDest->iMem==0 ){
-    pDest->iMem = sqlite3GetTempRange(pParse, nResultCol);
+    pDest->iMem = pParse->nMem+1;
     pDest->nMem = nResultCol;
+    pParse->nMem += nResultCol;
   }else if( pDest->nMem!=nResultCol ){
     /* This happens when two SELECTs of a compound SELECT have differing
     ** numbers of result columns.  The error message will be generated by
@@ -1787,6 +1788,7 @@ static void computeLimitRegisters(Parse *pParse, Select *p, int iBreak){
   int iLimit = 0;
   int iOffset;
   int addr1;
+  if( p->iLimit ) return;
 
   /* 
   ** "LIMIT -1" always shows all rows.  There is some
@@ -1971,7 +1973,7 @@ static int multiSelect(
     goto multi_select_end;
   }
 
-#if 0
+#if 1
   if( p->pOrderBy ){
     return multiSelectOrderBy(pParse, p, pDest, aff);
   }
@@ -2312,13 +2314,32 @@ multi_select_end:
 /*
 ** Code an output subroutine for a coroutine implementation of a
 ** SELECT statment.
+**
+** The data to be output is contained in pIn->iMem.  There are
+** pIn->nMem columns to be output.  pDest is where the output should
+** be sent.
+**
+** regReturn is the number of the register holding the subroutine
+** return address.
+**
+** If regPrev>0 then it is a the first register in a vector that
+** records the previous output.  mem[regPrev] is a flag that is false
+** if there has been no previous output.  If regPrev>0 then code is
+** generated to suppress duplicates.  pKeyInfo is used for comparing
+** keys.
+**
+** If the LIMIT found in p->iLimit is reached, jump immediately to
+** iBreak.
 */
-static int outputSubroutine(
+static int generateOutputSubroutine(
   Parse *pParse,          /* Parsing context */
   Select *p,              /* The SELECT statement */
   SelectDest *pIn,        /* Coroutine supplying data */
   SelectDest *pDest,      /* Where to send the data */
   int regReturn,          /* The return address register */
+  int regPrev,            /* Previous result register.  No uniqueness if 0 */
+  KeyInfo *pKeyInfo,      /* For comparing with previous entry */
+  int p4type,             /* The p4 type for pKeyInfo */
   int iBreak              /* Jump here if we hit the LIMIT */
 ){
   Vdbe *v = pParse->pVdbe;
@@ -2328,6 +2349,22 @@ static int outputSubroutine(
 
   addr = sqlite3VdbeCurrentAddr(v);
   iContinue = sqlite3VdbeMakeLabel(v);
+
+  /* Suppress duplicates for UNION, EXCEPT, and INTERSECT 
+  */
+  if( regPrev ){
+    int j1, j2;
+    j1 = sqlite3VdbeAddOp1(v, OP_IfNot, regPrev);
+    j2 = sqlite3VdbeAddOp4(v, OP_Compare, pIn->iMem, regPrev+1, pIn->nMem,
+                              (char*)pKeyInfo, p4type);
+    sqlite3VdbeAddOp3(v, OP_Jump, j2+2, iContinue, j2+2);
+    sqlite3VdbeJumpHere(v, j1);
+    sqlite3ExprCodeCopy(pParse, pIn->iMem, regPrev+1, pIn->nMem);
+    sqlite3VdbeAddOp2(v, OP_Integer, 1, regPrev);
+  }
+
+  /* Suppress the the first OFFSET entries if there is an OFFSET clause
+  */
   codeOffset(v, p, iContinue);
 
   switch( pDest->eDest ){
@@ -2425,13 +2462,9 @@ static int outputSubroutine(
     sqlite3VdbeAddOp2(v, OP_IfZero, p->iLimit, iBreak);
   }
 
-  /* Advance the coroutine to its next value.
-  */
-  sqlite3VdbeResolveLabel(v, iContinue);
-  sqlite3VdbeAddOp1(v, OP_Yield, pIn->iParm);
-
   /* Generate the subroutine return
   */
+  sqlite3VdbeResolveLabel(v, iContinue);
   sqlite3VdbeAddOp1(v, OP_Return, regReturn);
 
   return addr;
@@ -2475,18 +2508,25 @@ static int outputSubroutine(
 **
 **             UNION ALL         UNION            EXCEPT          INTERSECT
 **          -------------  -----------------  --------------  -----------------
-**   AltB:   outA, nextA      outA, nextA       outA, nextA        nextA
+**   AltB:   outA, nextA      outA, nextA       outA, nextA         nextA
+**
+**   AeqB:   outA, nextA         nextA             nextA         outA, nextA
+**
+**   AgtB:   outB, nextB      outB, nextB          nextB            nextB
 **
-**   AeqB:   outA, nextA         nextA             nextA           outA
-**                                                            nextA while A==B
+**   EofA:   outB, nextB      outB, nextB          halt             halt
 **
-**   AgtB:   outB, nextB      outB, nextB          nextB           nextB
+**   EofB:   outA, nextA      outA, nextA       outA, nextA         halt
 **
-**   EofA:   outB, nextB      A<-B, outB,          halt            halt
-**                          nextB while A==B
+** In the AltB, AeqB, and AgtB subroutines, an EOF on A following nextA
+** causes an immediate jump to EofA and an EOF on B following nextB causes
+** an immediate jump to EofB.  Within EofA and EofB, and EOF on entry or
+** following nextX causes a jump to the end of the select processing.
 **
-**   EofB:   outA, nextA      B<-A, outA        outA, nextA        halt
-**                          nextA while A==B
+** Duplicate removal in the UNION, EXCEPT, and INTERSECT cases is handled
+** within the output subroutine.  The regPrev register set holds the previously
+** output value.  A comparison is made against this value and the output
+** is skipped if the next results would be the same as the previous.
 **
 ** The implementation plan is to implement the two coroutines and seven
 ** subroutines first, then put the control logic at the bottom.  Like this:
@@ -2521,6 +2561,7 @@ static int multiSelectOrderBy(
   SelectDest *pDest,    /* What to do with query results */
   char *aff             /* If eDest is SRT_Union, the affinity string */
 ){
+  int i, j;             /* Loop counters */
   Select *pPrior;       /* Another SELECT immediately to our left */
   Vdbe *v;              /* Generate code to this VDBE */
   SelectDest destA;     /* Destination for coroutine A */
@@ -2542,41 +2583,147 @@ static int multiSelectOrderBy(
   int addrAgtB;         /* Address of the A>B subroutine */
   int regLimitA;        /* Limit register for select-A */
   int regLimitB;        /* Limit register for select-A */
+  int regPrev;          /* A range of registers to hold previous output */
   int savedLimit;       /* Saved value of p->iLimit */
   int savedOffset;      /* Saved value of p->iOffset */
   int labelCmpr;        /* Label for the start of the merge algorithm */
   int labelEnd;         /* Label for the end of the overall SELECT stmt */
-  int j1, j2, j3;       /* Jump instructions that get retargetted */
+  int j1;               /* Jump instructions that get retargetted */
   int op;               /* One of TK_ALL, TK_UNION, TK_EXCEPT, TK_INTERSECT */
-  KeyInfo *pKeyInfo;    /* Type data for comparisons */
-  int p4type;           /* P4 type used for pKeyInfo */
+  KeyInfo *pKeyDup;     /* Comparison information for duplicate removal */
+  KeyInfo *pKeyMerge;   /* Comparison information for merging rows */
+  sqlite3 *db;          /* Database connection */
+  ExprList *pOrderBy;   /* The ORDER BY clause */
+  int nOrderBy;         /* Number of terms in the ORDER BY clause */
+  int *aPermute;        /* Mapping from ORDER BY terms to result set columns */
+  u8 NotUsed;           /* Dummy variables */
 
   assert( p->pOrderBy!=0 );
+  db = pParse->db;
   v = pParse->pVdbe;
   if( v==0 ) return SQLITE_NOMEM;
   labelEnd = sqlite3VdbeMakeLabel(v);
   labelCmpr = sqlite3VdbeMakeLabel(v);
-  pKeyInfo = keyInfoFromExprList(pParse, p->pEList);
-  p4type = P4_KEYINFO_HANDOFF;
+
 
   /* Patch up the ORDER BY clause
   */
   op = p->op;  
   pPrior = p->pPrior;
   assert( pPrior->pOrderBy==0 );
-  if( processCompoundOrderBy(pParse, p, 0) ){
-    return SQLITE_ERROR;
+  pOrderBy = p->pOrderBy;
+  if( pOrderBy ){
+    if( processCompoundOrderBy(pParse, p, 0) ){
+      return SQLITE_ERROR;
+    }
+    nOrderBy = pOrderBy->nExpr;
+  }else{
+    nOrderBy = 0;
+  }
+  /* For operators other than UNION ALL we have to make sure that
+  ** the ORDER BY clause covers every term of the result set.  Add
+  ** terms to the ORDER BY clause as necessary.
+  */
+  if( op!=TK_ALL ){
+    for(i=1; db->mallocFailed==0 && i<=p->pEList->nExpr; i++){
+      for(j=0; j<nOrderBy; j++){
+        Expr *pTerm = pOrderBy->a[j].pExpr;
+        assert( pTerm->op==TK_INTEGER );
+        assert( (pTerm->flags & EP_IntValue)!=0 );
+        if( pTerm->iTable==i ) break;
+      }
+      if( j==nOrderBy ){
+        Expr *pNew = sqlite3PExpr(pParse, TK_INTEGER, 0, 0, 0);
+        if( pNew==0 ) return SQLITE_NOMEM;
+        pNew->flags |= EP_IntValue;
+        pNew->iTable = i;
+        pOrderBy = sqlite3ExprListAppend(pParse, pOrderBy, pNew, 0);
+        nOrderBy++;
+      }
+    }
+  }
+
+  /* Compute the comparison permutation and keyinfo that is used with
+  ** the permutation in order to comparisons to determine if the next
+  ** row of results comes from selectA or selectB.  Also add explicit
+  ** collations to the ORDER BY clause terms so that when the subqueries
+  ** to the right and the left are evaluated, they use the correct
+  ** collation.
+  */
+  aPermute = sqlite3DbMallocRaw(db, sizeof(int)*nOrderBy);
+  if( aPermute ){
+    for(i=0; i<nOrderBy; i++){
+      Expr *pTerm = pOrderBy->a[i].pExpr;
+      assert( pTerm->op==TK_INTEGER );
+      assert( (pTerm->flags & EP_IntValue)!=0 );
+      aPermute[i] = pTerm->iTable-1;
+      assert( aPermute[i]>=0 && aPermute[i]<p->pEList->nExpr );
+    }
+    pKeyMerge =
+      sqlite3DbMallocRaw(db, sizeof(*pKeyMerge)+nOrderBy*(sizeof(CollSeq*)+1));
+    if( pKeyMerge ){
+      pKeyMerge->aSortOrder = (u8*)&pKeyMerge->aColl[nOrderBy];
+      pKeyMerge->nField = nOrderBy;
+      pKeyMerge->enc = ENC(db);
+      for(i=0; i<nOrderBy; i++){
+        CollSeq *pColl;
+        Expr *pTerm = pOrderBy->a[i].pExpr;
+        if( pTerm->flags & EP_ExpCollate ){
+          pColl = pTerm->pColl;
+        }else{
+          pColl = multiSelectCollSeq(pParse, p, aPermute[i]);
+          pTerm->flags |= EP_ExpCollate;
+          pTerm->pColl = pColl;
+        }
+        pKeyMerge->aColl[i] = pColl;
+        pKeyMerge->aSortOrder[i] = pOrderBy->a[i].sortOrder;
+      }
+    }
+  }else{
+    pKeyMerge = 0;
+  }
+
+  /* Reattach the ORDER BY clause to the query.
+  */
+  p->pOrderBy = pOrderBy;
+  pPrior->pOrderBy = sqlite3ExprListDup(pParse->db, pOrderBy);
+
+  /* Allocate a range of temporary registers and the KeyInfo needed
+  ** for the logic that removes duplicate result rows when the
+  ** operator is UNION, EXCEPT, or INTERSECT (but not UNION ALL).
+  */
+  if( op==TK_ALL ){
+    regPrev = 0;
+  }else{
+    int nExpr = p->pEList->nExpr;
+    assert( nOrderBy>=nExpr );
+    regPrev = sqlite3GetTempRange(pParse, nExpr+1);
+    sqlite3VdbeAddOp2(v, OP_Integer, 0, regPrev);
+    pKeyDup = sqlite3DbMallocZero(db,
+                  sizeof(*pKeyDup) + nExpr*(sizeof(CollSeq*)+1) );
+    if( pKeyDup ){
+      pKeyDup->aSortOrder = (u8*)&pKeyDup->aColl[nExpr];
+      pKeyDup->nField = nExpr;
+      pKeyDup->enc = ENC(db);
+      for(i=0; i<nExpr; i++){
+        pKeyDup->aColl[i] = multiSelectCollSeq(pParse, p, i);
+        pKeyDup->aSortOrder[i] = 0;
+      }
+    }
   }
-  pPrior->pOrderBy = sqlite3ExprListDup(pParse->db, p->pOrderBy);
  
   /* Separate the left and the right query from one another
   */
   p->pPrior = 0;
   pPrior->pRightmost = 0;
-  
+  processOrderGroupBy(pParse, p, p->pOrderBy, 1, &NotUsed);
+  if( pPrior->pPrior==0 ){
+    processOrderGroupBy(pParse, pPrior, pPrior->pOrderBy, 1, &NotUsed);
+  }
+
   /* Compute the limit registers */
   computeLimitRegisters(pParse, p, labelEnd);
-  if( p->iLimit ){
+  if( p->iLimit && op==TK_ALL ){
     regLimitA = ++pParse->nMem;
     regLimitB = ++pParse->nMem;
     sqlite3VdbeAddOp2(v, OP_Copy, p->iOffset ? p->iOffset+1 : p->iLimit,
@@ -2585,6 +2732,10 @@ static int multiSelectOrderBy(
   }else{
     regLimitA = regLimitB = 0;
   }
+  sqlite3ExprDelete(p->pLimit);
+  p->pLimit = 0;
+  sqlite3ExprDelete(p->pOffset);
+  p->pOffset = 0;
 
   regAddrA = ++pParse->nMem;
   regEofA = ++pParse->nMem;
@@ -2601,8 +2752,10 @@ static int multiSelectOrderBy(
   j1 = sqlite3VdbeAddOp0(v, OP_Goto);
   addrSelectA = sqlite3VdbeCurrentAddr(v);
 
+
   /* Generate a coroutine to evaluate the SELECT statement to the
-  ** left of the compound operator - the "A" select. */
+  ** left of the compound operator - the "A" select.
+  */
   VdbeNoopComment((v, "Begin coroutine for left SELECT"));
   pPrior->iLimit = regLimitA;
   sqlite3Select(pParse, pPrior, &destA, 0, 0, 0, 0);
@@ -2627,43 +2780,34 @@ static int multiSelectOrderBy(
   VdbeNoopComment((v, "End coroutine for right SELECT"));
 
   /* Generate a subroutine that outputs the current row of the A
-  ** select as the next output row of the compound select and then
-  ** advances the A select to its next row
+  ** select as the next output row of the compound select.
   */
   VdbeNoopComment((v, "Output routine for A"));
-  addrOutA = outputSubroutine(pParse, p, &destA, pDest, regOutA, labelEnd);
+  addrOutA = generateOutputSubroutine(pParse,
+                 p, &destA, pDest, regOutA,
+                 regPrev, pKeyDup, P4_KEYINFO_HANDOFF, labelEnd);
   
   /* Generate a subroutine that outputs the current row of the B
-  ** select as the next output row of the compound select and then
-  ** advances the B select to its next row
+  ** select as the next output row of the compound select.
   */
-  VdbeNoopComment((v, "Output routine for B"));
-  addrOutB = outputSubroutine(pParse, p, &destB, pDest, regOutB, labelEnd);
+  if( op==TK_ALL || op==TK_UNION ){
+    VdbeNoopComment((v, "Output routine for B"));
+    addrOutB = generateOutputSubroutine(pParse,
+                 p, &destB, pDest, regOutB,
+                 regPrev, pKeyDup, P4_KEYINFO_STATIC, labelEnd);
+  }
 
   /* Generate a subroutine to run when the results from select A
   ** are exhausted and only data in select B remains.
   */
   VdbeNoopComment((v, "eof-A subroutine"));
-  addrEofA = sqlite3VdbeCurrentAddr(v);
   if( op==TK_EXCEPT || op==TK_INTERSECT ){
-    sqlite3VdbeAddOp2(v, OP_Goto, 0, labelEnd);
+    addrEofA = sqlite3VdbeAddOp2(v, OP_Goto, 0, labelEnd);
   }else{  
-    if( op==TK_ALL ){
-      j2 = sqlite3VdbeAddOp2(v, OP_If, regEofB, labelEnd);
-      sqlite3VdbeAddOp2(v, OP_Gosub, regOutB, addrOutB);
-      sqlite3VdbeAddOp2(v, OP_Goto, 0, j2);
-    }else{
-      assert( op==TK_UNION );
-      sqlite3ExprCodeCopy(pParse, destB.iMem, destA.iMem, destB.nMem);
-      j2 = sqlite3VdbeAddOp2(v, OP_Gosub, regOutB, addrOutB);
-      sqlite3VdbeAddOp2(v, OP_If, regEofB, labelEnd);
-      sqlite3VdbeAddOp4(v, OP_Compare, destA.iMem, destB.iMem, destB.nMem,
-                           (char*)pKeyInfo, p4type);
-      p4type = P4_KEYINFO_STATIC;
-      sqlite3VdbeAddOp3(v, OP_Jump, j2, j2+4, j2);
-      sqlite3VdbeAddOp1(v, OP_Yield, regAddrB);
-      sqlite3VdbeAddOp2(v, OP_Goto, 0, j2+1);
-    }
+    addrEofA = sqlite3VdbeAddOp2(v, OP_If, regEofB, labelEnd);
+    sqlite3VdbeAddOp2(v, OP_Gosub, regOutB, addrOutB);
+    sqlite3VdbeAddOp1(v, OP_Yield, regAddrB);
+    sqlite3VdbeAddOp2(v, OP_Goto, 0, addrEofA);
   }
 
   /* Generate a subroutine to run when the results from select B
@@ -2673,35 +2817,17 @@ static int multiSelectOrderBy(
     addrEofB = addrEofA;
   }else{  
     VdbeNoopComment((v, "eof-B subroutine"));
-    addrEofB = sqlite3VdbeCurrentAddr(v);
-    if( op==TK_ALL || op==TK_EXCEPT ){
-      j2 = sqlite3VdbeAddOp2(v, OP_If, regEofA, labelEnd);
-      sqlite3VdbeAddOp2(v, OP_Gosub, regOutA, addrOutA);
-      sqlite3VdbeAddOp2(v, OP_Goto, 0, j2);
-    }else{
-      assert( op==TK_UNION );
-      sqlite3VdbeAddOp2(v, OP_If, regEofA, labelEnd);
-      sqlite3ExprCodeCopy(pParse, destA.iMem, destB.iMem, destA.nMem);
-      j2 = sqlite3VdbeAddOp2(v, OP_Gosub, regOutA, addrOutA);
-      sqlite3VdbeAddOp2(v, OP_If, regEofA, labelEnd);
-      sqlite3VdbeAddOp4(v, OP_Compare, destA.iMem, destB.iMem, destB.nMem,
-                           (char*)pKeyInfo, p4type);
-      p4type = P4_KEYINFO_STATIC;
-      sqlite3VdbeAddOp3(v, OP_Jump, j2, j2+4, j2);
-      sqlite3VdbeAddOp1(v, OP_Yield, regAddrA);
-      sqlite3VdbeAddOp2(v, OP_Goto, 0, j2+1);
-    }
+    addrEofB = sqlite3VdbeAddOp2(v, OP_If, regEofA, labelEnd);
+    sqlite3VdbeAddOp2(v, OP_Gosub, regOutA, addrOutA);
+    sqlite3VdbeAddOp1(v, OP_Yield, regAddrA);
+    sqlite3VdbeAddOp2(v, OP_Goto, 0, addrEofB);
   }
 
   /* Generate code to handle the case of A<B
   */
   VdbeNoopComment((v, "A-lt-B subroutine"));
-  addrAltB = sqlite3VdbeCurrentAddr(v);
-  if( op==TK_INTERSECT ){
-    sqlite3VdbeAddOp1(v, OP_Yield, regAddrA);
-  }else{
-    sqlite3VdbeAddOp2(v, OP_Gosub, regOutA, addrOutA);
-  }
+  addrAltB = sqlite3VdbeAddOp2(v, OP_Gosub, regOutA, addrOutA);
+  sqlite3VdbeAddOp1(v, OP_Yield, regAddrA);
   sqlite3VdbeAddOp2(v, OP_If, regEofA, addrEofA);
   sqlite3VdbeAddOp2(v, OP_Goto, 0, labelCmpr);
 
@@ -2709,24 +2835,15 @@ static int multiSelectOrderBy(
   */
   if( op==TK_ALL ){
     addrAeqB = addrAltB;
+  }else if( op==TK_INTERSECT ){
+    addrAeqB = addrAltB;
+    addrAltB++;
   }else{
     VdbeNoopComment((v, "A-eq-B subroutine"));
-    addrAeqB = sqlite3VdbeCurrentAddr(v);
-    if( op==TK_INTERSECT ){
-      sqlite3VdbeAddOp2(v, OP_Gosub, regOutA, addrOutA);
-      j2 = sqlite3VdbeAddOp1(v, OP_Yield, regAddrA);
-      sqlite3VdbeAddOp2(v, OP_If, regEofA, addrEofA);
-      sqlite3VdbeAddOp4(v, OP_Compare, destA.iMem, destB.iMem, destA.nMem,
-                           (char*)pKeyInfo, p4type);
-      p4type = P4_KEYINFO_STATIC;
-      j3 = sqlite3VdbeCurrentAddr(v)+1;
-      sqlite3VdbeAddOp3(v, OP_Jump, j3, j2, j3);
-      sqlite3VdbeAddOp2(v, OP_Goto, 0, labelCmpr);
-    }else{
-      sqlite3VdbeAddOp1(v, OP_Yield, regAddrA);
-      sqlite3VdbeAddOp2(v, OP_If, regEofA, addrEofA);
-      sqlite3VdbeAddOp2(v, OP_Goto, 0, labelCmpr);
-    }
+    addrAeqB =
+    sqlite3VdbeAddOp1(v, OP_Yield, regAddrA);
+    sqlite3VdbeAddOp2(v, OP_If, regEofA, addrEofA);
+    sqlite3VdbeAddOp2(v, OP_Goto, 0, labelCmpr);
   }
 
   /* Generate code to handle the case of A>B
@@ -2735,9 +2852,8 @@ static int multiSelectOrderBy(
   addrAgtB = sqlite3VdbeCurrentAddr(v);
   if( op==TK_ALL || op==TK_UNION ){
     sqlite3VdbeAddOp2(v, OP_Gosub, regOutB, addrOutB);
-  }else{
-    sqlite3VdbeAddOp1(v, OP_Yield, regAddrB);
   }
+  sqlite3VdbeAddOp1(v, OP_Yield, regAddrB);
   sqlite3VdbeAddOp2(v, OP_If, regEofB, addrEofB);
   sqlite3VdbeAddOp2(v, OP_Goto, 0, labelCmpr);
 
@@ -2746,20 +2862,25 @@ static int multiSelectOrderBy(
   sqlite3VdbeJumpHere(v, j1);
   sqlite3VdbeAddOp2(v, OP_Integer, 0, regEofA);
   sqlite3VdbeAddOp2(v, OP_Integer, 0, regEofB);
-  sqlite3VdbeAddOp2(v, OP_Integer, addrSelectB, regAddrB);
   sqlite3VdbeAddOp2(v, OP_Gosub, regAddrA, addrSelectA);
+  sqlite3VdbeAddOp2(v, OP_Gosub, regAddrB, addrSelectB);
   sqlite3VdbeAddOp2(v, OP_If, regEofA, addrEofA);
-  sqlite3VdbeAddOp1(v, OP_Yield, regAddrB);
   sqlite3VdbeAddOp2(v, OP_If, regEofB, addrEofB);
 
   /* Implement the main merge loop
   */
   sqlite3VdbeResolveLabel(v, labelCmpr);
-  sqlite3VdbeAddOp4(v, OP_Compare, destA.iMem, destB.iMem, destA.nMem,
-                         (char*)pKeyInfo, p4type);
-  p4type = P4_KEYINFO_STATIC;
+  sqlite3VdbeAddOp4(v, OP_Permutation, 0, 0, 0, (char*)aPermute, P4_INTARRAY);
+  sqlite3VdbeAddOp4(v, OP_Compare, destA.iMem, destB.iMem, nOrderBy,
+                         (char*)pKeyMerge, P4_KEYINFO_HANDOFF);
   sqlite3VdbeAddOp3(v, OP_Jump, addrAltB, addrAeqB, addrAgtB);
 
+  /* Release temporary registers
+  */
+  if( regPrev ){
+    sqlite3ReleaseTempRange(pParse, regPrev, nOrderBy+1);
+  }
+
   /* Jump to the this point in order to terminate the query.
   */
   sqlite3VdbeResolveLabel(v, labelEnd);
@@ -2767,16 +2888,14 @@ static int multiSelectOrderBy(
   /* Set the number of output columns
   */
   if( pDest->eDest==SRT_Callback ){
-    Select *pFirst = p;
+    Select *pFirst = pPrior;
     while( pFirst->pPrior ) pFirst = pFirst->pPrior;
     generateColumnNames(pParse, 0, pFirst->pEList);
   }
 
-  /* Free the KeyInfo if unused.
-  */
-  if( p4type==P4_KEYINFO_HANDOFF ){
-    sqlite3_free(pKeyInfo);
-  }
+  /* Reassembly the compound query so that it will be freed correctly
+  ** by the calling function */
+  p->pPrior = pPrior;
 
   /*** TBD:  Insert subroutine calls to close cursors on incomplete
   **** subqueries ****/
diff --git a/src/vdbe.c b/src/vdbe.c
index 07c64f2e6..20207343d 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.754 2008/06/23 18:49:44 danielk1977 Exp $
+** $Id: vdbe.c,v 1.755 2008/06/25 00:12:41 drh Exp $
 */
 #include "sqliteInt.h"
 #include <ctype.h>
@@ -546,6 +546,8 @@ int sqlite3VdbeExec(
   Mem *pIn1, *pIn2, *pIn3;   /* Input operands */
   Mem *pOut;                 /* Output operand */
   u8 opProperty;
+  int iCompare = 0;          /* Result of last OP_Compare operation */
+  int *aPermute = 0;         /* Permuation of columns for OP_Compare */
 #ifdef VDBE_PROFILE
   u64 start;                 /* CPU clock count at start of opcode */
   int origPc;                /* Program counter at start of opcode */
@@ -997,7 +999,6 @@ case OP_Move: {
   pOut = &p->aMem[p2];
   assert( p1+n<=p2 || p2+n<=p1 );
   while( n-- ){
-    REGISTER_TRACE(p1++, pIn1);
     zMalloc = pOut->zMalloc;
     pOut->zMalloc = 0;
     sqlite3VdbeMemMove(pOut, pIn1);
@@ -1020,7 +1021,6 @@ case OP_Copy: {
   assert( pOp->p1>0 );
   assert( pOp->p1<=p->nMem );
   pIn1 = &p->aMem[pOp->p1];
-  REGISTER_TRACE(pOp->p1, pIn1);
   assert( pOp->p2>0 );
   assert( pOp->p2<=p->nMem );
   pOut = &p->aMem[pOp->p2];
@@ -1083,6 +1083,7 @@ case OP_ResultRow: {
   for(i=0; i<pOp->p2; i++){
     sqlite3VdbeMemNulTerminate(&pMem[i]);
     storeTypeInfo(&pMem[i], encoding);
+    REGISTER_TRACE(pOp->p1+i, &pMem[i]);
   }
   if( db->mallocFailed ) goto no_mem;
 
@@ -1740,15 +1741,34 @@ case OP_Ge: {             /* same as TK_GE, jump, in1, in3 */
   break;
 }
 
+/* Opcode: Permutation * * * P4 *
+**
+** Set the permuation used by the OP_Compare operator to be the array
+** of integers in P4.
+**
+** The permutation is only valid until the next OP_Permutation, OP_Compare,
+** OP_Halt, or OP_ResultRow.  Typically the OP_Permutation should occur
+** immediately prior to the OP_Compare.
+*/
+case OP_Permutation: {
+  assert( pOp->p4type==P4_INTARRAY );
+  assert( pOp->p4.ai );
+  aPermute = pOp->p4.ai;
+  break;
+}
+
 /* Opcode: Compare P1 P2 P3 P4 *
 **
 ** Compare to vectors of registers in reg(P1)..reg(P1+P3-1) (all this
 ** one "A") and in reg(P2)..reg(P2+P3-1) ("B").  Save the result of
 ** the comparison for use by the next OP_Jump instruct.
 **
-** P4 is a KeyInfo structure that defines collating sequences usedused for affinity purposes.  The
-** comparison is done for sorting purposes, so NULLs compare
-** equal, NULLs are less than numbers, numbers are less than strings,
+** P4 is a KeyInfo structure that defines collating sequences and sort
+** orders for the comparison.  The permutation applies to registers
+** only.  The KeyInfo elements are used sequentially.
+**
+** The comparison is a sort comparison, so NULLs compare equal,
+** NULLs are less than numbers, numbers are less than strings,
 ** and strings are less than blobs.
 */
 case OP_Compare: {
@@ -1760,13 +1780,28 @@ case OP_Compare: {
   assert( p1>0 && p1+n-1<p->nMem );
   p2 = pOp->p2;
   assert( p2>0 && p2+n-1<p->nMem );
-  for(i=0; i<n; i++, p1++, p2++){
-    REGISTER_TRACE(p1, &p->aMem[p1]);
-    REGISTER_TRACE(p2, &p->aMem[p2]);
-    p->iCompare = sqlite3MemCompare(&p->aMem[p1], &p->aMem[p2],
-                   pKeyInfo && i<pKeyInfo->nField ? pKeyInfo->aColl[i] : 0);
-    if( p->iCompare ) break;
+  for(i=0; i<n; i++){
+    int idx = aPermute ? aPermute[i] : i;
+    CollSeq *pColl;    /* Collating sequence to use on this term */
+    int bRev;          /* True for DESCENDING sort order */
+    assert( pKeyInfo==0 || i<pKeyInfo->nField );
+    REGISTER_TRACE(p1+idx, &p->aMem[p1+idx]);
+    REGISTER_TRACE(p2+idx, &p->aMem[p2+idx]);
+    if( pKeyInfo ){
+      assert( i<pKeyInfo->nField );
+      pColl = pKeyInfo->aColl[i];
+      bRev = pKeyInfo->aSortOrder[i];
+    }else{
+      pColl = 0;
+      bRev = 0;
+    }
+    iCompare = sqlite3MemCompare(&p->aMem[p1+idx], &p->aMem[p2+idx], pColl);
+    if( iCompare ){
+      if( bRev ) iCompare = -iCompare;
+      break;
+    }
   }
+  aPermute = 0;
   break;
 }
 
@@ -1776,10 +1811,10 @@ case OP_Compare: {
 ** in the most recent OP_Compare instruction the P1 vector was less than
 ** equal to, or greater than the P2 vector, respectively.
 */
-case OP_Jump: {
-  if( p->iCompare<0 ){
+case OP_Jump: {             /* jump */
+  if( iCompare<0 ){
     pc = pOp->p1 - 1;
-  }else if( p->iCompare==0 ){
+  }else if( iCompare==0 ){
     pc = pOp->p2 - 1;
   }else{
     pc = pOp->p3 - 1;
diff --git a/src/vdbe.h b/src/vdbe.h
index b19e66dc2..91fdaf6d8 100644
--- a/src/vdbe.h
+++ b/src/vdbe.h
@@ -15,7 +15,7 @@
 ** or VDBE.  The VDBE implements an abstract machine that runs a
 ** simple program to access and modify the underlying database.
 **
-** $Id: vdbe.h,v 1.133 2008/06/20 18:13:25 drh Exp $
+** $Id: vdbe.h,v 1.134 2008/06/25 00:12:41 drh Exp $
 */
 #ifndef _SQLITE_VDBE_H_
 #define _SQLITE_VDBE_H_
@@ -61,6 +61,7 @@ struct VdbeOp {
     Mem *pMem;             /* Used when p4type is P4_MEM */
     sqlite3_vtab *pVtab;   /* Used when p4type is P4_VTAB */
     KeyInfo *pKeyInfo;     /* Used when p4type is P4_KEYINFO */
+    int *ai;               /* Used when p4type is P4_INTARRAY */
   } p4;
 #ifdef SQLITE_DEBUG
   char *zComment;          /* Comment to improve readability */
@@ -101,6 +102,7 @@ typedef struct VdbeOpList VdbeOpList;
 #define P4_REAL     (-12) /* P4 is a 64-bit floating point value */
 #define P4_INT64    (-13) /* P4 is a 64-bit signed integer */
 #define P4_INT32    (-14) /* P4 is a 32-bit signed integer */
+#define P4_INTARRAY (-15) /* P4 is a vector of 32-bit integers */
 
 /* When adding a P4 argument using P4_KEYINFO, a copy of the KeyInfo structure
 ** is made.  That copy is freed when the Vdbe is finalized.  But if the
@@ -109,8 +111,8 @@ typedef struct VdbeOpList VdbeOpList;
 ** from a single sqliteMalloc().  But no copy is made and the calling
 ** function should *not* try to free the KeyInfo.
 */
-#define P4_KEYINFO_HANDOFF (-15)
-#define P4_KEYINFO_STATIC  (-16)
+#define P4_KEYINFO_HANDOFF (-16)
+#define P4_KEYINFO_STATIC  (-17)
 
 /*
 ** The Vdbe.aColName array contains 5n Mem structures, where n is the 
diff --git a/src/vdbeInt.h b/src/vdbeInt.h
index 7cbcb8e3a..2100e3830 100644
--- a/src/vdbeInt.h
+++ b/src/vdbeInt.h
@@ -15,7 +15,7 @@
 ** 6000 lines long) it was split up into several smaller files and
 ** this header information was factored out.
 **
-** $Id: vdbeInt.h,v 1.148 2008/06/20 18:13:25 drh Exp $
+** $Id: vdbeInt.h,v 1.149 2008/06/25 00:12:41 drh Exp $
 */
 #ifndef _VDBEINT_H_
 #define _VDBEINT_H_
@@ -321,7 +321,6 @@ struct Vdbe {
   u8 expired;             /* True if the VM needs to be recompiled */
   u8 minWriteFileFormat;  /* Minimum file format for writable database files */
   u8 inVtabMethod;        /* See comments above */
-  int iCompare;           /* Result of most recent OP_Compare comparison */
   int nChange;            /* Number of db changes made since last reset */
   i64 startTime;          /* Time when query started - used for profiling */
   int btreeMask;          /* Bitmask of db->aDb[] entries referenced */
diff --git a/src/vdbeaux.c b/src/vdbeaux.c
index c45e6708e..e2b235bc0 100644
--- a/src/vdbeaux.c
+++ b/src/vdbeaux.c
@@ -14,7 +14,7 @@
 ** to version 2.8.7, all this code was combined into the vdbe.c source file.
 ** But that file was getting too big so this subroutines were split out.
 **
-** $Id: vdbeaux.c,v 1.392 2008/06/23 13:57:22 danielk1977 Exp $
+** $Id: vdbeaux.c,v 1.393 2008/06/25 00:12:42 drh Exp $
 */
 #include "sqliteInt.h"
 #include <ctype.h>
@@ -441,31 +441,32 @@ static void freeEphemeralFunction(FuncDef *pDef){
 /*
 ** Delete a P4 value if necessary.
 */
-static void freeP4(int p4type, void *p3){
-  if( p3 ){
+static void freeP4(int p4type, void *p4){
+  if( p4 ){
     switch( p4type ){
       case P4_REAL:
       case P4_INT64:
       case P4_MPRINTF:
       case P4_DYNAMIC:
       case P4_KEYINFO:
+      case P4_INTARRAY:
       case P4_KEYINFO_HANDOFF: {
-        sqlite3_free(p3);
+        sqlite3_free(p4);
         break;
       }
       case P4_VDBEFUNC: {
-        VdbeFunc *pVdbeFunc = (VdbeFunc *)p3;
+        VdbeFunc *pVdbeFunc = (VdbeFunc *)p4;
         freeEphemeralFunction(pVdbeFunc->pFunc);
         sqlite3VdbeDeleteAuxData(pVdbeFunc, 0);
         sqlite3_free(pVdbeFunc);
         break;
       }
       case P4_FUNCDEF: {
-        freeEphemeralFunction((FuncDef*)p3);
+        freeEphemeralFunction((FuncDef*)p4);
         break;
       }
       case P4_MEM: {
-        sqlite3ValueFree((sqlite3_value*)p3);
+        sqlite3ValueFree((sqlite3_value*)p4);
         break;
       }
     }
@@ -697,6 +698,10 @@ static char *displayP4(Op *pOp, char *zTemp, int nTemp){
       break;
     }
 #endif
+    case P4_INTARRAY: {
+      sqlite3_snprintf(nTemp, zTemp, "intarray");
+      break;
+    }
     default: {
       zP4 = pOp->p4.z;
       if( zP4==0 ){