SQLite

/*
** 2011-07-09
**
** The author disclaims copyright to this source code.  In place of
** a legal notice, here is a blessing:
**
**    May you do good and not evil.
**    May you find forgiveness for yourself and forgive others.
**    May you share freely, never taking more than you give.
**
*************************************************************************
** This file contains code for the VdbeSorter object, used in concert with
** a VdbeCursor to sort large numbers of keys for CREATE TABLE statements
** or by SELECT statements with ORDER BY clauses that cannot be satisfied
** using indexes and without LIMIT clauses.
**
** The VdbeSorter object implements a multi-threaded external merge sort
** algorithm that is efficient even if the number of element being sorted
** exceeds the available memory.
**
** Here is the (internal, non-API) interface between this module and the
** rest of the SQLite system:
**
**    sqlite3VdbeSorterInit()       Create a new VdbeSorter object.
**
**    sqlite3VdbeSorterWrite()      Add a single new row to the VdbeSorter
**                                  object.  The row is a binary blob in the
**                                  OP_MakeRecord format that contains both
**                                  the ORDER BY key columns and result columns
**                                  in the case of a SELECT w/ ORDER BY, or
**                                  the complete record for an index entry
**                                  in the case of a CREATE INDEX.
**
**    sqlite3VdbeSorterRewind()     Sort all content previously added.
**                                  Position the read cursor on the
**                                  first sorted element.
**
**    sqlite3VdbeSorterNext()       Advance the read cursor to the next sorted
**                                  element.
**
**    sqlite3VdbeSorterRowkey()     Return the complete binary blob for the
**                                  row currently under the read cursor.
**
**    sqlite3VdbeSorterCompare()    Compare the binary blob for the row
**                                  currently under the read cursor against
**                                  another binary blob X and report if
**                                  X is strictly less than the read cursor.
**                                  Used to enforce uniqueness in a
**                                  CREATE UNIQUE INDEX statement.
**
**    sqlite3VdbeSorterClose()      Close the VdbeSorter object and reclaim
**                                  all resources.
**
**    sqlite3VdbeSorterReset()      Refurbish the VdbeSorter for reuse.  This
**                                  is like Close() followed by Init() only
**                                  much faster.
**
** The interfaces above must be called in a particular order.  Write() can 
** only occur in between Init()/Reset() and Rewind().  Next(), Rowkey(), and
** Compare() can only occur in between Rewind() and Close()/Reset().
**
** Algorithm:
**
** Records to be sorted are initially held in memory, in the order in
** which they arrive from Write().  When the amount of memory needed exceeds
** a threshold, all in-memory records are sorted and then appended to
** a temporary file as a "Packed-Memory-Array" or "PMA" and the memory is
** reset.  There is a single temporary file used for all PMAs.  The PMAs
** are packed one after another in the file.  The VdbeSorter object keeps
** track of the number of PMAs written.
**
** When the Rewind() is seen, any records still held in memory are sorted.
** If no PMAs have been written (if all records are still held in memory)
** then subsequent Rowkey(), Next(), and Compare() operations work directly
** from memory.  But if PMAs have been written things get a little more
** complicated.
**
** When Rewind() is seen after PMAs have been written, any records still
** in memory are sorted and written as a final PMA.  Then all the PMAs
** are merged together into a single massive PMA that Next(), Rowkey(),
** and Compare() walk to extract the records in sorted order.
**
** If SQLITE_MAX_WORKER_THREADS is non-zero, various steps of the above
** algorithm might be performed in parallel by separate threads.  Threads
** are only used when one or more PMA spill to disk.  If the sort is small
** enough to fit entirely in memory, everything happens on the main thread.
*/

#include "sqliteInt.h"
#include "vdbeInt.h"

/*
** Private objects used by the sorter
*/
typedef struct VdbeSorterIter VdbeSorterIter;
typedef struct SortSubtask SortSubtask;
typedef struct SorterRecord SorterRecord;
typedef struct SorterMerger SorterMerger;
typedef struct FileWriter FileWriter;


/*
** 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 */

/*
** Sorting is divided up into smaller subtasks.  Each subtask is controlled

** by an instance of this object.  Subtask might run in either the main thread

** or in a background thread.
**
** Exactly VdbeSorter.nThread instances of this object are allocated
** as part of each VdbeSorter object. Instances are never allocated any other
** way. VdbeSorter.nThread is set to the number of worker threads allowed
** (see SQLITE_CONFIG_WORKER_THREADS) plus one (the main thread).
**
** When a background thread is launched to perform work, SortSubtask.bDone
** is set to 0 and the SortSubtask.pThread variable set to point to the
** thread handle. SortSubtask.bDone is set to 1 (to indicate to the main
** thread that joining SortSubtask.pThread will not block) before the thread
** exits. SortSubtask.pThread and bDone are always cleared after the 
** background thread has been joined.
**
** One object (specifically, VdbeSorter.aThread[VdbeSorter.nThread-1])
** is reserved for the foreground thread.
**
** The nature of the work performed is determined by SortSubtask.eWork,
** as follows:
**
**   SORT_SUBTASK_SORT:
**     Sort the linked list of records at SortSubtask.pList.
**
**   SORT_SUBTASK_TO_PMA:
**     Sort the linked list of records at SortSubtask.pList, and write
**     the results to a new PMA in temp file SortSubtask.pTemp1. Open
**     the temp file if it is not already open.
**
**   SORT_SUBTASK_CONS:
**     Merge existing PMAs until SortSubtask.nConsolidate or fewer
**     remain in temp file SortSubtask.pTemp1.
*/
struct SortSubtask {
  SQLiteThread *pThread;          /* Thread handle, or NULL */
  int bDone;                      /* Set to true by pThread when finished */

