SQLite

** Private objects used by the sorter
*/
typedef struct MergeEngine MergeEngine;     /* Merge PMAs together */
typedef struct PmaReader PmaReader;         /* Incrementally read one PMA */
typedef struct PmaWriter PmaWriter;         /* Incrementally write on PMA */
typedef struct SorterRecord SorterRecord;   /* A record being sorted */
typedef struct SortSubtask SortSubtask;     /* A sub-task in the sort process */
typedef struct SorterFile SorterFile;
typedef struct IncrMerger IncrMerger;


/*
** Candidate values for SortSubtask.eWork
*/
#define SORT_SUBTASK_SORT   1     /* Sort records on pList */
#define SORT_SUBTASK_TO_PMA 2     /* Xfer pList to Packed-Memory-Array pTemp1 */
#define SORT_SUBTASK_CONS   3     /* Consolidate multiple PMAs */

struct SorterFile {
  sqlite3_file *pFd;
  i64 iEof;
};

/*
** Sorting is divided up into smaller subtasks.  Each subtask is controlled
** by an instance of this object. A Subtask might run in either the main thread
** or in a background thread.
**
** Exactly VdbeSorter.nTask instances of this object are allocated

**     remain in temp file SortSubtask.pTemp1.
*/
struct SortSubtask {
  SQLiteThread *pThread;          /* Thread handle, or NULL */
  int bDone;                      /* Set to true by pTask when finished */

  sqlite3 *db;                    /* Database connection */
  VdbeSorter *pSorter;            /* Sorter */
  KeyInfo *pKeyInfo;              /* How to compare records */
  UnpackedRecord *pUnpacked;      /* Space to unpack a record */
  int pgsz;                       /* Main database page size */

  u8 eWork;                       /* One of the SORT_SUBTASK_* constants */
  int nConsolidate;               /* For SORT_SUBTASK_CONS, max final PMAs */
  SorterRecord *pList;            /* List of records for pTask to sort */
  int nInMemory;                  /* Expected size of PMA based on pList */
  u8 *aListMemory;                /* Records memory (or NULL) */

  int nPMA;                       /* Number of PMAs currently in file */

  SorterFile file;
};


/*
** The MergeEngine object is used to combine two or more smaller PMAs into
** one big PMA using a merge operation.  Separate PMAs all need to be
** combined into one big PMA in order to be able to step through the sorted

** sorter cursor created by the VDBE.
*/
struct VdbeSorter {
  int nInMemory;                  /* Current size of pRecord list as PMA */
  int mnPmaSize;                  /* Minimum PMA size, in bytes */
  int mxPmaSize;                  /* Maximum PMA size, in bytes.  0==no limit */
  int bUsePMA;                    /* True if one or more PMAs created */
  int bUseThreads;                /* True if one or more PMAs created */
  SorterRecord *pRecord;          /* Head of in-memory record list */
  PmaReader *pReader;             /* Read data from here after Rewind() */
  UnpackedRecord *pUnpacked;      /* Used by VdbeSorterCompare() */
  u8 *aMemory;                    /* Block of memory to alloc records from */
  int iMemory;                    /* Offset of first free byte in aMemory */
  int nMemory;                    /* Size of aMemory allocation in bytes */
  int iPrev;                      /* Previous thread used to flush PMA */
  int nTask;                      /* Size of aTask[] array */
  SortSubtask aTask[1];           /* One or more subtasks */
};

  int nKey;                       /* Number of bytes in key */
  sqlite3_file *pFile;            /* File iterator is reading from */
  u8 *aAlloc;                     /* Allocated space */
  u8 *aKey;                       /* Pointer to current key */
  u8 *aBuffer;                    /* Current read buffer */
  int nBuffer;                    /* Size of read buffer in bytes */
  u8 *aMap;                       /* Pointer to mapping of entire file */
  IncrMerger *pIncr;              /* Incremental merger */
};

struct IncrMerger {
  int mxSz;                       /* Maximum size of files */
  SortSubtask *pTask;             /* Task that owns this merger */
  int bEof;                       /* Set to true when merge is finished */
  SorterFile aFile[2];            /* aFile[0] for reading, [1] for writing */
  MergeEngine *pMerger;           /* Merge engine thread reads data from */
  SQLiteThread *pThread;          /* Thread currently populating aFile[1] */
};

