Chromium Code Reviews
chromiumcodereview-hr@appspot.gserviceaccount.com (chromiumcodereview-hr) | Please choose your nickname with Settings | Help | Chromium Project | Gerrit Changes | Sign out
(2)

Side by Side Diff: third_party/sqlite/sqlite-src-3170000/src/vdbesort.c

Issue 2747283002: [sql] Import reference version of SQLite 3.17.. (Closed)
Patch Set: Created 3 years, 9 months ago
Use n/p to move between diff chunks; N/P to move between comments. Draft comments are only viewable by you.
Jump to:
View unified diff | Download patch
OLDNEW
(Empty)
1 /*
2 ** 2011-07-09
3 **
4 ** The author disclaims copyright to this source code. In place of
5 ** a legal notice, here is a blessing:
6 **
7 ** May you do good and not evil.
8 ** May you find forgiveness for yourself and forgive others.
9 ** May you share freely, never taking more than you give.
10 **
11 *************************************************************************
12 ** This file contains code for the VdbeSorter object, used in concert with
13 ** a VdbeCursor to sort large numbers of keys for CREATE INDEX statements
14 ** or by SELECT statements with ORDER BY clauses that cannot be satisfied
15 ** using indexes and without LIMIT clauses.
16 **
17 ** The VdbeSorter object implements a multi-threaded external merge sort
18 ** algorithm that is efficient even if the number of elements being sorted
19 ** exceeds the available memory.
20 **
21 ** Here is the (internal, non-API) interface between this module and the
22 ** rest of the SQLite system:
23 **
24 ** sqlite3VdbeSorterInit() Create a new VdbeSorter object.
25 **
26 ** sqlite3VdbeSorterWrite() Add a single new row to the VdbeSorter
27 ** object. The row is a binary blob in the
28 ** OP_MakeRecord format that contains both
29 ** the ORDER BY key columns and result columns
30 ** in the case of a SELECT w/ ORDER BY, or
31 ** the complete record for an index entry
32 ** in the case of a CREATE INDEX.
33 **
34 ** sqlite3VdbeSorterRewind() Sort all content previously added.
35 ** Position the read cursor on the
36 ** first sorted element.
37 **
38 ** sqlite3VdbeSorterNext() Advance the read cursor to the next sorted
39 ** element.
40 **
41 ** sqlite3VdbeSorterRowkey() Return the complete binary blob for the
42 ** row currently under the read cursor.
43 **
44 ** sqlite3VdbeSorterCompare() Compare the binary blob for the row
45 ** currently under the read cursor against
46 ** another binary blob X and report if
47 ** X is strictly less than the read cursor.
48 ** Used to enforce uniqueness in a
49 ** CREATE UNIQUE INDEX statement.
50 **
51 ** sqlite3VdbeSorterClose() Close the VdbeSorter object and reclaim
52 ** all resources.
53 **
54 ** sqlite3VdbeSorterReset() Refurbish the VdbeSorter for reuse. This
55 ** is like Close() followed by Init() only
56 ** much faster.
57 **
58 ** The interfaces above must be called in a particular order. Write() can
59 ** only occur in between Init()/Reset() and Rewind(). Next(), Rowkey(), and
60 ** Compare() can only occur in between Rewind() and Close()/Reset(). i.e.
61 **
62 ** Init()
63 ** for each record: Write()
64 ** Rewind()
65 ** Rowkey()/Compare()
66 ** Next()
67 ** Close()
68 **
69 ** Algorithm:
70 **
71 ** Records passed to the sorter via calls to Write() are initially held
72 ** unsorted in main memory. Assuming the amount of memory used never exceeds
73 ** a threshold, when Rewind() is called the set of records is sorted using
74 ** an in-memory merge sort. In this case, no temporary files are required
75 ** and subsequent calls to Rowkey(), Next() and Compare() read records
76 ** directly from main memory.
77 **
78 ** If the amount of space used to store records in main memory exceeds the
79 ** threshold, then the set of records currently in memory are sorted and
80 ** written to a temporary file in "Packed Memory Array" (PMA) format.
81 ** A PMA created at this point is known as a "level-0 PMA". Higher levels
82 ** of PMAs may be created by merging existing PMAs together - for example
83 ** merging two or more level-0 PMAs together creates a level-1 PMA.
84 **
85 ** The threshold for the amount of main memory to use before flushing
86 ** records to a PMA is roughly the same as the limit configured for the
87 ** page-cache of the main database. Specifically, the threshold is set to
88 ** the value returned by "PRAGMA main.page_size" multipled by
89 ** that returned by "PRAGMA main.cache_size", in bytes.
90 **
91 ** If the sorter is running in single-threaded mode, then all PMAs generated
92 ** are appended to a single temporary file. Or, if the sorter is running in
93 ** multi-threaded mode then up to (N+1) temporary files may be opened, where
94 ** N is the configured number of worker threads. In this case, instead of
95 ** sorting the records and writing the PMA to a temporary file itself, the
96 ** calling thread usually launches a worker thread to do so. Except, if
97 ** there are already N worker threads running, the main thread does the work
98 ** itself.
99 **
100 ** The sorter is running in multi-threaded mode if (a) the library was built
101 ** with pre-processor symbol SQLITE_MAX_WORKER_THREADS set to a value greater
102 ** than zero, and (b) worker threads have been enabled at runtime by calling
103 ** "PRAGMA threads=N" with some value of N greater than 0.
104 **
105 ** When Rewind() is called, any data remaining in memory is flushed to a
106 ** final PMA. So at this point the data is stored in some number of sorted
107 ** PMAs within temporary files on disk.
108 **
109 ** If there are fewer than SORTER_MAX_MERGE_COUNT PMAs in total and the
110 ** sorter is running in single-threaded mode, then these PMAs are merged
111 ** incrementally as keys are retreived from the sorter by the VDBE. The
112 ** MergeEngine object, described in further detail below, performs this
113 ** merge.
114 **
115 ** Or, if running in multi-threaded mode, then a background thread is
116 ** launched to merge the existing PMAs. Once the background thread has
117 ** merged T bytes of data into a single sorted PMA, the main thread
118 ** begins reading keys from that PMA while the background thread proceeds
119 ** with merging the next T bytes of data. And so on.
120 **
121 ** Parameter T is set to half the value of the memory threshold used
122 ** by Write() above to determine when to create a new PMA.
123 **
124 ** If there are more than SORTER_MAX_MERGE_COUNT PMAs in total when
125 ** Rewind() is called, then a hierarchy of incremental-merges is used.
126 ** First, T bytes of data from the first SORTER_MAX_MERGE_COUNT PMAs on
127 ** disk are merged together. Then T bytes of data from the second set, and
128 ** so on, such that no operation ever merges more than SORTER_MAX_MERGE_COUNT
129 ** PMAs at a time. This done is to improve locality.
130 **
131 ** If running in multi-threaded mode and there are more than
132 ** SORTER_MAX_MERGE_COUNT PMAs on disk when Rewind() is called, then more
133 ** than one background thread may be created. Specifically, there may be
134 ** one background thread for each temporary file on disk, and one background
135 ** thread to merge the output of each of the others to a single PMA for
136 ** the main thread to read from.
137 */
138 #include "sqliteInt.h"
139 #include "vdbeInt.h"
140
141 /*
142 ** If SQLITE_DEBUG_SORTER_THREADS is defined, this module outputs various
143 ** messages to stderr that may be helpful in understanding the performance
144 ** characteristics of the sorter in multi-threaded mode.
145 */
146 #if 0
147 # define SQLITE_DEBUG_SORTER_THREADS 1
148 #endif
149
150 /*
151 ** Hard-coded maximum amount of data to accumulate in memory before flushing
152 ** to a level 0 PMA. The purpose of this limit is to prevent various integer
153 ** overflows. 512MiB.
154 */
155 #define SQLITE_MAX_PMASZ (1<<29)
156
157 /*
158 ** Private objects used by the sorter
159 */
160 typedef struct MergeEngine MergeEngine; /* Merge PMAs together */
161 typedef struct PmaReader PmaReader; /* Incrementally read one PMA */
162 typedef struct PmaWriter PmaWriter; /* Incrementally write one PMA */
163 typedef struct SorterRecord SorterRecord; /* A record being sorted */
164 typedef struct SortSubtask SortSubtask; /* A sub-task in the sort process */
165 typedef struct SorterFile SorterFile; /* Temporary file object wrapper */
166 typedef struct SorterList SorterList; /* In-memory list of records */
167 typedef struct IncrMerger IncrMerger; /* Read & merge multiple PMAs */
168
169 /*
170 ** A container for a temp file handle and the current amount of data
171 ** stored in the file.
172 */
173 struct SorterFile {
174 sqlite3_file *pFd; /* File handle */
175 i64 iEof; /* Bytes of data stored in pFd */
176 };
177
178 /*
179 ** An in-memory list of objects to be sorted.
180 **
181 ** If aMemory==0 then each object is allocated separately and the objects
182 ** are connected using SorterRecord.u.pNext. If aMemory!=0 then all objects
183 ** are stored in the aMemory[] bulk memory, one right after the other, and
184 ** are connected using SorterRecord.u.iNext.
185 */
186 struct SorterList {
187 SorterRecord *pList; /* Linked list of records */
188 u8 *aMemory; /* If non-NULL, bulk memory to hold pList */
189 int szPMA; /* Size of pList as PMA in bytes */
190 };
191
192 /*
193 ** The MergeEngine object is used to combine two or more smaller PMAs into
194 ** one big PMA using a merge operation. Separate PMAs all need to be
195 ** combined into one big PMA in order to be able to step through the sorted
196 ** records in order.
197 **
198 ** The aReadr[] array contains a PmaReader object for each of the PMAs being
199 ** merged. An aReadr[] object either points to a valid key or else is at EOF.
200 ** ("EOF" means "End Of File". When aReadr[] is at EOF there is no more data.)
201 ** For the purposes of the paragraphs below, we assume that the array is
202 ** actually N elements in size, where N is the smallest power of 2 greater
203 ** to or equal to the number of PMAs being merged. The extra aReadr[] elements
204 ** are treated as if they are empty (always at EOF).
205 **
206 ** The aTree[] array is also N elements in size. The value of N is stored in
207 ** the MergeEngine.nTree variable.
208 **
209 ** The final (N/2) elements of aTree[] contain the results of comparing
210 ** pairs of PMA keys together. Element i contains the result of
211 ** comparing aReadr[2*i-N] and aReadr[2*i-N+1]. Whichever key is smaller, the
212 ** aTree element is set to the index of it.
213 **
214 ** For the purposes of this comparison, EOF is considered greater than any
215 ** other key value. If the keys are equal (only possible with two EOF
216 ** values), it doesn't matter which index is stored.
217 **
218 ** The (N/4) elements of aTree[] that precede the final (N/2) described
219 ** above contains the index of the smallest of each block of 4 PmaReaders
220 ** And so on. So that aTree[1] contains the index of the PmaReader that
221 ** currently points to the smallest key value. aTree[0] is unused.
222 **
223 ** Example:
224 **
225 ** aReadr[0] -> Banana
226 ** aReadr[1] -> Feijoa
227 ** aReadr[2] -> Elderberry
228 ** aReadr[3] -> Currant
229 ** aReadr[4] -> Grapefruit
230 ** aReadr[5] -> Apple
231 ** aReadr[6] -> Durian
232 ** aReadr[7] -> EOF
233 **
234 ** aTree[] = { X, 5 0, 5 0, 3, 5, 6 }
235 **
236 ** The current element is "Apple" (the value of the key indicated by
237 ** PmaReader 5). When the Next() operation is invoked, PmaReader 5 will
238 ** be advanced to the next key in its segment. Say the next key is
239 ** "Eggplant":
240 **
241 ** aReadr[5] -> Eggplant
242 **
243 ** The contents of aTree[] are updated first by comparing the new PmaReader
244 ** 5 key to the current key of PmaReader 4 (still "Grapefruit"). The PmaReader
245 ** 5 value is still smaller, so aTree[6] is set to 5. And so on up the tree.
246 ** The value of PmaReader 6 - "Durian" - is now smaller than that of PmaReader
247 ** 5, so aTree[3] is set to 6. Key 0 is smaller than key 6 (Banana<Durian),
248 ** so the value written into element 1 of the array is 0. As follows:
249 **
250 ** aTree[] = { X, 0 0, 6 0, 3, 5, 6 }
251 **
252 ** In other words, each time we advance to the next sorter element, log2(N)
253 ** key comparison operations are required, where N is the number of segments
254 ** being merged (rounded up to the next power of 2).
255 */
256 struct MergeEngine {
257 int nTree; /* Used size of aTree/aReadr (power of 2) */
258 SortSubtask *pTask; /* Used by this thread only */
259 int *aTree; /* Current state of incremental merge */
260 PmaReader *aReadr; /* Array of PmaReaders to merge data from */
261 };
262
263 /*
264 ** This object represents a single thread of control in a sort operation.
265 ** Exactly VdbeSorter.nTask instances of this object are allocated
266 ** as part of each VdbeSorter object. Instances are never allocated any
267 ** other way. VdbeSorter.nTask is set to the number of worker threads allowed
268 ** (see SQLITE_CONFIG_WORKER_THREADS) plus one (the main thread). Thus for
269 ** single-threaded operation, there is exactly one instance of this object
270 ** and for multi-threaded operation there are two or more instances.
271 **
272 ** Essentially, this structure contains all those fields of the VdbeSorter
273 ** structure for which each thread requires a separate instance. For example,
274 ** each thread requries its own UnpackedRecord object to unpack records in
275 ** as part of comparison operations.
276 **
277 ** Before a background thread is launched, variable bDone is set to 0. Then,
278 ** right before it exits, the thread itself sets bDone to 1. This is used for
279 ** two purposes:
280 **
281 ** 1. When flushing the contents of memory to a level-0 PMA on disk, to
282 ** attempt to select a SortSubtask for which there is not already an
283 ** active background thread (since doing so causes the main thread
284 ** to block until it finishes).
285 **
286 ** 2. If SQLITE_DEBUG_SORTER_THREADS is defined, to determine if a call
287 ** to sqlite3ThreadJoin() is likely to block. Cases that are likely to
288 ** block provoke debugging output.
289 **
290 ** In both cases, the effects of the main thread seeing (bDone==0) even
291 ** after the thread has finished are not dire. So we don't worry about
292 ** memory barriers and such here.
293 */
294 typedef int (*SorterCompare)(SortSubtask*,int*,const void*,int,const void*,int);
295 struct SortSubtask {
296 SQLiteThread *pThread; /* Background thread, if any */
297 int bDone; /* Set if thread is finished but not joined */
298 VdbeSorter *pSorter; /* Sorter that owns this sub-task */
299 UnpackedRecord *pUnpacked; /* Space to unpack a record */
300 SorterList list; /* List for thread to write to a PMA */
301 int nPMA; /* Number of PMAs currently in file */
302 SorterCompare xCompare; /* Compare function to use */
303 SorterFile file; /* Temp file for level-0 PMAs */
304 SorterFile file2; /* Space for other PMAs */
305 };
306
307
308 /*
309 ** Main sorter structure. A single instance of this is allocated for each
310 ** sorter cursor created by the VDBE.
311 **
312 ** mxKeysize:
313 ** As records are added to the sorter by calls to sqlite3VdbeSorterWrite(),
314 ** this variable is updated so as to be set to the size on disk of the
315 ** largest record in the sorter.
316 */
317 struct VdbeSorter {
318 int mnPmaSize; /* Minimum PMA size, in bytes */
319 int mxPmaSize; /* Maximum PMA size, in bytes. 0==no limit */
320 int mxKeysize; /* Largest serialized key seen so far */
321 int pgsz; /* Main database page size */
322 PmaReader *pReader; /* Readr data from here after Rewind() */
323 MergeEngine *pMerger; /* Or here, if bUseThreads==0 */
324 sqlite3 *db; /* Database connection */
325 KeyInfo *pKeyInfo; /* How to compare records */
326 UnpackedRecord *pUnpacked; /* Used by VdbeSorterCompare() */
327 SorterList list; /* List of in-memory records */
328 int iMemory; /* Offset of free space in list.aMemory */
329 int nMemory; /* Size of list.aMemory allocation in bytes */
330 u8 bUsePMA; /* True if one or more PMAs created */
331 u8 bUseThreads; /* True to use background threads */
332 u8 iPrev; /* Previous thread used to flush PMA */
333 u8 nTask; /* Size of aTask[] array */
334 u8 typeMask;
335 SortSubtask aTask[1]; /* One or more subtasks */
336 };
337
338 #define SORTER_TYPE_INTEGER 0x01
339 #define SORTER_TYPE_TEXT 0x02
340
341 /*
342 ** An instance of the following object is used to read records out of a
343 ** PMA, in sorted order. The next key to be read is cached in nKey/aKey.
344 ** aKey might point into aMap or into aBuffer. If neither of those locations
345 ** contain a contiguous representation of the key, then aAlloc is allocated
346 ** and the key is copied into aAlloc and aKey is made to poitn to aAlloc.
347 **
348 ** pFd==0 at EOF.
349 */
350 struct PmaReader {
351 i64 iReadOff; /* Current read offset */
352 i64 iEof; /* 1 byte past EOF for this PmaReader */
353 int nAlloc; /* Bytes of space at aAlloc */
354 int nKey; /* Number of bytes in key */
355 sqlite3_file *pFd; /* File handle we are reading from */
356 u8 *aAlloc; /* Space for aKey if aBuffer and pMap wont work */
357 u8 *aKey; /* Pointer to current key */
358 u8 *aBuffer; /* Current read buffer */
359 int nBuffer; /* Size of read buffer in bytes */
360 u8 *aMap; /* Pointer to mapping of entire file */
361 IncrMerger *pIncr; /* Incremental merger */
362 };
363
364 /*
365 ** Normally, a PmaReader object iterates through an existing PMA stored
366 ** within a temp file. However, if the PmaReader.pIncr variable points to
367 ** an object of the following type, it may be used to iterate/merge through
368 ** multiple PMAs simultaneously.
369 **
370 ** There are two types of IncrMerger object - single (bUseThread==0) and
371 ** multi-threaded (bUseThread==1).
372 **
373 ** A multi-threaded IncrMerger object uses two temporary files - aFile[0]
374 ** and aFile[1]. Neither file is allowed to grow to more than mxSz bytes in
375 ** size. When the IncrMerger is initialized, it reads enough data from
376 ** pMerger to populate aFile[0]. It then sets variables within the
377 ** corresponding PmaReader object to read from that file and kicks off
378 ** a background thread to populate aFile[1] with the next mxSz bytes of
379 ** sorted record data from pMerger.
380 **
381 ** When the PmaReader reaches the end of aFile[0], it blocks until the
382 ** background thread has finished populating aFile[1]. It then exchanges
383 ** the contents of the aFile[0] and aFile[1] variables within this structure,
384 ** sets the PmaReader fields to read from the new aFile[0] and kicks off
385 ** another background thread to populate the new aFile[1]. And so on, until
386 ** the contents of pMerger are exhausted.
387 **
388 ** A single-threaded IncrMerger does not open any temporary files of its
389 ** own. Instead, it has exclusive access to mxSz bytes of space beginning
390 ** at offset iStartOff of file pTask->file2. And instead of using a
391 ** background thread to prepare data for the PmaReader, with a single
392 ** threaded IncrMerger the allocate part of pTask->file2 is "refilled" with
393 ** keys from pMerger by the calling thread whenever the PmaReader runs out
394 ** of data.
395 */
396 struct IncrMerger {
397 SortSubtask *pTask; /* Task that owns this merger */
398 MergeEngine *pMerger; /* Merge engine thread reads data from */
399 i64 iStartOff; /* Offset to start writing file at */
400 int mxSz; /* Maximum bytes of data to store */
401 int bEof; /* Set to true when merge is finished */
402 int bUseThread; /* True to use a bg thread for this object */
403 SorterFile aFile[2]; /* aFile[0] for reading, [1] for writing */
404 };
405
406 /*
407 ** An instance of this object is used for writing a PMA.
408 **
409 ** The PMA is written one record at a time. Each record is of an arbitrary
410 ** size. But I/O is more efficient if it occurs in page-sized blocks where
411 ** each block is aligned on a page boundary. This object caches writes to
412 ** the PMA so that aligned, page-size blocks are written.
413 */
414 struct PmaWriter {
415 int eFWErr; /* Non-zero if in an error state */
416 u8 *aBuffer; /* Pointer to write buffer */
417 int nBuffer; /* Size of write buffer in bytes */
418 int iBufStart; /* First byte of buffer to write */
419 int iBufEnd; /* Last byte of buffer to write */
420 i64 iWriteOff; /* Offset of start of buffer in file */
421 sqlite3_file *pFd; /* File handle to write to */
422 };
423
424 /*
425 ** This object is the header on a single record while that record is being
426 ** held in memory and prior to being written out as part of a PMA.
427 **
428 ** How the linked list is connected depends on how memory is being managed
429 ** by this module. If using a separate allocation for each in-memory record
430 ** (VdbeSorter.list.aMemory==0), then the list is always connected using the
431 ** SorterRecord.u.pNext pointers.
432 **
433 ** Or, if using the single large allocation method (VdbeSorter.list.aMemory!=0),
434 ** then while records are being accumulated the list is linked using the
435 ** SorterRecord.u.iNext offset. This is because the aMemory[] array may
436 ** be sqlite3Realloc()ed while records are being accumulated. Once the VM
437 ** has finished passing records to the sorter, or when the in-memory buffer
438 ** is full, the list is sorted. As part of the sorting process, it is
439 ** converted to use the SorterRecord.u.pNext pointers. See function
440 ** vdbeSorterSort() for details.
441 */
442 struct SorterRecord {
443 int nVal; /* Size of the record in bytes */
444 union {
445 SorterRecord *pNext; /* Pointer to next record in list */
446 int iNext; /* Offset within aMemory of next record */
447 } u;
448 /* The data for the record immediately follows this header */
449 };
450
451 /* Return a pointer to the buffer containing the record data for SorterRecord
452 ** object p. Should be used as if:
453 **
454 ** void *SRVAL(SorterRecord *p) { return (void*)&p[1]; }
455 */
456 #define SRVAL(p) ((void*)((SorterRecord*)(p) + 1))
457
458
459 /* Maximum number of PMAs that a single MergeEngine can merge */
460 #define SORTER_MAX_MERGE_COUNT 16
461
462 static int vdbeIncrSwap(IncrMerger*);
463 static void vdbeIncrFree(IncrMerger *);
464
465 /*
466 ** Free all memory belonging to the PmaReader object passed as the
467 ** argument. All structure fields are set to zero before returning.
468 */
469 static void vdbePmaReaderClear(PmaReader *pReadr){
470 sqlite3_free(pReadr->aAlloc);
471 sqlite3_free(pReadr->aBuffer);
472 if( pReadr->aMap ) sqlite3OsUnfetch(pReadr->pFd, 0, pReadr->aMap);
473 vdbeIncrFree(pReadr->pIncr);
474 memset(pReadr, 0, sizeof(PmaReader));
475 }
476
477 /*
478 ** Read the next nByte bytes of data from the PMA p.
479 ** If successful, set *ppOut to point to a buffer containing the data
480 ** and return SQLITE_OK. Otherwise, if an error occurs, return an SQLite
481 ** error code.
482 **
483 ** The buffer returned in *ppOut is only valid until the
484 ** next call to this function.
485 */
486 static int vdbePmaReadBlob(
487 PmaReader *p, /* PmaReader from which to take the blob */
488 int nByte, /* Bytes of data to read */
489 u8 **ppOut /* OUT: Pointer to buffer containing data */
490 ){
491 int iBuf; /* Offset within buffer to read from */
492 int nAvail; /* Bytes of data available in buffer */
493
494 if( p->aMap ){
495 *ppOut = &p->aMap[p->iReadOff];
496 p->iReadOff += nByte;
497 return SQLITE_OK;
498 }
499
500 assert( p->aBuffer );
501
502 /* If there is no more data to be read from the buffer, read the next
503 ** p->nBuffer bytes of data from the file into it. Or, if there are less
504 ** than p->nBuffer bytes remaining in the PMA, read all remaining data. */
505 iBuf = p->iReadOff % p->nBuffer;
506 if( iBuf==0 ){
507 int nRead; /* Bytes to read from disk */
508 int rc; /* sqlite3OsRead() return code */
509
510 /* Determine how many bytes of data to read. */
511 if( (p->iEof - p->iReadOff) > (i64)p->nBuffer ){
512 nRead = p->nBuffer;
513 }else{
514 nRead = (int)(p->iEof - p->iReadOff);
515 }
516 assert( nRead>0 );
517
518 /* Readr data from the file. Return early if an error occurs. */
519 rc = sqlite3OsRead(p->pFd, p->aBuffer, nRead, p->iReadOff);
520 assert( rc!=SQLITE_IOERR_SHORT_READ );
521 if( rc!=SQLITE_OK ) return rc;
522 }
523 nAvail = p->nBuffer - iBuf;
524
525 if( nByte<=nAvail ){
526 /* The requested data is available in the in-memory buffer. In this
527 ** case there is no need to make a copy of the data, just return a
528 ** pointer into the buffer to the caller. */
529 *ppOut = &p->aBuffer[iBuf];
530 p->iReadOff += nByte;
531 }else{
532 /* The requested data is not all available in the in-memory buffer.
533 ** In this case, allocate space at p->aAlloc[] to copy the requested
534 ** range into. Then return a copy of pointer p->aAlloc to the caller. */
535 int nRem; /* Bytes remaining to copy */
536
537 /* Extend the p->aAlloc[] allocation if required. */
538 if( p->nAlloc<nByte ){
539 u8 *aNew;
540 int nNew = MAX(128, p->nAlloc*2);
541 while( nByte>nNew ) nNew = nNew*2;
542 aNew = sqlite3Realloc(p->aAlloc, nNew);
543 if( !aNew ) return SQLITE_NOMEM_BKPT;
544 p->nAlloc = nNew;
545 p->aAlloc = aNew;
546 }
547
548 /* Copy as much data as is available in the buffer into the start of
549 ** p->aAlloc[]. */
550 memcpy(p->aAlloc, &p->aBuffer[iBuf], nAvail);
551 p->iReadOff += nAvail;
552 nRem = nByte - nAvail;
553
554 /* The following loop copies up to p->nBuffer bytes per iteration into
555 ** the p->aAlloc[] buffer. */
556 while( nRem>0 ){
557 int rc; /* vdbePmaReadBlob() return code */
558 int nCopy; /* Number of bytes to copy */
559 u8 *aNext; /* Pointer to buffer to copy data from */
560
561 nCopy = nRem;
562 if( nRem>p->nBuffer ) nCopy = p->nBuffer;
563 rc = vdbePmaReadBlob(p, nCopy, &aNext);
564 if( rc!=SQLITE_OK ) return rc;
565 assert( aNext!=p->aAlloc );
566 memcpy(&p->aAlloc[nByte - nRem], aNext, nCopy);
567 nRem -= nCopy;
568 }
569
570 *ppOut = p->aAlloc;
571 }
572
573 return SQLITE_OK;
574 }
575
576 /*
577 ** Read a varint from the stream of data accessed by p. Set *pnOut to
578 ** the value read.
579 */
580 static int vdbePmaReadVarint(PmaReader *p, u64 *pnOut){
581 int iBuf;
582
583 if( p->aMap ){
584 p->iReadOff += sqlite3GetVarint(&p->aMap[p->iReadOff], pnOut);
585 }else{
586 iBuf = p->iReadOff % p->nBuffer;
587 if( iBuf && (p->nBuffer-iBuf)>=9 ){
588 p->iReadOff += sqlite3GetVarint(&p->aBuffer[iBuf], pnOut);
589 }else{
590 u8 aVarint[16], *a;
591 int i = 0, rc;
592 do{
593 rc = vdbePmaReadBlob(p, 1, &a);
594 if( rc ) return rc;
595 aVarint[(i++)&0xf] = a[0];
596 }while( (a[0]&0x80)!=0 );
597 sqlite3GetVarint(aVarint, pnOut);
598 }
599 }
600
601 return SQLITE_OK;
602 }
603
604 /*
605 ** Attempt to memory map file pFile. If successful, set *pp to point to the
606 ** new mapping and return SQLITE_OK. If the mapping is not attempted
607 ** (because the file is too large or the VFS layer is configured not to use
608 ** mmap), return SQLITE_OK and set *pp to NULL.
609 **
610 ** Or, if an error occurs, return an SQLite error code. The final value of
611 ** *pp is undefined in this case.
612 */
613 static int vdbeSorterMapFile(SortSubtask *pTask, SorterFile *pFile, u8 **pp){
614 int rc = SQLITE_OK;
615 if( pFile->iEof<=(i64)(pTask->pSorter->db->nMaxSorterMmap) ){
616 sqlite3_file *pFd = pFile->pFd;
617 if( pFd->pMethods->iVersion>=3 ){
618 rc = sqlite3OsFetch(pFd, 0, (int)pFile->iEof, (void**)pp);
619 testcase( rc!=SQLITE_OK );
620 }
621 }
622 return rc;
623 }
624
625 /*
626 ** Attach PmaReader pReadr to file pFile (if it is not already attached to
627 ** that file) and seek it to offset iOff within the file. Return SQLITE_OK
628 ** if successful, or an SQLite error code if an error occurs.
629 */
630 static int vdbePmaReaderSeek(
631 SortSubtask *pTask, /* Task context */
632 PmaReader *pReadr, /* Reader whose cursor is to be moved */
633 SorterFile *pFile, /* Sorter file to read from */
634 i64 iOff /* Offset in pFile */
635 ){
636 int rc = SQLITE_OK;
637
638 assert( pReadr->pIncr==0 || pReadr->pIncr->bEof==0 );
639
640 if( sqlite3FaultSim(201) ) return SQLITE_IOERR_READ;
641 if( pReadr->aMap ){
642 sqlite3OsUnfetch(pReadr->pFd, 0, pReadr->aMap);
643 pReadr->aMap = 0;
644 }
645 pReadr->iReadOff = iOff;
646 pReadr->iEof = pFile->iEof;
647 pReadr->pFd = pFile->pFd;
648
649 rc = vdbeSorterMapFile(pTask, pFile, &pReadr->aMap);
650 if( rc==SQLITE_OK && pReadr->aMap==0 ){
651 int pgsz = pTask->pSorter->pgsz;
652 int iBuf = pReadr->iReadOff % pgsz;
653 if( pReadr->aBuffer==0 ){
654 pReadr->aBuffer = (u8*)sqlite3Malloc(pgsz);
655 if( pReadr->aBuffer==0 ) rc = SQLITE_NOMEM_BKPT;
656 pReadr->nBuffer = pgsz;
657 }
658 if( rc==SQLITE_OK && iBuf ){
659 int nRead = pgsz - iBuf;
660 if( (pReadr->iReadOff + nRead) > pReadr->iEof ){
661 nRead = (int)(pReadr->iEof - pReadr->iReadOff);
662 }
663 rc = sqlite3OsRead(
664 pReadr->pFd, &pReadr->aBuffer[iBuf], nRead, pReadr->iReadOff
665 );
666 testcase( rc!=SQLITE_OK );
667 }
668 }
669
670 return rc;
671 }
672
673 /*
674 ** Advance PmaReader pReadr to the next key in its PMA. Return SQLITE_OK if
675 ** no error occurs, or an SQLite error code if one does.
676 */
677 static int vdbePmaReaderNext(PmaReader *pReadr){
678 int rc = SQLITE_OK; /* Return Code */
679 u64 nRec = 0; /* Size of record in bytes */
680
681
682 if( pReadr->iReadOff>=pReadr->iEof ){
683 IncrMerger *pIncr = pReadr->pIncr;
684 int bEof = 1;
685 if( pIncr ){
686 rc = vdbeIncrSwap(pIncr);
687 if( rc==SQLITE_OK && pIncr->bEof==0 ){
688 rc = vdbePmaReaderSeek(
689 pIncr->pTask, pReadr, &pIncr->aFile[0], pIncr->iStartOff
690 );
691 bEof = 0;
692 }
693 }
694
695 if( bEof ){
696 /* This is an EOF condition */
697 vdbePmaReaderClear(pReadr);
698 testcase( rc!=SQLITE_OK );
699 return rc;
700 }
701 }
702
703 if( rc==SQLITE_OK ){
704 rc = vdbePmaReadVarint(pReadr, &nRec);
705 }
706 if( rc==SQLITE_OK ){
707 pReadr->nKey = (int)nRec;
708 rc = vdbePmaReadBlob(pReadr, (int)nRec, &pReadr->aKey);
709 testcase( rc!=SQLITE_OK );
710 }
711
712 return rc;
713 }
714
715 /*
716 ** Initialize PmaReader pReadr to scan through the PMA stored in file pFile
717 ** starting at offset iStart and ending at offset iEof-1. This function
718 ** leaves the PmaReader pointing to the first key in the PMA (or EOF if the
719 ** PMA is empty).
720 **
721 ** If the pnByte parameter is NULL, then it is assumed that the file
722 ** contains a single PMA, and that that PMA omits the initial length varint.
723 */
724 static int vdbePmaReaderInit(
725 SortSubtask *pTask, /* Task context */
726 SorterFile *pFile, /* Sorter file to read from */
727 i64 iStart, /* Start offset in pFile */
728 PmaReader *pReadr, /* PmaReader to populate */
729 i64 *pnByte /* IN/OUT: Increment this value by PMA size */
730 ){
731 int rc;
732
733 assert( pFile->iEof>iStart );
734 assert( pReadr->aAlloc==0 && pReadr->nAlloc==0 );
735 assert( pReadr->aBuffer==0 );
736 assert( pReadr->aMap==0 );
737
738 rc = vdbePmaReaderSeek(pTask, pReadr, pFile, iStart);
739 if( rc==SQLITE_OK ){
740 u64 nByte = 0; /* Size of PMA in bytes */
741 rc = vdbePmaReadVarint(pReadr, &nByte);
742 pReadr->iEof = pReadr->iReadOff + nByte;
743 *pnByte += nByte;
744 }
745
746 if( rc==SQLITE_OK ){
747 rc = vdbePmaReaderNext(pReadr);
748 }
749 return rc;
750 }
751
752 /*
753 ** A version of vdbeSorterCompare() that assumes that it has already been
754 ** determined that the first field of key1 is equal to the first field of
755 ** key2.
756 */
757 static int vdbeSorterCompareTail(
758 SortSubtask *pTask, /* Subtask context (for pKeyInfo) */
759 int *pbKey2Cached, /* True if pTask->pUnpacked is pKey2 */
760 const void *pKey1, int nKey1, /* Left side of comparison */
761 const void *pKey2, int nKey2 /* Right side of comparison */
762 ){
763 UnpackedRecord *r2 = pTask->pUnpacked;
764 if( *pbKey2Cached==0 ){
765 sqlite3VdbeRecordUnpack(pTask->pSorter->pKeyInfo, nKey2, pKey2, r2);
766 *pbKey2Cached = 1;
767 }
768 return sqlite3VdbeRecordCompareWithSkip(nKey1, pKey1, r2, 1);
769 }
770
771 /*
772 ** Compare key1 (buffer pKey1, size nKey1 bytes) with key2 (buffer pKey2,
773 ** size nKey2 bytes). Use (pTask->pKeyInfo) for the collation sequences
774 ** used by the comparison. Return the result of the comparison.
775 **
776 ** If IN/OUT parameter *pbKey2Cached is true when this function is called,
777 ** it is assumed that (pTask->pUnpacked) contains the unpacked version
778 ** of key2. If it is false, (pTask->pUnpacked) is populated with the unpacked
779 ** version of key2 and *pbKey2Cached set to true before returning.
780 **
781 ** If an OOM error is encountered, (pTask->pUnpacked->error_rc) is set
782 ** to SQLITE_NOMEM.
783 */
784 static int vdbeSorterCompare(
785 SortSubtask *pTask, /* Subtask context (for pKeyInfo) */
786 int *pbKey2Cached, /* True if pTask->pUnpacked is pKey2 */
787 const void *pKey1, int nKey1, /* Left side of comparison */
788 const void *pKey2, int nKey2 /* Right side of comparison */
789 ){
790 UnpackedRecord *r2 = pTask->pUnpacked;
791 if( !*pbKey2Cached ){
792 sqlite3VdbeRecordUnpack(pTask->pSorter->pKeyInfo, nKey2, pKey2, r2);
793 *pbKey2Cached = 1;
794 }
795 return sqlite3VdbeRecordCompare(nKey1, pKey1, r2);
796 }
797
798 /*
799 ** A specially optimized version of vdbeSorterCompare() that assumes that
800 ** the first field of each key is a TEXT value and that the collation
801 ** sequence to compare them with is BINARY.
802 */
803 static int vdbeSorterCompareText(
804 SortSubtask *pTask, /* Subtask context (for pKeyInfo) */
805 int *pbKey2Cached, /* True if pTask->pUnpacked is pKey2 */
806 const void *pKey1, int nKey1, /* Left side of comparison */
807 const void *pKey2, int nKey2 /* Right side of comparison */
808 ){
809 const u8 * const p1 = (const u8 * const)pKey1;
810 const u8 * const p2 = (const u8 * const)pKey2;
811 const u8 * const v1 = &p1[ p1[0] ]; /* Pointer to value 1 */
812 const u8 * const v2 = &p2[ p2[0] ]; /* Pointer to value 2 */
813
814 int n1;
815 int n2;
816 int res;
817
818 getVarint32(&p1[1], n1); n1 = (n1 - 13) / 2;
819 getVarint32(&p2[1], n2); n2 = (n2 - 13) / 2;
820 res = memcmp(v1, v2, MIN(n1, n2));
821 if( res==0 ){
822 res = n1 - n2;
823 }
824
825 if( res==0 ){
826 if( pTask->pSorter->pKeyInfo->nField>1 ){
827 res = vdbeSorterCompareTail(
828 pTask, pbKey2Cached, pKey1, nKey1, pKey2, nKey2
829 );
830 }
831 }else{
832 if( pTask->pSorter->pKeyInfo->aSortOrder[0] ){
833 res = res * -1;
834 }
835 }
836
837 return res;
838 }
839
840 /*
841 ** A specially optimized version of vdbeSorterCompare() that assumes that
842 ** the first field of each key is an INTEGER value.
843 */
844 static int vdbeSorterCompareInt(
845 SortSubtask *pTask, /* Subtask context (for pKeyInfo) */
846 int *pbKey2Cached, /* True if pTask->pUnpacked is pKey2 */
847 const void *pKey1, int nKey1, /* Left side of comparison */
848 const void *pKey2, int nKey2 /* Right side of comparison */
849 ){
850 const u8 * const p1 = (const u8 * const)pKey1;
851 const u8 * const p2 = (const u8 * const)pKey2;
852 const int s1 = p1[1]; /* Left hand serial type */
853 const int s2 = p2[1]; /* Right hand serial type */
854 const u8 * const v1 = &p1[ p1[0] ]; /* Pointer to value 1 */
855 const u8 * const v2 = &p2[ p2[0] ]; /* Pointer to value 2 */
856 int res; /* Return value */
857
858 assert( (s1>0 && s1<7) || s1==8 || s1==9 );
859 assert( (s2>0 && s2<7) || s2==8 || s2==9 );
860
861 if( s1>7 && s2>7 ){
862 res = s1 - s2;
863 }else{
864 if( s1==s2 ){
865 if( (*v1 ^ *v2) & 0x80 ){
866 /* The two values have different signs */
867 res = (*v1 & 0x80) ? -1 : +1;
868 }else{
869 /* The two values have the same sign. Compare using memcmp(). */
870 static const u8 aLen[] = {0, 1, 2, 3, 4, 6, 8 };
871 int i;
872 res = 0;
873 for(i=0; i<aLen[s1]; i++){
874 if( (res = v1[i] - v2[i]) ) break;
875 }
876 }
877 }else{
878 if( s2>7 ){
879 res = +1;
880 }else if( s1>7 ){
881 res = -1;
882 }else{
883 res = s1 - s2;
884 }
885 assert( res!=0 );
886
887 if( res>0 ){
888 if( *v1 & 0x80 ) res = -1;
889 }else{
890 if( *v2 & 0x80 ) res = +1;
891 }
892 }
893 }
894
895 if( res==0 ){
896 if( pTask->pSorter->pKeyInfo->nField>1 ){
897 res = vdbeSorterCompareTail(
898 pTask, pbKey2Cached, pKey1, nKey1, pKey2, nKey2
899 );
900 }
901 }else if( pTask->pSorter->pKeyInfo->aSortOrder[0] ){
902 res = res * -1;
903 }
904
905 return res;
906 }
907
908 /*
909 ** Initialize the temporary index cursor just opened as a sorter cursor.
910 **
911 ** Usually, the sorter module uses the value of (pCsr->pKeyInfo->nField)
912 ** to determine the number of fields that should be compared from the
913 ** records being sorted. However, if the value passed as argument nField
914 ** is non-zero and the sorter is able to guarantee a stable sort, nField
915 ** is used instead. This is used when sorting records for a CREATE INDEX
916 ** statement. In this case, keys are always delivered to the sorter in
917 ** order of the primary key, which happens to be make up the final part
918 ** of the records being sorted. So if the sort is stable, there is never
919 ** any reason to compare PK fields and they can be ignored for a small
920 ** performance boost.
921 **
922 ** The sorter can guarantee a stable sort when running in single-threaded
923 ** mode, but not in multi-threaded mode.
924 **
925 ** SQLITE_OK is returned if successful, or an SQLite error code otherwise.
926 */
927 int sqlite3VdbeSorterInit(
928 sqlite3 *db, /* Database connection (for malloc()) */
929 int nField, /* Number of key fields in each record */
930 VdbeCursor *pCsr /* Cursor that holds the new sorter */
931 ){
932 int pgsz; /* Page size of main database */
933 int i; /* Used to iterate through aTask[] */
934 VdbeSorter *pSorter; /* The new sorter */
935 KeyInfo *pKeyInfo; /* Copy of pCsr->pKeyInfo with db==0 */
936 int szKeyInfo; /* Size of pCsr->pKeyInfo in bytes */
937 int sz; /* Size of pSorter in bytes */
938 int rc = SQLITE_OK;
939 #if SQLITE_MAX_WORKER_THREADS==0
940 # define nWorker 0
941 #else
942 int nWorker;
943 #endif
944
945 /* Initialize the upper limit on the number of worker threads */
946 #if SQLITE_MAX_WORKER_THREADS>0
947 if( sqlite3TempInMemory(db) || sqlite3GlobalConfig.bCoreMutex==0 ){
948 nWorker = 0;
949 }else{
950 nWorker = db->aLimit[SQLITE_LIMIT_WORKER_THREADS];
951 }
952 #endif
953
954 /* Do not allow the total number of threads (main thread + all workers)
955 ** to exceed the maximum merge count */
956 #if SQLITE_MAX_WORKER_THREADS>=SORTER_MAX_MERGE_COUNT
957 if( nWorker>=SORTER_MAX_MERGE_COUNT ){
958 nWorker = SORTER_MAX_MERGE_COUNT-1;
959 }
960 #endif
961
962 assert( pCsr->pKeyInfo && pCsr->pBtx==0 );
963 assert( pCsr->eCurType==CURTYPE_SORTER );
964 szKeyInfo = sizeof(KeyInfo) + (pCsr->pKeyInfo->nField-1)*sizeof(CollSeq*);
965 sz = sizeof(VdbeSorter) + nWorker * sizeof(SortSubtask);
966
967 pSorter = (VdbeSorter*)sqlite3DbMallocZero(db, sz + szKeyInfo);
968 pCsr->uc.pSorter = pSorter;
969 if( pSorter==0 ){
970 rc = SQLITE_NOMEM_BKPT;
971 }else{
972 pSorter->pKeyInfo = pKeyInfo = (KeyInfo*)((u8*)pSorter + sz);
973 memcpy(pKeyInfo, pCsr->pKeyInfo, szKeyInfo);
974 pKeyInfo->db = 0;
975 if( nField && nWorker==0 ){
976 pKeyInfo->nXField += (pKeyInfo->nField - nField);
977 pKeyInfo->nField = nField;
978 }
979 pSorter->pgsz = pgsz = sqlite3BtreeGetPageSize(db->aDb[0].pBt);
980 pSorter->nTask = nWorker + 1;
981 pSorter->iPrev = (u8)(nWorker - 1);
982 pSorter->bUseThreads = (pSorter->nTask>1);
983 pSorter->db = db;
984 for(i=0; i<pSorter->nTask; i++){
985 SortSubtask *pTask = &pSorter->aTask[i];
986 pTask->pSorter = pSorter;
987 }
988
989 if( !sqlite3TempInMemory(db) ){
990 i64 mxCache; /* Cache size in bytes*/
991 u32 szPma = sqlite3GlobalConfig.szPma;
992 pSorter->mnPmaSize = szPma * pgsz;
993
994 mxCache = db->aDb[0].pSchema->cache_size;
995 if( mxCache<0 ){
996 /* A negative cache-size value C indicates that the cache is abs(C)
997 ** KiB in size. */
998 mxCache = mxCache * -1024;
999 }else{
1000 mxCache = mxCache * pgsz;
1001 }
1002 mxCache = MIN(mxCache, SQLITE_MAX_PMASZ);
1003 pSorter->mxPmaSize = MAX(pSorter->mnPmaSize, (int)mxCache);
1004
1005 /* EVIDENCE-OF: R-26747-61719 When the application provides any amount of
1006 ** scratch memory using SQLITE_CONFIG_SCRATCH, SQLite avoids unnecessary
1007 ** large heap allocations.
1008 */
1009 if( sqlite3GlobalConfig.pScratch==0 ){
1010 assert( pSorter->iMemory==0 );
1011 pSorter->nMemory = pgsz;
1012 pSorter->list.aMemory = (u8*)sqlite3Malloc(pgsz);
1013 if( !pSorter->list.aMemory ) rc = SQLITE_NOMEM_BKPT;
1014 }
1015 }
1016
1017 if( (pKeyInfo->nField+pKeyInfo->nXField)<13
1018 && (pKeyInfo->aColl[0]==0 || pKeyInfo->aColl[0]==db->pDfltColl)
1019 ){
1020 pSorter->typeMask = SORTER_TYPE_INTEGER | SORTER_TYPE_TEXT;
1021 }
1022 }
1023
1024 return rc;
1025 }
1026 #undef nWorker /* Defined at the top of this function */
1027
1028 /*
1029 ** Free the list of sorted records starting at pRecord.
1030 */
1031 static void vdbeSorterRecordFree(sqlite3 *db, SorterRecord *pRecord){
1032 SorterRecord *p;
1033 SorterRecord *pNext;
1034 for(p=pRecord; p; p=pNext){
1035 pNext = p->u.pNext;
1036 sqlite3DbFree(db, p);
1037 }
1038 }
1039
1040 /*
1041 ** Free all resources owned by the object indicated by argument pTask. All
1042 ** fields of *pTask are zeroed before returning.
1043 */
1044 static void vdbeSortSubtaskCleanup(sqlite3 *db, SortSubtask *pTask){
1045 sqlite3DbFree(db, pTask->pUnpacked);
1046 #if SQLITE_MAX_WORKER_THREADS>0
1047 /* pTask->list.aMemory can only be non-zero if it was handed memory
1048 ** from the main thread. That only occurs SQLITE_MAX_WORKER_THREADS>0 */
1049 if( pTask->list.aMemory ){
1050 sqlite3_free(pTask->list.aMemory);
1051 }else
1052 #endif
1053 {
1054 assert( pTask->list.aMemory==0 );
1055 vdbeSorterRecordFree(0, pTask->list.pList);
1056 }
1057 if( pTask->file.pFd ){
1058 sqlite3OsCloseFree(pTask->file.pFd);
1059 }
1060 if( pTask->file2.pFd ){
1061 sqlite3OsCloseFree(pTask->file2.pFd);
1062 }
1063 memset(pTask, 0, sizeof(SortSubtask));
1064 }
1065
1066 #ifdef SQLITE_DEBUG_SORTER_THREADS
1067 static void vdbeSorterWorkDebug(SortSubtask *pTask, const char *zEvent){
1068 i64 t;
1069 int iTask = (pTask - pTask->pSorter->aTask);
1070 sqlite3OsCurrentTimeInt64(pTask->pSorter->db->pVfs, &t);
1071 fprintf(stderr, "%lld:%d %s\n", t, iTask, zEvent);
1072 }
1073 static void vdbeSorterRewindDebug(const char *zEvent){
1074 i64 t;
1075 sqlite3OsCurrentTimeInt64(sqlite3_vfs_find(0), &t);
1076 fprintf(stderr, "%lld:X %s\n", t, zEvent);
1077 }
1078 static void vdbeSorterPopulateDebug(
1079 SortSubtask *pTask,
1080 const char *zEvent
1081 ){
1082 i64 t;
1083 int iTask = (pTask - pTask->pSorter->aTask);
1084 sqlite3OsCurrentTimeInt64(pTask->pSorter->db->pVfs, &t);
1085 fprintf(stderr, "%lld:bg%d %s\n", t, iTask, zEvent);
1086 }
1087 static void vdbeSorterBlockDebug(
1088 SortSubtask *pTask,
1089 int bBlocked,
1090 const char *zEvent
1091 ){
1092 if( bBlocked ){
1093 i64 t;
1094 sqlite3OsCurrentTimeInt64(pTask->pSorter->db->pVfs, &t);
1095 fprintf(stderr, "%lld:main %s\n", t, zEvent);
1096 }
1097 }
1098 #else
1099 # define vdbeSorterWorkDebug(x,y)
1100 # define vdbeSorterRewindDebug(y)
1101 # define vdbeSorterPopulateDebug(x,y)
1102 # define vdbeSorterBlockDebug(x,y,z)
1103 #endif
1104
1105 #if SQLITE_MAX_WORKER_THREADS>0
1106 /*
1107 ** Join thread pTask->thread.
1108 */
1109 static int vdbeSorterJoinThread(SortSubtask *pTask){
1110 int rc = SQLITE_OK;
1111 if( pTask->pThread ){
1112 #ifdef SQLITE_DEBUG_SORTER_THREADS
1113 int bDone = pTask->bDone;
1114 #endif
1115 void *pRet = SQLITE_INT_TO_PTR(SQLITE_ERROR);
1116 vdbeSorterBlockDebug(pTask, !bDone, "enter");
1117 (void)sqlite3ThreadJoin(pTask->pThread, &pRet);
1118 vdbeSorterBlockDebug(pTask, !bDone, "exit");
1119 rc = SQLITE_PTR_TO_INT(pRet);
1120 assert( pTask->bDone==1 );
1121 pTask->bDone = 0;
1122 pTask->pThread = 0;
1123 }
1124 return rc;
1125 }
1126
1127 /*
1128 ** Launch a background thread to run xTask(pIn).
1129 */
1130 static int vdbeSorterCreateThread(
1131 SortSubtask *pTask, /* Thread will use this task object */
1132 void *(*xTask)(void*), /* Routine to run in a separate thread */
1133 void *pIn /* Argument passed into xTask() */
1134 ){
1135 assert( pTask->pThread==0 && pTask->bDone==0 );
1136 return sqlite3ThreadCreate(&pTask->pThread, xTask, pIn);
1137 }
1138
1139 /*
1140 ** Join all outstanding threads launched by SorterWrite() to create
1141 ** level-0 PMAs.
1142 */
1143 static int vdbeSorterJoinAll(VdbeSorter *pSorter, int rcin){
1144 int rc = rcin;
1145 int i;
1146
1147 /* This function is always called by the main user thread.
1148 **
1149 ** If this function is being called after SorterRewind() has been called,
1150 ** it is possible that thread pSorter->aTask[pSorter->nTask-1].pThread
1151 ** is currently attempt to join one of the other threads. To avoid a race
1152 ** condition where this thread also attempts to join the same object, join
1153 ** thread pSorter->aTask[pSorter->nTask-1].pThread first. */
1154 for(i=pSorter->nTask-1; i>=0; i--){
1155 SortSubtask *pTask = &pSorter->aTask[i];
1156 int rc2 = vdbeSorterJoinThread(pTask);
1157 if( rc==SQLITE_OK ) rc = rc2;
1158 }
1159 return rc;
1160 }
1161 #else
1162 # define vdbeSorterJoinAll(x,rcin) (rcin)
1163 # define vdbeSorterJoinThread(pTask) SQLITE_OK
1164 #endif
1165
1166 /*
1167 ** Allocate a new MergeEngine object capable of handling up to
1168 ** nReader PmaReader inputs.
1169 **
1170 ** nReader is automatically rounded up to the next power of two.
1171 ** nReader may not exceed SORTER_MAX_MERGE_COUNT even after rounding up.
1172 */
1173 static MergeEngine *vdbeMergeEngineNew(int nReader){
1174 int N = 2; /* Smallest power of two >= nReader */
1175 int nByte; /* Total bytes of space to allocate */
1176 MergeEngine *pNew; /* Pointer to allocated object to return */
1177
1178 assert( nReader<=SORTER_MAX_MERGE_COUNT );
1179
1180 while( N<nReader ) N += N;
1181 nByte = sizeof(MergeEngine) + N * (sizeof(int) + sizeof(PmaReader));
1182
1183 pNew = sqlite3FaultSim(100) ? 0 : (MergeEngine*)sqlite3MallocZero(nByte);
1184 if( pNew ){
1185 pNew->nTree = N;
1186 pNew->pTask = 0;
1187 pNew->aReadr = (PmaReader*)&pNew[1];
1188 pNew->aTree = (int*)&pNew->aReadr[N];
1189 }
1190 return pNew;
1191 }
1192
1193 /*
1194 ** Free the MergeEngine object passed as the only argument.
1195 */
1196 static void vdbeMergeEngineFree(MergeEngine *pMerger){
1197 int i;
1198 if( pMerger ){
1199 for(i=0; i<pMerger->nTree; i++){
1200 vdbePmaReaderClear(&pMerger->aReadr[i]);
1201 }
1202 }
1203 sqlite3_free(pMerger);
1204 }
1205
1206 /*
1207 ** Free all resources associated with the IncrMerger object indicated by
1208 ** the first argument.
1209 */
1210 static void vdbeIncrFree(IncrMerger *pIncr){
1211 if( pIncr ){
1212 #if SQLITE_MAX_WORKER_THREADS>0
1213 if( pIncr->bUseThread ){
1214 vdbeSorterJoinThread(pIncr->pTask);
1215 if( pIncr->aFile[0].pFd ) sqlite3OsCloseFree(pIncr->aFile[0].pFd);
1216 if( pIncr->aFile[1].pFd ) sqlite3OsCloseFree(pIncr->aFile[1].pFd);
1217 }
1218 #endif
1219 vdbeMergeEngineFree(pIncr->pMerger);
1220 sqlite3_free(pIncr);
1221 }
1222 }
1223
1224 /*
1225 ** Reset a sorting cursor back to its original empty state.
1226 */
1227 void sqlite3VdbeSorterReset(sqlite3 *db, VdbeSorter *pSorter){
1228 int i;
1229 (void)vdbeSorterJoinAll(pSorter, SQLITE_OK);
1230 assert( pSorter->bUseThreads || pSorter->pReader==0 );
1231 #if SQLITE_MAX_WORKER_THREADS>0
1232 if( pSorter->pReader ){
1233 vdbePmaReaderClear(pSorter->pReader);
1234 sqlite3DbFree(db, pSorter->pReader);
1235 pSorter->pReader = 0;
1236 }
1237 #endif
1238 vdbeMergeEngineFree(pSorter->pMerger);
1239 pSorter->pMerger = 0;
1240 for(i=0; i<pSorter->nTask; i++){
1241 SortSubtask *pTask = &pSorter->aTask[i];
1242 vdbeSortSubtaskCleanup(db, pTask);
1243 pTask->pSorter = pSorter;
1244 }
1245 if( pSorter->list.aMemory==0 ){
1246 vdbeSorterRecordFree(0, pSorter->list.pList);
1247 }
1248 pSorter->list.pList = 0;
1249 pSorter->list.szPMA = 0;
1250 pSorter->bUsePMA = 0;
1251 pSorter->iMemory = 0;
1252 pSorter->mxKeysize = 0;
1253 sqlite3DbFree(db, pSorter->pUnpacked);
1254 pSorter->pUnpacked = 0;
1255 }
1256
1257 /*
1258 ** Free any cursor components allocated by sqlite3VdbeSorterXXX routines.
1259 */
1260 void sqlite3VdbeSorterClose(sqlite3 *db, VdbeCursor *pCsr){
1261 VdbeSorter *pSorter;
1262 assert( pCsr->eCurType==CURTYPE_SORTER );
1263 pSorter = pCsr->uc.pSorter;
1264 if( pSorter ){
1265 sqlite3VdbeSorterReset(db, pSorter);
1266 sqlite3_free(pSorter->list.aMemory);
1267 sqlite3DbFree(db, pSorter);
1268 pCsr->uc.pSorter = 0;
1269 }
1270 }
1271
1272 #if SQLITE_MAX_MMAP_SIZE>0
1273 /*
1274 ** The first argument is a file-handle open on a temporary file. The file
1275 ** is guaranteed to be nByte bytes or smaller in size. This function
1276 ** attempts to extend the file to nByte bytes in size and to ensure that
1277 ** the VFS has memory mapped it.
1278 **
1279 ** Whether or not the file does end up memory mapped of course depends on
1280 ** the specific VFS implementation.
1281 */
1282 static void vdbeSorterExtendFile(sqlite3 *db, sqlite3_file *pFd, i64 nByte){
1283 if( nByte<=(i64)(db->nMaxSorterMmap) && pFd->pMethods->iVersion>=3 ){
1284 void *p = 0;
1285 int chunksize = 4*1024;
1286 sqlite3OsFileControlHint(pFd, SQLITE_FCNTL_CHUNK_SIZE, &chunksize);
1287 sqlite3OsFileControlHint(pFd, SQLITE_FCNTL_SIZE_HINT, &nByte);
1288 sqlite3OsFetch(pFd, 0, (int)nByte, &p);
1289 sqlite3OsUnfetch(pFd, 0, p);
1290 }
1291 }
1292 #else
1293 # define vdbeSorterExtendFile(x,y,z)
1294 #endif
1295
1296 /*
1297 ** Allocate space for a file-handle and open a temporary file. If successful,
1298 ** set *ppFd to point to the malloc'd file-handle and return SQLITE_OK.
1299 ** Otherwise, set *ppFd to 0 and return an SQLite error code.
1300 */
1301 static int vdbeSorterOpenTempFile(
1302 sqlite3 *db, /* Database handle doing sort */
1303 i64 nExtend, /* Attempt to extend file to this size */
1304 sqlite3_file **ppFd
1305 ){
1306 int rc;
1307 if( sqlite3FaultSim(202) ) return SQLITE_IOERR_ACCESS;
1308 rc = sqlite3OsOpenMalloc(db->pVfs, 0, ppFd,
1309 SQLITE_OPEN_TEMP_JOURNAL |
1310 SQLITE_OPEN_READWRITE | SQLITE_OPEN_CREATE |
1311 SQLITE_OPEN_EXCLUSIVE | SQLITE_OPEN_DELETEONCLOSE, &rc
1312 );
1313 if( rc==SQLITE_OK ){
1314 i64 max = SQLITE_MAX_MMAP_SIZE;
1315 sqlite3OsFileControlHint(*ppFd, SQLITE_FCNTL_MMAP_SIZE, (void*)&max);
1316 if( nExtend>0 ){
1317 vdbeSorterExtendFile(db, *ppFd, nExtend);
1318 }
1319 }
1320 return rc;
1321 }
1322
1323 /*
1324 ** If it has not already been allocated, allocate the UnpackedRecord
1325 ** structure at pTask->pUnpacked. Return SQLITE_OK if successful (or
1326 ** if no allocation was required), or SQLITE_NOMEM otherwise.
1327 */
1328 static int vdbeSortAllocUnpacked(SortSubtask *pTask){
1329 if( pTask->pUnpacked==0 ){
1330 pTask->pUnpacked = sqlite3VdbeAllocUnpackedRecord(pTask->pSorter->pKeyInfo);
1331 if( pTask->pUnpacked==0 ) return SQLITE_NOMEM_BKPT;
1332 pTask->pUnpacked->nField = pTask->pSorter->pKeyInfo->nField;
1333 pTask->pUnpacked->errCode = 0;
1334 }
1335 return SQLITE_OK;
1336 }
1337
1338
1339 /*
1340 ** Merge the two sorted lists p1 and p2 into a single list.
1341 */
1342 static SorterRecord *vdbeSorterMerge(
1343 SortSubtask *pTask, /* Calling thread context */
1344 SorterRecord *p1, /* First list to merge */
1345 SorterRecord *p2 /* Second list to merge */
1346 ){
1347 SorterRecord *pFinal = 0;
1348 SorterRecord **pp = &pFinal;
1349 int bCached = 0;
1350
1351 assert( p1!=0 && p2!=0 );
1352 for(;;){
1353 int res;
1354 res = pTask->xCompare(
1355 pTask, &bCached, SRVAL(p1), p1->nVal, SRVAL(p2), p2->nVal
1356 );
1357
1358 if( res<=0 ){
1359 *pp = p1;
1360 pp = &p1->u.pNext;
1361 p1 = p1->u.pNext;
1362 if( p1==0 ){
1363 *pp = p2;
1364 break;
1365 }
1366 }else{
1367 *pp = p2;
1368 pp = &p2->u.pNext;
1369 p2 = p2->u.pNext;
1370 bCached = 0;
1371 if( p2==0 ){
1372 *pp = p1;
1373 break;
1374 }
1375 }
1376 }
1377 return pFinal;
1378 }
1379
1380 /*
1381 ** Return the SorterCompare function to compare values collected by the
1382 ** sorter object passed as the only argument.
1383 */
1384 static SorterCompare vdbeSorterGetCompare(VdbeSorter *p){
1385 if( p->typeMask==SORTER_TYPE_INTEGER ){
1386 return vdbeSorterCompareInt;
1387 }else if( p->typeMask==SORTER_TYPE_TEXT ){
1388 return vdbeSorterCompareText;
1389 }
1390 return vdbeSorterCompare;
1391 }
1392
1393 /*
1394 ** Sort the linked list of records headed at pTask->pList. Return
1395 ** SQLITE_OK if successful, or an SQLite error code (i.e. SQLITE_NOMEM) if
1396 ** an error occurs.
1397 */
1398 static int vdbeSorterSort(SortSubtask *pTask, SorterList *pList){
1399 int i;
1400 SorterRecord **aSlot;
1401 SorterRecord *p;
1402 int rc;
1403
1404 rc = vdbeSortAllocUnpacked(pTask);
1405 if( rc!=SQLITE_OK ) return rc;
1406
1407 p = pList->pList;
1408 pTask->xCompare = vdbeSorterGetCompare(pTask->pSorter);
1409
1410 aSlot = (SorterRecord **)sqlite3MallocZero(64 * sizeof(SorterRecord *));
1411 if( !aSlot ){
1412 return SQLITE_NOMEM_BKPT;
1413 }
1414
1415 while( p ){
1416 SorterRecord *pNext;
1417 if( pList->aMemory ){
1418 if( (u8*)p==pList->aMemory ){
1419 pNext = 0;
1420 }else{
1421 assert( p->u.iNext<sqlite3MallocSize(pList->aMemory) );
1422 pNext = (SorterRecord*)&pList->aMemory[p->u.iNext];
1423 }
1424 }else{
1425 pNext = p->u.pNext;
1426 }
1427
1428 p->u.pNext = 0;
1429 for(i=0; aSlot[i]; i++){
1430 p = vdbeSorterMerge(pTask, p, aSlot[i]);
1431 aSlot[i] = 0;
1432 }
1433 aSlot[i] = p;
1434 p = pNext;
1435 }
1436
1437 p = 0;
1438 for(i=0; i<64; i++){
1439 if( aSlot[i]==0 ) continue;
1440 p = p ? vdbeSorterMerge(pTask, p, aSlot[i]) : aSlot[i];
1441 }
1442 pList->pList = p;
1443
1444 sqlite3_free(aSlot);
1445 assert( pTask->pUnpacked->errCode==SQLITE_OK
1446 || pTask->pUnpacked->errCode==SQLITE_NOMEM
1447 );
1448 return pTask->pUnpacked->errCode;
1449 }
1450
1451 /*
1452 ** Initialize a PMA-writer object.
1453 */
1454 static void vdbePmaWriterInit(
1455 sqlite3_file *pFd, /* File handle to write to */
1456 PmaWriter *p, /* Object to populate */
1457 int nBuf, /* Buffer size */
1458 i64 iStart /* Offset of pFd to begin writing at */
1459 ){
1460 memset(p, 0, sizeof(PmaWriter));
1461 p->aBuffer = (u8*)sqlite3Malloc(nBuf);
1462 if( !p->aBuffer ){
1463 p->eFWErr = SQLITE_NOMEM_BKPT;
1464 }else{
1465 p->iBufEnd = p->iBufStart = (iStart % nBuf);
1466 p->iWriteOff = iStart - p->iBufStart;
1467 p->nBuffer = nBuf;
1468 p->pFd = pFd;
1469 }
1470 }
1471
1472 /*
1473 ** Write nData bytes of data to the PMA. Return SQLITE_OK
1474 ** if successful, or an SQLite error code if an error occurs.
1475 */
1476 static void vdbePmaWriteBlob(PmaWriter *p, u8 *pData, int nData){
1477 int nRem = nData;
1478 while( nRem>0 && p->eFWErr==0 ){
1479 int nCopy = nRem;
1480 if( nCopy>(p->nBuffer - p->iBufEnd) ){
1481 nCopy = p->nBuffer - p->iBufEnd;
1482 }
1483
1484 memcpy(&p->aBuffer[p->iBufEnd], &pData[nData-nRem], nCopy);
1485 p->iBufEnd += nCopy;
1486 if( p->iBufEnd==p->nBuffer ){
1487 p->eFWErr = sqlite3OsWrite(p->pFd,
1488 &p->aBuffer[p->iBufStart], p->iBufEnd - p->iBufStart,
1489 p->iWriteOff + p->iBufStart
1490 );
1491 p->iBufStart = p->iBufEnd = 0;
1492 p->iWriteOff += p->nBuffer;
1493 }
1494 assert( p->iBufEnd<p->nBuffer );
1495
1496 nRem -= nCopy;
1497 }
1498 }
1499
1500 /*
1501 ** Flush any buffered data to disk and clean up the PMA-writer object.
1502 ** The results of using the PMA-writer after this call are undefined.
1503 ** Return SQLITE_OK if flushing the buffered data succeeds or is not
1504 ** required. Otherwise, return an SQLite error code.
1505 **
1506 ** Before returning, set *piEof to the offset immediately following the
1507 ** last byte written to the file.
1508 */
1509 static int vdbePmaWriterFinish(PmaWriter *p, i64 *piEof){
1510 int rc;
1511 if( p->eFWErr==0 && ALWAYS(p->aBuffer) && p->iBufEnd>p->iBufStart ){
1512 p->eFWErr = sqlite3OsWrite(p->pFd,
1513 &p->aBuffer[p->iBufStart], p->iBufEnd - p->iBufStart,
1514 p->iWriteOff + p->iBufStart
1515 );
1516 }
1517 *piEof = (p->iWriteOff + p->iBufEnd);
1518 sqlite3_free(p->aBuffer);
1519 rc = p->eFWErr;
1520 memset(p, 0, sizeof(PmaWriter));
1521 return rc;
1522 }
1523
1524 /*
1525 ** Write value iVal encoded as a varint to the PMA. Return
1526 ** SQLITE_OK if successful, or an SQLite error code if an error occurs.
1527 */
1528 static void vdbePmaWriteVarint(PmaWriter *p, u64 iVal){
1529 int nByte;
1530 u8 aByte[10];
1531 nByte = sqlite3PutVarint(aByte, iVal);
1532 vdbePmaWriteBlob(p, aByte, nByte);
1533 }
1534
1535 /*
1536 ** Write the current contents of in-memory linked-list pList to a level-0
1537 ** PMA in the temp file belonging to sub-task pTask. Return SQLITE_OK if
1538 ** successful, or an SQLite error code otherwise.
1539 **
1540 ** The format of a PMA is:
1541 **
1542 ** * A varint. This varint contains the total number of bytes of content
1543 ** in the PMA (not including the varint itself).
1544 **
1545 ** * One or more records packed end-to-end in order of ascending keys.
1546 ** Each record consists of a varint followed by a blob of data (the
1547 ** key). The varint is the number of bytes in the blob of data.
1548 */
1549 static int vdbeSorterListToPMA(SortSubtask *pTask, SorterList *pList){
1550 sqlite3 *db = pTask->pSorter->db;
1551 int rc = SQLITE_OK; /* Return code */
1552 PmaWriter writer; /* Object used to write to the file */
1553
1554 #ifdef SQLITE_DEBUG
1555 /* Set iSz to the expected size of file pTask->file after writing the PMA.
1556 ** This is used by an assert() statement at the end of this function. */
1557 i64 iSz = pList->szPMA + sqlite3VarintLen(pList->szPMA) + pTask->file.iEof;
1558 #endif
1559
1560 vdbeSorterWorkDebug(pTask, "enter");
1561 memset(&writer, 0, sizeof(PmaWriter));
1562 assert( pList->szPMA>0 );
1563
1564 /* If the first temporary PMA file has not been opened, open it now. */
1565 if( pTask->file.pFd==0 ){
1566 rc = vdbeSorterOpenTempFile(db, 0, &pTask->file.pFd);
1567 assert( rc!=SQLITE_OK || pTask->file.pFd );
1568 assert( pTask->file.iEof==0 );
1569 assert( pTask->nPMA==0 );
1570 }
1571
1572 /* Try to get the file to memory map */
1573 if( rc==SQLITE_OK ){
1574 vdbeSorterExtendFile(db, pTask->file.pFd, pTask->file.iEof+pList->szPMA+9);
1575 }
1576
1577 /* Sort the list */
1578 if( rc==SQLITE_OK ){
1579 rc = vdbeSorterSort(pTask, pList);
1580 }
1581
1582 if( rc==SQLITE_OK ){
1583 SorterRecord *p;
1584 SorterRecord *pNext = 0;
1585
1586 vdbePmaWriterInit(pTask->file.pFd, &writer, pTask->pSorter->pgsz,
1587 pTask->file.iEof);
1588 pTask->nPMA++;
1589 vdbePmaWriteVarint(&writer, pList->szPMA);
1590 for(p=pList->pList; p; p=pNext){
1591 pNext = p->u.pNext;
1592 vdbePmaWriteVarint(&writer, p->nVal);
1593 vdbePmaWriteBlob(&writer, SRVAL(p), p->nVal);
1594 if( pList->aMemory==0 ) sqlite3_free(p);
1595 }
1596 pList->pList = p;
1597 rc = vdbePmaWriterFinish(&writer, &pTask->file.iEof);
1598 }
1599
1600 vdbeSorterWorkDebug(pTask, "exit");
1601 assert( rc!=SQLITE_OK || pList->pList==0 );
1602 assert( rc!=SQLITE_OK || pTask->file.iEof==iSz );
1603 return rc;
1604 }
1605
1606 /*
1607 ** Advance the MergeEngine to its next entry.
1608 ** Set *pbEof to true there is no next entry because
1609 ** the MergeEngine has reached the end of all its inputs.
1610 **
1611 ** Return SQLITE_OK if successful or an error code if an error occurs.
1612 */
1613 static int vdbeMergeEngineStep(
1614 MergeEngine *pMerger, /* The merge engine to advance to the next row */
1615 int *pbEof /* Set TRUE at EOF. Set false for more content */
1616 ){
1617 int rc;
1618 int iPrev = pMerger->aTree[1];/* Index of PmaReader to advance */
1619 SortSubtask *pTask = pMerger->pTask;
1620
1621 /* Advance the current PmaReader */
1622 rc = vdbePmaReaderNext(&pMerger->aReadr[iPrev]);
1623
1624 /* Update contents of aTree[] */
1625 if( rc==SQLITE_OK ){
1626 int i; /* Index of aTree[] to recalculate */
1627 PmaReader *pReadr1; /* First PmaReader to compare */
1628 PmaReader *pReadr2; /* Second PmaReader to compare */
1629 int bCached = 0;
1630
1631 /* Find the first two PmaReaders to compare. The one that was just
1632 ** advanced (iPrev) and the one next to it in the array. */
1633 pReadr1 = &pMerger->aReadr[(iPrev & 0xFFFE)];
1634 pReadr2 = &pMerger->aReadr[(iPrev | 0x0001)];
1635
1636 for(i=(pMerger->nTree+iPrev)/2; i>0; i=i/2){
1637 /* Compare pReadr1 and pReadr2. Store the result in variable iRes. */
1638 int iRes;
1639 if( pReadr1->pFd==0 ){
1640 iRes = +1;
1641 }else if( pReadr2->pFd==0 ){
1642 iRes = -1;
1643 }else{
1644 iRes = pTask->xCompare(pTask, &bCached,
1645 pReadr1->aKey, pReadr1->nKey, pReadr2->aKey, pReadr2->nKey
1646 );
1647 }
1648
1649 /* If pReadr1 contained the smaller value, set aTree[i] to its index.
1650 ** Then set pReadr2 to the next PmaReader to compare to pReadr1. In this
1651 ** case there is no cache of pReadr2 in pTask->pUnpacked, so set
1652 ** pKey2 to point to the record belonging to pReadr2.
1653 **
1654 ** Alternatively, if pReadr2 contains the smaller of the two values,
1655 ** set aTree[i] to its index and update pReadr1. If vdbeSorterCompare()
1656 ** was actually called above, then pTask->pUnpacked now contains
1657 ** a value equivalent to pReadr2. So set pKey2 to NULL to prevent
1658 ** vdbeSorterCompare() from decoding pReadr2 again.
1659 **
1660 ** If the two values were equal, then the value from the oldest
1661 ** PMA should be considered smaller. The VdbeSorter.aReadr[] array
1662 ** is sorted from oldest to newest, so pReadr1 contains older values
1663 ** than pReadr2 iff (pReadr1<pReadr2). */
1664 if( iRes<0 || (iRes==0 && pReadr1<pReadr2) ){
1665 pMerger->aTree[i] = (int)(pReadr1 - pMerger->aReadr);
1666 pReadr2 = &pMerger->aReadr[ pMerger->aTree[i ^ 0x0001] ];
1667 bCached = 0;
1668 }else{
1669 if( pReadr1->pFd ) bCached = 0;
1670 pMerger->aTree[i] = (int)(pReadr2 - pMerger->aReadr);
1671 pReadr1 = &pMerger->aReadr[ pMerger->aTree[i ^ 0x0001] ];
1672 }
1673 }
1674 *pbEof = (pMerger->aReadr[pMerger->aTree[1]].pFd==0);
1675 }
1676
1677 return (rc==SQLITE_OK ? pTask->pUnpacked->errCode : rc);
1678 }
1679
1680 #if SQLITE_MAX_WORKER_THREADS>0
1681 /*
1682 ** The main routine for background threads that write level-0 PMAs.
1683 */
1684 static void *vdbeSorterFlushThread(void *pCtx){
1685 SortSubtask *pTask = (SortSubtask*)pCtx;
1686 int rc; /* Return code */
1687 assert( pTask->bDone==0 );
1688 rc = vdbeSorterListToPMA(pTask, &pTask->list);
1689 pTask->bDone = 1;
1690 return SQLITE_INT_TO_PTR(rc);
1691 }
1692 #endif /* SQLITE_MAX_WORKER_THREADS>0 */
1693
1694 /*
1695 ** Flush the current contents of VdbeSorter.list to a new PMA, possibly
1696 ** using a background thread.
1697 */
1698 static int vdbeSorterFlushPMA(VdbeSorter *pSorter){
1699 #if SQLITE_MAX_WORKER_THREADS==0
1700 pSorter->bUsePMA = 1;
1701 return vdbeSorterListToPMA(&pSorter->aTask[0], &pSorter->list);
1702 #else
1703 int rc = SQLITE_OK;
1704 int i;
1705 SortSubtask *pTask = 0; /* Thread context used to create new PMA */
1706 int nWorker = (pSorter->nTask-1);
1707
1708 /* Set the flag to indicate that at least one PMA has been written.
1709 ** Or will be, anyhow. */
1710 pSorter->bUsePMA = 1;
1711
1712 /* Select a sub-task to sort and flush the current list of in-memory
1713 ** records to disk. If the sorter is running in multi-threaded mode,
1714 ** round-robin between the first (pSorter->nTask-1) tasks. Except, if
1715 ** the background thread from a sub-tasks previous turn is still running,
1716 ** skip it. If the first (pSorter->nTask-1) sub-tasks are all still busy,
1717 ** fall back to using the final sub-task. The first (pSorter->nTask-1)
1718 ** sub-tasks are prefered as they use background threads - the final
1719 ** sub-task uses the main thread. */
1720 for(i=0; i<nWorker; i++){
1721 int iTest = (pSorter->iPrev + i + 1) % nWorker;
1722 pTask = &pSorter->aTask[iTest];
1723 if( pTask->bDone ){
1724 rc = vdbeSorterJoinThread(pTask);
1725 }
1726 if( rc!=SQLITE_OK || pTask->pThread==0 ) break;
1727 }
1728
1729 if( rc==SQLITE_OK ){
1730 if( i==nWorker ){
1731 /* Use the foreground thread for this operation */
1732 rc = vdbeSorterListToPMA(&pSorter->aTask[nWorker], &pSorter->list);
1733 }else{
1734 /* Launch a background thread for this operation */
1735 u8 *aMem = pTask->list.aMemory;
1736 void *pCtx = (void*)pTask;
1737
1738 assert( pTask->pThread==0 && pTask->bDone==0 );
1739 assert( pTask->list.pList==0 );
1740 assert( pTask->list.aMemory==0 || pSorter->list.aMemory!=0 );
1741
1742 pSorter->iPrev = (u8)(pTask - pSorter->aTask);
1743 pTask->list = pSorter->list;
1744 pSorter->list.pList = 0;
1745 pSorter->list.szPMA = 0;
1746 if( aMem ){
1747 pSorter->list.aMemory = aMem;
1748 pSorter->nMemory = sqlite3MallocSize(aMem);
1749 }else if( pSorter->list.aMemory ){
1750 pSorter->list.aMemory = sqlite3Malloc(pSorter->nMemory);
1751 if( !pSorter->list.aMemory ) return SQLITE_NOMEM_BKPT;
1752 }
1753
1754 rc = vdbeSorterCreateThread(pTask, vdbeSorterFlushThread, pCtx);
1755 }
1756 }
1757
1758 return rc;
1759 #endif /* SQLITE_MAX_WORKER_THREADS!=0 */
1760 }
1761
1762 /*
1763 ** Add a record to the sorter.
1764 */
1765 int sqlite3VdbeSorterWrite(
1766 const VdbeCursor *pCsr, /* Sorter cursor */
1767 Mem *pVal /* Memory cell containing record */
1768 ){
1769 VdbeSorter *pSorter;
1770 int rc = SQLITE_OK; /* Return Code */
1771 SorterRecord *pNew; /* New list element */
1772 int bFlush; /* True to flush contents of memory to PMA */
1773 int nReq; /* Bytes of memory required */
1774 int nPMA; /* Bytes of PMA space required */
1775 int t; /* serial type of first record field */
1776
1777 assert( pCsr->eCurType==CURTYPE_SORTER );
1778 pSorter = pCsr->uc.pSorter;
1779 getVarint32((const u8*)&pVal->z[1], t);
1780 if( t>0 && t<10 && t!=7 ){
1781 pSorter->typeMask &= SORTER_TYPE_INTEGER;
1782 }else if( t>10 && (t & 0x01) ){
1783 pSorter->typeMask &= SORTER_TYPE_TEXT;
1784 }else{
1785 pSorter->typeMask = 0;
1786 }
1787
1788 assert( pSorter );
1789
1790 /* Figure out whether or not the current contents of memory should be
1791 ** flushed to a PMA before continuing. If so, do so.
1792 **
1793 ** If using the single large allocation mode (pSorter->aMemory!=0), then
1794 ** flush the contents of memory to a new PMA if (a) at least one value is
1795 ** already in memory and (b) the new value will not fit in memory.
1796 **
1797 ** Or, if using separate allocations for each record, flush the contents
1798 ** of memory to a PMA if either of the following are true:
1799 **
1800 ** * The total memory allocated for the in-memory list is greater
1801 ** than (page-size * cache-size), or
1802 **
1803 ** * The total memory allocated for the in-memory list is greater
1804 ** than (page-size * 10) and sqlite3HeapNearlyFull() returns true.
1805 */
1806 nReq = pVal->n + sizeof(SorterRecord);
1807 nPMA = pVal->n + sqlite3VarintLen(pVal->n);
1808 if( pSorter->mxPmaSize ){
1809 if( pSorter->list.aMemory ){
1810 bFlush = pSorter->iMemory && (pSorter->iMemory+nReq) > pSorter->mxPmaSize;
1811 }else{
1812 bFlush = (
1813 (pSorter->list.szPMA > pSorter->mxPmaSize)
1814 || (pSorter->list.szPMA > pSorter->mnPmaSize && sqlite3HeapNearlyFull())
1815 );
1816 }
1817 if( bFlush ){
1818 rc = vdbeSorterFlushPMA(pSorter);
1819 pSorter->list.szPMA = 0;
1820 pSorter->iMemory = 0;
1821 assert( rc!=SQLITE_OK || pSorter->list.pList==0 );
1822 }
1823 }
1824
1825 pSorter->list.szPMA += nPMA;
1826 if( nPMA>pSorter->mxKeysize ){
1827 pSorter->mxKeysize = nPMA;
1828 }
1829
1830 if( pSorter->list.aMemory ){
1831 int nMin = pSorter->iMemory + nReq;
1832
1833 if( nMin>pSorter->nMemory ){
1834 u8 *aNew;
1835 int iListOff = (u8*)pSorter->list.pList - pSorter->list.aMemory;
1836 int nNew = pSorter->nMemory * 2;
1837 while( nNew < nMin ) nNew = nNew*2;
1838 if( nNew > pSorter->mxPmaSize ) nNew = pSorter->mxPmaSize;
1839 if( nNew < nMin ) nNew = nMin;
1840
1841 aNew = sqlite3Realloc(pSorter->list.aMemory, nNew);
1842 if( !aNew ) return SQLITE_NOMEM_BKPT;
1843 pSorter->list.pList = (SorterRecord*)&aNew[iListOff];
1844 pSorter->list.aMemory = aNew;
1845 pSorter->nMemory = nNew;
1846 }
1847
1848 pNew = (SorterRecord*)&pSorter->list.aMemory[pSorter->iMemory];
1849 pSorter->iMemory += ROUND8(nReq);
1850 if( pSorter->list.pList ){
1851 pNew->u.iNext = (int)((u8*)(pSorter->list.pList) - pSorter->list.aMemory);
1852 }
1853 }else{
1854 pNew = (SorterRecord *)sqlite3Malloc(nReq);
1855 if( pNew==0 ){
1856 return SQLITE_NOMEM_BKPT;
1857 }
1858 pNew->u.pNext = pSorter->list.pList;
1859 }
1860
1861 memcpy(SRVAL(pNew), pVal->z, pVal->n);
1862 pNew->nVal = pVal->n;
1863 pSorter->list.pList = pNew;
1864
1865 return rc;
1866 }
1867
1868 /*
1869 ** Read keys from pIncr->pMerger and populate pIncr->aFile[1]. The format
1870 ** of the data stored in aFile[1] is the same as that used by regular PMAs,
1871 ** except that the number-of-bytes varint is omitted from the start.
1872 */
1873 static int vdbeIncrPopulate(IncrMerger *pIncr){
1874 int rc = SQLITE_OK;
1875 int rc2;
1876 i64 iStart = pIncr->iStartOff;
1877 SorterFile *pOut = &pIncr->aFile[1];
1878 SortSubtask *pTask = pIncr->pTask;
1879 MergeEngine *pMerger = pIncr->pMerger;
1880 PmaWriter writer;
1881 assert( pIncr->bEof==0 );
1882
1883 vdbeSorterPopulateDebug(pTask, "enter");
1884
1885 vdbePmaWriterInit(pOut->pFd, &writer, pTask->pSorter->pgsz, iStart);
1886 while( rc==SQLITE_OK ){
1887 int dummy;
1888 PmaReader *pReader = &pMerger->aReadr[ pMerger->aTree[1] ];
1889 int nKey = pReader->nKey;
1890 i64 iEof = writer.iWriteOff + writer.iBufEnd;
1891
1892 /* Check if the output file is full or if the input has been exhausted.
1893 ** In either case exit the loop. */
1894 if( pReader->pFd==0 ) break;
1895 if( (iEof + nKey + sqlite3VarintLen(nKey))>(iStart + pIncr->mxSz) ) break;
1896
1897 /* Write the next key to the output. */
1898 vdbePmaWriteVarint(&writer, nKey);
1899 vdbePmaWriteBlob(&writer, pReader->aKey, nKey);
1900 assert( pIncr->pMerger->pTask==pTask );
1901 rc = vdbeMergeEngineStep(pIncr->pMerger, &dummy);
1902 }
1903
1904 rc2 = vdbePmaWriterFinish(&writer, &pOut->iEof);
1905 if( rc==SQLITE_OK ) rc = rc2;
1906 vdbeSorterPopulateDebug(pTask, "exit");
1907 return rc;
1908 }
1909
1910 #if SQLITE_MAX_WORKER_THREADS>0
1911 /*
1912 ** The main routine for background threads that populate aFile[1] of
1913 ** multi-threaded IncrMerger objects.
1914 */
1915 static void *vdbeIncrPopulateThread(void *pCtx){
1916 IncrMerger *pIncr = (IncrMerger*)pCtx;
1917 void *pRet = SQLITE_INT_TO_PTR( vdbeIncrPopulate(pIncr) );
1918 pIncr->pTask->bDone = 1;
1919 return pRet;
1920 }
1921
1922 /*
1923 ** Launch a background thread to populate aFile[1] of pIncr.
1924 */
1925 static int vdbeIncrBgPopulate(IncrMerger *pIncr){
1926 void *p = (void*)pIncr;
1927 assert( pIncr->bUseThread );
1928 return vdbeSorterCreateThread(pIncr->pTask, vdbeIncrPopulateThread, p);
1929 }
1930 #endif
1931
1932 /*
1933 ** This function is called when the PmaReader corresponding to pIncr has
1934 ** finished reading the contents of aFile[0]. Its purpose is to "refill"
1935 ** aFile[0] such that the PmaReader should start rereading it from the
1936 ** beginning.
1937 **
1938 ** For single-threaded objects, this is accomplished by literally reading
1939 ** keys from pIncr->pMerger and repopulating aFile[0].
1940 **
1941 ** For multi-threaded objects, all that is required is to wait until the
1942 ** background thread is finished (if it is not already) and then swap
1943 ** aFile[0] and aFile[1] in place. If the contents of pMerger have not
1944 ** been exhausted, this function also launches a new background thread
1945 ** to populate the new aFile[1].
1946 **
1947 ** SQLITE_OK is returned on success, or an SQLite error code otherwise.
1948 */
1949 static int vdbeIncrSwap(IncrMerger *pIncr){
1950 int rc = SQLITE_OK;
1951
1952 #if SQLITE_MAX_WORKER_THREADS>0
1953 if( pIncr->bUseThread ){
1954 rc = vdbeSorterJoinThread(pIncr->pTask);
1955
1956 if( rc==SQLITE_OK ){
1957 SorterFile f0 = pIncr->aFile[0];
1958 pIncr->aFile[0] = pIncr->aFile[1];
1959 pIncr->aFile[1] = f0;
1960 }
1961
1962 if( rc==SQLITE_OK ){
1963 if( pIncr->aFile[0].iEof==pIncr->iStartOff ){
1964 pIncr->bEof = 1;
1965 }else{
1966 rc = vdbeIncrBgPopulate(pIncr);
1967 }
1968 }
1969 }else
1970 #endif
1971 {
1972 rc = vdbeIncrPopulate(pIncr);
1973 pIncr->aFile[0] = pIncr->aFile[1];
1974 if( pIncr->aFile[0].iEof==pIncr->iStartOff ){
1975 pIncr->bEof = 1;
1976 }
1977 }
1978
1979 return rc;
1980 }
1981
1982 /*
1983 ** Allocate and return a new IncrMerger object to read data from pMerger.
1984 **
1985 ** If an OOM condition is encountered, return NULL. In this case free the
1986 ** pMerger argument before returning.
1987 */
1988 static int vdbeIncrMergerNew(
1989 SortSubtask *pTask, /* The thread that will be using the new IncrMerger */
1990 MergeEngine *pMerger, /* The MergeEngine that the IncrMerger will control */
1991 IncrMerger **ppOut /* Write the new IncrMerger here */
1992 ){
1993 int rc = SQLITE_OK;
1994 IncrMerger *pIncr = *ppOut = (IncrMerger*)
1995 (sqlite3FaultSim(100) ? 0 : sqlite3MallocZero(sizeof(*pIncr)));
1996 if( pIncr ){
1997 pIncr->pMerger = pMerger;
1998 pIncr->pTask = pTask;
1999 pIncr->mxSz = MAX(pTask->pSorter->mxKeysize+9,pTask->pSorter->mxPmaSize/2);
2000 pTask->file2.iEof += pIncr->mxSz;
2001 }else{
2002 vdbeMergeEngineFree(pMerger);
2003 rc = SQLITE_NOMEM_BKPT;
2004 }
2005 return rc;
2006 }
2007
2008 #if SQLITE_MAX_WORKER_THREADS>0
2009 /*
2010 ** Set the "use-threads" flag on object pIncr.
2011 */
2012 static void vdbeIncrMergerSetThreads(IncrMerger *pIncr){
2013 pIncr->bUseThread = 1;
2014 pIncr->pTask->file2.iEof -= pIncr->mxSz;
2015 }
2016 #endif /* SQLITE_MAX_WORKER_THREADS>0 */
2017
2018
2019
2020 /*
2021 ** Recompute pMerger->aTree[iOut] by comparing the next keys on the
2022 ** two PmaReaders that feed that entry. Neither of the PmaReaders
2023 ** are advanced. This routine merely does the comparison.
2024 */
2025 static void vdbeMergeEngineCompare(
2026 MergeEngine *pMerger, /* Merge engine containing PmaReaders to compare */
2027 int iOut /* Store the result in pMerger->aTree[iOut] */
2028 ){
2029 int i1;
2030 int i2;
2031 int iRes;
2032 PmaReader *p1;
2033 PmaReader *p2;
2034
2035 assert( iOut<pMerger->nTree && iOut>0 );
2036
2037 if( iOut>=(pMerger->nTree/2) ){
2038 i1 = (iOut - pMerger->nTree/2) * 2;
2039 i2 = i1 + 1;
2040 }else{
2041 i1 = pMerger->aTree[iOut*2];
2042 i2 = pMerger->aTree[iOut*2+1];
2043 }
2044
2045 p1 = &pMerger->aReadr[i1];
2046 p2 = &pMerger->aReadr[i2];
2047
2048 if( p1->pFd==0 ){
2049 iRes = i2;
2050 }else if( p2->pFd==0 ){
2051 iRes = i1;
2052 }else{
2053 SortSubtask *pTask = pMerger->pTask;
2054 int bCached = 0;
2055 int res;
2056 assert( pTask->pUnpacked!=0 ); /* from vdbeSortSubtaskMain() */
2057 res = pTask->xCompare(
2058 pTask, &bCached, p1->aKey, p1->nKey, p2->aKey, p2->nKey
2059 );
2060 if( res<=0 ){
2061 iRes = i1;
2062 }else{
2063 iRes = i2;
2064 }
2065 }
2066
2067 pMerger->aTree[iOut] = iRes;
2068 }
2069
2070 /*
2071 ** Allowed values for the eMode parameter to vdbeMergeEngineInit()
2072 ** and vdbePmaReaderIncrMergeInit().
2073 **
2074 ** Only INCRINIT_NORMAL is valid in single-threaded builds (when
2075 ** SQLITE_MAX_WORKER_THREADS==0). The other values are only used
2076 ** when there exists one or more separate worker threads.
2077 */
2078 #define INCRINIT_NORMAL 0
2079 #define INCRINIT_TASK 1
2080 #define INCRINIT_ROOT 2
2081
2082 /*
2083 ** Forward reference required as the vdbeIncrMergeInit() and
2084 ** vdbePmaReaderIncrInit() routines are called mutually recursively when
2085 ** building a merge tree.
2086 */
2087 static int vdbePmaReaderIncrInit(PmaReader *pReadr, int eMode);
2088
2089 /*
2090 ** Initialize the MergeEngine object passed as the second argument. Once this
2091 ** function returns, the first key of merged data may be read from the
2092 ** MergeEngine object in the usual fashion.
2093 **
2094 ** If argument eMode is INCRINIT_ROOT, then it is assumed that any IncrMerge
2095 ** objects attached to the PmaReader objects that the merger reads from have
2096 ** already been populated, but that they have not yet populated aFile[0] and
2097 ** set the PmaReader objects up to read from it. In this case all that is
2098 ** required is to call vdbePmaReaderNext() on each PmaReader to point it at
2099 ** its first key.
2100 **
2101 ** Otherwise, if eMode is any value other than INCRINIT_ROOT, then use
2102 ** vdbePmaReaderIncrMergeInit() to initialize each PmaReader that feeds data
2103 ** to pMerger.
2104 **
2105 ** SQLITE_OK is returned if successful, or an SQLite error code otherwise.
2106 */
2107 static int vdbeMergeEngineInit(
2108 SortSubtask *pTask, /* Thread that will run pMerger */
2109 MergeEngine *pMerger, /* MergeEngine to initialize */
2110 int eMode /* One of the INCRINIT_XXX constants */
2111 ){
2112 int rc = SQLITE_OK; /* Return code */
2113 int i; /* For looping over PmaReader objects */
2114 int nTree = pMerger->nTree;
2115
2116 /* eMode is always INCRINIT_NORMAL in single-threaded mode */
2117 assert( SQLITE_MAX_WORKER_THREADS>0 || eMode==INCRINIT_NORMAL );
2118
2119 /* Verify that the MergeEngine is assigned to a single thread */
2120 assert( pMerger->pTask==0 );
2121 pMerger->pTask = pTask;
2122
2123 for(i=0; i<nTree; i++){
2124 if( SQLITE_MAX_WORKER_THREADS>0 && eMode==INCRINIT_ROOT ){
2125 /* PmaReaders should be normally initialized in order, as if they are
2126 ** reading from the same temp file this makes for more linear file IO.
2127 ** However, in the INCRINIT_ROOT case, if PmaReader aReadr[nTask-1] is
2128 ** in use it will block the vdbePmaReaderNext() call while it uses
2129 ** the main thread to fill its buffer. So calling PmaReaderNext()
2130 ** on this PmaReader before any of the multi-threaded PmaReaders takes
2131 ** better advantage of multi-processor hardware. */
2132 rc = vdbePmaReaderNext(&pMerger->aReadr[nTree-i-1]);
2133 }else{
2134 rc = vdbePmaReaderIncrInit(&pMerger->aReadr[i], INCRINIT_NORMAL);
2135 }
2136 if( rc!=SQLITE_OK ) return rc;
2137 }
2138
2139 for(i=pMerger->nTree-1; i>0; i--){
2140 vdbeMergeEngineCompare(pMerger, i);
2141 }
2142 return pTask->pUnpacked->errCode;
2143 }
2144
2145 /*
2146 ** The PmaReader passed as the first argument is guaranteed to be an
2147 ** incremental-reader (pReadr->pIncr!=0). This function serves to open
2148 ** and/or initialize the temp file related fields of the IncrMerge
2149 ** object at (pReadr->pIncr).
2150 **
2151 ** If argument eMode is set to INCRINIT_NORMAL, then all PmaReaders
2152 ** in the sub-tree headed by pReadr are also initialized. Data is then
2153 ** loaded into the buffers belonging to pReadr and it is set to point to
2154 ** the first key in its range.
2155 **
2156 ** If argument eMode is set to INCRINIT_TASK, then pReadr is guaranteed
2157 ** to be a multi-threaded PmaReader and this function is being called in a
2158 ** background thread. In this case all PmaReaders in the sub-tree are
2159 ** initialized as for INCRINIT_NORMAL and the aFile[1] buffer belonging to
2160 ** pReadr is populated. However, pReadr itself is not set up to point
2161 ** to its first key. A call to vdbePmaReaderNext() is still required to do
2162 ** that.
2163 **
2164 ** The reason this function does not call vdbePmaReaderNext() immediately
2165 ** in the INCRINIT_TASK case is that vdbePmaReaderNext() assumes that it has
2166 ** to block on thread (pTask->thread) before accessing aFile[1]. But, since
2167 ** this entire function is being run by thread (pTask->thread), that will
2168 ** lead to the current background thread attempting to join itself.
2169 **
2170 ** Finally, if argument eMode is set to INCRINIT_ROOT, it may be assumed
2171 ** that pReadr->pIncr is a multi-threaded IncrMerge objects, and that all
2172 ** child-trees have already been initialized using IncrInit(INCRINIT_TASK).
2173 ** In this case vdbePmaReaderNext() is called on all child PmaReaders and
2174 ** the current PmaReader set to point to the first key in its range.
2175 **
2176 ** SQLITE_OK is returned if successful, or an SQLite error code otherwise.
2177 */
2178 static int vdbePmaReaderIncrMergeInit(PmaReader *pReadr, int eMode){
2179 int rc = SQLITE_OK;
2180 IncrMerger *pIncr = pReadr->pIncr;
2181 SortSubtask *pTask = pIncr->pTask;
2182 sqlite3 *db = pTask->pSorter->db;
2183
2184 /* eMode is always INCRINIT_NORMAL in single-threaded mode */
2185 assert( SQLITE_MAX_WORKER_THREADS>0 || eMode==INCRINIT_NORMAL );
2186
2187 rc = vdbeMergeEngineInit(pTask, pIncr->pMerger, eMode);
2188
2189 /* Set up the required files for pIncr. A multi-theaded IncrMerge object
2190 ** requires two temp files to itself, whereas a single-threaded object
2191 ** only requires a region of pTask->file2. */
2192 if( rc==SQLITE_OK ){
2193 int mxSz = pIncr->mxSz;
2194 #if SQLITE_MAX_WORKER_THREADS>0
2195 if( pIncr->bUseThread ){
2196 rc = vdbeSorterOpenTempFile(db, mxSz, &pIncr->aFile[0].pFd);
2197 if( rc==SQLITE_OK ){
2198 rc = vdbeSorterOpenTempFile(db, mxSz, &pIncr->aFile[1].pFd);
2199 }
2200 }else
2201 #endif
2202 /*if( !pIncr->bUseThread )*/{
2203 if( pTask->file2.pFd==0 ){
2204 assert( pTask->file2.iEof>0 );
2205 rc = vdbeSorterOpenTempFile(db, pTask->file2.iEof, &pTask->file2.pFd);
2206 pTask->file2.iEof = 0;
2207 }
2208 if( rc==SQLITE_OK ){
2209 pIncr->aFile[1].pFd = pTask->file2.pFd;
2210 pIncr->iStartOff = pTask->file2.iEof;
2211 pTask->file2.iEof += mxSz;
2212 }
2213 }
2214 }
2215
2216 #if SQLITE_MAX_WORKER_THREADS>0
2217 if( rc==SQLITE_OK && pIncr->bUseThread ){
2218 /* Use the current thread to populate aFile[1], even though this
2219 ** PmaReader is multi-threaded. If this is an INCRINIT_TASK object,
2220 ** then this function is already running in background thread
2221 ** pIncr->pTask->thread.
2222 **
2223 ** If this is the INCRINIT_ROOT object, then it is running in the
2224 ** main VDBE thread. But that is Ok, as that thread cannot return
2225 ** control to the VDBE or proceed with anything useful until the
2226 ** first results are ready from this merger object anyway.
2227 */
2228 assert( eMode==INCRINIT_ROOT || eMode==INCRINIT_TASK );
2229 rc = vdbeIncrPopulate(pIncr);
2230 }
2231 #endif
2232
2233 if( rc==SQLITE_OK && (SQLITE_MAX_WORKER_THREADS==0 || eMode!=INCRINIT_TASK) ){
2234 rc = vdbePmaReaderNext(pReadr);
2235 }
2236
2237 return rc;
2238 }
2239
2240 #if SQLITE_MAX_WORKER_THREADS>0
2241 /*
2242 ** The main routine for vdbePmaReaderIncrMergeInit() operations run in
2243 ** background threads.
2244 */
2245 static void *vdbePmaReaderBgIncrInit(void *pCtx){
2246 PmaReader *pReader = (PmaReader*)pCtx;
2247 void *pRet = SQLITE_INT_TO_PTR(
2248 vdbePmaReaderIncrMergeInit(pReader,INCRINIT_TASK)
2249 );
2250 pReader->pIncr->pTask->bDone = 1;
2251 return pRet;
2252 }
2253 #endif
2254
2255 /*
2256 ** If the PmaReader passed as the first argument is not an incremental-reader
2257 ** (if pReadr->pIncr==0), then this function is a no-op. Otherwise, it invokes
2258 ** the vdbePmaReaderIncrMergeInit() function with the parameters passed to
2259 ** this routine to initialize the incremental merge.
2260 **
2261 ** If the IncrMerger object is multi-threaded (IncrMerger.bUseThread==1),
2262 ** then a background thread is launched to call vdbePmaReaderIncrMergeInit().
2263 ** Or, if the IncrMerger is single threaded, the same function is called
2264 ** using the current thread.
2265 */
2266 static int vdbePmaReaderIncrInit(PmaReader *pReadr, int eMode){
2267 IncrMerger *pIncr = pReadr->pIncr; /* Incremental merger */
2268 int rc = SQLITE_OK; /* Return code */
2269 if( pIncr ){
2270 #if SQLITE_MAX_WORKER_THREADS>0
2271 assert( pIncr->bUseThread==0 || eMode==INCRINIT_TASK );
2272 if( pIncr->bUseThread ){
2273 void *pCtx = (void*)pReadr;
2274 rc = vdbeSorterCreateThread(pIncr->pTask, vdbePmaReaderBgIncrInit, pCtx);
2275 }else
2276 #endif
2277 {
2278 rc = vdbePmaReaderIncrMergeInit(pReadr, eMode);
2279 }
2280 }
2281 return rc;
2282 }
2283
2284 /*
2285 ** Allocate a new MergeEngine object to merge the contents of nPMA level-0
2286 ** PMAs from pTask->file. If no error occurs, set *ppOut to point to
2287 ** the new object and return SQLITE_OK. Or, if an error does occur, set *ppOut
2288 ** to NULL and return an SQLite error code.
2289 **
2290 ** When this function is called, *piOffset is set to the offset of the
2291 ** first PMA to read from pTask->file. Assuming no error occurs, it is
2292 ** set to the offset immediately following the last byte of the last
2293 ** PMA before returning. If an error does occur, then the final value of
2294 ** *piOffset is undefined.
2295 */
2296 static int vdbeMergeEngineLevel0(
2297 SortSubtask *pTask, /* Sorter task to read from */
2298 int nPMA, /* Number of PMAs to read */
2299 i64 *piOffset, /* IN/OUT: Readr offset in pTask->file */
2300 MergeEngine **ppOut /* OUT: New merge-engine */
2301 ){
2302 MergeEngine *pNew; /* Merge engine to return */
2303 i64 iOff = *piOffset;
2304 int i;
2305 int rc = SQLITE_OK;
2306
2307 *ppOut = pNew = vdbeMergeEngineNew(nPMA);
2308 if( pNew==0 ) rc = SQLITE_NOMEM_BKPT;
2309
2310 for(i=0; i<nPMA && rc==SQLITE_OK; i++){
2311 i64 nDummy = 0;
2312 PmaReader *pReadr = &pNew->aReadr[i];
2313 rc = vdbePmaReaderInit(pTask, &pTask->file, iOff, pReadr, &nDummy);
2314 iOff = pReadr->iEof;
2315 }
2316
2317 if( rc!=SQLITE_OK ){
2318 vdbeMergeEngineFree(pNew);
2319 *ppOut = 0;
2320 }
2321 *piOffset = iOff;
2322 return rc;
2323 }
2324
2325 /*
2326 ** Return the depth of a tree comprising nPMA PMAs, assuming a fanout of
2327 ** SORTER_MAX_MERGE_COUNT. The returned value does not include leaf nodes.
2328 **
2329 ** i.e.
2330 **
2331 ** nPMA<=16 -> TreeDepth() == 0
2332 ** nPMA<=256 -> TreeDepth() == 1
2333 ** nPMA<=65536 -> TreeDepth() == 2
2334 */
2335 static int vdbeSorterTreeDepth(int nPMA){
2336 int nDepth = 0;
2337 i64 nDiv = SORTER_MAX_MERGE_COUNT;
2338 while( nDiv < (i64)nPMA ){
2339 nDiv = nDiv * SORTER_MAX_MERGE_COUNT;
2340 nDepth++;
2341 }
2342 return nDepth;
2343 }
2344
2345 /*
2346 ** pRoot is the root of an incremental merge-tree with depth nDepth (according
2347 ** to vdbeSorterTreeDepth()). pLeaf is the iSeq'th leaf to be added to the
2348 ** tree, counting from zero. This function adds pLeaf to the tree.
2349 **
2350 ** If successful, SQLITE_OK is returned. If an error occurs, an SQLite error
2351 ** code is returned and pLeaf is freed.
2352 */
2353 static int vdbeSorterAddToTree(
2354 SortSubtask *pTask, /* Task context */
2355 int nDepth, /* Depth of tree according to TreeDepth() */
2356 int iSeq, /* Sequence number of leaf within tree */
2357 MergeEngine *pRoot, /* Root of tree */
2358 MergeEngine *pLeaf /* Leaf to add to tree */
2359 ){
2360 int rc = SQLITE_OK;
2361 int nDiv = 1;
2362 int i;
2363 MergeEngine *p = pRoot;
2364 IncrMerger *pIncr;
2365
2366 rc = vdbeIncrMergerNew(pTask, pLeaf, &pIncr);
2367
2368 for(i=1; i<nDepth; i++){
2369 nDiv = nDiv * SORTER_MAX_MERGE_COUNT;
2370 }
2371
2372 for(i=1; i<nDepth && rc==SQLITE_OK; i++){
2373 int iIter = (iSeq / nDiv) % SORTER_MAX_MERGE_COUNT;
2374 PmaReader *pReadr = &p->aReadr[iIter];
2375
2376 if( pReadr->pIncr==0 ){
2377 MergeEngine *pNew = vdbeMergeEngineNew(SORTER_MAX_MERGE_COUNT);
2378 if( pNew==0 ){
2379 rc = SQLITE_NOMEM_BKPT;
2380 }else{
2381 rc = vdbeIncrMergerNew(pTask, pNew, &pReadr->pIncr);
2382 }
2383 }
2384 if( rc==SQLITE_OK ){
2385 p = pReadr->pIncr->pMerger;
2386 nDiv = nDiv / SORTER_MAX_MERGE_COUNT;
2387 }
2388 }
2389
2390 if( rc==SQLITE_OK ){
2391 p->aReadr[iSeq % SORTER_MAX_MERGE_COUNT].pIncr = pIncr;
2392 }else{
2393 vdbeIncrFree(pIncr);
2394 }
2395 return rc;
2396 }
2397
2398 /*
2399 ** This function is called as part of a SorterRewind() operation on a sorter
2400 ** that has already written two or more level-0 PMAs to one or more temp
2401 ** files. It builds a tree of MergeEngine/IncrMerger/PmaReader objects that
2402 ** can be used to incrementally merge all PMAs on disk.
2403 **
2404 ** If successful, SQLITE_OK is returned and *ppOut set to point to the
2405 ** MergeEngine object at the root of the tree before returning. Or, if an
2406 ** error occurs, an SQLite error code is returned and the final value
2407 ** of *ppOut is undefined.
2408 */
2409 static int vdbeSorterMergeTreeBuild(
2410 VdbeSorter *pSorter, /* The VDBE cursor that implements the sort */
2411 MergeEngine **ppOut /* Write the MergeEngine here */
2412 ){
2413 MergeEngine *pMain = 0;
2414 int rc = SQLITE_OK;
2415 int iTask;
2416
2417 #if SQLITE_MAX_WORKER_THREADS>0
2418 /* If the sorter uses more than one task, then create the top-level
2419 ** MergeEngine here. This MergeEngine will read data from exactly
2420 ** one PmaReader per sub-task. */
2421 assert( pSorter->bUseThreads || pSorter->nTask==1 );
2422 if( pSorter->nTask>1 ){
2423 pMain = vdbeMergeEngineNew(pSorter->nTask);
2424 if( pMain==0 ) rc = SQLITE_NOMEM_BKPT;
2425 }
2426 #endif
2427
2428 for(iTask=0; rc==SQLITE_OK && iTask<pSorter->nTask; iTask++){
2429 SortSubtask *pTask = &pSorter->aTask[iTask];
2430 assert( pTask->nPMA>0 || SQLITE_MAX_WORKER_THREADS>0 );
2431 if( SQLITE_MAX_WORKER_THREADS==0 || pTask->nPMA ){
2432 MergeEngine *pRoot = 0; /* Root node of tree for this task */
2433 int nDepth = vdbeSorterTreeDepth(pTask->nPMA);
2434 i64 iReadOff = 0;
2435
2436 if( pTask->nPMA<=SORTER_MAX_MERGE_COUNT ){
2437 rc = vdbeMergeEngineLevel0(pTask, pTask->nPMA, &iReadOff, &pRoot);
2438 }else{
2439 int i;
2440 int iSeq = 0;
2441 pRoot = vdbeMergeEngineNew(SORTER_MAX_MERGE_COUNT);
2442 if( pRoot==0 ) rc = SQLITE_NOMEM_BKPT;
2443 for(i=0; i<pTask->nPMA && rc==SQLITE_OK; i += SORTER_MAX_MERGE_COUNT){
2444 MergeEngine *pMerger = 0; /* New level-0 PMA merger */
2445 int nReader; /* Number of level-0 PMAs to merge */
2446
2447 nReader = MIN(pTask->nPMA - i, SORTER_MAX_MERGE_COUNT);
2448 rc = vdbeMergeEngineLevel0(pTask, nReader, &iReadOff, &pMerger);
2449 if( rc==SQLITE_OK ){
2450 rc = vdbeSorterAddToTree(pTask, nDepth, iSeq++, pRoot, pMerger);
2451 }
2452 }
2453 }
2454
2455 if( rc==SQLITE_OK ){
2456 #if SQLITE_MAX_WORKER_THREADS>0
2457 if( pMain!=0 ){
2458 rc = vdbeIncrMergerNew(pTask, pRoot, &pMain->aReadr[iTask].pIncr);
2459 }else
2460 #endif
2461 {
2462 assert( pMain==0 );
2463 pMain = pRoot;
2464 }
2465 }else{
2466 vdbeMergeEngineFree(pRoot);
2467 }
2468 }
2469 }
2470
2471 if( rc!=SQLITE_OK ){
2472 vdbeMergeEngineFree(pMain);
2473 pMain = 0;
2474 }
2475 *ppOut = pMain;
2476 return rc;
2477 }
2478
2479 /*
2480 ** This function is called as part of an sqlite3VdbeSorterRewind() operation
2481 ** on a sorter that has written two or more PMAs to temporary files. It sets
2482 ** up either VdbeSorter.pMerger (for single threaded sorters) or pReader
2483 ** (for multi-threaded sorters) so that it can be used to iterate through
2484 ** all records stored in the sorter.
2485 **
2486 ** SQLITE_OK is returned if successful, or an SQLite error code otherwise.
2487 */
2488 static int vdbeSorterSetupMerge(VdbeSorter *pSorter){
2489 int rc; /* Return code */
2490 SortSubtask *pTask0 = &pSorter->aTask[0];
2491 MergeEngine *pMain = 0;
2492 #if SQLITE_MAX_WORKER_THREADS
2493 sqlite3 *db = pTask0->pSorter->db;
2494 int i;
2495 SorterCompare xCompare = vdbeSorterGetCompare(pSorter);
2496 for(i=0; i<pSorter->nTask; i++){
2497 pSorter->aTask[i].xCompare = xCompare;
2498 }
2499 #endif
2500
2501 rc = vdbeSorterMergeTreeBuild(pSorter, &pMain);
2502 if( rc==SQLITE_OK ){
2503 #if SQLITE_MAX_WORKER_THREADS
2504 assert( pSorter->bUseThreads==0 || pSorter->nTask>1 );
2505 if( pSorter->bUseThreads ){
2506 int iTask;
2507 PmaReader *pReadr = 0;
2508 SortSubtask *pLast = &pSorter->aTask[pSorter->nTask-1];
2509 rc = vdbeSortAllocUnpacked(pLast);
2510 if( rc==SQLITE_OK ){
2511 pReadr = (PmaReader*)sqlite3DbMallocZero(db, sizeof(PmaReader));
2512 pSorter->pReader = pReadr;
2513 if( pReadr==0 ) rc = SQLITE_NOMEM_BKPT;
2514 }
2515 if( rc==SQLITE_OK ){
2516 rc = vdbeIncrMergerNew(pLast, pMain, &pReadr->pIncr);
2517 if( rc==SQLITE_OK ){
2518 vdbeIncrMergerSetThreads(pReadr->pIncr);
2519 for(iTask=0; iTask<(pSorter->nTask-1); iTask++){
2520 IncrMerger *pIncr;
2521 if( (pIncr = pMain->aReadr[iTask].pIncr) ){
2522 vdbeIncrMergerSetThreads(pIncr);
2523 assert( pIncr->pTask!=pLast );
2524 }
2525 }
2526 for(iTask=0; rc==SQLITE_OK && iTask<pSorter->nTask; iTask++){
2527 /* Check that:
2528 **
2529 ** a) The incremental merge object is configured to use the
2530 ** right task, and
2531 ** b) If it is using task (nTask-1), it is configured to run
2532 ** in single-threaded mode. This is important, as the
2533 ** root merge (INCRINIT_ROOT) will be using the same task
2534 ** object.
2535 */
2536 PmaReader *p = &pMain->aReadr[iTask];
2537 assert( p->pIncr==0 || (
2538 (p->pIncr->pTask==&pSorter->aTask[iTask]) /* a */
2539 && (iTask!=pSorter->nTask-1 || p->pIncr->bUseThread==0) /* b */
2540 ));
2541 rc = vdbePmaReaderIncrInit(p, INCRINIT_TASK);
2542 }
2543 }
2544 pMain = 0;
2545 }
2546 if( rc==SQLITE_OK ){
2547 rc = vdbePmaReaderIncrMergeInit(pReadr, INCRINIT_ROOT);
2548 }
2549 }else
2550 #endif
2551 {
2552 rc = vdbeMergeEngineInit(pTask0, pMain, INCRINIT_NORMAL);
2553 pSorter->pMerger = pMain;
2554 pMain = 0;
2555 }
2556 }
2557
2558 if( rc!=SQLITE_OK ){
2559 vdbeMergeEngineFree(pMain);
2560 }
2561 return rc;
2562 }
2563
2564
2565 /*
2566 ** Once the sorter has been populated by calls to sqlite3VdbeSorterWrite,
2567 ** this function is called to prepare for iterating through the records
2568 ** in sorted order.
2569 */
2570 int sqlite3VdbeSorterRewind(const VdbeCursor *pCsr, int *pbEof){
2571 VdbeSorter *pSorter;
2572 int rc = SQLITE_OK; /* Return code */
2573
2574 assert( pCsr->eCurType==CURTYPE_SORTER );
2575 pSorter = pCsr->uc.pSorter;
2576 assert( pSorter );
2577
2578 /* If no data has been written to disk, then do not do so now. Instead,
2579 ** sort the VdbeSorter.pRecord list. The vdbe layer will read data directly
2580 ** from the in-memory list. */
2581 if( pSorter->bUsePMA==0 ){
2582 if( pSorter->list.pList ){
2583 *pbEof = 0;
2584 rc = vdbeSorterSort(&pSorter->aTask[0], &pSorter->list);
2585 }else{
2586 *pbEof = 1;
2587 }
2588 return rc;
2589 }
2590
2591 /* Write the current in-memory list to a PMA. When the VdbeSorterWrite()
2592 ** function flushes the contents of memory to disk, it immediately always
2593 ** creates a new list consisting of a single key immediately afterwards.
2594 ** So the list is never empty at this point. */
2595 assert( pSorter->list.pList );
2596 rc = vdbeSorterFlushPMA(pSorter);
2597
2598 /* Join all threads */
2599 rc = vdbeSorterJoinAll(pSorter, rc);
2600
2601 vdbeSorterRewindDebug("rewind");
2602
2603 /* Assuming no errors have occurred, set up a merger structure to
2604 ** incrementally read and merge all remaining PMAs. */
2605 assert( pSorter->pReader==0 );
2606 if( rc==SQLITE_OK ){
2607 rc = vdbeSorterSetupMerge(pSorter);
2608 *pbEof = 0;
2609 }
2610
2611 vdbeSorterRewindDebug("rewinddone");
2612 return rc;
2613 }
2614
2615 /*
2616 ** Advance to the next element in the sorter.
2617 */
2618 int sqlite3VdbeSorterNext(sqlite3 *db, const VdbeCursor *pCsr, int *pbEof){
2619 VdbeSorter *pSorter;
2620 int rc; /* Return code */
2621
2622 assert( pCsr->eCurType==CURTYPE_SORTER );
2623 pSorter = pCsr->uc.pSorter;
2624 assert( pSorter->bUsePMA || (pSorter->pReader==0 && pSorter->pMerger==0) );
2625 if( pSorter->bUsePMA ){
2626 assert( pSorter->pReader==0 || pSorter->pMerger==0 );
2627 assert( pSorter->bUseThreads==0 || pSorter->pReader );
2628 assert( pSorter->bUseThreads==1 || pSorter->pMerger );
2629 #if SQLITE_MAX_WORKER_THREADS>0
2630 if( pSorter->bUseThreads ){
2631 rc = vdbePmaReaderNext(pSorter->pReader);
2632 *pbEof = (pSorter->pReader->pFd==0);
2633 }else
2634 #endif
2635 /*if( !pSorter->bUseThreads )*/ {
2636 assert( pSorter->pMerger!=0 );
2637 assert( pSorter->pMerger->pTask==(&pSorter->aTask[0]) );
2638 rc = vdbeMergeEngineStep(pSorter->pMerger, pbEof);
2639 }
2640 }else{
2641 SorterRecord *pFree = pSorter->list.pList;
2642 pSorter->list.pList = pFree->u.pNext;
2643 pFree->u.pNext = 0;
2644 if( pSorter->list.aMemory==0 ) vdbeSorterRecordFree(db, pFree);
2645 *pbEof = !pSorter->list.pList;
2646 rc = SQLITE_OK;
2647 }
2648 return rc;
2649 }
2650
2651 /*
2652 ** Return a pointer to a buffer owned by the sorter that contains the
2653 ** current key.
2654 */
2655 static void *vdbeSorterRowkey(
2656 const VdbeSorter *pSorter, /* Sorter object */
2657 int *pnKey /* OUT: Size of current key in bytes */
2658 ){
2659 void *pKey;
2660 if( pSorter->bUsePMA ){
2661 PmaReader *pReader;
2662 #if SQLITE_MAX_WORKER_THREADS>0
2663 if( pSorter->bUseThreads ){
2664 pReader = pSorter->pReader;
2665 }else
2666 #endif
2667 /*if( !pSorter->bUseThreads )*/{
2668 pReader = &pSorter->pMerger->aReadr[pSorter->pMerger->aTree[1]];
2669 }
2670 *pnKey = pReader->nKey;
2671 pKey = pReader->aKey;
2672 }else{
2673 *pnKey = pSorter->list.pList->nVal;
2674 pKey = SRVAL(pSorter->list.pList);
2675 }
2676 return pKey;
2677 }
2678
2679 /*
2680 ** Copy the current sorter key into the memory cell pOut.
2681 */
2682 int sqlite3VdbeSorterRowkey(const VdbeCursor *pCsr, Mem *pOut){
2683 VdbeSorter *pSorter;
2684 void *pKey; int nKey; /* Sorter key to copy into pOut */
2685
2686 assert( pCsr->eCurType==CURTYPE_SORTER );
2687 pSorter = pCsr->uc.pSorter;
2688 pKey = vdbeSorterRowkey(pSorter, &nKey);
2689 if( sqlite3VdbeMemClearAndResize(pOut, nKey) ){
2690 return SQLITE_NOMEM_BKPT;
2691 }
2692 pOut->n = nKey;
2693 MemSetTypeFlag(pOut, MEM_Blob);
2694 memcpy(pOut->z, pKey, nKey);
2695
2696 return SQLITE_OK;
2697 }
2698
2699 /*
2700 ** Compare the key in memory cell pVal with the key that the sorter cursor
2701 ** passed as the first argument currently points to. For the purposes of
2702 ** the comparison, ignore the rowid field at the end of each record.
2703 **
2704 ** If the sorter cursor key contains any NULL values, consider it to be
2705 ** less than pVal. Even if pVal also contains NULL values.
2706 **
2707 ** If an error occurs, return an SQLite error code (i.e. SQLITE_NOMEM).
2708 ** Otherwise, set *pRes to a negative, zero or positive value if the
2709 ** key in pVal is smaller than, equal to or larger than the current sorter
2710 ** key.
2711 **
2712 ** This routine forms the core of the OP_SorterCompare opcode, which in
2713 ** turn is used to verify uniqueness when constructing a UNIQUE INDEX.
2714 */
2715 int sqlite3VdbeSorterCompare(
2716 const VdbeCursor *pCsr, /* Sorter cursor */
2717 Mem *pVal, /* Value to compare to current sorter key */
2718 int nKeyCol, /* Compare this many columns */
2719 int *pRes /* OUT: Result of comparison */
2720 ){
2721 VdbeSorter *pSorter;
2722 UnpackedRecord *r2;
2723 KeyInfo *pKeyInfo;
2724 int i;
2725 void *pKey; int nKey; /* Sorter key to compare pVal with */
2726
2727 assert( pCsr->eCurType==CURTYPE_SORTER );
2728 pSorter = pCsr->uc.pSorter;
2729 r2 = pSorter->pUnpacked;
2730 pKeyInfo = pCsr->pKeyInfo;
2731 if( r2==0 ){
2732 r2 = pSorter->pUnpacked = sqlite3VdbeAllocUnpackedRecord(pKeyInfo);
2733 if( r2==0 ) return SQLITE_NOMEM_BKPT;
2734 r2->nField = nKeyCol;
2735 }
2736 assert( r2->nField==nKeyCol );
2737
2738 pKey = vdbeSorterRowkey(pSorter, &nKey);
2739 sqlite3VdbeRecordUnpack(pKeyInfo, nKey, pKey, r2);
2740 for(i=0; i<nKeyCol; i++){
2741 if( r2->aMem[i].flags & MEM_Null ){
2742 *pRes = -1;
2743 return SQLITE_OK;
2744 }
2745 }
2746
2747 *pRes = sqlite3VdbeRecordCompare(pVal->n, pVal->z, r2);
2748 return SQLITE_OK;
2749 }
OLDNEW
« no previous file with comments | « third_party/sqlite/sqlite-src-3170000/src/vdbemem.c ('k') | third_party/sqlite/sqlite-src-3170000/src/vdbetrace.c » ('j') | no next file with comments »

Powered by Google App Engine
This is Rietveld 408576698