Further changes to sqlite3VdbeRecordCompare().

FossilOrigin-Name: 570893740067a7caa952f259fa078cdf67017d71

manifest

@@ -1,5 +1,5 @@
-C Attempt\sto\sspeed\sup\ssqlite3VdbeRecordCompare()\sby\svarious\smeans.\sThis\scode\sis\sin\san\sinterim\sstate.
+C Further\schanges\sto\ssqlite3VdbeRecordCompare().
-D 2014-02-25T21:01:25.824
+D 2014-02-27T20:44:18.479
 F Makefile.arm-wince-mingw32ce-gcc d6df77f1f48d690bd73162294bbba7f59507c72f
 F Makefile.in 2ef13430cd359f7b361bb863504e227b25cc7f81
 F Makefile.linux-gcc 91d710bdc4998cb015f39edf3cb314ec4f4d7e23
@@ -163,7 +163,7 @@ F src/auth.c 523da7fb4979469955d822ff9298352d6b31de34
 F src/backup.c a729e63cf5cd1829507cb7b8e89f99b95141bb53
 F src/bitvec.c 19a4ba637bd85f8f63fc8c9bae5ade9fb05ec1cb
 F src/btmutex.c 976f45a12e37293e32cae0281b15a21d48a8aaa7
-F src/btree.c f4d85c3e2e189a219965d6d4525330333735fd1d
+F src/btree.c 77f175987c80ebec063f8653cb7d300776411413
 F src/btree.h 9e0f97c01b972f779eb7655cfb4f8727fd6dc26f
 F src/btreeInt.h 0be66063468a520e4d66b80c7a1dc26d04ee6ea4
 F src/build.c 00ce613bc2256e525c9195cb10d0df7bcc48d1f0
@@ -221,7 +221,7 @@ F src/shell.c 3dd86bf73ccd079f0e32ef5069600586085e8239
 F src/sqlite.h.in a2ef671f92747a5a1c8a47bad5c585a8dd9eca80
 F src/sqlite3.rc 11094cc6a157a028b301a9f06b3d03089ea37c3e
 F src/sqlite3ext.h 886f5a34de171002ad46fae8c36a7d8051c190fc
-F src/sqliteInt.h 6d868994f476b616ddb1795a51aa83c331ef5a62
+F src/sqliteInt.h 7b42e02c1ca4599b5420d44cb549460b2348139a
 F src/sqliteLimit.h 164b0e6749d31e0daa1a4589a169d31c0dec7b3d
 F src/status.c 7ac05a5c7017d0b9f0b4bcd701228b784f987158
 F src/table.c 2cd62736f845d82200acfa1287e33feb3c15d62e
@@ -278,13 +278,13 @@ F src/utf.c 6fc6c88d50448c469c5c196acf21617a24f90269
 F src/util.c c46c90459ef9bdc0c6c73803cf4c55425b4771cf
 F src/vacuum.c 3728d74919d4fb1356f9e9a13e27773db60b7179
 F src/vdbe.c ab910206dd8c9c5c1455f82953934bdbfe0bcc2a
-F src/vdbe.h 0758eff7f1bf939bcafa377b2fafba4f5be63007
+F src/vdbe.h 6833579fc0fbdc1c933e34519064841abda5b9b3
 F src/vdbeInt.h 5286af9067cabdb8ba57b87c0c988a931be6c6c8
 F src/vdbeapi.c 5bc41aaea448a7fc250902c418f1795859be3820
-F src/vdbeaux.c 988269c675ebb75a3610479840b65bcb8ea92647
+F src/vdbeaux.c 80e5315957377554c9011858f5afde61afedc181
 F src/vdbeblob.c d939997de046b8fcc607cfee4248f3d33dbcca50
 F src/vdbemem.c 25cc487244bf6ad647105c5adbc3052403dfd143
-F src/vdbesort.c 5144d84bd7d0d2545af1c6322edbbf07f97e3892
+F src/vdbesort.c 72290f12428973c2c6b9d4f95ad0a7c8181e1280
 F src/vdbetrace.c 6f52bc0c51e144b7efdcfb2a8f771167a8816767
 F src/vtab.c 21b932841e51ebd7d075e2d0ad1415dce8d2d5fd
 F src/wal.c 76e7fc6de229bea8b30bb2539110f03a494dc3a8