/*
** An instance of this object is used for writing a PMA.
**
** The PMA is written one record at a time.  Each record is of an arbitrary
** size.  But I/O is more efficient if it occurs in page-sized blocks where

/* The minimum PMA size is set to this value multiplied by the database
** page size in bytes.  */
#define SORTER_MIN_WORKING 10

/* Maximum number of PMAs that a single MergeEngine can merge */
#define SORTER_MAX_MERGE_COUNT 16

static int vdbeIncrSwap(IncrMerger*);
static void vdbeIncrFree(IncrMerger*);

/*
** Free all memory belonging to the PmaReader object passed as the second
** argument. All structure fields are set to zero before returning.
*/
static void vdbePmaReaderClear(PmaReader *pIter){
  sqlite3_free(pIter->aAlloc);
  sqlite3_free(pIter->aBuffer);
  if( pIter->aMap ) sqlite3OsUnfetch(pIter->pFile, 0, pIter->aMap);
  if( pIter->pIncr ) vdbeIncrFree(pIter->pIncr);
  memset(pIter, 0, sizeof(PmaReader));
}

/*
** Read nByte bytes of data from the stream of data iterated by object p.
** If successful, set *ppOut to point to a buffer containing the data
** and return SQLITE_OK. Otherwise, if an error occurs, return an SQLite

    ** In this case, allocate space at p->aAlloc[] to copy the requested
    ** range into. Then return a copy of pointer p->aAlloc to the caller.  */
    int nRem;                     /* Bytes remaining to copy */

    /* Extend the p->aAlloc[] allocation if required. */
    if( p->nAlloc<nByte ){
      u8 *aNew;
      int nNew = MAX(128, p->nAlloc*2);
      while( nByte>nNew ) nNew = nNew*2;
      aNew = sqlite3Realloc(p->aAlloc, nNew);
      if( !aNew ) return SQLITE_NOMEM;
      p->nAlloc = nNew;
      p->aAlloc = aNew;
    }

      sqlite3GetVarint(aVarint, pnOut);
    }
  }

  return SQLITE_OK;
}

static int vdbeSorterMapFile(SortSubtask *pTask, SorterFile *pFile, u8 **pp){
  int rc = SQLITE_OK;
  if( pFile->iEof<=(i64)(pTask->db->nMaxSorterMmap) ){
    rc = sqlite3OsFetch(pFile->pFd, 0, pFile->iEof, (void**)pp);
  }
  return rc;
}

static int vdbePmaReaderReinit(PmaReader *pIter){
  IncrMerger *pIncr = pIter->pIncr;
  SortSubtask *pTask = pIncr->pTask;
  int rc = SQLITE_OK;

  assert( pIncr->bEof==0 );

  if( pIter->aMap ){
    sqlite3OsUnfetch(pIter->pFile, 0, pIter->aMap);
    pIter->aMap = 0;
  }
  pIter->iReadOff = 0;
  pIter->iEof = pIncr->aFile[0].iEof;
  pIter->pFile = pIncr->aFile[0].pFd;

  rc = vdbeSorterMapFile(pTask, &pIncr->aFile[0], &pIter->aMap);
  if( rc==SQLITE_OK ){
    if( pIter->aMap==0 && pIter->aBuffer==0 ){
      pIter->aBuffer = (u8*)sqlite3Malloc(pTask->pgsz);
      if( pIter->aBuffer==0 ) rc = SQLITE_NOMEM;
      pIter->nBuffer = pTask->pgsz;
    }
  }

  return rc;
}


/*
** Advance iterator pIter to the next key in its PMA. Return SQLITE_OK if
** no error occurs, or an SQLite error code if one does.
*/
static int vdbePmaReaderNext(PmaReader *pIter){
  int rc = SQLITE_OK;             /* Return Code */
  u64 nRec = 0;                   /* Size of record in bytes */

  if( pIter->iReadOff>=pIter->iEof ){
    int bEof = 1;
    if( pIter->pIncr ){
      rc = vdbeIncrSwap(pIter->pIncr);
      if( rc==SQLITE_OK && pIter->pIncr->bEof==0 ){
        rc = vdbePmaReaderReinit(pIter);
        bEof = 0;
      }
    }

    if( bEof ){
      /* This is an EOF condition */
      vdbePmaReaderClear(pIter);
      return rc;
    }
  }

  if( rc==SQLITE_OK ){
    rc = vdbePmaReadVarint(pIter, &nRec);
  }
  if( rc==SQLITE_OK ){
    pIter->nKey = (int)nRec;
    rc = vdbePmaReadBlob(pIter, (int)nRec, &pIter->aKey);
  }

  return rc;
}

