/* ** 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" /* ** If SQLITE_DEBUG_SORTER_THREADS is defined, this module outputs various ** messages to stderr that may be helpful in understanding the performance ** characteristics of the sorter in multi-threaded mode. */ #if 0 # define SQLITE_DEBUG_SORTER_THREADS 1 #endif /* ** 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 SorterThread SorterThread; typedef struct SorterList SorterList; typedef struct IncrMerger IncrMerger; /* ** A container for a temp file handle and the current amount of data ** stored in the file. */ struct SorterFile { sqlite3_file *pFd; /* File handle */ i64 iEof; /* Bytes of data stored in pFd */ }; /* ** An object of this type is used to store the thread handle for each ** background thread launched by the sorter. Before the thread is launched, ** variable bDone is set to 0. Then, right before it exits, the thread ** itself sets bDone to 1. ** ** This is then used for two purposes: ** ** 1. When flushing the contents of memory to a level-0 PMA on disk, to ** attempt to select a SortSubtask for which there is not already an ** active background thread (since doing so causes the main thread ** to block until it finishes). ** ** 2. If SQLITE_DEBUG_SORTER_THREADS is defined, to determine if a call ** to sqlite3ThreadJoin() is likely to block. ** ** In both cases, the effects of the main thread seeing (bDone==0) even ** after the thread has finished are not dire. So we don't worry about ** memory barriers and such here. */ struct SorterThread { SQLiteThread *pThread; int bDone; }; struct SorterList { SorterRecord *pList; /* Linked list of records */ u8 *aMemory; /* If non-NULL, blob of memory for pList */ int szPMA; /* Size of pList as PMA in bytes */ }; /* ** 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 ** as part of each VdbeSorter object. Instances are never allocated any other ** way. VdbeSorter.nTask 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.pTask variable set to point to the ** thread handle. SortSubtask.bDone is set to 1 (to indicate to the main ** thread that joining SortSubtask.pTask will not block) before the thread ** exits. SortSubtask.pTask and bDone are always cleared after the ** background thread has been joined. ** ** One object (specifically, VdbeSorter.aTask[VdbeSorter.nTask-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 { SorterThread thread; 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 */ SorterList list; /* List for thread to write to a PMA */ int nPMA; /* Number of PMAs currently in file */ SorterFile file; /* Temp file for level-0 PMAs */ SorterFile file2; /* Space for other PMAs */ }; /* ** 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 ** records in order. ** ** The aIter[] array contains a PmaReader object for each of the PMAs being ** merged. An aIter[] object either points to a valid key or else is at EOF. ** For the purposes of the paragraphs below, we assume that the array is ** actually N elements in size, where N is the smallest power of 2 greater ** to or equal to the number of PMAs being merged. The extra aIter[] elements ** are treated as if they are empty (always at EOF). ** ** The aTree[] array is also N elements in size. The value of N is stored in ** the MergeEngine.nTree variable. ** ** The final (N/2) elements of aTree[] contain the results of comparing ** pairs of PMA keys together. Element i contains the result of ** comparing aIter[2*i-N] and aIter[2*i-N+1]. Whichever key is smaller, the ** aTree element is set to the index of it. ** ** For the purposes of this comparison, EOF is considered greater than any ** other key value. If the keys are equal (only possible with two EOF ** values), it doesn't matter which index is stored. ** ** The (N/4) elements of aTree[] that precede the final (N/2) described ** above contains the index of the smallest of each block of 4 iterators. ** And so on. So that aTree[1] contains the index of the iterator that ** currently points to the smallest key value. aTree[0] is unused. ** ** Example: ** ** aIter[0] -> Banana ** aIter[1] -> Feijoa ** aIter[2] -> Elderberry ** aIter[3] -> Currant ** aIter[4] -> Grapefruit ** aIter[5] -> Apple ** aIter[6] -> Durian ** aIter[7] -> EOF ** ** aTree[] = { X, 5 0, 5 0, 3, 5, 6 } ** ** The current element is "Apple" (the value of the key indicated by ** iterator 5). When the Next() operation is invoked, iterator 5 will ** be advanced to the next key in its segment. Say the next key is ** "Eggplant": ** ** aIter[5] -> Eggplant ** ** The contents of aTree[] are updated first by comparing the new iterator ** 5 key to the current key of iterator 4 (still "Grapefruit"). The iterator ** 5 value is still smaller, so aTree[6] is set to 5. And so on up the tree. ** The value of iterator 6 - "Durian" - is now smaller than that of iterator ** 5, so aTree[3] is set to 6. Key 0 is smaller than key 6 (BananaaAlloc); 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 ** error code. ** ** The buffer indicated by *ppOut may only be considered valid until the ** next call to this function. */ static int vdbePmaReadBlob( PmaReader *p, /* Iterator */ int nByte, /* Bytes of data to read */ u8 **ppOut /* OUT: Pointer to buffer containing data */ ){ int iBuf; /* Offset within buffer to read from */ int nAvail; /* Bytes of data available in buffer */ if( p->aMap ){ *ppOut = &p->aMap[p->iReadOff]; p->iReadOff += nByte; return SQLITE_OK; } assert( p->aBuffer ); /* If there is no more data to be read from the buffer, read the next ** p->nBuffer bytes of data from the file into it. Or, if there are less ** than p->nBuffer bytes remaining in the PMA, read all remaining data. */ iBuf = p->iReadOff % p->nBuffer; if( iBuf==0 ){ int nRead; /* Bytes to read from disk */ int rc; /* sqlite3OsRead() return code */ /* Determine how many bytes of data to read. */ if( (p->iEof - p->iReadOff) > (i64)p->nBuffer ){ nRead = p->nBuffer; }else{ nRead = (int)(p->iEof - p->iReadOff); } assert( nRead>0 ); /* Read data from the file. Return early if an error occurs. */ rc = sqlite3OsRead(p->pFile, p->aBuffer, nRead, p->iReadOff); assert( rc!=SQLITE_IOERR_SHORT_READ ); if( rc!=SQLITE_OK ) return rc; } nAvail = p->nBuffer - iBuf; if( nByte<=nAvail ){ /* The requested data is available in the in-memory buffer. In this ** case there is no need to make a copy of the data, just return a ** pointer into the buffer to the caller. */ *ppOut = &p->aBuffer[iBuf]; p->iReadOff += nByte; }else{ /* The requested data is not all available in the in-memory buffer. ** 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->nAllocnAlloc*2); while( nByte>nNew ) nNew = nNew*2; aNew = sqlite3Realloc(p->aAlloc, nNew); if( !aNew ) return SQLITE_NOMEM; p->nAlloc = nNew; p->aAlloc = aNew; } /* Copy as much data as is available in the buffer into the start of ** p->aAlloc[]. */ memcpy(p->aAlloc, &p->aBuffer[iBuf], nAvail); p->iReadOff += nAvail; nRem = nByte - nAvail; /* The following loop copies up to p->nBuffer bytes per iteration into ** the p->aAlloc[] buffer. */ while( nRem>0 ){ int rc; /* vdbePmaReadBlob() return code */ int nCopy; /* Number of bytes to copy */ u8 *aNext; /* Pointer to buffer to copy data from */ nCopy = nRem; if( nRem>p->nBuffer ) nCopy = p->nBuffer; rc = vdbePmaReadBlob(p, nCopy, &aNext); if( rc!=SQLITE_OK ) return rc; assert( aNext!=p->aAlloc ); memcpy(&p->aAlloc[nByte - nRem], aNext, nCopy); nRem -= nCopy; } *ppOut = p->aAlloc; } return SQLITE_OK; } /* ** Read a varint from the stream of data accessed by p. Set *pnOut to ** the value read. */ static int vdbePmaReadVarint(PmaReader *p, u64 *pnOut){ int iBuf; if( p->aMap ){ p->iReadOff += sqlite3GetVarint(&p->aMap[p->iReadOff], pnOut); }else{ iBuf = p->iReadOff % p->nBuffer; if( iBuf && (p->nBuffer-iBuf)>=9 ){ p->iReadOff += sqlite3GetVarint(&p->aBuffer[iBuf], pnOut); }else{ u8 aVarint[16], *a; int i = 0, rc; do{ rc = vdbePmaReadBlob(p, 1, &a); if( rc ) return rc; aVarint[(i++)&0xf] = a[0]; }while( (a[0]&0x80)!=0 ); 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 = pIncr->iStartOff; 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 ){ /* TODO: Combine this code with similar code in vdbePmaReaderInit() */ int iBuf = pIter->iReadOff % pTask->pgsz; if( pIter->aBuffer==0 ){ pIter->aBuffer = (u8*)sqlite3Malloc(pTask->pgsz); if( pIter->aBuffer==0 ) rc = SQLITE_NOMEM; pIter->nBuffer = pTask->pgsz; } if( iBuf ){ int nRead = pTask->pgsz - iBuf; if( (pIter->iReadOff + nRead) > pIter->iEof ){ nRead = (int)(pIter->iEof - pIter->iReadOff); } rc = sqlite3OsRead( pIter->pFile, &pIter->aBuffer[iBuf], nRead, pIter->iReadOff ); assert( rc!=SQLITE_IOERR_SHORT_READ ); } } } 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; 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(). */ rc = vdbeSorterMapFile(pTask, pFile, &pIter->aMap); }else{ rc = SQLITE_NOMEM; } if( rc==SQLITE_OK && pIter->aMap==0 ){ 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); } return rc; } /* ** Compare key1 (buffer pKey1, size nKey1 bytes) with key2 (buffer pKey2, ** size nKey2 bytes). Use (pTask->pKeyInfo) for the collation sequences ** used by the comparison. Return the result of the comparison. ** ** Before returning, object (pTask->pUnpacked) is populated with the ** unpacked version of key2. Or, if pKey2 is passed a NULL pointer, then it ** is assumed that the (pTask->pUnpacked) structure already contains the ** unpacked key to use as key2. ** ** If an OOM error is encountered, (pTask->pUnpacked->error_rc) is set ** to SQLITE_NOMEM. */ static int vdbeSorterCompare( SortSubtask *pTask, /* Subtask context (for pKeyInfo) */ const void *pKey1, int nKey1, /* Left side of comparison */ const void *pKey2, int nKey2 /* Right side of comparison */ ){ UnpackedRecord *r2 = pTask->pUnpacked; if( pKey2 ){ sqlite3VdbeRecordUnpack(pTask->pKeyInfo, nKey2, pKey2, r2); } return sqlite3VdbeRecordCompare(nKey1, pKey1, r2, 0); } /* ** This function is called to compare two iterator keys when merging ** multiple b-tree segments. Parameter iOut is the index of the aTree[] ** value to recalculate. */ static int vdbeSorterDoCompare( SortSubtask *pTask, MergeEngine *pMerger, int iOut ){ int i1; int i2; int iRes; PmaReader *p1; PmaReader *p2; assert( iOutnTree && iOut>0 ); if( iOut>=(pMerger->nTree/2) ){ i1 = (iOut - pMerger->nTree/2) * 2; i2 = i1 + 1; }else{ i1 = pMerger->aTree[iOut*2]; i2 = pMerger->aTree[iOut*2+1]; } p1 = &pMerger->aIter[i1]; p2 = &pMerger->aIter[i2]; if( p1->pFile==0 ){ iRes = i2; }else if( p2->pFile==0 ){ iRes = i1; }else{ int res; assert( pTask->pUnpacked!=0 ); /* allocated in vdbeSortSubtaskMain() */ res = vdbeSorterCompare( pTask, p1->aKey, p1->nKey, p2->aKey, p2->nKey ); if( res<=0 ){ iRes = i1; }else{ iRes = i2; } } pMerger->aTree[iOut] = iRes; return SQLITE_OK; } /* ** Initialize the temporary index cursor just opened as a sorter cursor. */ int sqlite3VdbeSorterInit( sqlite3 *db, /* Database connection (for malloc()) */ int nField, /* Number of key fields in each record */ VdbeCursor *pCsr /* Cursor that holds the new sorter */ ){ int pgsz; /* Page size of main database */ int i; /* Used to iterate through aTask[] */ int mxCache; /* Cache size */ VdbeSorter *pSorter; /* The new sorter */ KeyInfo *pKeyInfo; /* Copy of pCsr->pKeyInfo with db==0 */ int szKeyInfo; /* Size of pCsr->pKeyInfo in bytes */ int sz; /* Size of pSorter in bytes */ int rc = SQLITE_OK; int nWorker = (sqlite3GlobalConfig.bCoreMutex?sqlite3GlobalConfig.nWorker:0); assert( pCsr->pKeyInfo && pCsr->pBt==0 ); szKeyInfo = sizeof(KeyInfo) + (pCsr->pKeyInfo->nField-1)*sizeof(CollSeq*); sz = sizeof(VdbeSorter) + nWorker * sizeof(SortSubtask); pSorter = (VdbeSorter*)sqlite3DbMallocZero(db, sz + szKeyInfo); pCsr->pSorter = pSorter; if( pSorter==0 ){ rc = SQLITE_NOMEM; }else{ 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; inTask; 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( mxCachemxPmaSize = mxCache * pgsz; /* If the application is using memsys3 or memsys5, use a separate ** allocation for each sort-key in memory. Otherwise, use a single big ** allocation at pSorter->aMemory for all sort-keys. */ if( sqlite3GlobalConfig.pHeap==0 ){ assert( pSorter->iMemory==0 ); pSorter->nMemory = pgsz; pSorter->list.aMemory = (u8*)sqlite3Malloc(pgsz); if( !pSorter->list.aMemory ) rc = SQLITE_NOMEM; } } } return rc; } /* ** Free the list of sorted records starting at pRecord. */ static void vdbeSorterRecordFree(sqlite3 *db, SorterRecord *pRecord){ SorterRecord *p; SorterRecord *pNext; for(p=pRecord; p; p=pNext){ pNext = p->u.pNext; sqlite3DbFree(db, p); } } /* ** Free all resources owned by the object indicated by argument pTask. All ** fields of *pTask are zeroed before returning. */ static void vdbeSortSubtaskCleanup(sqlite3 *db, SortSubtask *pTask){ sqlite3DbFree(db, pTask->pUnpacked); pTask->pUnpacked = 0; if( pTask->list.aMemory==0 ){ vdbeSorterRecordFree(0, pTask->list.pList); }else{ sqlite3_free(pTask->list.aMemory); pTask->list.aMemory = 0; } pTask->list.pList = 0; if( pTask->file.pFd ){ sqlite3OsCloseFree(pTask->file.pFd); pTask->file.pFd = 0; pTask->file.iEof = 0; } if( pTask->file2.pFd ){ sqlite3OsCloseFree(pTask->file2.pFd); pTask->file2.pFd = 0; pTask->file2.iEof = 0; } } #ifdef SQLITE_DEBUG_SORTER_THREADS 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); } static void vdbeSorterPopulateDebug( SortSubtask *pTask, const char *zEvent ){ i64 t; int iTask = (pTask - pTask->pSorter->aTask); sqlite3OsCurrentTimeInt64(pTask->db->pVfs, &t); fprintf(stderr, "%lld:bg%d %s\n", t, iTask, zEvent); } static void vdbeSorterBlockDebug( SortSubtask *pTask, int bBlocked, const char *zEvent ){ if( bBlocked ){ i64 t; sqlite3OsCurrentTimeInt64(pTask->db->pVfs, &t); fprintf(stderr, "%lld:main %s\n", t, zEvent); } } #else # define vdbeSorterWorkDebug(x,y) # define vdbeSorterRewindDebug(x,y) # define vdbeSorterPopulateDebug(x,y) # define vdbeSorterBlockDebug(x,y,z) #endif #if SQLITE_MAX_WORKER_THREADS>0 /* ** Join thread p. */ static int vdbeSorterJoinThread(SortSubtask *pTask, SorterThread *p){ int rc = SQLITE_OK; if( p->pThread ){ #ifdef SQLITE_DEBUG_SORTER_THREADS int bDone = p->bDone; #endif void *pRet; vdbeSorterBlockDebug(pTask, !bDone, "enter"); rc = sqlite3ThreadJoin(p->pThread, &pRet); vdbeSorterBlockDebug(pTask, !bDone, "exit"); if( rc==SQLITE_OK ) rc = SQLITE_PTR_TO_INT(pRet); assert( p->bDone==1 ); p->bDone = 0; p->pThread = 0; } return rc; } /* ** Launch a background thread to run xTask(pIn). */ static int vdbeSorterCreateThread( SorterThread *p, /* Thread object to populate */ void *(*xTask)(void*), /* Routine to run in a separate thread */ void *pIn /* Argument passed into xTask() */ ){ assert( p->pThread==0 && p->bDone==0 ); return sqlite3ThreadCreate(&p->pThread, xTask, pIn); } /* ** Join all outstanding threads launched by SorterWrite() to create ** level-0 PMAs. */ static int vdbeSorterJoinAll(VdbeSorter *pSorter, int rcin){ int rc = rcin; int i; for(i=0; inTask; i++){ SortSubtask *pTask = &pSorter->aTask[i]; int rc2 = vdbeSorterJoinThread(pTask, &pTask->thread); if( rc==SQLITE_OK ) rc = rc2; } return rc; } #else # define vdbeSorterJoinAll(x,rcin) (rcin) # define vdbeSorterJoinThread(pTask,p) SQLITE_OK #endif /* ** 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( NnTree = N; pNew->aIter = (PmaReader*)&pNew[1]; pNew->aTree = (int*)&pNew->aIter[N]; } return pNew; } /* ** Free the MergeEngine object passed as the only argument. */ static void vdbeMergeEngineFree(MergeEngine *pMerger){ int i; if( pMerger ){ for(i=0; inTree; i++){ vdbePmaReaderClear(&pMerger->aIter[i]); } } sqlite3_free(pMerger); } /* ** 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; inTask; i++){ SortSubtask *pTask = &pSorter->aTask[i]; vdbeSortSubtaskCleanup(db, pTask); } if( pSorter->list.aMemory==0 ){ vdbeSorterRecordFree(0, pSorter->list.pList); } pSorter->list.pList = 0; pSorter->list.szPMA = 0; pSorter->bUsePMA = 0; pSorter->iMemory = 0; pSorter->mxKeysize = 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->list.aMemory); sqlite3DbFree(db, pSorter); pCsr->pSorter = 0; } } /* ** Allocate space for a file-handle and open a temporary file. If successful, ** set *ppFile to point to the malloc'd file-handle and return SQLITE_OK. ** Otherwise, set *ppFile to 0 and return an SQLite error code. */ static int vdbeSorterOpenTempFile(sqlite3_vfs *pVfs, sqlite3_file **ppFile){ int rc; rc = sqlite3OsOpenMalloc(pVfs, 0, ppFile, SQLITE_OPEN_TEMP_JOURNAL | SQLITE_OPEN_READWRITE | SQLITE_OPEN_CREATE | SQLITE_OPEN_EXCLUSIVE | SQLITE_OPEN_DELETEONCLOSE, &rc ); if( rc==SQLITE_OK ){ i64 max = SQLITE_MAX_MMAP_SIZE; sqlite3OsFileControlHint( *ppFile, SQLITE_FCNTL_MMAP_SIZE, (void*)&max); } return rc; } static int vdbeSortAllocUnpacked(SortSubtask *pTask){ if( pTask->pUnpacked==0 ){ char *pFree; pTask->pUnpacked = sqlite3VdbeAllocUnpackedRecord( pTask->pKeyInfo, 0, 0, &pFree ); assert( pTask->pUnpacked==(UnpackedRecord*)pFree ); if( pFree==0 ) return SQLITE_NOMEM; pTask->pUnpacked->nField = pTask->pKeyInfo->nField; pTask->pUnpacked->errCode = 0; } return SQLITE_OK; } /* ** Merge the two sorted lists p1 and p2 into a single list. ** Set *ppOut to the head of the new list. */ static void vdbeSorterMerge( SortSubtask *pTask, /* Calling thread context */ SorterRecord *p1, /* First list to merge */ SorterRecord *p2, /* Second list to merge */ SorterRecord **ppOut /* OUT: Head of merged list */ ){ SorterRecord *pFinal = 0; SorterRecord **pp = &pFinal; void *pVal2 = p2 ? SRVAL(p2) : 0; while( p1 && p2 ){ int res; res = vdbeSorterCompare(pTask, SRVAL(p1), p1->nVal, pVal2, p2->nVal); if( res<=0 ){ *pp = p1; pp = &p1->u.pNext; p1 = p1->u.pNext; pVal2 = 0; }else{ *pp = p2; pp = &p2->u.pNext; p2 = p2->u.pNext; if( p2==0 ) break; pVal2 = SRVAL(p2); } } *pp = p1 ? p1 : p2; *ppOut = pFinal; } /* ** Sort the linked list of records headed at pTask->pList. Return ** SQLITE_OK if successful, or an SQLite error code (i.e. SQLITE_NOMEM) if ** an error occurs. */ static int vdbeSorterSort(SortSubtask *pTask, SorterList *pList){ int i; SorterRecord **aSlot; SorterRecord *p; int rc; rc = vdbeSortAllocUnpacked(pTask); if( rc!=SQLITE_OK ) return rc; aSlot = (SorterRecord **)sqlite3MallocZero(64 * sizeof(SorterRecord *)); if( !aSlot ){ return SQLITE_NOMEM; } p = pList->pList; while( p ){ SorterRecord *pNext; if( pList->aMemory ){ if( (u8*)p==pList->aMemory ){ pNext = 0; }else{ assert( p->u.iNextaMemory) ); pNext = (SorterRecord*)&pList->aMemory[p->u.iNext]; } }else{ pNext = p->u.pNext; } p->u.pNext = 0; for(i=0; aSlot[i]; i++){ vdbeSorterMerge(pTask, p, aSlot[i], &p); aSlot[i] = 0; } aSlot[i] = p; p = pNext; } p = 0; for(i=0; i<64; i++){ vdbeSorterMerge(pTask, p, aSlot[i], &p); } pList->pList = p; sqlite3_free(aSlot); if( pTask->pUnpacked->errCode ){ assert( pTask->pUnpacked->errCode==SQLITE_NOMEM ); return SQLITE_NOMEM; } return SQLITE_OK; } /* ** Initialize a PMA-writer object. */ static void vdbePmaWriterInit( sqlite3_file *pFile, /* File to write to */ PmaWriter *p, /* Object to populate */ int nBuf, /* Buffer size */ i64 iStart /* Offset of pFile to begin writing at */ ){ memset(p, 0, sizeof(PmaWriter)); p->aBuffer = (u8*)sqlite3Malloc(nBuf); if( !p->aBuffer ){ p->eFWErr = SQLITE_NOMEM; }else{ p->iBufEnd = p->iBufStart = (iStart % nBuf); p->iWriteOff = iStart - p->iBufStart; p->nBuffer = nBuf; p->pFile = pFile; } } /* ** Write nData bytes of data to the PMA. Return SQLITE_OK ** if successful, or an SQLite error code if an error occurs. */ static void vdbePmaWriteBlob(PmaWriter *p, u8 *pData, int nData){ int nRem = nData; while( nRem>0 && p->eFWErr==0 ){ int nCopy = nRem; if( nCopy>(p->nBuffer - p->iBufEnd) ){ nCopy = p->nBuffer - p->iBufEnd; } memcpy(&p->aBuffer[p->iBufEnd], &pData[nData-nRem], nCopy); p->iBufEnd += nCopy; if( p->iBufEnd==p->nBuffer ){ p->eFWErr = sqlite3OsWrite(p->pFile, &p->aBuffer[p->iBufStart], p->iBufEnd - p->iBufStart, p->iWriteOff + p->iBufStart ); p->iBufStart = p->iBufEnd = 0; p->iWriteOff += p->nBuffer; } assert( p->iBufEndnBuffer ); nRem -= nCopy; } } /* ** Flush any buffered data to disk and clean up the PMA-writer object. ** The results of using the PMA-writer after this call are undefined. ** Return SQLITE_OK if flushing the buffered data succeeds or is not ** required. Otherwise, return an SQLite error code. ** ** Before returning, set *piEof to the offset immediately following the ** last byte written to the file. */ static int vdbePmaWriterFinish(PmaWriter *p, i64 *piEof){ int rc; if( p->eFWErr==0 && ALWAYS(p->aBuffer) && p->iBufEnd>p->iBufStart ){ p->eFWErr = sqlite3OsWrite(p->pFile, &p->aBuffer[p->iBufStart], p->iBufEnd - p->iBufStart, p->iWriteOff + p->iBufStart ); } *piEof = (p->iWriteOff + p->iBufEnd); sqlite3_free(p->aBuffer); rc = p->eFWErr; memset(p, 0, sizeof(PmaWriter)); return rc; } /* ** Write value iVal encoded as a varint to the PMA. Return ** SQLITE_OK if successful, or an SQLite error code if an error occurs. */ static void vdbePmaWriteVarint(PmaWriter *p, u64 iVal){ int nByte; u8 aByte[10]; nByte = sqlite3PutVarint(aByte, iVal); vdbePmaWriteBlob(p, aByte, nByte); } #if SQLITE_MAX_MMAP_SIZE>0 /* ** The first argument is a file-handle open on a temporary file. The file ** is guaranteed to be nByte bytes or smaller in size. This function ** attempts to extend the file to nByte bytes in size and to ensure that ** the VFS has memory mapped it. ** ** Whether or not the file does end up memory mapped of course depends on ** the specific VFS implementation. */ static void vdbeSorterExtendFile(sqlite3 *db, sqlite3_file *pFile, i64 nByte){ if( nByte<=(i64)(db->nMaxSorterMmap) ){ int rc = sqlite3OsTruncate(pFile, nByte); if( rc==SQLITE_OK ){ void *p = 0; sqlite3OsFetch(pFile, 0, nByte, &p); sqlite3OsUnfetch(pFile, 0, p); } } } #else # define vdbeSorterExtendFile(x,y,z) SQLITE_OK #endif /* ** Write the current contents of in-memory linked-list pList to a level-0 ** PMA in the temp file belonging to sub-task pTask. Return SQLITE_OK if ** successful, or an SQLite error code otherwise. ** ** The format of a PMA is: ** ** * A varint. This varint contains the total number of bytes of content ** in the PMA (not including the varint itself). ** ** * One or more records packed end-to-end in order of ascending keys. ** Each record consists of a varint followed by a blob of data (the ** key). The varint is the number of bytes in the blob of data. */ static int vdbeSorterListToPMA(SortSubtask *pTask, SorterList *pList){ int rc = SQLITE_OK; /* Return code */ PmaWriter writer; /* Object used to write to the file */ #ifdef SQLITE_DEBUG /* Set iSz to the expected size of file pTask->file after writing the PMA. ** This is used by an assert() statement at the end of this function. */ i64 iSz = pList->szPMA + sqlite3VarintLen(pList->szPMA) + pTask->file.iEof; #endif vdbeSorterWorkDebug(pTask, "enter"); memset(&writer, 0, sizeof(PmaWriter)); assert( pList->szPMA>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 + pList->szPMA + 9 ); } /* Sort the list */ if( rc==SQLITE_OK ){ rc = vdbeSorterSort(pTask, pList); } if( rc==SQLITE_OK ){ SorterRecord *p; SorterRecord *pNext = 0; vdbePmaWriterInit(pTask->file.pFd, &writer, pTask->pgsz, pTask->file.iEof); pTask->nPMA++; vdbePmaWriteVarint(&writer, pList->szPMA); for(p=pList->pList; p; p=pNext){ pNext = p->u.pNext; vdbePmaWriteVarint(&writer, p->nVal); vdbePmaWriteBlob(&writer, SRVAL(p), p->nVal); if( pList->aMemory==0 ) sqlite3_free(p); } pList->pList = p; rc = vdbePmaWriterFinish(&writer, &pTask->file.iEof); } vdbeSorterWorkDebug(pTask, "exit"); assert( rc!=SQLITE_OK || pList->pList==0 ); assert( rc!=SQLITE_OK || pTask->file.iEof==iSz ); return rc; } /* ** Advance the MergeEngine iterator passed as the second argument to ** the next entry. Set *pbEof to true if this means the iterator has ** reached EOF. ** ** Return SQLITE_OK if successful or an error code if an error occurs. */ static int vdbeSorterNext( SortSubtask *pTask, MergeEngine *pMerger, int *pbEof ){ int rc; int iPrev = pMerger->aTree[1];/* Index of iterator to advance */ /* Advance the current iterator */ rc = vdbePmaReaderNext(&pMerger->aIter[iPrev]); /* Update contents of aTree[] */ if( rc==SQLITE_OK ){ int i; /* Index of aTree[] to recalculate */ PmaReader *pIter1; /* First iterator to compare */ PmaReader *pIter2; /* Second iterator to compare */ u8 *pKey2; /* To pIter2->aKey, or 0 if record cached */ /* Find the first two iterators to compare. The one that was just ** advanced (iPrev) and the one next to it in the array. */ pIter1 = &pMerger->aIter[(iPrev & 0xFFFE)]; pIter2 = &pMerger->aIter[(iPrev | 0x0001)]; pKey2 = pIter2->aKey; for(i=(pMerger->nTree+iPrev)/2; i>0; i=i/2){ /* Compare pIter1 and pIter2. Store the result in variable iRes. */ int iRes; if( pIter1->pFile==0 ){ iRes = +1; }else if( pIter2->pFile==0 ){ iRes = -1; }else{ iRes = vdbeSorterCompare(pTask, pIter1->aKey, pIter1->nKey, pKey2, pIter2->nKey ); } /* If pIter1 contained the smaller value, set aTree[i] to its index. ** Then set pIter2 to the next iterator to compare to pIter1. In this ** case there is no cache of pIter2 in pTask->pUnpacked, so set ** pKey2 to point to the record belonging to pIter2. ** ** Alternatively, if pIter2 contains the smaller of the two values, ** set aTree[i] to its index and update pIter1. If vdbeSorterCompare() ** was actually called above, then pTask->pUnpacked now contains ** a value equivalent to pIter2. So set pKey2 to NULL to prevent ** vdbeSorterCompare() from decoding pIter2 again. ** ** If the two values were equal, then the value from the oldest ** PMA should be considered smaller. The VdbeSorter.aIter[] array ** is sorted from oldest to newest, so pIter1 contains older values ** than pIter2 iff (pIter1aTree[i] = (int)(pIter1 - pMerger->aIter); pIter2 = &pMerger->aIter[ pMerger->aTree[i ^ 0x0001] ]; pKey2 = pIter2->aKey; }else{ if( pIter1->pFile ) pKey2 = 0; pMerger->aTree[i] = (int)(pIter2 - pMerger->aIter); pIter1 = &pMerger->aIter[ pMerger->aTree[i ^ 0x0001] ]; } } *pbEof = (pMerger->aIter[pMerger->aTree[1]].pFile==0); } return rc; } /* ** The main routine for sorter-thread operations. */ static void *vdbeSorterFlushThread(void *pCtx){ SortSubtask *pTask = (SortSubtask*)pCtx; int rc; /* Return code */ assert( pTask->thread.bDone==0 ); rc = vdbeSorterListToPMA(pTask, &pTask->list); pTask->thread.bDone = 1; return SQLITE_INT_TO_PTR(rc); } /* ** Flush the current contents of VdbeSorter.list to a new PMA, possibly ** using a background thread. */ static int vdbeSorterFlushPMA(VdbeSorter *pSorter){ #if SQLITE_MAX_WORKER_THREADS==0 pSorter->bUsePMA = 1; return vdbeSorterListToPMA(&pSorter->aTask[0], &pSorter->list); #else int rc = SQLITE_OK; int i; SortSubtask *pTask = 0; /* Thread context used to create new PMA */ int nWorker = (pSorter->nTask-1); /* Set the flag to indicate that at least one PMA has been written. ** Or will be, anyhow. */ pSorter->bUsePMA = 1; /* Select a sub-task to sort and flush the current list of in-memory ** records to disk. If the sorter is running in multi-threaded mode, ** round-robin between the first (pSorter->nTask-1) tasks. Except, if ** the background thread from a sub-tasks previous turn is still running, ** skip it. If the first (pSorter->nTask-1) sub-tasks are all still busy, ** fall back to using the final sub-task. The first (pSorter->nTask-1) ** sub-tasks are prefered as they use background threads - the final ** sub-task uses the main thread. */ for(i=0; iiPrev + i + 1) % nWorker; pTask = &pSorter->aTask[iTest]; if( pTask->thread.bDone ){ rc = vdbeSorterJoinThread(pTask, &pTask->thread); } if( pTask->thread.pThread==0 || rc!=SQLITE_OK ) break; } if( rc==SQLITE_OK ){ if( i==nWorker ){ /* Use the foreground thread for this operation */ rc = vdbeSorterListToPMA(&pSorter->aTask[nWorker], &pSorter->list); }else{ /* Launch a background thread for this operation */ u8 *aMem = pTask->list.aMemory; void *pCtx = (void*)pTask; assert( pTask->thread.pThread==0 && pTask->thread.bDone==0 ); assert( pTask->list.pList==0 ); assert( pTask->list.aMemory==0 || pSorter->list.aMemory!=0 ); pSorter->iPrev = (pTask - pSorter->aTask); pTask->list = pSorter->list; pSorter->list.pList = 0; pSorter->list.szPMA = 0; if( aMem ){ pSorter->list.aMemory = aMem; pSorter->nMemory = sqlite3MallocSize(aMem); }else{ pSorter->list.aMemory = sqlite3Malloc(pSorter->nMemory); if( !pSorter->list.aMemory ) return SQLITE_NOMEM; } rc = vdbeSorterCreateThread(&pTask->thread, vdbeSorterFlushThread, pCtx); } } return rc; #endif } /* ** Add a record to the sorter. */ int sqlite3VdbeSorterWrite( sqlite3 *db, /* Database handle */ const VdbeCursor *pCsr, /* Sorter cursor */ Mem *pVal /* Memory cell containing record */ ){ VdbeSorter *pSorter = pCsr->pSorter; int rc = SQLITE_OK; /* Return Code */ SorterRecord *pNew; /* New list element */ int bFlush; /* True to flush contents of memory to PMA */ int nReq; /* Bytes of memory required */ int nPMA; /* Bytes of PMA space required */ assert( pSorter ); /* Figure out whether or not the current contents of memory should be ** flushed to a PMA before continuing. If so, do so. ** ** If using the single large allocation mode (pSorter->aMemory!=0), then ** flush the contents of memory to a new PMA if (a) at least one value is ** already in memory and (b) the new value will not fit in memory. ** ** Or, if using separate allocations for each record, flush the contents ** of memory to a PMA if either of the following are true: ** ** * The total memory allocated for the in-memory list is greater ** than (page-size * cache-size), or ** ** * The total memory allocated for the in-memory list is greater ** than (page-size * 10) and sqlite3HeapNearlyFull() returns true. */ nReq = pVal->n + sizeof(SorterRecord); nPMA = pVal->n + sqlite3VarintLen(pVal->n); if( pSorter->mxPmaSize ){ if( pSorter->list.aMemory ){ bFlush = pSorter->iMemory && (pSorter->iMemory+nReq) > pSorter->mxPmaSize; }else{ bFlush = ( (pSorter->list.szPMA > pSorter->mxPmaSize) || (pSorter->list.szPMA > pSorter->mnPmaSize && sqlite3HeapNearlyFull()) ); } if( bFlush ){ rc = vdbeSorterFlushPMA(pSorter); pSorter->list.szPMA = 0; pSorter->iMemory = 0; assert( rc!=SQLITE_OK || pSorter->list.pList==0 ); } } pSorter->list.szPMA += nPMA; if( nPMA>pSorter->mxKeysize ){ pSorter->mxKeysize = nPMA; } if( pSorter->list.aMemory ){ int nMin = pSorter->iMemory + nReq; if( nMin>pSorter->nMemory ){ u8 *aNew; int nNew = pSorter->nMemory * 2; while( nNew < nMin ) nNew = nNew*2; if( nNew > pSorter->mxPmaSize ) nNew = pSorter->mxPmaSize; if( nNew < nMin ) nNew = nMin; aNew = sqlite3Realloc(pSorter->list.aMemory, nNew); if( !aNew ) return SQLITE_NOMEM; pSorter->list.pList = (SorterRecord*)( aNew + ((u8*)pSorter->list.pList - pSorter->list.aMemory) ); pSorter->list.aMemory = aNew; pSorter->nMemory = nNew; } pNew = (SorterRecord*)&pSorter->list.aMemory[pSorter->iMemory]; pSorter->iMemory += ROUND8(nReq); pNew->u.iNext = (u8*)(pSorter->list.pList) - pSorter->list.aMemory; }else{ pNew = (SorterRecord *)sqlite3Malloc(nReq); if( pNew==0 ){ return SQLITE_NOMEM; } pNew->u.pNext = pSorter->list.pList; } memcpy(SRVAL(pNew), pVal->z, pVal->n); pNew->nVal = pVal->n; pSorter->list.pList = pNew; return rc; } /* ** 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; i64 iStart = pIncr->iStartOff; SorterFile *pOut = &pIncr->aFile[1]; MergeEngine *pMerger = pIncr->pMerger; PmaWriter writer; assert( pIncr->bEof==0 ); vdbeSorterPopulateDebug(pIncr->pTask, "enter"); vdbePmaWriterInit(pOut->pFd, &writer, pIncr->pTask->pgsz, iStart); 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 + nKey + sqlite3VarintLen(nKey))>(iStart + 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; vdbeSorterPopulateDebug(pIncr->pTask, "exit"); return rc; } static void *vdbeIncrPopulateThread(void *pCtx){ IncrMerger *pIncr = (IncrMerger*)pCtx; void *pRet = SQLITE_INT_TO_PTR( vdbeIncrPopulate(pIncr) ); pIncr->thread.bDone = 1; return pRet; } #if SQLITE_MAX_WORKER_THREADS>0 static int vdbeIncrBgPopulate(IncrMerger *pIncr){ void *pCtx = (void*)pIncr; assert( pIncr->bUseThread ); return vdbeSorterCreateThread(&pIncr->thread, vdbeIncrPopulateThread, pCtx); } #endif static int vdbeIncrSwap(IncrMerger *pIncr){ int rc = SQLITE_OK; #if SQLITE_MAX_WORKER_THREADS>0 if( pIncr->bUseThread ){ rc = vdbeSorterJoinThread(pIncr->pTask, &pIncr->thread); if( rc==SQLITE_OK ){ SorterFile f0 = pIncr->aFile[0]; pIncr->aFile[0] = pIncr->aFile[1]; pIncr->aFile[1] = f0; } if( rc==SQLITE_OK ){ if( pIncr->aFile[0].iEof==pIncr->iStartOff ){ pIncr->bEof = 1; }else{ rc = vdbeIncrBgPopulate(pIncr); } } }else #endif { rc = vdbeIncrPopulate(pIncr); pIncr->aFile[0] = pIncr->aFile[1]; if( pIncr->aFile[0].iEof==pIncr->iStartOff ){ pIncr->bEof = 1; } } return rc; } static void vdbeIncrFree(IncrMerger *pIncr){ if( pIncr ){ #if SQLITE_MAX_WORKER_THREADS>0 vdbeSorterJoinThread(pIncr->pTask, &pIncr->thread); if( pIncr->bUseThread ){ if( pIncr->aFile[0].pFd ) sqlite3OsCloseFree(pIncr->aFile[0].pFd); if( pIncr->aFile[1].pFd ) sqlite3OsCloseFree(pIncr->aFile[1].pFd); } #endif vdbeMergeEngineFree(pIncr->pMerger); sqlite3_free(pIncr); } } static IncrMerger *vdbeIncrNew(SortSubtask *pTask, MergeEngine *pMerger){ IncrMerger *pIncr = sqlite3_malloc(sizeof(IncrMerger)); if( pIncr ){ memset(pIncr, 0, sizeof(IncrMerger)); pIncr->pMerger = pMerger; pIncr->pTask = pTask; pIncr->mxSz = MAX(pTask->pSorter->mxKeysize+9,pTask->pSorter->mxPmaSize/2); pTask->file2.iEof += pIncr->mxSz; } return pIncr; } static void vdbeIncrSetThreads(IncrMerger *pIncr, int bUseThread){ if( bUseThread ){ pIncr->bUseThread = 1; pIncr->pTask->file2.iEof -= pIncr->mxSz; } } #define INCRINIT2_NORMAL 0 #define INCRINIT2_TASK 1 #define INCRINIT2_ROOT 2 static int vdbeIncrInit2(PmaReader *pIter, int eMode){ int rc = SQLITE_OK; IncrMerger *pIncr = pIter->pIncr; if( pIncr ){ SortSubtask *pTask = pIncr->pTask; int i; MergeEngine *pMerger = pIncr->pMerger; for(i=0; rc==SQLITE_OK && inTree; i++){ IncrMerger *p; if( eMode==INCRINIT2_ROOT ){ rc = vdbePmaReaderNext(&pMerger->aIter[i]); }else{ rc = vdbeIncrInit2(&pMerger->aIter[i], INCRINIT2_NORMAL); } } /* Set up the required files for pIncr */ if( rc==SQLITE_OK ){ if( pIncr->bUseThread==0 ){ if( pTask->file2.pFd==0 ){ rc = vdbeSorterOpenTempFile(pTask->db->pVfs, &pTask->file2.pFd); assert( pTask->file2.iEof>0 ); if( rc==SQLITE_OK ){ vdbeSorterExtendFile(pTask->db,pTask->file2.pFd,pTask->file2.iEof); pTask->file2.iEof = 0; } } if( rc==SQLITE_OK ){ pIncr->aFile[1].pFd = pTask->file2.pFd; pIncr->iStartOff = pTask->file2.iEof; pTask->file2.iEof += pIncr->mxSz; } }else{ rc = vdbeSorterOpenTempFile(pTask->db->pVfs, &pIncr->aFile[0].pFd); if( rc==SQLITE_OK ){ rc = vdbeSorterOpenTempFile(pTask->db->pVfs, &pIncr->aFile[1].pFd); } } } for(i=pMerger->nTree-1; rc==SQLITE_OK && i>0; i--){ rc = vdbeSorterDoCompare(pIncr->pTask, pMerger, i); } if( rc==SQLITE_OK && pIncr->bUseThread ){ /* Use the current thread */ assert( eMode==INCRINIT2_ROOT || eMode==INCRINIT2_TASK ); rc = vdbeIncrPopulate(pIncr); } if( rc==SQLITE_OK && eMode!=INCRINIT2_TASK ){ rc = vdbePmaReaderNext(pIter); } } return rc; } #if SQLITE_MAX_WORKER_THREADS>0 static void *vdbeIncrInit2Thread(void *pCtx){ PmaReader *pReader = (PmaReader*)pCtx; void *pRet = SQLITE_INT_TO_PTR( vdbeIncrInit2(pReader, INCRINIT2_TASK) ); pReader->pIncr->thread.bDone = 1; return pRet; } static int vdbeIncrBgInit2(PmaReader *pIter){ void *pCtx = (void*)pIter; return vdbeSorterCreateThread( &pIter->pIncr->thread, vdbeIncrInit2Thread, pCtx ); } #endif /* ** Allocate a new MergeEngine object to merge the contents of nPMA level-0 ** PMAs from pTask->file. If no error occurs, set *ppOut to point to ** the new object and return SQLITE_OK. Or, if an error does occur, set *ppOut ** to NULL and return an SQLite error code. ** ** When this function is called, *piOffset is set to the offset of the ** first PMA to read from pTask->file. Assuming no error occurs, it is ** set to the offset immediately following the last byte of the last ** PMA before returning. If an error does occur, then the final value of ** *piOffset is undefined. */ static int vdbeMergeEngineLevel0( SortSubtask *pTask, /* Sorter task to read from */ int nPMA, /* Number of PMAs to read */ i64 *piOffset, /* IN/OUT: Read offset in pTask->file */ MergeEngine **ppOut /* OUT: New merge-engine */ ){ MergeEngine *pNew; /* Merge engine to return */ i64 iOff = *piOffset; int i; int rc = SQLITE_OK; *ppOut = pNew = vdbeMergeEngineNew(nPMA); if( pNew==0 ) rc = SQLITE_NOMEM; for(i=0; iaIter[i]; rc = vdbePmaReaderInit(pTask, &pTask->file, iOff, pIter, &nDummy); iOff = pIter->iEof; } if( rc!=SQLITE_OK ){ vdbeMergeEngineFree(pNew); *ppOut = 0; } *piOffset = iOff; return rc; } typedef struct IncrBuilder IncrBuilder; struct IncrBuilder { int nPMA; /* Number of iterators used so far */ MergeEngine *pMerger; /* Merge engine to populate. */ }; static int vdbeAddToBuilder( SortSubtask *pTask, IncrBuilder *pBuilder, MergeEngine *pMerger ){ int rc = SQLITE_OK; IncrMerger *pIncr; assert( pMerger ); if( pBuilder->nPMA==SORTER_MAX_MERGE_COUNT ){ rc = vdbeAddToBuilder(pTask, &pBuilder[1], pBuilder->pMerger); pBuilder->pMerger = 0; pBuilder->nPMA = 0; } if( rc==SQLITE_OK && pBuilder->pMerger==0 ){ pBuilder->pMerger = vdbeMergeEngineNew(SORTER_MAX_MERGE_COUNT); if( pBuilder->pMerger==0 ) rc = SQLITE_NOMEM; } if( rc==SQLITE_OK ){ pIncr = vdbeIncrNew(pTask, pMerger); if( pIncr==0 ) rc = SQLITE_NOMEM; pBuilder->pMerger->aIter[pBuilder->nPMA++].pIncr = pIncr; } if( rc!=SQLITE_OK ){ vdbeMergeEngineFree(pMerger); } return rc; } /* ** Populate iterator *pIter so that it may be used to iterate through all ** keys stored in all PMAs created by this sorter. */ static int vdbePmaReaderIncrInit(VdbeSorter *pSorter, PmaReader *pIter){ SortSubtask *pTask0 = &pSorter->aTask[0]; MergeEngine *pMain = 0; sqlite3 *db = pTask0->db; int rc = SQLITE_OK; int iTask; IncrBuilder *aMerge; const int nMerge = 32; aMerge = sqlite3DbMallocZero(db, sizeof(aMerge[0])*nMerge); if( aMerge==0 ) return SQLITE_NOMEM; if( pSorter->nTask>1 ){ pMain = vdbeMergeEngineNew(pSorter->nTask); if( pMain==0 ) rc = SQLITE_NOMEM; } for(iTask=0; iTasknTask && rc==SQLITE_OK; iTask++){ MergeEngine *pRoot = 0; int iPMA; i64 iReadOff = 0; SortSubtask *pTask = &pSorter->aTask[iTask]; if( pTask->nPMA==0 ) continue; for(iPMA=0; iPMAnPMA; iPMA += SORTER_MAX_MERGE_COUNT){ MergeEngine *pMerger = 0; int nReader = MIN(pTask->nPMA - iPMA, SORTER_MAX_MERGE_COUNT); rc = vdbeMergeEngineLevel0(pTask, nReader, &iReadOff, &pMerger); if( rc!=SQLITE_OK ) break; if( iPMA==0 ){ pRoot = pMerger; }else{ if( pRoot ){ rc = vdbeAddToBuilder(pTask, &aMerge[0], pRoot); pRoot = 0; if( rc!=SQLITE_OK ){ vdbeMergeEngineFree(pMerger); break; } } rc = vdbeAddToBuilder(pTask, &aMerge[0], pMerger); } } if( pRoot==0 ){ int i; for(i=0; rc==SQLITE_OK && iaIter[iTask].pIncr = pNew; if( pNew==0 ) rc = SQLITE_NOMEM; } memset(aMerge, 0, nMerge*sizeof(aMerge[0])); } } if( rc==SQLITE_OK ){ SortSubtask *pLast = &pSorter->aTask[pSorter->nTask-1]; rc = vdbeSortAllocUnpacked(pLast); if( rc==SQLITE_OK ){ pIter->pIncr = vdbeIncrNew(pLast, pMain); if( pIter->pIncr==0 ){ rc = SQLITE_NOMEM; } #if SQLITE_MAX_WORKER_THREADS>0 else{ vdbeIncrSetThreads(pIter->pIncr, pSorter->bUseThreads); for(iTask=0; iTask<(pSorter->nTask-1); iTask++){ IncrMerger *pIncr; if( (pIncr = pMain->aIter[iTask].pIncr) ){ vdbeIncrSetThreads(pIncr, pSorter->bUseThreads); assert( pIncr->pTask!=pLast ); } } if( pSorter->nTask>1 ){ for(iTask=0; rc==SQLITE_OK && iTasknTask; iTask++){ PmaReader *p = &pMain->aIter[iTask]; assert( p->pIncr==0 || p->pIncr->pTask==&pSorter->aTask[iTask] ); if( p->pIncr ){ rc = vdbeIncrBgInit2(p); } } } } #endif } } if( rc==SQLITE_OK ){ int eMode = (pSorter->nTask>1 ? INCRINIT2_ROOT : INCRINIT2_NORMAL); rc = vdbeIncrInit2(pIter, eMode); } sqlite3_free(aMerge); 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){ VdbeSorter *pSorter = pCsr->pSorter; int rc = SQLITE_OK; /* Return code */ assert( pSorter ); /* If no data has been written to disk, then do not do so now. Instead, ** sort the VdbeSorter.pRecord list. The vdbe layer will read data directly ** from the in-memory list. */ if( pSorter->bUsePMA==0 ){ if( pSorter->list.pList ){ *pbEof = 0; rc = vdbeSorterSort(&pSorter->aTask[0], &pSorter->list); }else{ *pbEof = 1; } return rc; } /* Write the current in-memory list to a PMA. */ if( pSorter->list.pList ){ rc = vdbeSorterFlushPMA(pSorter); } /* 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->list.pList; pSorter->list.pList = pFree->u.pNext; pFree->u.pNext = 0; if( pSorter->list.aMemory==0 ) vdbeSorterRecordFree(db, pFree); *pbEof = !pSorter->list.pList; 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->list.pList->nVal; pKey = SRVAL(pSorter->list.pList); } return pKey; } /* ** Copy the current sorter key into the memory cell pOut. */ int sqlite3VdbeSorterRowkey(const VdbeCursor *pCsr, Mem *pOut){ VdbeSorter *pSorter = pCsr->pSorter; void *pKey; int nKey; /* Sorter key to copy into pOut */ pKey = vdbeSorterRowkey(pSorter, &nKey); if( sqlite3VdbeMemGrow(pOut, nKey, 0) ){ return SQLITE_NOMEM; } pOut->n = nKey; MemSetTypeFlag(pOut, MEM_Blob); memcpy(pOut->z, pKey, nKey); return SQLITE_OK; } /* ** 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 */ ){ 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; inField; i++){ if( r2->aMem[i].flags & MEM_Null ){ *pRes = -1; return SQLITE_OK; } } *pRes = sqlite3VdbeRecordCompare(pVal->n, pVal->z, r2, 0); return SQLITE_OK; }