SQLite

Check-in [d284e30eb1]
Login

Many hyperlinks are disabled.
Use anonymous login to enable hyperlinks.

Overview
Comment:Refactor local object and method names in vdbesort.c so that their names more closely reflect their actual use.
Downloads: Tarball | ZIP archive
Timelines: family | ancestors | descendants | both | threads
Files: files | file ages | folders
SHA1: d284e30eb1db144965fa85566e4234e30464350b
User & Date: drh 2014-04-03 02:54:27.677
Context
2014-04-03
14:29
Fix minor errors causing compilation to fail with SQLITE_MAX_WORKER_THREADS set to a value greater than zero. (check-in: 0561272abf user: dan tags: threads)
02:54
Refactor local object and method names in vdbesort.c so that their names more closely reflect their actual use. (check-in: d284e30eb1 user: drh tags: threads)
2014-04-02
18:58
Add a big introductory comment to vdbesort.c explaining its operation at a high level. Also adjust some symbolic names and fix other comment issues in that file. (check-in: eef60f1bf5 user: drh tags: threads)
Changes
Side-by-Side Diff Ignore Whitespace Patch
Changes to src/vdbesort.c.
87
88
89
90
91
92
93
94
95
96
97





98
99
100
101
102
103
104
105
106
107
108
109
110

111
112
113

114
115

116
117
118
119

120
121
122


123
124
125

126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145

146
147
148
149
150
151
152
153
154

155
156
157
158
159
160
161
162
163
164
165
166
167



168
169
170
171
172
173
174

175
176
177
178
179
180
181






182
183
184

185
186
187

188
189
190
191
192
193
194
87
88
89
90
91
92
93




94
95
96
97
98

99
100
101
102
103
104
105
106
107
108
109

110
111
112

113
114

115
116
117
118

119
120


121
122
123
124

125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144

145
146
147
148
149
150
151
152
153

154
155
156
157
158
159
160
161
162
163
164



165
166
167







168
169






170
171
172
173
174
175
176
177

178
179
180

181
182
183
184
185
186
187
188







-
-
-
-
+
+
+
+
+
-











-
+


-
+

-
+



-
+

-
-
+
+


-
+



















-
+








-
+










-
-
-
+
+
+
-
-
-
-
-
-
-
+

-
-
-
-
-
-
+
+
+
+
+
+


-
+


-
+







*/
#include "sqliteInt.h"
#include "vdbeInt.h"

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


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

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

  sqlite3_vfs *pVfs;              /* VFS used to open temporary files */
  KeyInfo *pKeyInfo;              /* How to compare records */
  UnpackedRecord *pUnpacked;      /* Space to unpack a record */
  int pgsz;                       /* Main database page size */

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

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