/*
** Initialize iterator pIter to scan through the PMA stored in file pFile
** starting at offset iStart and ending at offset iEof-1. This function 
** leaves the iterator pointing to the first key in the PMA (or EOF if the 
** PMA is empty).
**
** If the pnByte parameter is NULL, then it is assumed that the file 
** contains a single PMA, and that that PMA omits the initial length varint.
*/
static int vdbePmaReaderInit(
  SortSubtask *pTask,             /* Task context */
  SorterFile *pFile,              /* Sorter file to read from */
  i64 iStart,                     /* Start offset in pFile */
  PmaReader *pIter,               /* Iterator to populate */
  i64 *pnByte                     /* IN/OUT: Increment this value by PMA size */
){
  int rc = SQLITE_OK;
  int nBuf = pTask->pgsz;
  void *pMap = 0;                 /* Mapping of temp file */

  assert( pFile->iEof>iStart );
  assert( pIter->aAlloc==0 );
  assert( pIter->aBuffer==0 );
  pIter->pFile = pFile->pFd;
  pIter->iReadOff = iStart;
  pIter->nAlloc = 128;
  pIter->aAlloc = (u8*)sqlite3Malloc(pIter->nAlloc);
  if( pIter->aAlloc ){
    /* Try to xFetch() a mapping of the entire temp file. If this is possible,
    ** the PMA will be read via the mapping. Otherwise, use xRead().  */
    if( pFile->iEof<=(i64)(pTask->db->nMaxSorterMmap) ){
      rc = sqlite3OsFetch(pIter->pFile, 0, pFile->iEof, &pMap);
    }
  }else{
    rc = SQLITE_NOMEM;
  }

  if( rc==SQLITE_OK ){
    if( pMap ){
      pIter->aMap = (u8*)pMap;
    }else{
      pIter->nBuffer = nBuf;
      pIter->aBuffer = (u8*)sqlite3Malloc(nBuf);
      if( !pIter->aBuffer ){
        rc = SQLITE_NOMEM;
      }else{
        int iBuf = iStart % nBuf;
        if( iBuf ){
          int nRead = nBuf - iBuf;
          if( (iStart + nRead) > pFile->iEof ){
            nRead = (int)(pFile->iEof - iStart);
          }
          rc = sqlite3OsRead(
              pIter->pFile, &pIter->aBuffer[iBuf], nRead, iStart
          );
          assert( rc!=SQLITE_IOERR_SHORT_READ );
        }
      }
    }
  }

  if( rc==SQLITE_OK ){
    u64 nByte;                    /* Size of PMA in bytes */
    pIter->iEof = pFile->iEof;
    rc = vdbePmaReadVarint(pIter, &nByte);
    pIter->iEof = pIter->iReadOff + nByte;
    *pnByte += nByte;
  }

  if( rc==SQLITE_OK ){
    rc = vdbePmaReaderNext(pIter);

    pKeyInfo = (KeyInfo*)((u8*)pSorter + sz);
    memcpy(pKeyInfo, pCsr->pKeyInfo, szKeyInfo);
    pKeyInfo->db = 0;
    if( nField && nWorker==0 ) pKeyInfo->nField = nField;
    pgsz = sqlite3BtreeGetPageSize(db->aDb[0].pBt);

    pSorter->nTask = nWorker + 1;
    pSorter->bUseThreads = (pSorter->nTask>1);
    for(i=0; i<pSorter->nTask; i++){
      SortSubtask *pTask = &pSorter->aTask[i];
      pTask->pKeyInfo = pKeyInfo;
      pTask->pgsz = pgsz;
      pTask->db = db;
      pTask->pSorter = pSorter;
    }

    if( !sqlite3TempInMemory(db) ){
      pSorter->mnPmaSize = SORTER_MIN_WORKING * pgsz;
      mxCache = db->aDb[0].pSchema->cache_size;
      if( mxCache<SORTER_MIN_WORKING ) mxCache = SORTER_MIN_WORKING;
      pSorter->mxPmaSize = mxCache * pgsz;

  if( pTask->aListMemory==0 ){
    vdbeSorterRecordFree(0, pTask->pList);
  }else{
    sqlite3_free(pTask->aListMemory);
    pTask->aListMemory = 0;
  }
  pTask->pList = 0;
  if( pTask->file.pFd ){
    sqlite3OsCloseFree(pTask->file.pFd);
    pTask->file.pFd = 0;
    pTask->file.iEof = 0;
  }
}

/*
** Join all threads.  
*/
#if SQLITE_MAX_WORKER_THREADS>0

** Allocate a new MergeEngine object with space for nIter iterators.
*/
static MergeEngine *vdbeMergeEngineNew(int nIter){
  int N = 2;                      /* Smallest power of two >= nIter */
  int nByte;                      /* Total bytes of space to allocate */
  MergeEngine *pNew;              /* Pointer to allocated object to return */

  /* assert( nIter<=SORTER_MAX_MERGE_COUNT ); */

  while( N<nIter ) N += N;
  nByte = sizeof(MergeEngine) + N * (sizeof(int) + sizeof(PmaReader));

  pNew = (MergeEngine*)sqlite3MallocZero(nByte);
  if( pNew ){
    pNew->nTree = N;
    pNew->aIter = (PmaReader*)&pNew[1];

/*
** Reset a sorting cursor back to its original empty state.
*/
void sqlite3VdbeSorterReset(sqlite3 *db, VdbeSorter *pSorter){
  int i;
  (void)vdbeSorterJoinAll(pSorter, SQLITE_OK);
  if( pSorter->pReader ){
    vdbePmaReaderClear(pSorter->pReader);
    sqlite3DbFree(db, pSorter->pReader);
    pSorter->pReader = 0;
  }
  for(i=0; i<pSorter->nTask; i++){
    SortSubtask *pTask = &pSorter->aTask[i];
    vdbeSortSubtaskCleanup(db, pTask);
  }
  if( pSorter->aMemory==0 ){
    vdbeSorterRecordFree(0, pSorter->pRecord);
  }
  pSorter->pRecord = 0;
  pSorter->nInMemory = 0;
  pSorter->bUsePMA = 0;
  pSorter->iMemory = 0;
  sqlite3DbFree(db, pSorter->pUnpacked);
  pSorter->pUnpacked = 0;
}

/*
** Free any cursor components allocated by sqlite3VdbeSorterXXX routines.
*/
void sqlite3VdbeSorterClose(sqlite3 *db, VdbeCursor *pCsr){
  VdbeSorter *pSorter = pCsr->pSorter;
  if( pSorter ){
    sqlite3VdbeSorterReset(db, pSorter);

    sqlite3_free(pSorter->aMemory);
    sqlite3DbFree(db, pSorter);
    pCsr->pSorter = 0;
  }
}

/*

  int rc = SQLITE_OK;             /* Return code */
  PmaWriter writer;               /* Object used to write to the file */

  memset(&writer, 0, sizeof(PmaWriter));
  assert( pTask->nInMemory>0 );

  /* If the first temporary PMA file has not been opened, open it now. */
  if( pTask->file.pFd==0 ){
    rc = vdbeSorterOpenTempFile(pTask->db->pVfs, &pTask->file.pFd);
    assert( rc!=SQLITE_OK || pTask->file.pFd );
    assert( pTask->file.iEof==0 );
    assert( pTask->nPMA==0 );
  }

  /* Try to get the file to memory map */
  if( rc==SQLITE_OK ){
    vdbeSorterExtendFile(pTask->db, 
        pTask->file.pFd, pTask->file.iEof + pTask->nInMemory + 9
    );
  }

  if( rc==SQLITE_OK ){
    SorterRecord *p;
    SorterRecord *pNext = 0;

    vdbePmaWriterInit(pTask->file.pFd, &writer, pTask->pgsz,
                      pTask->file.iEof);
    pTask->nPMA++;
    vdbePmaWriteVarint(&writer, pTask->nInMemory);
    for(p=pTask->pList; p; p=pNext){
      pNext = p->u.pNext;
      vdbePmaWriteVarint(&writer, p->nVal);
      vdbePmaWriteBlob(&writer, SRVAL(p), p->nVal);
      if( pTask->aListMemory==0 ) sqlite3_free(p);
    }
    pTask->pList = p;
    rc = vdbePmaWriterFinish(&writer, &pTask->file.iEof);
  }

  assert( pTask->pList==0 || rc!=SQLITE_OK );
  return rc;
}

/*

    }
    *pbEof = (pMerger->aIter[pMerger->aTree[1]].pFile==0);
  }

  return rc;
}

#if 0
static void vdbeSorterWorkDebug(SortSubtask *pTask, const char *zEvent){
  i64 t;
  int iTask = (pTask - pTask->pSorter->aTask);
  sqlite3OsCurrentTimeInt64(pTask->db->pVfs, &t);
  fprintf(stderr, "%lld:%d %s\n", t, iTask, zEvent);
}
static void vdbeSorterRewindDebug(sqlite3 *db, const char *zEvent){
  i64 t;
  sqlite3OsCurrentTimeInt64(db->pVfs, &t);
  fprintf(stderr, "%lld:X %s\n", t, zEvent);
}
#else
# define vdbeSorterWorkDebug(x,y)
# define vdbeSorterRewindDebug(x,y)
#endif

/*
** The main routine for sorter-thread operations.
*/
static void *vdbeSortSubtaskMain(void *pCtx){
  int rc = SQLITE_OK;
  SortSubtask *pTask = (SortSubtask*)pCtx;

  assert( pTask->eWork==SORT_SUBTASK_SORT
       || pTask->eWork==SORT_SUBTASK_TO_PMA
       || pTask->eWork==SORT_SUBTASK_CONS
  );
  assert( pTask->bDone==0 );

  vdbeSorterWorkDebug(pTask, "enter");

  if( pTask->pUnpacked==0 ){
    char *pFree;
    pTask->pUnpacked = sqlite3VdbeAllocUnpackedRecord(
        pTask->pKeyInfo, 0, 0, &pFree
    );
    assert( pTask->pUnpacked==(UnpackedRecord*)pFree );

        rc = SQLITE_NOMEM;
        break;
      }

      /* Open a second temp file to write merged data to */
      rc = vdbeSorterOpenTempFile(pTask->db->pVfs, &pTemp2);
      if( rc==SQLITE_OK ){
        vdbeSorterExtendFile(pTask->db, pTemp2, pTask->file.iEof);
      }else{
        vdbeMergeEngineFree(pMerger);
        break;
      }

      /* This loop runs once for each output PMA. Each output PMA is made
      ** of data merged from up to SORTER_MAX_MERGE_COUNT input PMAs. */
      for(i=0; rc==SQLITE_OK && i<pTask->nPMA; i+=SORTER_MAX_MERGE_COUNT){
        PmaWriter writer;         /* Object for writing data to pTemp2 */
        i64 nOut = 0;             /* Bytes of data in output PMA */
        int bEof = 0;
        int rc2;

        /* Configure the merger object to read and merge data from the next 
        ** SORTER_MAX_MERGE_COUNT PMAs in pTemp1 (or from all remaining PMAs,
        ** if that is fewer). */
        int iIter;
        for(iIter=0; iIter<SORTER_MAX_MERGE_COUNT; iIter++){
          PmaReader *pIter = &pMerger->aIter[iIter];
          rc = vdbePmaReaderInit(pTask, &pTask->file, iReadOff, pIter, &nOut);
          iReadOff = pIter->iEof;
          if( iReadOff>=pTask->file.iEof || rc!=SQLITE_OK ) break;
        }
        for(iIter=pMerger->nTree-1; rc==SQLITE_OK && iIter>0; iIter--){
          rc = vdbeSorterDoCompare(pTask, pMerger, iIter);
        }

        vdbePmaWriterInit(pTemp2, &writer, pTask->pgsz, iWriteOff);
        vdbePmaWriteVarint(&writer, nOut);
        while( rc==SQLITE_OK && bEof==0 ){
          PmaReader *pIter = &pMerger->aIter[ pMerger->aTree[1] ];
          assert( pIter->pFile!=0 );        /* pIter is not at EOF */
          vdbePmaWriteVarint(&writer, pIter->nKey);
          vdbePmaWriteBlob(&writer, pIter->aKey, pIter->nKey);
          rc = vdbeSorterNext(pTask, pMerger, &bEof);
        }
        rc2 = vdbePmaWriterFinish(&writer, &iWriteOff);
        if( rc==SQLITE_OK ) rc = rc2;
      }

      vdbeMergeEngineFree(pMerger);
      sqlite3OsCloseFree(pTask->file.pFd);
      pTask->file.pFd = pTemp2;
      pTask->nPMA = (i / SORTER_MAX_MERGE_COUNT);
      pTask->file.iEof = iWriteOff;
    }
  }else{
    /* Sort the pTask->pList list */
    rc = vdbeSorterSort(pTask);

    /* If required, write the list out to a PMA. */
    if( rc==SQLITE_OK && pTask->eWork==SORT_SUBTASK_TO_PMA ){
#ifdef SQLITE_DEBUG
      i64 nExpect = pTask->nInMemory
        + sqlite3VarintLen(pTask->nInMemory)
        + pTask->file.iEof;
#endif
      rc = vdbeSorterListToPMA(pTask);
      assert( rc!=SQLITE_OK || (nExpect==pTask->file.iEof) );
    }
  }

 thread_out:
  pTask->bDone = 1;
  if( rc==SQLITE_OK && pTask->pUnpacked->errCode ){
    assert( pTask->pUnpacked->errCode==SQLITE_NOMEM );
    rc = SQLITE_NOMEM;
  }
  vdbeSorterWorkDebug(pTask, "exit");
  return SQLITE_INT_TO_PTR(rc);
}

/*
** Run the activity scheduled by the object passed as the only argument
** in the current thread.
*/

  int nPMA = 0;
  int i;
  for(i=0; i<pSorter->nTask; i++){
    nPMA += pSorter->aTask[i].nPMA;
  }
  return nPMA;
}