@@ -1151,10 +1151,7 @@ F tool/vdbe_profile.tcl 67746953071a9f8f2f668b73fe899074e2c6d8c1
 F tool/warnings-clang.sh f6aa929dc20ef1f856af04a730772f59283631d4
 F tool/warnings.sh d1a6de74685f360ab718efda6265994b99bbea01
 F tool/win/sqlite.vsix 030f3eeaf2cb811a3692ab9c14d021a75ce41fff
-P 23001a85cd334090cf6c70d4d7e722a01f4f6899
+P 85206e0bbac29adab52bef795f6d1479f2ae2c0e
-R 04ade79c9b23add6f2d6adf0c34db1e4
+R eb22cd7a0c8c7d3bc8c75c472c26dd6a
-T *branch * experimental
-T *sym-experimental *
-T -sym-trunk *
 U dan
-Z d3f147085a7b8a8eca4578239439871b
+Z 91158d7f280c804b2fe508b4511e22e9

manifest.uuid

@@ -1 +1 @@
-85206e0bbac29adab52bef795f6d1479f2ae2c0e
+570893740067a7caa952f259fa078cdf67017d71

src/btree.c

@@ -4547,7 +4547,7 @@ int sqlite3BtreeMovetoUnpacked(
   int *pRes                /* Write search results here */
 ){
   int rc;
-  int (*xRecordCompare)(int, const void*, UnpackedRecord*);
+  RecordCompare xRecordCompare;
 
   assert( cursorHoldsMutex(pCur) );
   assert( sqlite3_mutex_held(pCur->pBtree->db->mutex) );
@@ -4571,7 +4571,10 @@ int sqlite3BtreeMovetoUnpacked(
 
   if( pIdxKey ){
     xRecordCompare = sqlite3VdbeFindCompare(pIdxKey);
-    assert( pIdxKey->default_rc==1 || pIdxKey->default_rc==0 || pIdxKey->default_rc==-1);
+    assert( pIdxKey->default_rc==1 
+         || pIdxKey->default_rc==0 
+         || pIdxKey->default_rc==-1
+    );
   }
 
   rc = moveToRoot(pCur);
@@ -4658,14 +4661,14 @@ int sqlite3BtreeMovetoUnpacked(
           ** single byte varint and the record fits entirely on the main
           ** b-tree page.  */
           testcase( pCell+nCell+1==pPage->aDataEnd );
-          c = xRecordCompare(nCell, (void*)&pCell[1], pIdxKey);
+          c = xRecordCompare(nCell, (void*)&pCell[1], pCell[1], 1, pIdxKey);
         }else if( !(pCell[1] & 0x80) 
           && (nCell = ((nCell&0x7f)<<7) + pCell[1])<=pPage->maxLocal
         ){
           /* The record-size field is a 2 byte varint and the record 
           ** fits entirely on the main b-tree page.  */
           testcase( pCell+nCell+2==pPage->aDataEnd );
-          c = xRecordCompare(nCell, (void*)&pCell[2], pIdxKey);
+          c = xRecordCompare(nCell, (void*)&pCell[2], pCell[2], 1, pIdxKey);
         }else{
           /* The record flows over onto one or more overflow pages. In
           ** this case the whole cell needs to be parsed, a buffer allocated
@@ -4686,7 +4689,7 @@ int sqlite3BtreeMovetoUnpacked(
             sqlite3_free(pCellKey);
             goto moveto_finish;
           }
-          c = xRecordCompare(nCell, pCellKey, pIdxKey);
+          c = xRecordCompare(nCell, pCellKey, ((u8*)pCellKey)[0], 1, pIdxKey);
           sqlite3_free(pCellKey);
         }
         if( c<0 ){

src/sqliteInt.h

@@ -1592,6 +1592,8 @@ struct UnpackedRecord {
   u16 nField;         /* Number of entries in apMem[] */
   char default_rc;    /* Comparison result if keys are equal */
   Mem *aMem;          /* Values */
+  int r1;
+  int r2;
 };
 
 

src/vdbe.h

@@ -214,8 +214,9 @@ void sqlite3VdbeRecordUnpack(KeyInfo*,int,const void*,UnpackedRecord*);
 int sqlite3VdbeRecordCompare(int,const void*,UnpackedRecord*);
 UnpackedRecord *sqlite3VdbeAllocUnpackedRecord(KeyInfo *, char *, int, char **);
 
-typedef int (*RecordCompare)(int,const void*, UnpackedRecord*);
+typedef int (*RecordCompare)(int,const void*,int,u32,UnpackedRecord*);
 RecordCompare sqlite3VdbeFindCompare(UnpackedRecord*);
+RecordCompare sqlite3VdbeFindSorterCompare(KeyInfo*);
 
 #ifndef SQLITE_OMIT_TRIGGER
 void sqlite3VdbeLinkSubProgram(Vdbe *, SubProgram *);

src/vdbeaux.c

@@ -3123,90 +3123,102 @@ void sqlite3VdbeRecordUnpack(
   p->nField = u;
 }
 
-static int vdbeRecordCompareString(
+/*
+** This function compares the two table rows or index records
+** specified by {nKey1, pKey1} and pPKey2.  It returns a negative, zero
+** or positive integer if key1 is less than, equal to or 
+** greater than key2.  The {nKey1, pKey1} key must be a blob
+** created by th OP_MakeRecord opcode of the VDBE.  The pPKey2
+** key must be a parsed key such as obtained from
+** sqlite3VdbeParseRecord.
+**
+** Key1 and Key2 do not have to contain the same number of fields.
+** The key with fewer fields is usually compares less than the 
+** longer key.  However if the UNPACKED_INCRKEY flags in pPKey2 is set
+** and the common prefixes are equal, then key1 is less than key2.
+** Or if the UNPACKED_MATCH_PREFIX flag is set and the prefixes are
+** equal, then the keys are considered to be equal and
+** the parts beyond the common prefix are ignored.
+*/
+static int vdbeRecordComparePrev(
   int nKey1, const void *pKey1, /* Left key */
   UnpackedRecord *pPKey2        /* Right key */
 ){
-  const u8 *aKey1 = (const u8*)pKey1;
+  u32 d1;            /* Offset into aKey[] of next data element */
-  int szHdr;
+  u32 idx1;          /* Offset into aKey[] of next header element */
-  int serial_type;
+  u32 szHdr1;        /* Number of bytes in header */
-  int res;
+  int i = 0;
+  int rc = 0;
+  const unsigned char *aKey1 = (const unsigned char *)pKey1;
+  KeyInfo *pKeyInfo;
+  Mem mem1;
 
-  szHdr = aKey1[0];
+  pKeyInfo = pPKey2->pKeyInfo;
-  getVarint32(&aKey1[1], serial_type);
+  mem1.enc = pKeyInfo->enc;
+  mem1.db = pKeyInfo->db;
+  /* mem1.flags = 0;  // Will be initialized by sqlite3VdbeSerialGet() */
+  VVA_ONLY( mem1.zMalloc = 0; ) /* Only needed by assert() statements */
 
-  if( serial_type<12 ){
+  /* Compilers may complain that mem1.u.i is potentially uninitialized.
-    res = -1;      /* (pKey1/nKey1) is a number or a null */
+  ** We could initialize it, as shown here, to silence those complaints.
-  }else if( !(serial_type & 0x01) ){ 
+  ** But in fact, mem1.u.i will never actually be used uninitialized, and doing 
-    res = +1;      /* (pKey1/nKey1) is a blob */
+  ** the unnecessary initialization has a measurable negative performance
-  }else{
+  ** impact, since this routine is a very high runner.  And so, we choose
-    int nCmp;
+  ** to ignore the compiler warnings and leave this variable uninitialized.
-    int nStr;
+  */
-    aKey1 = &aKey1[szHdr];
+  /*  mem1.u.i = 0;  // not needed, here to silence compiler warning */
   
-    nStr = (serial_type-12) / 2;
+  idx1 = getVarint32(aKey1, szHdr1);
-    if( (szHdr + nStr) > nKey1 ) return 0;    /* Corruption */
+  d1 = szHdr1;
-    nCmp = MIN( pPKey2->aMem[0].n, nStr );
+  assert( pKeyInfo->nField+pKeyInfo->nXField>=pPKey2->nField || CORRUPT_DB );
-    res = memcmp(aKey1, pPKey2->aMem[0].z, nCmp);
+  assert( pKeyInfo->aSortOrder!=0 );
+  assert( pKeyInfo->nField>0 );
+  assert( idx1<=szHdr1 || CORRUPT_DB );
+  do{
+    u32 serial_type1;
 
-    if( res==0 ){
+    /* Read the serial types for the next element in each key. */
-      res = nStr - pPKey2->aMem[0].n;
+    idx1 += getVarint32( aKey1+idx1, serial_type1 );
-      if( res==0 ) res = pPKey2->default_rc;
-    }
-  }
 
-  assert( (res==0 && sqlite3VdbeRecordCompare(nKey1, pKey1, pPKey2)==0)
+    /* Verify that there is enough key space remaining to avoid
-       || (res<0 && sqlite3VdbeRecordCompare(nKey1, pKey1, pPKey2)<0)
+    ** a buffer overread.  The "d1+serial_type1+2" subexpression will
-       || (res>0 && sqlite3VdbeRecordCompare(nKey1, pKey1, pPKey2)>0)
+    ** always be greater than or equal to the amount of required key space.
-  );
+    ** Use that approximation to avoid the more expensive call to
-  return res;
+    ** sqlite3VdbeSerialTypeLen() in the common case.
-}
+    */
-
+    if( d1+serial_type1+2>(u32)nKey1
-static int vdbeRecordCompareInt(
+     && d1+sqlite3VdbeSerialTypeLen(serial_type1)>(u32)nKey1 
-  int nKey1, const void *pKey1, /* Left key */
-  UnpackedRecord *pPKey2        /* Right key */
     ){
-  const u8 *aKey1 = (const u8*)pKey1;
+      break;
-  int szHdr;
-  int serial_type;
-  int res;
-
-  szHdr = aKey1[0];
-  getVarint32(&aKey1[1], serial_type);
-
-  if( serial_type==0 ){
-    res = -1;      /* NULL values are smaller than integers */
-  }else if( serial_type>=12 ){
-    res = +1;      /* text/blob values are greater */
-  }else{
-    Mem mem;
-    sqlite3VdbeSerialGet(&aKey1[szHdr], serial_type, &mem);
-    if( mem.flags & MEM_Int ){
-      i64 v = pPKey2->aMem[0].u.i;
-      if( v>mem.u.i ){
-        res = -1;
-      }else if( v<mem.u.i ){
-        res = +1;
-      }else{
-        res = pPKey2->default_rc;
-      }
-    }else{
-      double v = (double)pPKey2->aMem[0].u.i;
-      if( v>mem.r ){
-        res = -1;
-      }else if( v<mem.r ){
-        res = +1;
-      }else{
-        res = pPKey2->default_rc;
-      }
-    }
     }
 
-  assert( (res==0 && sqlite3VdbeRecordCompare(nKey1, pKey1, pPKey2)==0)
+    /* Extract the values to be compared.
-       || (res<0 && sqlite3VdbeRecordCompare(nKey1, pKey1, pPKey2)<0)
+    */
-       || (res>0 && sqlite3VdbeRecordCompare(nKey1, pKey1, pPKey2)>0)
+    d1 += sqlite3VdbeSerialGet(&aKey1[d1], serial_type1, &mem1);
-  );
+
-  return res;
+    /* Do the comparison
+    */
+    rc = sqlite3MemCompare(&mem1, &pPKey2->aMem[i], pKeyInfo->aColl[i]);
+    if( rc!=0 ){
+      assert( mem1.zMalloc==0 );  /* See comment below */
+      if( pKeyInfo->aSortOrder[i] ){
+        rc = -rc;  /* Invert the result for DESC sort order. */
+      }
+      return rc;
+    }
+    i++;
+  }while( idx1<szHdr1 && i<pPKey2->nField );
+
+  /* No memory allocation is ever used on mem1.  Prove this using
+  ** the following assert().  If the assert() fails, it indicates a
+  ** memory leak and a need to call sqlite3VdbeMemRelease(&mem1).
+  */
+  assert( mem1.zMalloc==0 );
+
+  /* rc==0 here means that one of the keys ran out of fields and
+  ** all the fields up to that point were equal. Return the the default_rc
+  ** value.  */
+  return pPKey2->default_rc;
 }
 
 static int vdbeCompareMemString(
@@ -3332,114 +3344,43 @@ int sqlite3MemCompare(const Mem *pMem1, const Mem *pMem2, const CollSeq *pColl){
 }
 
 
-/*
+static i64 vdbeRecordDecodeInt(u32 serial_type, const u8 *aKey){
-** This function compares the two table rows or index records
+  switch( serial_type ){
-** specified by {nKey1, pKey1} and pPKey2.  It returns a negative, zero
+    case 1:
-** or positive integer if key1 is less than, equal to or 
+      return (char)aKey[0];
-** greater than key2.  The {nKey1, pKey1} key must be a blob
+    case 2:
-** created by th OP_MakeRecord opcode of the VDBE.  The pPKey2
+      return ((char)aKey[0] << 8) | aKey[1];
-** key must be a parsed key such as obtained from
+    case 3:
-** sqlite3VdbeParseRecord.
+      return ((char)aKey[0] << 16) | (aKey[1] << 8) | aKey[2];
-**
+    case 4:
-** Key1 and Key2 do not have to contain the same number of fields.
+      return ((char)aKey[0]<<24) | (aKey[1]<<16) | (aKey[2]<<8)| aKey[3];
-** The key with fewer fields is usually compares less than the 
+
-** longer key.  However if the UNPACKED_INCRKEY flags in pPKey2 is set
+    case 5: {
-** and the common prefixes are equal, then key1 is less than key2.
+      i64 msw = ((char)aKey[0]<<24)|(aKey[1]<<16)|(aKey[2]<<8)|aKey[3];
-** Or if the UNPACKED_MATCH_PREFIX flag is set and the prefixes are
+      u32 lsw = (aKey[4] << 8) | aKey[5];
-** equal, then the keys are considered to be equal and
+      return (i64)( msw << 16 | (u64)lsw );
-** the parts beyond the common prefix are ignored.
+    }
-*/
+
-static int vdbeRecordComparePrev(
+    case 6: {
+      i64 msw = ((char)aKey[0]<<24)|(aKey[1]<<16)|(aKey[2]<<8)|aKey[3];
+      u32 lsw = ((unsigned)aKey[4]<<24)|(aKey[5]<<16)|(aKey[6]<<8)|aKey[7];
+      return (i64)( msw << 32 | (u64)lsw );
+    }
+  }
+
+  return (serial_type - 8);
+}
+
+static int vdbeRecordCompare(
   int nKey1, const void *pKey1, /* Left key */
-  UnpackedRecord *pPKey2        /* Right key */
+  int szHdr1,                   /* Size of record header in bytes */
+  u32 idx1,                     /* Offset of first type in header */
+  UnpackedRecord *const pPKey2  /* Right key */
 ){
-  u32 d1;            /* Offset into aKey[] of next data element */
+  u32 d1 = szHdr1;   /* Offset into aKey[] of next data element */
-  u32 idx1;          /* Offset into aKey[] of next header element */
-  u32 szHdr1;        /* Number of bytes in header */
-  int i = 0;
-  int rc = 0;
-  const unsigned char *aKey1 = (const unsigned char *)pKey1;
-  KeyInfo *pKeyInfo;
-  Mem mem1;
-
-  pKeyInfo = pPKey2->pKeyInfo;
-  mem1.enc = pKeyInfo->enc;
-  mem1.db = pKeyInfo->db;
-  /* mem1.flags = 0;  // Will be initialized by sqlite3VdbeSerialGet() */
-  VVA_ONLY( mem1.zMalloc = 0; ) /* Only needed by assert() statements */
-
-  /* Compilers may complain that mem1.u.i is potentially uninitialized.
-  ** We could initialize it, as shown here, to silence those complaints.
-  ** But in fact, mem1.u.i will never actually be used uninitialized, and doing 
-  ** the unnecessary initialization has a measurable negative performance
-  ** impact, since this routine is a very high runner.  And so, we choose
-  ** to ignore the compiler warnings and leave this variable uninitialized.
-  */
-  /*  mem1.u.i = 0;  // not needed, here to silence compiler warning */
-  
-  idx1 = getVarint32(aKey1, szHdr1);
-  d1 = szHdr1;
-  assert( pKeyInfo->nField+pKeyInfo->nXField>=pPKey2->nField || CORRUPT_DB );
-  assert( pKeyInfo->aSortOrder!=0 );
-  assert( pKeyInfo->nField>0 );
-  assert( idx1<=szHdr1 || CORRUPT_DB );
-  do{
-    u32 serial_type1;
-
-    /* Read the serial types for the next element in each key. */
-    idx1 += getVarint32( aKey1+idx1, serial_type1 );
-
-    /* Verify that there is enough key space remaining to avoid
-    ** a buffer overread.  The "d1+serial_type1+2" subexpression will
-    ** always be greater than or equal to the amount of required key space.
-    ** Use that approximation to avoid the more expensive call to
-    ** sqlite3VdbeSerialTypeLen() in the common case.
-    */
-    if( d1+serial_type1+2>(u32)nKey1
-     && d1+sqlite3VdbeSerialTypeLen(serial_type1)>(u32)nKey1 
-    ){
-      break;
-    }
-
-    /* Extract the values to be compared.
-    */
-    d1 += sqlite3VdbeSerialGet(&aKey1[d1], serial_type1, &mem1);
-
-    /* Do the comparison
-    */
-    rc = sqlite3MemCompare(&mem1, &pPKey2->aMem[i], pKeyInfo->aColl[i]);
-    if( rc!=0 ){
-      assert( mem1.zMalloc==0 );  /* See comment below */
-      if( pKeyInfo->aSortOrder[i] ){
-        rc = -rc;  /* Invert the result for DESC sort order. */
-      }
-      return rc;
-    }
-    i++;
-  }while( idx1<szHdr1 && i<pPKey2->nField );
-
-  /* No memory allocation is ever used on mem1.  Prove this using
-  ** the following assert().  If the assert() fails, it indicates a
-  ** memory leak and a need to call sqlite3VdbeMemRelease(&mem1).
-  */
-  assert( mem1.zMalloc==0 );
-
-  /* rc==0 here means that one of the keys ran out of fields and
-  ** all the fields up to that point were equal. Return the the default_rc
-  ** value.  */
-  return pPKey2->default_rc;
-}
- 
-
-int sqlite3VdbeRecordCompare(
-  int nKey1, const void *pKey1, /* Left key */
-  UnpackedRecord *pPKey2        /* Right key */
-){
-  u32 d1;            /* Offset into aKey[] of next data element */
-  u32 idx1;          /* Offset into aKey[] of next header element */
-  u32 szHdr1;        /* Number of bytes in header */
   int i = 0;
   int rc = 0;
+  Mem *pRhs = pPKey2->aMem;
   KeyInfo *pKeyInfo = pPKey2->pKeyInfo;
   const unsigned char *aKey1 = (const unsigned char *)pKey1;
   Mem mem1;
@@ -3450,17 +3391,25 @@ int sqlite3VdbeRecordCompare(
   nCall++;
 #endif
 
-  VVA_ONLY( mem1.zMalloc = 0; ) /* Only needed by assert() statements */
+  /* If idx==0, then the caller has already determined that the first two
+  ** elements in the keys are equal. Fix the various stack variables so
+  ** that this routine begins comparing at the second field. */
+  if( idx1==0 ){
+    u32 s1;
+    assert( sqlite3VarintLen(szHdr1)==1 );
+    idx1 = 1 + getVarint32(&aKey1[1], s1);
+    d1 += sqlite3VdbeSerialTypeLen(s1);
+    i = 1;
+    pRhs++;
+  }
 
-  idx1 = getVarint32(aKey1, szHdr1);
+  VVA_ONLY( mem1.zMalloc = 0; ) /* Only needed by assert() statements */
-  d1 = szHdr1;
   assert( pPKey2->pKeyInfo->nField+pPKey2->pKeyInfo->nXField>=pPKey2->nField 
        || CORRUPT_DB );
   assert( pPKey2->pKeyInfo->aSortOrder!=0 );
   assert( pPKey2->pKeyInfo->nField>0 );
   assert( idx1<=szHdr1 || CORRUPT_DB );
   do{
-    Mem *pRhs = &pPKey2->aMem[i];
     u32 serial_type;
 
     /* RHS is an integer */
@@ -3470,25 +3419,24 @@ int sqlite3VdbeRecordCompare(
         rc = +1;
       }else if( serial_type==0 ){
         rc = -1;
-      }else{
+      }else if( serial_type==7 ){
-        sqlite3VdbeSerialGet(&aKey1[d1], serial_type, &mem1);
-        if( serial_type==7 ){
         double rhs = (double)pRhs->u.i;
+        sqlite3VdbeSerialGet(&aKey1[d1], serial_type, &mem1);
         if( mem1.r<rhs ){
           rc = -1;
         }else if( mem1.r>rhs ){
           rc = +1;
         }
       }else{
+        i64 lhs = vdbeRecordDecodeInt(serial_type, &aKey1[d1]);
         i64 rhs = pRhs->u.i;
-          if( mem1.u.i<rhs ){
+        if( lhs<rhs ){
           rc = -1;
-          }else if( mem1.u.i>rhs ){
+        }else if( lhs>rhs ){
           rc = +1;
         }
       }
     }
-    }
 
     /* RHS is real */
     else if( pRhs->flags & MEM_Real ){
@@ -3579,6 +3527,7 @@ int sqlite3VdbeRecordCompare(
     }
 
     i++;
+    pRhs++;
     d1 += sqlite3VdbeSerialTypeLen(serial_type);
     idx1 += sqlite3VarintLen(serial_type);
   }while( idx1<szHdr1 && i<pPKey2->nField && d1<=nKey1 );
@@ -3596,20 +3545,184 @@ int sqlite3VdbeRecordCompare(
   return pPKey2->default_rc;
 }
 
+static int vdbeRecordCompareInt(
+  int nKey1, const void *pKey1, /* Left key */
+  int szHdr,
+  u32 idx1,
+  UnpackedRecord *pPKey2        /* Right key */
+){
+  const u8 *aKey = &((const u8*)pKey1)[szHdr];
+  int serial_type = ((const u8*)pKey1)[1];
+  int res;
+  i64 v = pPKey2->aMem[0].u.i;
+  i64 lhs;
+
+  switch( serial_type ){
+    case 1:
+      lhs = (char)(aKey[0]);
+      break;
+    case 2:
+      lhs = 256*(signed char)aKey[0] + aKey[1];
+      break;
+    case 3:
+      lhs = 65536*(char)aKey[0] | (aKey[1]<<8) | aKey[2];
+      break;
+    case 4:
+      lhs = (int)(((u32)aKey[0]<<24) | (aKey[1]<<16) | (aKey[2]<<8)| aKey[3]);
+      break;
+
+    case 5: {
+      i64 msw = ((char)aKey[0]<<24)|(aKey[1]<<16)|(aKey[2]<<8)|aKey[3];
+      u32 lsw = (aKey[4] << 8) | aKey[5];
+      lhs = (i64)( msw << 16 | (u64)lsw );
+      break;
+    }
+
+    case 6: {
+      i64 msw = ((char)aKey[0]<<24)|(aKey[1]<<16)|(aKey[2]<<8)|aKey[3];
+      u32 lsw = ((unsigned)aKey[4]<<24)|(aKey[5]<<16)|(aKey[6]<<8)|aKey[7];
+      lhs = (i64)( msw << 32 | (u64)lsw );
+      break;
+    }
+
+    case 8: 
+      lhs = 0;
+      break;
+
+    case 9:
+      lhs = 1;
+      break;
+
+    default:
+      return vdbeRecordCompare(nKey1, pKey1, szHdr, 1, pPKey2);
+  }
+
+  if( v>lhs ){
+    res = pPKey2->r1;
+  }else if( v<lhs ){
+    res = pPKey2->r2;
+  }else if( pPKey2->nField>1 ){
+    res = vdbeRecordCompare(nKey1, pKey1, szHdr, 0, pPKey2);
+  }else{
+    res = pPKey2->default_rc;
+  }
+
+  assert( (res==0 && vdbeRecordComparePrev(nKey1, pKey1, pPKey2)==0)
+       || (res<0 && vdbeRecordComparePrev(nKey1, pKey1, pPKey2)<0)
+       || (res>0 && vdbeRecordComparePrev(nKey1, pKey1, pPKey2)>0)
+       || CORRUPT_DB
+  );
+  return res;
+}
+
+static int vdbeRecordCompareString(
+  int nKey1, const void *pKey1, /* Left key */
+  int szHdr,
+  u32 idx1,
+  UnpackedRecord *pPKey2        /* Right key */
+){
+  const u8 *aKey1 = (const u8*)pKey1;
+  int serial_type;
+  int res;
+
+  getVarint32(&aKey1[1], serial_type);
+
+  if( serial_type<12 ){
+    res = pPKey2->r1;      /* (pKey1/nKey1) is a number or a null */
+  }else if( !(serial_type & 0x01) ){ 
+    res = pPKey2->r2;      /* (pKey1/nKey1) is a blob */
+  }else{
+    int nCmp;
+    int nStr;
+    aKey1 = &aKey1[szHdr];
+
+    nStr = (serial_type-12) / 2;
+    if( (szHdr + nStr) > nKey1 ) return 0;    /* Corruption */
+    nCmp = MIN( pPKey2->aMem[0].n, nStr );
+    res = memcmp(aKey1, pPKey2->aMem[0].z, nCmp);
+
+    if( res==0 ){
+      res = nStr - pPKey2->aMem[0].n;
+      if( res==0 ){
+        if( pPKey2->nField>1 ){
+          res = vdbeRecordCompare(nKey1, pKey1, szHdr, 0, pPKey2);
+        }else{
+          res = pPKey2->default_rc;
+        }
+      }else if( res>0 ){
+        res = pPKey2->r2;
+      }else{
+        res = pPKey2->r1;
+      }
+    }else if( res>0 ){
+      res = pPKey2->r2;
+    }else{
+      res = pPKey2->r1;
+    }
+  }
+
+  assert( (res==0 && sqlite3VdbeRecordCompare(nKey1, pKey1, pPKey2)==0)
+       || (res<0 && sqlite3VdbeRecordCompare(nKey1, pKey1, pPKey2)<0)
+       || (res>0 && sqlite3VdbeRecordCompare(nKey1, pKey1, pPKey2)>0)
+       || CORRUPT_DB
+  );
+  return res;
+}
+
+
+int vdbeRecordCompareLargeHeader(
+  int nKey1, const void *pKey1, /* Left key */
+  int dummy1, u32 dummy2,       /* Unused arguments */
+  UnpackedRecord *pPKey2        /* Right key */
+){
+  int szHdr;
+  u32 idx1;
+  idx1 = getVarint32(((u8*)pKey1), szHdr);
+  return vdbeRecordCompare(nKey1, pKey1, szHdr, idx1, pPKey2);
+}
+
 RecordCompare sqlite3VdbeFindCompare(UnpackedRecord *p){
-  if( p->nField==1 && p->pKeyInfo->aSortOrder[0]==0 ){
+  if( (p->pKeyInfo->nField + p->pKeyInfo->nXField) > 10 ){
+    return vdbeRecordCompareLargeHeader;
+  }else{
     int flags = p->aMem[0].flags;
+    if( p->pKeyInfo->aSortOrder[0] ){
+      p->r1 = 1;
+      p->r2 = -1;
+    }else{
+      p->r1 = -1;
+      p->r2 = 1;
+    }
     if( (flags & MEM_Int) ){
       return vdbeRecordCompareInt;
-    }else if( (p->aMem[0].flags&(MEM_Int|MEM_Real|MEM_Null|MEM_Blob))==0 
+    }
+    if( (flags & (MEM_Int|MEM_Real|MEM_Null|MEM_Blob))==0 
         && p->pKeyInfo->aColl[0]==0 
     ){
       return vdbeRecordCompareString;
     }
   }
-  return sqlite3VdbeRecordCompare;
+
+  return vdbeRecordCompare;
 }
 
+RecordCompare sqlite3VdbeFindSorterCompare(KeyInfo *pKeyInfo){
+  if( (pKeyInfo->nField + pKeyInfo->nXField) > 10 ){
+    return vdbeRecordCompareLargeHeader;
+  }
+  return vdbeRecordCompare;
+}
+
+int sqlite3VdbeRecordCompare(
+  int nKey1, const void *pKey1, /* Left key */
+  UnpackedRecord *pPKey2        /* Right key */
+){
+  int szHdr;
+  u32 idx1;
+
+  idx1 = getVarint32(((u8*)pKey1), szHdr);
+  return vdbeRecordCompare(nKey1, pKey1, szHdr, idx1, pPKey2);
+}
 
 /*
 ** pCur points at an index entry created using the OP_MakeRecord opcode.

src/vdbesort.c

@@ -105,6 +105,7 @@ struct VdbeSorter {
   sqlite3_file *pTemp1;           /* PMA file 1 */
   SorterRecord *pRecord;          /* Head of in-memory record list */
   UnpackedRecord *pUnpacked;      /* Used to unpack keys */
+  RecordCompare xRecordCompare;   /* Record compare function */
 };
 
 /*
@@ -412,7 +413,10 @@ static void vdbeSorterCompare(
     assert( r2->default_rc==0 );
   }
 
+#if 0
   *pRes = sqlite3VdbeRecordCompare(nKey1, pKey1, r2);
+#endif
+  *pRes = pSorter->xRecordCompare(nKey1, pKey1, *((u8*)pKey1), 1, r2);
 }
 
 /*
@@ -488,6 +492,7 @@ int sqlite3VdbeSorterInit(sqlite3 *db, VdbeCursor *pCsr){
     if( mxCache<SORTER_MIN_WORKING ) mxCache = SORTER_MIN_WORKING;
     pSorter->mxPmaSize = mxCache * pgsz;
   }
+  pSorter->xRecordCompare = sqlite3VdbeFindSorterCompare(pCsr->pKeyInfo);
 
   return SQLITE_OK;
 }