  sqlite3_vfs *pVfs;              /* VFS used to open temporary files */
  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 pThread to sort */
  int nInMemory;                  /* Expected size of PMA based on pList */
  u8 *aListMemory;                /* Records memory (or NULL) */

  int nPMA;                       /* Number of PMAs currently in pTemp1 */
  i64 iTemp1Off;                  /* Offset to write to in pTemp1 */
  sqlite3_file *pTemp1;           /* File to write PMAs to, or NULL */

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

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

  if( pThread->pUnpacked==0 ){
    char *pFree;
    pThread->pUnpacked = sqlite3VdbeAllocUnpackedRecord(
        pThread->pKeyInfo, 0, 0, &pFree
    );
    assert( pThread->pUnpacked==(UnpackedRecord*)pFree );
    if( pFree==0 ){
      rc = SQLITE_NOMEM;
      goto thread_out;
    }
    pThread->pUnpacked->nField = pThread->pKeyInfo->nField;
    pThread->pUnpacked->errCode = 0;
  }

  if( pThread->eWork==SORT_SUBTASK_CONS ){
    assert( pThread->pList==0 );
    while( pThread->nPMA>pThread->nConsolidate && rc==SQLITE_OK ){
      int nIter = MIN(pThread->nPMA, SORTER_MAX_MERGE_COUNT);
      sqlite3_file *pTemp2 = 0;     /* Second temp file to use */
      SorterMerger *pMerger;        /* Object for reading/merging PMA data */
      i64 iReadOff = 0;             /* Offset in pTemp1 to read from */
      i64 iWriteOff = 0;            /* Offset in pTemp2 to write to */

      pThread->iTemp1Off = iWriteOff;
    }
  }else{
    /* Sort the pThread->pList list */
    rc = vdbeSorterSort(pThread);

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

  if( pThread==0 ){
    pThread = &pSorter->aThread[nWorker];
  }
  pSorter->iPrev = (pThread - pSorter->aThread);

  if( rc==SQLITE_OK ){
    assert( pThread->pThread==0 && pThread->bDone==0 );
    pThread->eWork = SORT_SUBTASK_TO_PMA;
    pThread->pList = pSorter->pRecord;
    pThread->nInMemory = pSorter->nInMemory;
    pSorter->nInMemory = 0;
    pSorter->pRecord = 0;

    if( pSorter->aMemory ){
      u8 *aMem = pThread->aListMemory;

  ** sort the VdbeSorter.pRecord list. The vdbe layer will read data directly
  ** from the in-memory list.  */
  if( pSorter->bUsePMA==0 ){
    if( pSorter->pRecord ){
      SortSubtask *pThread = &pSorter->aThread[0];
      *pbEof = 0;
      pThread->pList = pSorter->pRecord;
      pThread->eWork = SORT_SUBTASK_SORT;
      assert( pThread->aListMemory==0 );
      pThread->aListMemory = pSorter->aMemory;
      rc = vdbeSorterRunThread(pThread);
      pThread->aListMemory = 0;
      pSorter->pRecord = pThread->pList;
      pThread->pList = 0;
    }else{

  ** some of them together so that this is no longer the case. */
  if( vdbeSorterCountPMA(pSorter)>SORTER_MAX_MERGE_COUNT ){
    int i;
    for(i=0; rc==SQLITE_OK && i<pSorter->nThread; i++){
      SortSubtask *pThread = &pSorter->aThread[i];
      if( pThread->pTemp1 ){
        pThread->nConsolidate = SORTER_MAX_MERGE_COUNT / pSorter->nThread;
        pThread->eWork = SORT_SUBTASK_CONS;

#if SQLITE_MAX_WORKER_THREADS>0
        if( i<(pSorter->nThread-1) ){
          void *pCtx = (void*)pThread;
          rc = sqlite3ThreadCreate(&pThread->pThread,vdbeSortSubtaskMain,pCtx);
        }else
#endif

/*
** Compare the key in memory cell pVal with the key that the sorter cursor
** passed as the first argument currently points to. For the purposes of
** the comparison, ignore the rowid field at the end of each record.
**
** If the sorter cursor key contains any NULL values, consider it to be
** less than pVal. Even if pVal also contains NULL values.
**
** If an error occurs, return an SQLite error code (i.e. SQLITE_NOMEM).
** Otherwise, set *pRes to a negative, zero or positive value if the
** key in pVal is smaller than, equal to or larger than the current sorter
** key.
**
** This routine forms the core of the OP_SorterCompare opcode, which in
** turn is used to verify uniqueness when constructing a UNIQUE INDEX.
*/
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 */
){