/*
** Read keys from pIncr->pMerger and populate pIncr->aFile[1]. The format
** of the data stored in aFile[1] is the same as that used by regular PMAs,
** except that the number-of-bytes varint is omitted from the start.
*/
static int vdbeIncrPopulate(IncrMerger *pIncr){
  int rc = SQLITE_OK;
  int rc2;
  SorterFile *pOut = &pIncr->aFile[1];
  MergeEngine *pMerger = pIncr->pMerger;
  PmaWriter writer;
  assert( pIncr->bEof==0 );

  vdbePmaWriterInit(pIncr->aFile[1].pFd, &writer, pIncr->pTask->pgsz, 0);
  while( rc==SQLITE_OK ){
    int dummy;
    PmaReader *pReader = &pMerger->aIter[ pMerger->aTree[1] ];
    int nKey = pReader->nKey;
    i64 iEof = writer.iWriteOff + writer.iBufEnd;

    /* Check if the output file is full or if the input has been exhausted.
    ** In either case exit the loop. */
    if( pReader->pFile==0 ) break;
    if( iEof && (iEof + nKey)>pIncr->mxSz ) break;

    /* Write the next key to the output. */
    vdbePmaWriteVarint(&writer, nKey);
    vdbePmaWriteBlob(&writer, pReader->aKey, nKey);
    rc = vdbeSorterNext(pIncr->pTask, pIncr->pMerger, &dummy);
  }

  rc2 = vdbePmaWriterFinish(&writer, &pOut->iEof);
  if( rc==SQLITE_OK ) rc = rc2;
  return rc;
}