/*
** NOTES ON DATA STRUCTURE USED FOR N-WAY MERGES:
**
** As keys are added to the sorter, they are written to disk in a series
** The MergeEngine object is used to combine two or more smaller PMAs into
** one big PMA using a merge operation.  Separate PMAs all need to be
** combined into one big PMA in order to be able to step through the sorted
** of sorted packed-memory-arrays (PMAs). The size of each PMA is roughly
** the same as the cache-size allowed for temporary databases. In order
** to allow the caller to extract keys from the sorter in sorted order,
** all PMAs currently stored on disk must be merged together. This comment
** describes the data structure used to do so. The structure supports 
** merging any number of arrays in a single pass with no redundant comparison 
** operations.
** records in order.
**
** The aIter[] array contains an iterator for each of the PMAs being merged.
** An aIter[] iterator either points to a valid key or else is at EOF. For 
** the purposes of the paragraphs below, we assume that the array is actually 
** N elements in size, where N is the smallest power of 2 greater to or equal 
** to the number of iterators being merged. The extra aIter[] elements are 
** treated as if they are empty (always at EOF).
** The aIter[] array contains a PmaReader object for each of the PMAs being
** merged.  An aIter[] object either points to a valid key or else is at EOF.
** For the purposes of the paragraphs below, we assume that the array is
** actually N elements in size, where N is the smallest power of 2 greater
** to or equal to the number of PMAs being merged. The extra aIter[] elements
** are treated as if they are empty (always at EOF).
**
** The aTree[] array is also N elements in size. The value of N is stored in
** the VdbeSorter.nTree variable.
** the MergeEngine.nTree variable.
**
** The final (N/2) elements of aTree[] contain the results of comparing
** pairs of iterator keys together. Element i contains the result of 
** pairs of PMA keys together. Element i contains the result of 
** comparing aIter[2*i-N] and aIter[2*i-N+1]. Whichever key is smaller, the
** aTree element is set to the index of it. 
**
** For the purposes of this comparison, EOF is considered greater than any
** other key value. If the keys are equal (only possible with two EOF
** values), it doesn't matter which index is stored.
**
226
227
228
229
230
231
232
233
234
235
236




237
238
239
240
241
242
243
244
245
246
247
248
249

250
251
252
253
254
255


256
257
258
259
260



261
262

263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278




279


280
281

282
283
284
285
286
287
288
289
290
291
292
293


294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310

311
312
313
314

315
316
317
318
319
320
321
322
323
324
325
326
327
328

329
330
331
332

333
334
335

336
337
338
339

340
341
342
343
344
345
346
347
348
349
350
351
352


353
354
355
356
357
358
359
220
221
222
223
224
225
226




227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242

243
244
245
246
247


248
249
250
251
252


253
254
255
256

257
258
259
260
261
262
263
264
265
266
267
268
269
270



271
272
273
274

275
276
277

278
279
280
281
282
283
284
285
286
287
288


289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306

307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325

326
327
328
329

330
331
332

333
334
335
336

337
338
339
340
341
342
343
344
345
346
347
348


349
350
351
352
353
354
355
356
357







-
-
-
-
+
+
+
+












-
+




-
-
+
+



-
-
+
+
+

-
+













-
-
-
+
+
+
+
-
+
+

-
+










-
-
+
+
















-
+




+













-
+



-
+


-
+



-
+











-
-
+
+







**
**     aTree[] = { X, 0   0, 6    0, 3, 5, 6 }
**
** In other words, each time we advance to the next sorter element, log2(N)
** key comparison operations are required, where N is the number of segments
** being merged (rounded up to the next power of 2).
*/
struct SorterMerger {
  int nTree;                      /* Used size of aTree/aIter (power of 2) */
  int *aTree;                     /* Current state of incremental merge */
  VdbeSorterIter *aIter;          /* Array of iterators to merge data from */
struct MergeEngine {
  int nTree;                 /* Used size of aTree/aIter (power of 2) */
  int *aTree;                /* Current state of incremental merge */
  PmaReader *aIter;          /* Array of iterators to merge data from */
};

/*
** Main sorter structure. A single instance of this is allocated for each 
** sorter cursor created by the VDBE.
*/
struct VdbeSorter {
  int nInMemory;                  /* Current size of pRecord list as PMA */
  int mnPmaSize;                  /* Minimum PMA size, in bytes */
  int mxPmaSize;                  /* Maximum PMA size, in bytes.  0==no limit */
  int bUsePMA;                    /* True if one or more PMAs created */
  SorterRecord *pRecord;          /* Head of in-memory record list */
  SorterMerger *pMerger;          /* For final merge of PMAs (by caller) */ 
  MergeEngine *pMerger;           /* For final merge of PMAs (by caller) */ 
  u8 *aMemory;                    /* Block of memory to alloc records from */
  int iMemory;                    /* Offset of first free byte in aMemory */
  int nMemory;                    /* Size of aMemory allocation in bytes */
  int iPrev;                      /* Previous thread used to flush PMA */
  int nThread;                    /* Size of aThread[] array */
  SortSubtask aThread[1];
  int nTask;                      /* Size of aTask[] array */
  SortSubtask aTask[1];           /* One or more subtasks */
};

/*
** The following type is an iterator for a PMA. It caches the current key in 
** variables nKey/aKey. If the iterator is at EOF, pFile==0.
** An instance of the following object is used to read records out of a
** PMA, in sorted order.  The next key to be read is cached in nKey/aKey.
** pFile==0 at EOF.
*/
struct VdbeSorterIter {
struct PmaReader {
  i64 iReadOff;                   /* Current read offset */
  i64 iEof;                       /* 1 byte past EOF for this iterator */
  int nAlloc;                     /* Bytes of space at aAlloc */
  int nKey;                       /* Number of bytes in key */
  sqlite3_file *pFile;            /* File iterator is reading from */
  u8 *aAlloc;                     /* Allocated space */
  u8 *aKey;                       /* Pointer to current key */
  u8 *aBuffer;                    /* Current read buffer */
  int nBuffer;                    /* Size of read buffer in bytes */
  u8 *aMap;                       /* Pointer to mapping of entire file */
};

/*
** An instance of this structure is used to organize the stream of records
** being written to files by the merge-sort code into aligned, page-sized
** blocks.  Doing all I/O in aligned page-sized blocks helps I/O to go
** An instance of this object is used for writing a PMA.
**
** The PMA is written one record at a time.  Each record is of an arbitrary
** size.  But I/O is more efficient if it occurs in page-sized blocks where
** faster on many operating systems.
** each block is aligned on a page boundary.  This object caches writes to
** the PMA so that aligned, page-size blocks are written.
*/
struct FileWriter {
struct PmaWriter {
  int eFWErr;                     /* Non-zero if in an error state */
  u8 *aBuffer;                    /* Pointer to write buffer */
  int nBuffer;                    /* Size of write buffer in bytes */
  int iBufStart;                  /* First byte of buffer to write */
  int iBufEnd;                    /* Last byte of buffer to write */
  i64 iWriteOff;                  /* Offset of start of buffer in file */
  sqlite3_file *pFile;            /* File to write to */
};

/*
** A structure to store a single record. All in-memory records are connected
** together into a linked list headed at VdbeSorter.pRecord.
** This object is the header on a single record while that record is being
** held in memory and prior to being written out as part of a PMA.
**
** How the linked list is connected depends on how memory is being managed
** by this module. If using a separate allocation for each in-memory record
** (VdbeSorter.aMemory==0), then the list is always connected using the
** SorterRecord.u.pNext pointers.
**
** Or, if using the single large allocation method (VdbeSorter.aMemory!=0),
** then while records are being accumulated the list is linked using the
** SorterRecord.u.iNext offset. This is because the aMemory[] array may
** be sqlite3Realloc()ed while records are being accumulated. Once the VM
** has finished passing records to the sorter, or when the in-memory buffer
** is full, the list is sorted. As part of the sorting process, it is
** converted to use the SorterRecord.u.pNext pointers. See function
** vdbeSorterSort() for details.
*/
struct SorterRecord {
  int nVal;
  int nVal;                       /* Size of the record in bytes */
  union {
    SorterRecord *pNext;          /* Pointer to next record in list */
    int iNext;                    /* Offset within aMemory of next record */
  } u;
  /* The data for the record immediately follows this header */
};

/* Return a pointer to the buffer containing the record data for SorterRecord
** object p. Should be used as if:
**
**   void *SRVAL(SorterRecord *p) { return (void*)&p[1]; }
*/
#define SRVAL(p) ((void*)((SorterRecord*)(p) + 1))

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

/* Maximum number of segments to merge in a single pass. */
/* Maximum number of PMAs that a single MergeEngine can merge */
#define SORTER_MAX_MERGE_COUNT 16

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

/*
** Read nByte bytes of data from the stream of data iterated by object p.
** If successful, set *ppOut to point to a buffer containing the data
** and return SQLITE_OK. Otherwise, if an error occurs, return an SQLite
** error code.
**
** The buffer indicated by *ppOut may only be considered valid until the
** next call to this function.
*/
static int vdbeSorterIterRead(
  VdbeSorterIter *p,              /* Iterator */
static int vdbePmaReadBlob(
  PmaReader *p,                   /* Iterator */
  int nByte,                      /* Bytes of data to read */
  u8 **ppOut                      /* OUT: Pointer to buffer containing data */
){
  int iBuf;                       /* Offset within buffer to read from */
  int nAvail;                     /* Bytes of data available in buffer */

  if( p->aMap ){
415
416
417
418
419
420
421
422

423
424
425
426
427
428

429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445

446
447
448
449
450
451
452
453
454
455
456
457
458

459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474

475
476
477
478
479
480

481
482
483
484

485
486
487

488
489
490
491
492
493
494
495
496
497
498
499
500
501
502




503
504
505
506

507
508
509

510
511
512

513
514
515
516
517
518
519

520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537


538
539
540

541
542
543
544
545
546
547
548
549
550
551


552
553
554
555
556
557

558
559
560
561
562
563
564
565

566
567
568

569
570

571
572
573

574
575
576
577

578
579
580
581

582
583

584
585
586
587
588
589
590
591
592
593
594
595


596
597
598
599
600
601
602


603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623

624
625

626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641





642
643

644
645
646
647
648
649
650
413
414
415
416
417
418
419

420
421
422
423
424
425

426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442

443
444
445
446
447
448
449
450
451
452
453
454
455

456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471

472
473
474
475
476
477

478
479
480
481

482
483
484

485
486
487
488
489
490
491
492
493
494
495
496




497
498
499
500
501
502
503

504
505
506

507
508
509

510
511
512
513
514
515
516

517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533


534
535
536
537

538
539
540
541
542
543
544
545
546
547


548
549
550
551
552
553
554

555
556
557
558
559
560
561
562

563
564
565

566
567

568
569
570

571
572
573
574

575
576
577
578

579
580

581
582
583
584
585
586
587
588
589
590
591


592
593
594
595
596
597
598


599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620

621
622

623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638

639
640
641
642
643
644

645
646
647
648
649
650
651
652







-
+





-
+
















-
+












-
+















-
+





-
+



-
+


-
+











-
-
-
-
+
+
+
+



-
+


-
+


-
+






-
+
















-
-
+
+


-
+









-
-
+
+





-
+







-
+


-
+

-
+


-
+



-
+



-
+

-
+










-
-
+
+





-
-
+
+




















-
+

-
+















-
+
+
+
+
+

-
+







    memcpy(p->aAlloc, &p->aBuffer[iBuf], nAvail);
    p->iReadOff += nAvail;
    nRem = nByte - nAvail;

    /* The following loop copies up to p->nBuffer bytes per iteration into
    ** the p->aAlloc[] buffer.  */
    while( nRem>0 ){
      int rc;                     /* vdbeSorterIterRead() return code */
      int rc;                     /* vdbePmaReadBlob() return code */
      int nCopy;                  /* Number of bytes to copy */
      u8 *aNext;                  /* Pointer to buffer to copy data from */

      nCopy = nRem;
      if( nRem>p->nBuffer ) nCopy = p->nBuffer;
      rc = vdbeSorterIterRead(p, nCopy, &aNext);
      rc = vdbePmaReadBlob(p, nCopy, &aNext);
      if( rc!=SQLITE_OK ) return rc;
      assert( aNext!=p->aAlloc );
      memcpy(&p->aAlloc[nByte - nRem], aNext, nCopy);
      nRem -= nCopy;
    }

    *ppOut = p->aAlloc;
  }

  return SQLITE_OK;
}

/*
** Read a varint from the stream of data accessed by p. Set *pnOut to
** the value read.
*/
static int vdbeSorterIterVarint(VdbeSorterIter *p, u64 *pnOut){
static int vdbePmaReadVarint(PmaReader *p, u64 *pnOut){
  int iBuf;

  if( p->aMap ){
    p->iReadOff += sqlite3GetVarint(&p->aMap[p->iReadOff], pnOut);
  }else{
    iBuf = p->iReadOff % p->nBuffer;
    if( iBuf && (p->nBuffer-iBuf)>=9 ){
      p->iReadOff += sqlite3GetVarint(&p->aBuffer[iBuf], pnOut);
    }else{
      u8 aVarint[16], *a;
      int i = 0, rc;
      do{
        rc = vdbeSorterIterRead(p, 1, &a);
        rc = vdbePmaReadBlob(p, 1, &a);
        if( rc ) return rc;
        aVarint[(i++)&0xf] = a[0];
      }while( (a[0]&0x80)!=0 );
      sqlite3GetVarint(aVarint, pnOut);
    }
  }

  return SQLITE_OK;
}


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

  if( pIter->iReadOff>=pIter->iEof ){
    /* This is an EOF condition */
    vdbeSorterIterZero(pIter);
    vdbePmaReaderClear(pIter);
    return SQLITE_OK;
  }

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

  return rc;
}

/*
** Initialize iterator pIter to scan through the PMA stored in file pFile
** starting at offset iStart and ending at offset iEof-1. This function 
** leaves the iterator pointing to the first key in the PMA (or EOF if the 
** PMA is empty).
*/
static int vdbeSorterIterInit(
  SortSubtask *pThread,           /* Thread context */
  i64 iStart,                     /* Start offset in pThread->pTemp1 */
  VdbeSorterIter *pIter,          /* Iterator to populate */
static int vdbePmaReaderInit(
  SortSubtask *pTask,             /* Thread context */
  i64 iStart,                     /* Start offset in pTask->pTemp1 */
  PmaReader *pIter,               /* Iterator to populate */
  i64 *pnByte                     /* IN/OUT: Increment this value by PMA size */
){
  int rc = SQLITE_OK;
  int nBuf = pThread->pgsz;
  int nBuf = pTask->pgsz;
  void *pMap = 0;                 /* Mapping of temp file */

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

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

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

  if( rc==SQLITE_OK ){
    rc = vdbeSorterIterNext(pIter);
    rc = vdbePmaReaderNext(pIter);
  }
  return rc;
}


/*
** Compare key1 (buffer pKey1, size nKey1 bytes) with key2 (buffer pKey2, 
** size nKey2 bytes). Use (pThread->pKeyInfo) for the collation sequences
** size nKey2 bytes). Use (pTask->pKeyInfo) for the collation sequences
** used by the comparison. Return the result of the comparison.
**
** Before returning, object (pThread->pUnpacked) is populated with the
** Before returning, object (pTask->pUnpacked) is populated with the
** unpacked version of key2. Or, if pKey2 is passed a NULL pointer, then it 
** is assumed that the (pThread->pUnpacked) structure already contains the 
** is assumed that the (pTask->pUnpacked) structure already contains the 
** unpacked key to use as key2.
**
** If an OOM error is encountered, (pThread->pUnpacked->error_rc) is set
** If an OOM error is encountered, (pTask->pUnpacked->error_rc) is set
** to SQLITE_NOMEM.
*/
static int vdbeSorterCompare(
  SortSubtask *pThread,           /* Thread context (for pKeyInfo) */
  SortSubtask *pTask,             /* Subtask context (for pKeyInfo) */
  const void *pKey1, int nKey1,   /* Left side of comparison */
  const void *pKey2, int nKey2    /* Right side of comparison */
){
  UnpackedRecord *r2 = pThread->pUnpacked;
  UnpackedRecord *r2 = pTask->pUnpacked;
  if( pKey2 ){
    sqlite3VdbeRecordUnpack(pThread->pKeyInfo, nKey2, pKey2, r2);
    sqlite3VdbeRecordUnpack(pTask->pKeyInfo, nKey2, pKey2, r2);
  }
  return sqlite3VdbeRecordCompare(nKey1, pKey1, r2, 0);
}

/*
** This function is called to compare two iterator keys when merging 
** multiple b-tree segments. Parameter iOut is the index of the aTree[] 
** value to recalculate.
*/
static int vdbeSorterDoCompare(
  SortSubtask *pThread, 
  SorterMerger *pMerger, 
  SortSubtask *pTask, 
  MergeEngine *pMerger, 
  int iOut
){
  int i1;
  int i2;
  int iRes;
  VdbeSorterIter *p1;
  VdbeSorterIter *p2;
  PmaReader *p1;
  PmaReader *p2;

  assert( iOut<pMerger->nTree && iOut>0 );

  if( iOut>=(pMerger->nTree/2) ){
    i1 = (iOut - pMerger->nTree/2) * 2;
    i2 = i1 + 1;
  }else{
    i1 = pMerger->aTree[iOut*2];
    i2 = pMerger->aTree[iOut*2+1];
  }

  p1 = &pMerger->aIter[i1];
  p2 = &pMerger->aIter[i2];

  if( p1->pFile==0 ){
    iRes = i2;
  }else if( p2->pFile==0 ){
    iRes = i1;
  }else{
    int res;
    assert( pThread->pUnpacked!=0 );  /* allocated in vdbeSortSubtaskMain() */
    assert( pTask->pUnpacked!=0 );  /* allocated in vdbeSortSubtaskMain() */
    res = vdbeSorterCompare(
        pThread, p1->aKey, p1->nKey, p2->aKey, p2->nKey
        pTask, p1->aKey, p1->nKey, p2->aKey, p2->nKey
    );
    if( res<=0 ){
      iRes = i1;
    }else{
      iRes = i2;
    }
  }

  pMerger->aTree[iOut] = iRes;
  return SQLITE_OK;
}

/*
** Initialize the temporary index cursor just opened as a sorter cursor.
*/
int sqlite3VdbeSorterInit(sqlite3 *db, int nField, VdbeCursor *pCsr){
int sqlite3VdbeSorterInit(
  sqlite3 *db,                    /* Database connection (for malloc()) */
  int nField,                     /* Number of key fields in each record */
  VdbeCursor *pCsr                /* Cursor that holds the new sorter */
){
  int pgsz;                       /* Page size of main database */
  int i;                          /* Used to iterate through aThread[] */
  int i;                          /* Used to iterate through aTask[] */
  int mxCache;                    /* Cache size */
  VdbeSorter *pSorter;            /* The new sorter */
  KeyInfo *pKeyInfo;              /* Copy of pCsr->pKeyInfo with db==0 */
  int szKeyInfo;                  /* Size of pCsr->pKeyInfo in bytes */
  int sz;                         /* Size of pSorter in bytes */
  int rc = SQLITE_OK;
  int nWorker = (sqlite3GlobalConfig.bCoreMutex?sqlite3GlobalConfig.nWorker:0);
660
661
662
663
664
665
666
667
668
669
670
671
672






673
674
675
676
677
678
679
662
663
664
665
666
667
668






669
670
671
672
673
674
675
676
677
678
679
680
681







-
-
-
-
-
-
+
+
+
+
+
+







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

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

    if( !sqlite3TempInMemory(db) ){
      pSorter->mnPmaSize = SORTER_MIN_WORKING * pgsz;
      mxCache = db->aDb[0].pSchema->cache_size;
      if( mxCache<SORTER_MIN_WORKING ) mxCache = SORTER_MIN_WORKING;
      pSorter->mxPmaSize = mxCache * pgsz;
702
703
704
705
706
707
708
709
710


711
712
713
714
715
716





717
718
719


720
721
722
723
724




725
726
727
728
729
730
731
732
733
734
735
736
737



738
739
740
741



742
743
744
745
746
747
748
749
750
751
752
753

754
755

756
757
758

759
760
761
762

763
764

765
766
767

768
769
770
771
772
773
774

775
776

777
778
779
780

781
782
783
784
785
786
787
788
789
790
791
792


793
794
795
796



797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814

815
816
817
818
819
820
821
704
705
706
707
708
709
710


711
712
713





714
715
716
717
718
719


720
721
722




723
724
725
726
727
728
729
730
731
732
733
734
735
736



737
738
739
740



741
742
743
744
745
746
747
748
749
750
751
752
753
754

755
756

757
758
759

760
761
762
763

764
765

766
767
768

769
770
771
772
773
774
775

776
777

778
779
780
781

782
783
784
785
786
787
788
789
790
791
792


793
794
795



796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815

816
817
818
819
820
821
822
823







-
-
+
+

-
-
-
-
-
+
+
+
+
+

-
-
+
+

-
-
-
-
+
+
+
+










-
-
-
+
+
+

-
-
-
+
+
+











-
+

-
+


-
+



-
+

-
+


-
+






-
+

-
+



-
+










-
-
+
+

-
-
-
+
+
+

















-
+







  for(p=pRecord; p; p=pNext){
    pNext = p->u.pNext;
    sqlite3DbFree(db, p);
  }
}

/*
** Free all resources owned by the object indicated by argument pThread. All 
** fields of *pThread are zeroed before returning.
** Free all resources owned by the object indicated by argument pTask. All 
** fields of *pTask are zeroed before returning.
*/
static void vdbeSortSubtaskCleanup(sqlite3 *db, SortSubtask *pThread){
  sqlite3DbFree(db, pThread->pUnpacked);
  pThread->pUnpacked = 0;
  if( pThread->aListMemory==0 ){
    vdbeSorterRecordFree(0, pThread->pList);
static void vdbeSortSubtaskCleanup(sqlite3 *db, SortSubtask *pTask){
  sqlite3DbFree(db, pTask->pUnpacked);
  pTask->pUnpacked = 0;
  if( pTask->aListMemory==0 ){
    vdbeSorterRecordFree(0, pTask->pList);
  }else{
    sqlite3_free(pThread->aListMemory);
    pThread->aListMemory = 0;
    sqlite3_free(pTask->aListMemory);
    pTask->aListMemory = 0;
  }
  pThread->pList = 0;
  if( pThread->pTemp1 ){
    sqlite3OsCloseFree(pThread->pTemp1);
    pThread->pTemp1 = 0;
  pTask->pList = 0;
  if( pTask->pTemp1 ){
    sqlite3OsCloseFree(pTask->pTemp1);
    pTask->pTemp1 = 0;
  }
}

/*
** Join all threads.  
*/
#if SQLITE_MAX_WORKER_THREADS>0
static int vdbeSorterJoinAll(VdbeSorter *pSorter, int rcin){
  int rc = rcin;
  int i;
  for(i=0; i<pSorter->nThread; i++){
    SortSubtask *pThread = &pSorter->aThread[i];
    if( pThread->pThread ){
  for(i=0; i<pSorter->nTask; i++){
    SortSubtask *pTask = &pSorter->aTask[i];
    if( pTask->pTask ){
      void *pRet;
      int rc2 = sqlite3ThreadJoin(pThread->pThread, &pRet);
      pThread->pThread = 0;
      pThread->bDone = 0;
      int rc2 = sqlite3ThreadJoin(pTask->pTask, &pRet);
      pTask->pTask = 0;
      pTask->bDone = 0;
      if( rc==SQLITE_OK ) rc = rc2;
      if( rc==SQLITE_OK ) rc = SQLITE_PTR_TO_INT(pRet);
    }
  }
  return rc;
}
#else
# define vdbeSorterJoinAll(x,rcin) (rcin)
#endif

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

  assert( nIter<=SORTER_MAX_MERGE_COUNT );
  while( N<nIter ) N += N;
  nByte = sizeof(SorterMerger) + N * (sizeof(int) + sizeof(VdbeSorterIter));
  nByte = sizeof(MergeEngine) + N * (sizeof(int) + sizeof(PmaReader));

  pNew = (SorterMerger*)sqlite3MallocZero(nByte);
  pNew = (MergeEngine*)sqlite3MallocZero(nByte);
  if( pNew ){
    pNew->nTree = N;
    pNew->aIter = (VdbeSorterIter*)&pNew[1];
    pNew->aIter = (PmaReader*)&pNew[1];
    pNew->aTree = (int*)&pNew->aIter[N];
  }
  return pNew;
}

/*
** Free the SorterMerger object passed as the only argument.
** Free the MergeEngine object passed as the only argument.
*/
static void vdbeSorterMergerFree(SorterMerger *pMerger){
static void vdbeMergeEngineFree(MergeEngine *pMerger){
  int i;
  if( pMerger ){
    for(i=0; i<pMerger->nTree; i++){
      vdbeSorterIterZero(&pMerger->aIter[i]);
      vdbePmaReaderClear(&pMerger->aIter[i]);
    }
  }
  sqlite3_free(pMerger);
}

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

/*
** Free any cursor components allocated by sqlite3VdbeSorterXXX routines.
*/
void sqlite3VdbeSorterClose(sqlite3 *db, VdbeCursor *pCsr){
  VdbeSorter *pSorter = pCsr->pSorter;
  if( pSorter ){
    sqlite3VdbeSorterReset(db, pSorter);
    vdbeSorterMergerFree(pSorter->pMerger);
    vdbeMergeEngineFree(pSorter->pMerger);
    sqlite3_free(pSorter->aMemory);
    sqlite3DbFree(db, pSorter);
    pCsr->pSorter = 0;
  }
}

/*
838
839
840
841
842
843
844
845

846
847
848
849
850
851
852
853
854
855
856

857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875

876
877
878
879

880
881
882
883
884
885
886
887
888
889

890
891
892
893


894
895
896
897


898
899
900
901
902
903
904
905

906
907
908
909
910
911
912
913
914

915
916

917
918
919
920
921
922
923

924
925

926
927

928
929
930
931

932
933
934
935
936
937
938
939
940
941
942
943
944

945
946
947

948
949
950
951
952
953
954
840
841
842
843
844
845
846

847
848
849
850
851
852
853
854
855
856
857

858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876

877
878
879
880

881
882
883
884
885
886
887
888
889
890

891
892
893


894
895
896
897


898
899
900
901
902
903
904
905
906

907
908
909
910
911
912
913
914
915

916
917

918
919
920
921
922
923
924

925
926

927
928

929
930
931
932

933
934
935
936
937
938
939
940
941
942
943
944
945

946
947
948

949
950
951
952
953
954
955
956







-
+










-
+


















-
+



-
+









-
+


-
-
+
+


-
-
+
+







-
+








-
+

-
+






-
+

-
+

-
+



-
+












-
+


-
+







}

/*
** Merge the two sorted lists p1 and p2 into a single list.
** Set *ppOut to the head of the new list.
*/
static void vdbeSorterMerge(
  SortSubtask *pThread,          /* Calling thread context */
  SortSubtask *pTask,             /* Calling thread context */
  SorterRecord *p1,               /* First list to merge */
  SorterRecord *p2,               /* Second list to merge */
  SorterRecord **ppOut            /* OUT: Head of merged list */
){
  SorterRecord *pFinal = 0;
  SorterRecord **pp = &pFinal;
  void *pVal2 = p2 ? SRVAL(p2) : 0;

  while( p1 && p2 ){
    int res;
    res = vdbeSorterCompare(pThread, SRVAL(p1), p1->nVal, pVal2, p2->nVal);
    res = vdbeSorterCompare(pTask, SRVAL(p1), p1->nVal, pVal2, p2->nVal);
    if( res<=0 ){
      *pp = p1;
      pp = &p1->u.pNext;
      p1 = p1->u.pNext;
      pVal2 = 0;
    }else{
      *pp = p2;
       pp = &p2->u.pNext;
      p2 = p2->u.pNext;
      if( p2==0 ) break;
      pVal2 = SRVAL(p2);
    }
  }
  *pp = p1 ? p1 : p2;
  *ppOut = pFinal;
}

/*
** Sort the linked list of records headed at pThread->pList. Return 
** Sort the linked list of records headed at pTask->pList. Return 
** SQLITE_OK if successful, or an SQLite error code (i.e. SQLITE_NOMEM) if 
** an error occurs.
*/
static int vdbeSorterSort(SortSubtask *pThread){
static int vdbeSorterSort(SortSubtask *pTask){
  int i;
  SorterRecord **aSlot;
  SorterRecord *p;

  aSlot = (SorterRecord **)sqlite3MallocZero(64 * sizeof(SorterRecord *));
  if( !aSlot ){
    return SQLITE_NOMEM;
  }

  p = pThread->pList;
  p = pTask->pList;
  while( p ){
    SorterRecord *pNext;
    if( pThread->aListMemory ){
      if( (u8*)p==pThread->aListMemory ){
    if( pTask->aListMemory ){
      if( (u8*)p==pTask->aListMemory ){
        pNext = 0;
      }else{
        assert( p->u.iNext<sqlite3MallocSize(pThread->aListMemory) );
        pNext = (SorterRecord*)&pThread->aListMemory[p->u.iNext];
        assert( p->u.iNext<sqlite3MallocSize(pTask->aListMemory) );
        pNext = (SorterRecord*)&pTask->aListMemory[p->u.iNext];
      }
    }else{
      pNext = p->u.pNext;
    }

    p->u.pNext = 0;
    for(i=0; aSlot[i]; i++){
      vdbeSorterMerge(pThread, p, aSlot[i], &p);
      vdbeSorterMerge(pTask, p, aSlot[i], &p);
      aSlot[i] = 0;
    }
    aSlot[i] = p;
    p = pNext;
  }

  p = 0;
  for(i=0; i<64; i++){
    vdbeSorterMerge(pThread, p, aSlot[i], &p);
    vdbeSorterMerge(pTask, p, aSlot[i], &p);
  }
  pThread->pList = p;
  pTask->pList = p;

  sqlite3_free(aSlot);
  return SQLITE_OK;
}

/*
** Initialize a file-writer object.
** Initialize a PMA-writer object.
*/
static void fileWriterInit(
static void vdbePmaWriterInit(
  sqlite3_file *pFile,            /* File to write to */
  FileWriter *p,                  /* Object to populate */
  PmaWriter *p,                   /* Object to populate */
  int nBuf,                       /* Buffer size */
  i64 iStart                      /* Offset of pFile to begin writing at */
){
  memset(p, 0, sizeof(FileWriter));
  memset(p, 0, sizeof(PmaWriter));
  p->aBuffer = (u8*)sqlite3Malloc(nBuf);
  if( !p->aBuffer ){
    p->eFWErr = SQLITE_NOMEM;
  }else{
    p->iBufEnd = p->iBufStart = (iStart % nBuf);
    p->iWriteOff = iStart - p->iBufStart;
    p->nBuffer = nBuf;
    p->pFile = pFile;
  }
}

/*
** Write nData bytes of data to the file-write object. Return SQLITE_OK
** Write nData bytes of data to the PMA. Return SQLITE_OK
** if successful, or an SQLite error code if an error occurs.
*/
static void fileWriterWrite(FileWriter *p, u8 *pData, int nData){
static void vdbePmaWriteBlob(PmaWriter *p, u8 *pData, int nData){
  int nRem = nData;
  while( nRem>0 && p->eFWErr==0 ){
    int nCopy = nRem;
    if( nCopy>(p->nBuffer - p->iBufEnd) ){
      nCopy = p->nBuffer - p->iBufEnd;
    }

965
966
967
968
969
970
971
972
973


974
975
976
977
978
979
980

981
982
983
984
985
986
987
988
989
990
991

992
993
994
995
996

997
998
999

1000
1001
1002
1003

1004
1005
1006
1007
1008
1009
1010
967
968
969
970
971
972
973


974
975
976
977
978
979
980
981

982
983
984
985
986
987
988
989
990
991
992

993
994
995
996
997

998
999
1000

1001
1002
1003
1004

1005
1006
1007
1008
1009
1010
1011
1012







-
-
+
+






-
+










-
+




-
+


-
+



-
+







    assert( p->iBufEnd<p->nBuffer );

    nRem -= nCopy;
  }
}

/*
** Flush any buffered data to disk and clean up the file-writer object.
** The results of using the file-writer after this call are undefined.
** Flush any buffered data to disk and clean up the PMA-writer object.
** The results of using the PMA-writer after this call are undefined.
** Return SQLITE_OK if flushing the buffered data succeeds or is not 
** required. Otherwise, return an SQLite error code.
**
** Before returning, set *piEof to the offset immediately following the
** last byte written to the file.
*/
static int fileWriterFinish(FileWriter *p, i64 *piEof){
static int vdbePmaWriterFinish(PmaWriter *p, i64 *piEof){
  int rc;
  if( p->eFWErr==0 && ALWAYS(p->aBuffer) && p->iBufEnd>p->iBufStart ){
    p->eFWErr = sqlite3OsWrite(p->pFile, 
        &p->aBuffer[p->iBufStart], p->iBufEnd - p->iBufStart, 
        p->iWriteOff + p->iBufStart
    );
  }
  *piEof = (p->iWriteOff + p->iBufEnd);
  sqlite3_free(p->aBuffer);
  rc = p->eFWErr;
  memset(p, 0, sizeof(FileWriter));
  memset(p, 0, sizeof(PmaWriter));
  return rc;
}

/*
** Write value iVal encoded as a varint to the file-write object. Return 
** Write value iVal encoded as a varint to the PMA. Return 
** SQLITE_OK if successful, or an SQLite error code if an error occurs.
*/
static void fileWriterWriteVarint(FileWriter *p, u64 iVal){
static void vdbePmaWriteVarint(PmaWriter *p, u64 iVal){
  int nByte; 
  u8 aByte[10];
  nByte = sqlite3PutVarint(aByte, iVal);
  fileWriterWrite(p, aByte, nByte);
  vdbePmaWriteBlob(p, aByte, nByte);
}

#if SQLITE_MAX_MMAP_SIZE>0
/*
** The first argument is a file-handle open on a temporary file. The file
** is guaranteed to be nByte bytes or smaller in size. This function
** attempts to extend the file to nByte bytes in size and to ensure that
1036
1037
1038
1039
1040
1041
1042
1043

1044
1045

1046
1047
1048


1049
1050
1051
1052
1053
1054
1055





1056
1057
1058
1059
1060
1061

1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072





1073
1074
1075
1076



1077
1078
1079


1080
1081
1082

1083
1084
1085
1086
1087

1088
1089
1090
1091
1092
1093
1094
1095


1096
1097
1098
1099
1100
1101
1102

1103
1104
1105
1106
1107
1108


1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125

1126
1127
1128
1129
1130
1131
1132

1133
1134
1135
1136
1137

1138
1139
1140
1141
1142
1143
1144
1038
1039
1040
1041
1042
1043
1044

1045
1046

1047
1048


1049
1050
1051
1052





1053
1054
1055
1056
1057
1058
1059
1060
1061
1062

1063
1064
1065
1066
1067
1068
1069
1070




1071
1072
1073
1074
1075
1076



1077
1078
1079
1080


1081
1082
1083
1084

1085
1086
1087
1088
1089

1090
1091
1092
1093
1094
1095
1096


1097
1098
1099
1100
1101
1102
1103
1104

1105
1106
1107
1108
1109


1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127

1128
1129
1130
1131
1132
1133
1134

1135
1136
1137
1138
1139

1140
1141
1142
1143
1144
1145
1146
1147







-
+

-
+

-
-
+
+


-
-
-
-
-
+
+
+
+
+





-
+







-
-
-
-
+
+
+
+
+

-
-
-
+
+
+

-
-
+
+


-
+




-
+






-
-
+
+






-
+




-
-
+
+
















-
+






-
+




-
+







**     * A varint. This varint contains the total number of bytes of content
**       in the PMA (not including the varint itself).
**
**     * One or more records packed end-to-end in order of ascending keys. 
**       Each record consists of a varint followed by a blob of data (the 
**       key). The varint is the number of bytes in the blob of data.
*/
static int vdbeSorterListToPMA(SortSubtask *pThread){
static int vdbeSorterListToPMA(SortSubtask *pTask){
  int rc = SQLITE_OK;             /* Return code */
  FileWriter writer;              /* Object used to write to the file */
  PmaWriter writer;               /* Object used to write to the file */

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

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

  /* Try to get the file to memory map */
  if( rc==SQLITE_OK ){
    rc = vdbeSorterExtendFile(
        pThread->pTemp1, pThread->iTemp1Off + pThread->nInMemory + 9
        pTask->pTemp1, pTask->iTemp1Off + pTask->nInMemory + 9
    );
  }

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

    fileWriterInit(pThread->pTemp1, &writer, pThread->pgsz, pThread->iTemp1Off);
    pThread->nPMA++;
    fileWriterWriteVarint(&writer, pThread->nInMemory);
    for(p=pThread->pList; p; p=pNext){
    vdbePmaWriterInit(pTask->pTemp1, &writer, pTask->pgsz,
                      pTask->iTemp1Off);
    pTask->nPMA++;
    vdbePmaWriteVarint(&writer, pTask->nInMemory);
    for(p=pTask->pList; p; p=pNext){
      pNext = p->u.pNext;
      fileWriterWriteVarint(&writer, p->nVal);
      fileWriterWrite(&writer, SRVAL(p), p->nVal);
      if( pThread->aListMemory==0 ) sqlite3_free(p);
      vdbePmaWriteVarint(&writer, p->nVal);
      vdbePmaWriteBlob(&writer, SRVAL(p), p->nVal);
      if( pTask->aListMemory==0 ) sqlite3_free(p);
    }
    pThread->pList = p;
    rc = fileWriterFinish(&writer, &pThread->iTemp1Off);
    pTask->pList = p;
    rc = vdbePmaWriterFinish(&writer, &pTask->iTemp1Off);
  }

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

/*
** Advance the SorterMerger iterator passed as the second argument to
** Advance the MergeEngine iterator passed as the second argument to
** the next entry. Set *pbEof to true if this means the iterator has 
** reached EOF.
**
** Return SQLITE_OK if successful or an error code if an error occurs.
*/
static int vdbeSorterNext(
  SortSubtask *pThread, 
  SorterMerger *pMerger, 
  SortSubtask *pTask, 
  MergeEngine *pMerger, 
  int *pbEof
){
  int rc;
  int iPrev = pMerger->aTree[1];/* Index of iterator to advance */

  /* Advance the current iterator */
  rc = vdbeSorterIterNext(&pMerger->aIter[iPrev]);
  rc = vdbePmaReaderNext(&pMerger->aIter[iPrev]);

  /* Update contents of aTree[] */
  if( rc==SQLITE_OK ){
    int i;                      /* Index of aTree[] to recalculate */
    VdbeSorterIter *pIter1;     /* First iterator to compare */
    VdbeSorterIter *pIter2;     /* Second iterator to compare */
    PmaReader *pIter1;     /* First iterator to compare */
    PmaReader *pIter2;     /* Second iterator to compare */
    u8 *pKey2;                  /* To pIter2->aKey, or 0 if record cached */

    /* Find the first two iterators to compare. The one that was just
    ** advanced (iPrev) and the one next to it in the array.  */
    pIter1 = &pMerger->aIter[(iPrev & 0xFFFE)];
    pIter2 = &pMerger->aIter[(iPrev | 0x0001)];
    pKey2 = pIter2->aKey;

    for(i=(pMerger->nTree+iPrev)/2; i>0; i=i/2){
      /* Compare pIter1 and pIter2. Store the result in variable iRes. */
      int iRes;
      if( pIter1->pFile==0 ){
        iRes = +1;
      }else if( pIter2->pFile==0 ){
        iRes = -1;
      }else{
        iRes = vdbeSorterCompare(pThread, 
        iRes = vdbeSorterCompare(pTask, 
            pIter1->aKey, pIter1->nKey, pKey2, pIter2->nKey
        );
      }

      /* If pIter1 contained the smaller value, set aTree[i] to its index.
      ** Then set pIter2 to the next iterator to compare to pIter1. In this
      ** case there is no cache of pIter2 in pThread->pUnpacked, so set
      ** case there is no cache of pIter2 in pTask->pUnpacked, so set
      ** pKey2 to point to the record belonging to pIter2.
      **
      ** Alternatively, if pIter2 contains the smaller of the two values,
      ** set aTree[i] to its index and update pIter1. If vdbeSorterCompare()
      ** was actually called above, then pThread->pUnpacked now contains
      ** was actually called above, then pTask->pUnpacked now contains
      ** a value equivalent to pIter2. So set pKey2 to NULL to prevent
      ** vdbeSorterCompare() from decoding pIter2 again.
      **
      ** If the two values were equal, then the value from the oldest
      ** PMA should be considered smaller. The VdbeSorter.aIter[] array
      ** is sorted from oldest to newest, so pIter1 contains older values
      ** than pIter2 iff (pIter1<pIter2).  */
1159
1160
1161
1162
1163
1164
1165
1166

1167
1168
1169
1170



1171
1172

1173
1174

1175
1176
1177


1178
1179

1180
1181
1182
1183
1184
1185


1186
1187
1188
1189
1190
1191




1192
1193

1194
1195
1196
1197
1198
1199

1200
1201
1202
1203
1204
1205
1206

1207
1208

1209
1210
1211

1212
1213
1214
1215
1216
1217
1218


1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229


1230
1231

1232
1233
1234

1235
1236
1237
1238


1239
1240

1241
1242
1243
1244



1245
1246

1247
1248
1249
1250
1251
1252
1253
1254





1255
1256
1257
1258


1259
1260
1261

1262
1263
1264
1265



1266
1267
1268


1269
1270
1271
1272
1273
1274
1275



1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288




1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303


1304
1305
1306
1307
1308

1309
1310

1311
1312
1313
1314
1315




1316
1317
1318
1319
1320
1321
1322


1323
1324
1325


1326
1327

1328
1329
1330
1331
1332
1333




1334
1335
1336
1337
1338
1339


1340
1341
1342
1343
1344

1345
1346
1347
1348



1349
1350
1351
1352
1353
1354
1355
1356

1357
1358
1359
1360
1361

1362
1363
1364


1365
1366

1367
1368
1369
1370
1371
1372
1373
1162
1163
1164
1165
1166
1167
1168

1169
1170



1171
1172
1173
1174

1175
1176

1177
1178


1179
1180
1181

1182
1183
1184
1185
1186


1187
1188
1189
1190




1191
1192
1193
1194
1195

1196
1197
1198
1199
1200
1201

1202
1203
1204
1205
1206
1207
1208

1209
1210

1211
1212
1213

1214
1215
1216
1217
1218
1219


1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230


1231
1232
1233

1234
1235
1236

1237
1238
1239


1240
1241
1242

1243
1244



1245
1246
1247
1248

1249
1250
1251
1252





1253
1254
1255
1256
1257
1258
1259


1260
1261
1262
1263

1264
1265



1266
1267
1268
1269


1270
1271
1272
1273
1274
1275



1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287




1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304


1305
1306
1307
1308
1309
1310

1311
1312

1313
1314




1315
1316
1317
1318
1319
1320
1321
1322
1323


1324
1325
1326


1327
1328
1329

1330
1331
1332




1333
1334
1335
1336
1337
1338
1339
1340


1341
1342
1343
1344
1345
1346

1347
1348



1349
1350
1351
1352
1353
1354
1355
1356
1357
1358

1359
1360
1361
1362
1363

1364
1365


1366
1367
1368

1369
1370
1371
1372
1373
1374
1375
1376







-
+

-
-
-
+
+
+

-
+

-
+

-
-
+
+

-
+




-
-
+
+


-
-
-
-
+
+
+
+

-
+





-
+






-
+

-
+


-
+





-
-
+
+









-
-
+
+

-
+


-
+


-
-
+
+

-
+

-
-
-
+
+
+

-
+



-
-
-
-
-
+
+
+
+
+


-
-
+
+


-
+

-
-
-
+
+
+

-
-
+
+




-
-
-
+
+
+









-
-
-
-
+
+
+
+













-
-
+
+




-
+

-
+

-
-
-
-
+
+
+
+





-
-
+
+

-
-
+
+

-
+


-
-
-
-
+
+
+
+




-
-
+
+




-
+

-
-
-
+
+
+







-
+




-
+

-
-
+
+

-
+







}

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

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

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

  if( pThread->eWork==SORT_SUBTASK_CONS ){
    assert( pThread->pList==0 );
    while( pThread->nPMA>pThread->nConsolidate && rc==SQLITE_OK ){
      int nIter = MIN(pThread->nPMA, SORTER_MAX_MERGE_COUNT);
  if( pTask->eWork==SORT_SUBTASK_CONS ){
    assert( pTask->pList==0 );
    while( pTask->nPMA>pTask->nConsolidate && rc==SQLITE_OK ){
      int nIter = MIN(pTask->nPMA, SORTER_MAX_MERGE_COUNT);
      sqlite3_file *pTemp2 = 0;     /* Second temp file to use */
      SorterMerger *pMerger;        /* Object for reading/merging PMA data */
      MergeEngine *pMerger;         /* Object for reading/merging PMA data */
      i64 iReadOff = 0;             /* Offset in pTemp1 to read from */
      i64 iWriteOff = 0;            /* Offset in pTemp2 to write to */
      int i;
      
      /* Allocate a merger object to merge PMAs together. */
      pMerger = vdbeSorterMergerNew(nIter);
      pMerger = vdbeMergeEngineNew(nIter);
      if( pMerger==0 ){
        rc = SQLITE_NOMEM;
        break;
      }

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

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

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

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

      vdbeSorterMergerFree(pMerger);
      sqlite3OsCloseFree(pThread->pTemp1);
      pThread->pTemp1 = pTemp2;
      pThread->nPMA = (i / SORTER_MAX_MERGE_COUNT);
      pThread->iTemp1Off = iWriteOff;
      vdbeMergeEngineFree(pMerger);
      sqlite3OsCloseFree(pTask->pTemp1);
      pTask->pTemp1 = pTemp2;
      pTask->nPMA = (i / SORTER_MAX_MERGE_COUNT);
      pTask->iTemp1Off = iWriteOff;
    }
  }else{
    /* Sort the pThread->pList list */
    rc = vdbeSorterSort(pThread);
    /* Sort the pTask->pList list */
    rc = vdbeSorterSort(pTask);

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

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

/*
** Run the activity scheduled by the object passed as the only argument
** in the current thread.
*/
static int vdbeSorterRunThread(SortSubtask *pThread){
  int rc = SQLITE_PTR_TO_INT( vdbeSortSubtaskMain((void*)pThread) );
  assert( pThread->bDone );
  pThread->bDone = 0;
static int vdbeSorterRunTask(SortSubtask *pTask){
  int rc = SQLITE_PTR_TO_INT( vdbeSortSubtaskMain((void*)pTask) );
  assert( pTask->bDone );
  pTask->bDone = 0;
  return rc;
}

/*
** Flush the current contents of VdbeSorter.pRecord to a new PMA, possibly
** using a background thread.
**
** If argument bFg is non-zero, the operation always uses the calling thread.
*/
static int vdbeSorterFlushPMA(sqlite3 *db, const VdbeCursor *pCsr, int bFg){
  VdbeSorter *pSorter = pCsr->pSorter;
  int rc = SQLITE_OK;
  int i;
  SortSubtask *pThread = 0;    /* Thread context used to create new PMA */
  int nWorker = (pSorter->nThread-1);
  SortSubtask *pTask = 0;    /* Thread context used to create new PMA */
  int nWorker = (pSorter->nTask-1);

  pSorter->bUsePMA = 1;
  for(i=0; i<nWorker; i++){
    int iTest = (pSorter->iPrev + i + 1) % nWorker;
    pThread = &pSorter->aThread[iTest];
    pTask = &pSorter->aTask[iTest];
#if SQLITE_MAX_WORKER_THREADS>0
    if( pThread->bDone ){
    if( pTask->bDone ){
      void *pRet;
      assert( pThread->pThread );
      rc = sqlite3ThreadJoin(pThread->pThread, &pRet);
      pThread->pThread = 0;
      pThread->bDone = 0;
      assert( pTask->pTask );
      rc = sqlite3ThreadJoin(pTask->pTask, &pRet);
      pTask->pTask = 0;
      pTask->bDone = 0;
      if( rc==SQLITE_OK ){
        rc = SQLITE_PTR_TO_INT(pRet);
      }
    }
#endif
    if( pThread->pThread==0 ) break;
    pThread = 0;
    if( pTask->pThread==0 ) break;
    pTask = 0;
  }
  if( pThread==0 ){
    pThread = &pSorter->aThread[nWorker];
  if( pTask==0 ){
    pTask = &pSorter->aTask[nWorker];
  }
  pSorter->iPrev = (pThread - pSorter->aThread);
  pSorter->iPrev = (pTask - pSorter->aTask);

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

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

#if SQLITE_MAX_WORKER_THREADS>0
    if( !bFg && pThread!=&pSorter->aThread[nWorker] ){
    if( !bFg && pTask!=&pSorter->aTask[nWorker] ){
      /* Launch a background thread for this operation */
      void *pCtx = (void*)pThread;
      assert( pSorter->aMemory==0 || pThread->aListMemory!=0 );
      if( pThread->aListMemory ){
      void *pCtx = (void*)pTask;
      assert( pSorter->aMemory==0 || pTask->aListMemory!=0 );
      if( pTask->aListMemory ){
        if( pSorter->aMemory==0 ){
          pSorter->aMemory = sqlite3Malloc(pSorter->nMemory);
          if( pSorter->aMemory==0 ) return SQLITE_NOMEM;
        }else{
          pSorter->nMemory = sqlite3MallocSize(pSorter->aMemory);
        }
      }
      rc = sqlite3ThreadCreate(&pThread->pThread, vdbeSortSubtaskMain, pCtx);
      rc = sqlite3ThreadCreate(&pTask->pTask, vdbeSortSubtaskMain, pCtx);
    }else
#endif
    {
      /* Use the foreground thread for this operation */
      rc = vdbeSorterRunThread(pThread);
      rc = vdbeSorterRunTask(pTask);
      if( rc==SQLITE_OK ){
        u8 *aMem = pThread->aListMemory;
        pThread->aListMemory = pSorter->aMemory;
        u8 *aMem = pTask->aListMemory;
        pTask->aListMemory = pSorter->aMemory;
        pSorter->aMemory = aMem;
        assert( pThread->pList==0 );
        assert( pTask->pList==0 );
      }
    }
  }

  return rc;
}

1465
1466
1467
1468
1469
1470
1471
1472
1473


1474
1475
1476
1477
1478
1479
1480



1481
1482
1483
1484
1485
1486
1487
1488
1489
1490
1491
1492
1493

1494
1495
1496
1497
1498
1499
1500
1501
1502








1503
1504
1505
1506
1507
1508
1509
1510
1511
1512
1513
1514
1515
1516
1517
1518
1519
1520
1521
1522
1523
1524
1525





1526
1527
1528
1529
1530



1531
1532
1533
1534

1535
1536
1537
1538
1539
1540
1541
1542
1543
1544
1545
1546
1547
1548
1549
1550
1551



1552
1553
1554

1555
1556
1557
1558
1559
1560

1561
1562
1563
1564


1565
1566
1567


1568
1569
1570
1571
1572
1573

1574
1575
1576
1577
1578
1579
1580
1581
1582
1583
1584
1585
1586
1587
1588
1589
1590
1591
1592

1593
1594
1595
1596
1597
1598
1599
1468
1469
1470
1471
1472
1473
1474


1475
1476
1477
1478
1479
1480
1481


1482
1483
1484
1485
1486
1487
1488
1489
1490
1491
1492
1493
1494
1495
1496

1497
1498








1499
1500
1501
1502
1503
1504
1505
1506
1507
1508
1509
1510
1511
1512
1513
1514
1515
1516
1517
1518
1519
1520
1521
1522
1523
1524





1525
1526
1527
1528
1529
1530
1531



1532
1533
1534
1535
1536
1537

1538
1539
1540
1541
1542
1543
1544
1545
1546
1547
1548
1549
1550
1551
1552



1553
1554
1555
1556
1557

1558
1559
1560
1561
1562
1563

1564
1565
1566


1567
1568
1569


1570
1571
1572
1573
1574
1575
1576

1577
1578
1579
1580
1581
1582
1583
1584
1585
1586
1587
1588
1589
1590
1591
1592
1593
1594
1595

1596
1597
1598
1599
1600
1601
1602
1603







-
-
+
+





-
-
+
+
+












-
+

-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+


















-
-
-
-
-
+
+
+
+
+


-
-
-
+
+
+



-
+














-
-
-
+
+
+


-
+





-
+


-
-
+
+

-
-
+
+





-
+


















-
+








/*
** Return the total number of PMAs in all temporary files.
*/
static int vdbeSorterCountPMA(VdbeSorter *pSorter){
  int nPMA = 0;
  int i;
  for(i=0; i<pSorter->nThread; i++){
    nPMA += pSorter->aThread[i].nPMA;
  for(i=0; i<pSorter->nTask; i++){
    nPMA += pSorter->aTask[i].nPMA;
  }
  return nPMA;
}

/*
** Once the sorter has been populated, this function is called to prepare
** for iterating through its contents in sorted order.
** Once the sorter has been populated by calls to sqlite3VdbeSorterWrite,
** this function is called to prepare for iterating through the records
** in sorted order.
*/
int sqlite3VdbeSorterRewind(sqlite3 *db, const VdbeCursor *pCsr, int *pbEof){
  VdbeSorter *pSorter = pCsr->pSorter;
  int rc = SQLITE_OK;             /* Return code */

  assert( pSorter );

  /* If no data has been written to disk, then do not do so now. Instead,
  ** sort the VdbeSorter.pRecord list. The vdbe layer will read data directly
  ** from the in-memory list.  */
  if( pSorter->bUsePMA==0 ){
    if( pSorter->pRecord ){
      SortSubtask *pThread = &pSorter->aThread[0];
      SortSubtask *pTask = &pSorter->aTask[0];
      *pbEof = 0;
      pThread->pList = pSorter->pRecord;
      pThread->eWork = SORT_SUBTASK_SORT;
      assert( pThread->aListMemory==0 );
      pThread->aListMemory = pSorter->aMemory;
      rc = vdbeSorterRunThread(pThread);
      pThread->aListMemory = 0;
      pSorter->pRecord = pThread->pList;
      pThread->pList = 0;
      pTask->pList = pSorter->pRecord;
      pTask->eWork = SORT_SUBTASK_SORT;
      assert( pTask->aListMemory==0 );
      pTask->aListMemory = pSorter->aMemory;
      rc = vdbeSorterRunTask(pTask);
      pTask->aListMemory = 0;
      pSorter->pRecord = pTask->pList;
      pTask->pList = 0;
    }else{
      *pbEof = 1;
    }
    return rc;
  }

  /* Write the current in-memory list to a PMA. */
  if( pSorter->pRecord ){
    rc = vdbeSorterFlushPMA(db, pCsr, 1);
  }

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

  /* If there are more than SORTER_MAX_MERGE_COUNT PMAs on disk, merge
  ** some of them together so that this is no longer the case. */
  if( vdbeSorterCountPMA(pSorter)>SORTER_MAX_MERGE_COUNT ){
    int i;
    for(i=0; rc==SQLITE_OK && i<pSorter->nThread; i++){
      SortSubtask *pThread = &pSorter->aThread[i];
      if( pThread->pTemp1 ){
        pThread->nConsolidate = SORTER_MAX_MERGE_COUNT / pSorter->nThread;
        pThread->eWork = SORT_SUBTASK_CONS;
    for(i=0; rc==SQLITE_OK && i<pSorter->nTask; i++){
      SortSubtask *pTask = &pSorter->aTask[i];
      if( pTask->pTemp1 ){
        pTask->nConsolidate = SORTER_MAX_MERGE_COUNT / pSorter->nTask;
        pTask->eWork = SORT_SUBTASK_CONS;

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

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

  /* Assuming no errors have occurred, set up a merger structure to read
  ** and merge all remaining PMAs.  */
  assert( pSorter->pMerger==0 );
  if( rc==SQLITE_OK ){
    int nIter = 0;                /* Number of iterators used */
    int i;
    SorterMerger *pMerger;
    for(i=0; i<pSorter->nThread; i++){
      nIter += pSorter->aThread[i].nPMA;
    MergeEngine *pMerger;
    for(i=0; i<pSorter->nTask; i++){
      nIter += pSorter->aTask[i].nPMA;
    }

    pSorter->pMerger = pMerger = vdbeSorterMergerNew(nIter);
    pSorter->pMerger = pMerger = vdbeMergeEngineNew(nIter);
    if( pMerger==0 ){
      rc = SQLITE_NOMEM;
    }else{
      int iIter = 0;
      int iThread = 0;
      for(iThread=0; iThread<pSorter->nThread; iThread++){
      for(iThread=0; iThread<pSorter->nTask; iThread++){
        int iPMA;
        i64 iReadOff = 0;
        SortSubtask *pThread = &pSorter->aThread[iThread];
        for(iPMA=0; iPMA<pThread->nPMA && rc==SQLITE_OK; iPMA++){
        SortSubtask *pTask = &pSorter->aTask[iThread];
        for(iPMA=0; iPMA<pTask->nPMA && rc==SQLITE_OK; iPMA++){
          i64 nDummy = 0;
          VdbeSorterIter *pIter = &pMerger->aIter[iIter++];
          rc = vdbeSorterIterInit(pThread, iReadOff, pIter, &nDummy);
          PmaReader *pIter = &pMerger->aIter[iIter++];
          rc = vdbePmaReaderInit(pTask, iReadOff, pIter, &nDummy);
          iReadOff = pIter->iEof;
        }
      }

      for(i=pMerger->nTree-1; rc==SQLITE_OK && i>0; i--){
        rc = vdbeSorterDoCompare(&pSorter->aThread[0], pMerger, i);
        rc = vdbeSorterDoCompare(&pSorter->aTask[0], pMerger, i);
      }
    }
  }

  if( rc==SQLITE_OK ){
    *pbEof = (pSorter->pMerger->aIter[pSorter->pMerger->aTree[1]].pFile==0);
  }
  return rc;
}

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

  if( pSorter->pMerger ){
    rc = vdbeSorterNext(&pSorter->aThread[0], pSorter->pMerger, pbEof);
    rc = vdbeSorterNext(&pSorter->aTask[0], pSorter->pMerger, pbEof);
  }else{
    SorterRecord *pFree = pSorter->pRecord;
    pSorter->pRecord = pFree->u.pNext;
    pFree->u.pNext = 0;
    if( pSorter->aMemory==0 ) vdbeSorterRecordFree(db, pFree);
    *pbEof = !pSorter->pRecord;
    rc = SQLITE_OK;
1607
1608
1609
1610
1611
1612
1613
1614

1615
1616
1617
1618
1619
1620
1621
1611
1612
1613
1614
1615
1616
1617

1618
1619
1620
1621
1622
1623
1624
1625







-
+







*/
static void *vdbeSorterRowkey(
  const VdbeSorter *pSorter,      /* Sorter object */
  int *pnKey                      /* OUT: Size of current key in bytes */
){
  void *pKey;
  if( pSorter->pMerger ){
    VdbeSorterIter *pIter;
    PmaReader *pIter;
    pIter = &pSorter->pMerger->aIter[ pSorter->pMerger->aTree[1] ];
    *pnKey = pIter->nKey;
    pKey = pIter->aKey;
  }else{
    *pnKey = pSorter->pRecord->nVal;
    pKey = SRVAL(pSorter->pRecord);
  }
1659
1660
1661
1662
1663
1664
1665
1666

1667
1668
1669
1670
1671
1672
1673
1663
1664
1665
1666
1667
1668
1669

1670
1671
1672
1673
1674
1675
1676
1677







-
+







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

  assert( r2->nField>=pKeyInfo->nField-nIgnore );
  r2->nField = pKeyInfo->nField-nIgnore;