static void *vdbeIncrPopulateThreadMain(void *pCtx){
  IncrMerger *pIncr = (IncrMerger*)pCtx;
  return SQLITE_INT_TO_PTR( vdbeIncrPopulate(pIncr) );
}

static int vdbeIncrBgPopulate(IncrMerger *pIncr){
  int rc;
  assert( pIncr->pThread==0 );
  if( pIncr->pTask->pSorter->bUseThreads==0 ){
    rc = vdbeIncrPopulate(pIncr);
  }else{
    void *pCtx = (void*)pIncr;
    rc = sqlite3ThreadCreate(&pIncr->pThread, vdbeIncrPopulateThreadMain, pCtx);
  }
  return rc;
}

static int vdbeIncrSwap(IncrMerger *pIncr){
  int rc = SQLITE_OK;
  
  if( pIncr->pThread ){
    void *pRet;
    rc = sqlite3ThreadJoin(pIncr->pThread, &pRet);
    if( rc==SQLITE_OK ) rc = SQLITE_PTR_TO_INT(pRet);
    pIncr->pThread = 0;
  }

  if( rc==SQLITE_OK ){
    SorterFile f0 = pIncr->aFile[0];
    pIncr->aFile[0] = pIncr->aFile[1];
    pIncr->aFile[1] = f0;

    if( pIncr->aFile[0].iEof==0 ){
      pIncr->bEof = 1;
    }else{
      rc = vdbeIncrBgPopulate(pIncr);
    }
  }

  return rc;
}

static void vdbeIncrFree(IncrMerger *pIncr){
  if( pIncr->pThread ){
    void *pRet;
    sqlite3ThreadJoin(pIncr->pThread, &pRet);
  }
  if( pIncr->aFile[0].pFd ) sqlite3OsCloseFree(pIncr->aFile[0].pFd);
  if( pIncr->aFile[1].pFd ) sqlite3OsCloseFree(pIncr->aFile[1].pFd);
  vdbeMergeEngineFree(pIncr->pMerger);
  sqlite3_free(pIncr);
}

/*
** Populate iterator *pIter so that it may be used to iterate through all 
** keys stored in subtask pTask using the incremental merge method.
*/
static int vdbePmaReaderIncrInit(VdbeSorter *pSorter, PmaReader *pIter){
  SortSubtask *pTask0 = &pSorter->aTask[0];
  int rc = SQLITE_OK;
  MergeEngine *pMerger = 0;
  IncrMerger *pIncr = 0;
  int i;
  int nPMA = 0;

  for(i=0; i<pSorter->nTask; i++){
    nPMA += pSorter->aTask[i].nPMA;
  }
  pMerger = vdbeMergeEngineNew(nPMA);
  if( pMerger==0 ){
    rc = SQLITE_NOMEM;
  }else{
    int iIter = 0;
    int iPMA;
    for(i=0; i<pSorter->nTask; i++){
      i64 iReadOff = 0;
      SortSubtask *pTask = &pSorter->aTask[i];
      for(iPMA=0; iPMA<pTask->nPMA; iPMA++){
        i64 nDummy = 0;
        PmaReader *pIter = &pMerger->aIter[iIter++];
        rc = vdbePmaReaderInit(pTask, &pTask->file, iReadOff, pIter, &nDummy);
        iReadOff = pIter->iEof;
      }
    }
    for(i=pMerger->nTree-1; rc==SQLITE_OK && i>0; i--){
      rc = vdbeSorterDoCompare(pTask0, pMerger, i);
    }
  }

  if( rc==SQLITE_OK ){
    pIncr = (IncrMerger*)sqlite3_malloc(sizeof(IncrMerger));
    if( pIncr==0 ){
      rc = SQLITE_NOMEM;
    }else{
      memset(pIncr, 0, sizeof(IncrMerger));
      pIncr->mxSz = (pSorter->mxPmaSize / 2);
      pIncr->pMerger = pMerger;
      pIncr->pTask = pTask0;
    }
  }

  /* Open the two temp files. */
  if( rc==SQLITE_OK ){
    rc = vdbeSorterOpenTempFile(pTask0->db->pVfs, &pIncr->aFile[0].pFd);
  }
  if( rc==SQLITE_OK ){
    rc = vdbeSorterOpenTempFile(pTask0->db->pVfs, &pIncr->aFile[1].pFd);
  }

  /* Launch a background thread to populate aFile[1]. */
  if( rc==SQLITE_OK ){
    rc = vdbeIncrBgPopulate(pIncr);
  }

  pIter->pIncr = pIncr;
  if( rc==SQLITE_OK ){
    rc = vdbePmaReaderNext(pIter);
  }
  return rc;
}


/*
** Once the sorter has been populated by calls to sqlite3VdbeSorterWrite,
** this function is called to prepare for iterating through the records
** in sorted order.
*/
int sqlite3VdbeSorterRewind(sqlite3 *db, const VdbeCursor *pCsr, int *pbEof){

  if( pSorter->pRecord ){
    rc = vdbeSorterFlushPMA(db, pCsr, 1);
  }

  /* Join all threads */
  rc = vdbeSorterJoinAll(pSorter, rc);










  vdbeSorterRewindDebug(db, "rewind");
















  /* Assuming no errors have occurred, set up a merger structure to 
  ** incrementally read and merge all remaining PMAs.  */
  assert( pSorter->pReader==0 );
  if( rc==SQLITE_OK ){


    PmaReader *pReader;



    pReader = (PmaReader*)sqlite3DbMallocZero(db, sizeof(PmaReader));
    pSorter->pReader = pReader;












    rc = vdbePmaReaderIncrInit(pSorter, pReader);
    assert( rc!=SQLITE_OK || pReader->pFile );


    *pbEof = 0;


  }


  vdbeSorterRewindDebug(db, "rewinddone");



  return rc;
}

/*
** Advance to the next element in the sorter.
*/
int sqlite3VdbeSorterNext(sqlite3 *db, const VdbeCursor *pCsr, int *pbEof){
  VdbeSorter *pSorter = pCsr->pSorter;
  int rc;                         /* Return code */

  if( pSorter->pReader ){
    rc = vdbePmaReaderNext(pSorter->pReader);
    *pbEof = (pSorter->pReader->pFile==0);
  }else{
    SorterRecord *pFree = pSorter->pRecord;
    pSorter->pRecord = pFree->u.pNext;
    pFree->u.pNext = 0;
    if( pSorter->aMemory==0 ) vdbeSorterRecordFree(db, pFree);
    *pbEof = !pSorter->pRecord;
    rc = SQLITE_OK;
  }
  return rc;
}

/*
** Return a pointer to a buffer owned by the sorter that contains the 
** current key.
*/
static void *vdbeSorterRowkey(
  const VdbeSorter *pSorter,      /* Sorter object */
  int *pnKey                      /* OUT: Size of current key in bytes */
){
  void *pKey;
  if( pSorter->pReader ){


    *pnKey = pSorter->pReader->nKey;
    pKey = pSorter->pReader->aKey;
  }else{
    *pnKey = pSorter->pRecord->nVal;
    pKey = SRVAL(pSorter->pRecord);
  }
  return pKey;
}

int sqlite3VdbeSorterCompare(
  const VdbeCursor *pCsr,         /* Sorter cursor */
  Mem *pVal,                      /* Value to compare to current sorter key */
  int nIgnore,                    /* Ignore this many fields at the end */
  int *pRes                       /* OUT: Result of comparison */
){
  VdbeSorter *pSorter = pCsr->pSorter;
  UnpackedRecord *r2 = pSorter->pUnpacked;
  KeyInfo *pKeyInfo = pCsr->pKeyInfo;
  int i;
  void *pKey; int nKey;           /* Sorter key to compare pVal with */

  if( r2==0 ){
    char *p;
    r2 = pSorter->pUnpacked = sqlite3VdbeAllocUnpackedRecord(pKeyInfo,0,0,&p);
    assert( pSorter->pUnpacked==(UnpackedRecord*)p );
    if( r2==0 ) return SQLITE_NOMEM;
    r2->nField = pKeyInfo->nField-nIgnore;
  }
  assert( r2->nField>=pKeyInfo->nField-nIgnore );

  pKey = vdbeSorterRowkey(pSorter, &nKey);
  sqlite3VdbeRecordUnpack(pKeyInfo, nKey, pKey, r2);
  for(i=0; i<r2->nField; i++){
    if( r2->aMem[i].flags & MEM_Null ){
      *pRes = -1;
      return SQLITE_OK;
    }
  }

  *pRes = sqlite3VdbeRecordCompare(pVal->n, pVal->z, r2, 0);
  return SQLITE_OK;
}

  ) INSERT INTO t1 SELECT * FROM r;
}

do_execsql_test 2.2 {
  CREATE UNIQUE INDEX i1 ON t1(b, a);
}

do_execsql_test 2.3 {
  CREATE UNIQUE INDEX i2 ON t1(a);
}

do_execsql_test 2.4 { PRAGMA integrity_check } {ok}

db close
sqlite3_shutdown
sqlite3_config_worker_threads 0
sqlite3_initialize
finish_test