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

Changes to src/shell.c.

  3531   3531     memcpy(data->separator,"|", 2);
  3532   3532     data->showHeader = 0;
  3533   3533     sqlite3_config(SQLITE_CONFIG_URI, 1);
  3534   3534     sqlite3_config(SQLITE_CONFIG_LOG, shellLog, data);
  3535   3535     sqlite3_snprintf(sizeof(mainPrompt), mainPrompt,"sqlite> ");
  3536   3536     sqlite3_snprintf(sizeof(continuePrompt), continuePrompt,"   ...> ");
  3537   3537     sqlite3_config(SQLITE_CONFIG_MULTITHREAD);
  3538         -  sqlite3_config(SQLITE_CONFIG_WORKER_THREADS, 3);
         3538  +  sqlite3_config(SQLITE_CONFIG_WORKER_THREADS, 4);
  3539   3539   }
  3540   3540   
  3541   3541   /*
  3542   3542   ** Output text to the console in a font that attracts extra attention.
  3543   3543   */
  3544   3544   #ifdef _WIN32
  3545   3545   static void printBold(const char *zText){

Changes to src/vdbesort.c.

   160    160     u8 eWork;                       /* One of the SORT_SUBTASK_* constants */
   161    161     int nConsolidate;               /* For SORT_SUBTASK_CONS, max final PMAs */
   162    162     SorterRecord *pList;            /* List of records for pTask to sort */
   163    163     int nInMemory;                  /* Expected size of PMA based on pList */
   164    164     u8 *aListMemory;                /* Records memory (or NULL) */
   165    165   
   166    166     int nPMA;                       /* Number of PMAs currently in file */
   167         -  SorterFile file;
          167  +  SorterFile file;                /* Temp file for level-0 PMAs */
          168  +  SorterFile file2;               /* Space for other PMAs */
   168    169   };
   169    170   
   170    171   
   171    172   /*
   172    173   ** The MergeEngine object is used to combine two or more smaller PMAs into
   173    174   ** one big PMA using a merge operation.  Separate PMAs all need to be
   174    175   ** combined into one big PMA in order to be able to step through the sorted
................................................................................
   236    237     int *aTree;                /* Current state of incremental merge */
   237    238     PmaReader *aIter;          /* Array of iterators to merge data from */
   238    239   };
   239    240   
   240    241   /*
   241    242   ** Main sorter structure. A single instance of this is allocated for each 
   242    243   ** sorter cursor created by the VDBE.
          244  +**
          245  +** mxKeysize:
          246  +**   As records are added to the sorter by calls to sqlite3VdbeSorterWrite(),
          247  +**   this variable is updated so as to be set to the size on disk of the
          248  +**   largest record in the sorter.
   243    249   */
   244    250   struct VdbeSorter {
   245    251     int nInMemory;                  /* Current size of pRecord list as PMA */
   246    252     int mnPmaSize;                  /* Minimum PMA size, in bytes */
   247    253     int mxPmaSize;                  /* Maximum PMA size, in bytes.  0==no limit */
   248    254     int bUsePMA;                    /* True if one or more PMAs created */
   249    255     int bUseThreads;                /* True if one or more PMAs created */
   250    256     SorterRecord *pRecord;          /* Head of in-memory record list */
   251    257     PmaReader *pReader;             /* Read data from here after Rewind() */
          258  +  int mxKeysize;                  /* Largest serialized key seen so far */
   252    259     UnpackedRecord *pUnpacked;      /* Used by VdbeSorterCompare() */
   253    260     u8 *aMemory;                    /* Block of memory to alloc records from */
   254    261     int iMemory;                    /* Offset of first free byte in aMemory */
   255    262     int nMemory;                    /* Size of aMemory allocation in bytes */
   256    263     int iPrev;                      /* Previous thread used to flush PMA */
   257    264     int nTask;                      /* Size of aTask[] array */
   258    265     SortSubtask aTask[1];           /* One or more subtasks */
................................................................................
   273    280     u8 *aKey;                       /* Pointer to current key */
   274    281     u8 *aBuffer;                    /* Current read buffer */
   275    282     int nBuffer;                    /* Size of read buffer in bytes */
   276    283     u8 *aMap;                       /* Pointer to mapping of entire file */
   277    284     IncrMerger *pIncr;              /* Incremental merger */
   278    285   };
   279    286   
          287  +/*
          288  +** Normally, a PmaReader object iterates through an existing PMA stored 
          289  +** within a temp file. However, if the PmaReader.pIncr variable points to
          290  +** an object of the following type, it may be used to iterate/merge through
          291  +** multiple PMAs simultaneously.
          292  +*/
   280    293   struct IncrMerger {
   281         -  int mxSz;                       /* Maximum size of files */
   282    294     SortSubtask *pTask;             /* Task that owns this merger */
   283         -  int bEof;                       /* Set to true when merge is finished */
   284         -  SorterFile aFile[2];            /* aFile[0] for reading, [1] for writing */
   285         -  MergeEngine *pMerger;           /* Merge engine thread reads data from */
   286    295     SQLiteThread *pThread;          /* Thread currently populating aFile[1] */
          296  +  MergeEngine *pMerger;           /* Merge engine thread reads data from */
          297  +  i64 iStartOff;                  /* Offset to start writing file at */
          298  +  int mxSz;                       /* Maximum bytes of data to store */
          299  +  int bEof;                       /* Set to true when merge is finished */
          300  +  int bUseThread;                 /* True to use a bg thread for this object */
          301  +  SorterFile aFile[2];            /* aFile[0] for reading, [1] for writing */
   287    302   };
   288    303   
   289    304   /*
   290    305   ** An instance of this object is used for writing a PMA.
   291    306   **
   292    307   ** The PMA is written one record at a time.  Each record is of an arbitrary
   293    308   ** size.  But I/O is more efficient if it occurs in page-sized blocks where
................................................................................
   502    517   
   503    518     assert( pIncr->bEof==0 );
   504    519   
   505    520     if( pIter->aMap ){
   506    521       sqlite3OsUnfetch(pIter->pFile, 0, pIter->aMap);
   507    522       pIter->aMap = 0;
   508    523     }
   509         -  pIter->iReadOff = 0;
          524  +  pIter->iReadOff = pIncr->iStartOff;
   510    525     pIter->iEof = pIncr->aFile[0].iEof;
   511    526     pIter->pFile = pIncr->aFile[0].pFd;
   512    527   
   513    528     rc = vdbeSorterMapFile(pTask, &pIncr->aFile[0], &pIter->aMap);
   514    529     if( rc==SQLITE_OK ){
   515         -    if( pIter->aMap==0 && pIter->aBuffer==0 ){
   516         -      pIter->aBuffer = (u8*)sqlite3Malloc(pTask->pgsz);
   517         -      if( pIter->aBuffer==0 ) rc = SQLITE_NOMEM;
   518         -      pIter->nBuffer = pTask->pgsz;
          530  +    if( pIter->aMap==0 ){
          531  +      /* TODO: Combine this code with similar code in vdbePmaReaderInit() */
          532  +      int iBuf = pIter->iReadOff % pTask->pgsz;
          533  +      if( pIter->aBuffer==0 ){
          534  +        pIter->aBuffer = (u8*)sqlite3Malloc(pTask->pgsz);
          535  +        if( pIter->aBuffer==0 ) rc = SQLITE_NOMEM;
          536  +        pIter->nBuffer = pTask->pgsz;
          537  +      }
          538  +      if( iBuf ){
          539  +        int nRead = pTask->pgsz - iBuf;
          540  +        if( (pIter->iReadOff + nRead) > pIter->iEof ){
          541  +          nRead = (int)(pIter->iEof - pIter->iReadOff);
          542  +        }
          543  +        rc = sqlite3OsRead(
          544  +            pIter->pFile, &pIter->aBuffer[iBuf], nRead, pIter->iReadOff
          545  +        );
          546  +        assert( rc!=SQLITE_IOERR_SHORT_READ );
          547  +      }
   519    548       }
   520    549     }
   521    550   
   522    551     return rc;
   523    552   }
   524    553   
   525    554   
................................................................................
   573    602     SorterFile *pFile,              /* Sorter file to read from */
   574    603     i64 iStart,                     /* Start offset in pFile */
   575    604     PmaReader *pIter,               /* Iterator to populate */
   576    605     i64 *pnByte                     /* IN/OUT: Increment this value by PMA size */
   577    606   ){
   578    607     int rc = SQLITE_OK;
   579    608     int nBuf = pTask->pgsz;
   580         -  void *pMap = 0;                 /* Mapping of temp file */
   581    609   
   582    610     assert( pFile->iEof>iStart );
   583    611     assert( pIter->aAlloc==0 );
   584    612     assert( pIter->aBuffer==0 );
   585    613     pIter->pFile = pFile->pFd;
   586    614     pIter->iReadOff = iStart;
   587    615     pIter->nAlloc = 128;
   588    616     pIter->aAlloc = (u8*)sqlite3Malloc(pIter->nAlloc);
   589    617     if( pIter->aAlloc ){
   590    618       /* Try to xFetch() a mapping of the entire temp file. If this is possible,
   591    619       ** the PMA will be read via the mapping. Otherwise, use xRead().  */
   592         -    if( pFile->iEof<=(i64)(pTask->db->nMaxSorterMmap) ){
   593         -      rc = sqlite3OsFetch(pIter->pFile, 0, pFile->iEof, &pMap);
   594         -    }
          620  +    rc = vdbeSorterMapFile(pTask, pFile, &pIter->aMap);
   595    621     }else{
   596    622       rc = SQLITE_NOMEM;
   597    623     }
   598    624   
   599         -  if( rc==SQLITE_OK ){
   600         -    if( pMap ){
   601         -      pIter->aMap = (u8*)pMap;
          625  +  if( rc==SQLITE_OK && pIter->aMap==0 ){
          626  +    pIter->nBuffer = nBuf;
          627  +    pIter->aBuffer = (u8*)sqlite3Malloc(nBuf);
          628  +    if( !pIter->aBuffer ){
          629  +      rc = SQLITE_NOMEM;
   602    630       }else{
   603         -      pIter->nBuffer = nBuf;
   604         -      pIter->aBuffer = (u8*)sqlite3Malloc(nBuf);
   605         -      if( !pIter->aBuffer ){
   606         -        rc = SQLITE_NOMEM;
   607         -      }else{
   608         -        int iBuf = iStart % nBuf;
   609         -        if( iBuf ){
   610         -          int nRead = nBuf - iBuf;
   611         -          if( (iStart + nRead) > pFile->iEof ){
   612         -            nRead = (int)(pFile->iEof - iStart);
   613         -          }
   614         -          rc = sqlite3OsRead(
   615         -              pIter->pFile, &pIter->aBuffer[iBuf], nRead, iStart
   616         -          );
   617         -          assert( rc!=SQLITE_IOERR_SHORT_READ );
          631  +      int iBuf = iStart % nBuf;
          632  +      if( iBuf ){
          633  +        int nRead = nBuf - iBuf;
          634  +        if( (iStart + nRead) > pFile->iEof ){
          635  +          nRead = (int)(pFile->iEof - iStart);
   618    636           }
          637  +        rc = sqlite3OsRead(
          638  +            pIter->pFile, &pIter->aBuffer[iBuf], nRead, iStart
          639  +        );
          640  +        assert( rc!=SQLITE_IOERR_SHORT_READ );
   619    641         }
   620    642       }
   621    643     }
   622    644   
   623    645     if( rc==SQLITE_OK ){
   624    646       u64 nByte;                    /* Size of PMA in bytes */
   625    647       pIter->iEof = pFile->iEof;
................................................................................
   801    823     }
   802    824     pTask->pList = 0;
   803    825     if( pTask->file.pFd ){
   804    826       sqlite3OsCloseFree(pTask->file.pFd);
   805    827       pTask->file.pFd = 0;
   806    828       pTask->file.iEof = 0;
   807    829     }
          830  +  if( pTask->file2.pFd ){
          831  +    sqlite3OsCloseFree(pTask->file2.pFd);
          832  +    pTask->file2.pFd = 0;
          833  +    pTask->file2.iEof = 0;
          834  +  }
   808    835   }
   809    836   
   810    837   /*
   811    838   ** Join all threads.  
   812    839   */
   813    840   #if SQLITE_MAX_WORKER_THREADS>0
   814    841   static int vdbeSorterJoinAll(VdbeSorter *pSorter, int rcin){
................................................................................
   835    862   ** Allocate a new MergeEngine object with space for nIter iterators.
   836    863   */
   837    864   static MergeEngine *vdbeMergeEngineNew(int nIter){
   838    865     int N = 2;                      /* Smallest power of two >= nIter */
   839    866     int nByte;                      /* Total bytes of space to allocate */
   840    867     MergeEngine *pNew;              /* Pointer to allocated object to return */
   841    868   
   842         -  /* assert( nIter<=SORTER_MAX_MERGE_COUNT ); */
          869  +  assert( nIter<=SORTER_MAX_MERGE_COUNT );
   843    870   
   844    871     while( N<nIter ) N += N;
   845    872     nByte = sizeof(MergeEngine) + N * (sizeof(int) + sizeof(PmaReader));
   846    873   
   847    874     pNew = (MergeEngine*)sqlite3MallocZero(nByte);
   848    875     if( pNew ){
   849    876       pNew->nTree = N;
................................................................................
   884    911     if( pSorter->aMemory==0 ){
   885    912       vdbeSorterRecordFree(0, pSorter->pRecord);
   886    913     }
   887    914     pSorter->pRecord = 0;
   888    915     pSorter->nInMemory = 0;
   889    916     pSorter->bUsePMA = 0;
   890    917     pSorter->iMemory = 0;
          918  +  pSorter->mxKeysize = 0;
   891    919     sqlite3DbFree(db, pSorter->pUnpacked);
   892    920     pSorter->pUnpacked = 0;
   893    921   }
   894    922   
   895    923   /*
   896    924   ** Free any cursor components allocated by sqlite3VdbeSorterXXX routines.
   897    925   */
................................................................................
  1255   1283     fprintf(stderr, "%lld:%d %s\n", t, iTask, zEvent);
  1256   1284   }
  1257   1285   static void vdbeSorterRewindDebug(sqlite3 *db, const char *zEvent){
  1258   1286     i64 t;
  1259   1287     sqlite3OsCurrentTimeInt64(db->pVfs, &t);
  1260   1288     fprintf(stderr, "%lld:X %s\n", t, zEvent);
  1261   1289   }
         1290  +static void vdbeSorterPopulateDebug(
         1291  +  SortSubtask *pTask,
         1292  +  const char *zEvent
         1293  +){
         1294  +  i64 t;
         1295  +  int iTask = (pTask - pTask->pSorter->aTask);
         1296  +  sqlite3OsCurrentTimeInt64(pTask->db->pVfs, &t);
         1297  +  fprintf(stderr, "%lld:bg%d %s\n", t, iTask, zEvent);
         1298  +}
  1262   1299   #else
  1263   1300   # define vdbeSorterWorkDebug(x,y)
  1264   1301   # define vdbeSorterRewindDebug(x,y)
         1302  +# define vdbeSorterPopulateDebug(x,y)
  1265   1303   #endif
         1304  +
         1305  +static int vdbeSortAllocUnpacked(SortSubtask *pTask){
         1306  +  if( pTask->pUnpacked==0 ){
         1307  +    char *pFree;
         1308  +    pTask->pUnpacked = sqlite3VdbeAllocUnpackedRecord(
         1309  +        pTask->pKeyInfo, 0, 0, &pFree
         1310  +    );
         1311  +    assert( pTask->pUnpacked==(UnpackedRecord*)pFree );
         1312  +    if( pFree==0 ) return SQLITE_NOMEM;
         1313  +    pTask->pUnpacked->nField = pTask->pKeyInfo->nField;
         1314  +    pTask->pUnpacked->errCode = 0;
         1315  +  }
         1316  +  return SQLITE_OK;
         1317  +}
  1266   1318   
  1267   1319   /*
  1268   1320   ** The main routine for sorter-thread operations.
  1269   1321   */
  1270   1322   static void *vdbeSortSubtaskMain(void *pCtx){
  1271   1323     int rc = SQLITE_OK;
  1272   1324     SortSubtask *pTask = (SortSubtask*)pCtx;
................................................................................
  1275   1327          || pTask->eWork==SORT_SUBTASK_TO_PMA
  1276   1328          || pTask->eWork==SORT_SUBTASK_CONS
  1277   1329     );
  1278   1330     assert( pTask->bDone==0 );
  1279   1331   
  1280   1332     vdbeSorterWorkDebug(pTask, "enter");
  1281   1333   
  1282         -  if( pTask->pUnpacked==0 ){
  1283         -    char *pFree;
  1284         -    pTask->pUnpacked = sqlite3VdbeAllocUnpackedRecord(
  1285         -        pTask->pKeyInfo, 0, 0, &pFree
  1286         -    );
  1287         -    assert( pTask->pUnpacked==(UnpackedRecord*)pFree );
  1288         -    if( pFree==0 ){
  1289         -      rc = SQLITE_NOMEM;
  1290         -      goto thread_out;
  1291         -    }
  1292         -    pTask->pUnpacked->nField = pTask->pKeyInfo->nField;
  1293         -    pTask->pUnpacked->errCode = 0;
  1294         -  }
         1334  +  rc = vdbeSortAllocUnpacked(pTask);
         1335  +  if( rc!=SQLITE_OK ) goto thread_out;
  1295   1336   
  1296   1337     if( pTask->eWork==SORT_SUBTASK_CONS ){
  1297   1338       assert( pTask->pList==0 );
  1298   1339       while( pTask->nPMA>pTask->nConsolidate && rc==SQLITE_OK ){
  1299   1340         int nIter = MIN(pTask->nPMA, SORTER_MAX_MERGE_COUNT);
  1300   1341         sqlite3_file *pTemp2 = 0;     /* Second temp file to use */
  1301   1342         MergeEngine *pMerger;         /* Object for reading/merging PMA data */
................................................................................
  1529   1570         pSorter->nInMemory = 0;
  1530   1571         pSorter->iMemory = 0;
  1531   1572         assert( rc!=SQLITE_OK || pSorter->pRecord==0 );
  1532   1573       }
  1533   1574     }
  1534   1575   
  1535   1576     pSorter->nInMemory += nPMA;
         1577  +  if( nPMA>pSorter->mxKeysize ){
         1578  +    pSorter->mxKeysize = nPMA;
         1579  +  }
  1536   1580   
  1537   1581     if( pSorter->aMemory ){
  1538   1582       int nMin = pSorter->iMemory + nReq;
  1539   1583   
  1540   1584       if( nMin>pSorter->nMemory ){
  1541   1585         u8 *aNew;
  1542   1586         int nNew = pSorter->nMemory * 2;
................................................................................
  1587   1631   ** Read keys from pIncr->pMerger and populate pIncr->aFile[1]. The format
  1588   1632   ** of the data stored in aFile[1] is the same as that used by regular PMAs,
  1589   1633   ** except that the number-of-bytes varint is omitted from the start.
  1590   1634   */
  1591   1635   static int vdbeIncrPopulate(IncrMerger *pIncr){
  1592   1636     int rc = SQLITE_OK;
  1593   1637     int rc2;
         1638  +  i64 iStart = pIncr->iStartOff;
  1594   1639     SorterFile *pOut = &pIncr->aFile[1];
  1595   1640     MergeEngine *pMerger = pIncr->pMerger;
  1596   1641     PmaWriter writer;
  1597   1642     assert( pIncr->bEof==0 );
  1598   1643   
  1599         -  vdbePmaWriterInit(pIncr->aFile[1].pFd, &writer, pIncr->pTask->pgsz, 0);
         1644  +  vdbeSorterPopulateDebug(pIncr->pTask, "enter");
         1645  +
         1646  +  vdbePmaWriterInit(pOut->pFd, &writer, pIncr->pTask->pgsz, iStart);
  1600   1647     while( rc==SQLITE_OK ){
  1601   1648       int dummy;
  1602   1649       PmaReader *pReader = &pMerger->aIter[ pMerger->aTree[1] ];
  1603   1650       int nKey = pReader->nKey;
  1604   1651       i64 iEof = writer.iWriteOff + writer.iBufEnd;
  1605   1652   
  1606   1653       /* Check if the output file is full or if the input has been exhausted.
  1607   1654       ** In either case exit the loop. */
  1608   1655       if( pReader->pFile==0 ) break;
  1609         -    if( iEof && (iEof + nKey)>pIncr->mxSz ) break;
         1656  +    if( (iEof + nKey + sqlite3VarintLen(nKey))>(iStart + pIncr->mxSz) ) break;
  1610   1657   
  1611   1658       /* Write the next key to the output. */
  1612   1659       vdbePmaWriteVarint(&writer, nKey);
  1613   1660       vdbePmaWriteBlob(&writer, pReader->aKey, nKey);
  1614   1661       rc = vdbeSorterNext(pIncr->pTask, pIncr->pMerger, &dummy);
  1615   1662     }
  1616   1663   
  1617   1664     rc2 = vdbePmaWriterFinish(&writer, &pOut->iEof);
  1618   1665     if( rc==SQLITE_OK ) rc = rc2;
         1666  +  vdbeSorterPopulateDebug(pIncr->pTask, "exit");
  1619   1667     return rc;
  1620   1668   }
  1621   1669   
  1622   1670   static void *vdbeIncrPopulateThreadMain(void *pCtx){
  1623   1671     IncrMerger *pIncr = (IncrMerger*)pCtx;
  1624   1672     return SQLITE_INT_TO_PTR( vdbeIncrPopulate(pIncr) );
  1625   1673   }
  1626   1674   
  1627   1675   static int vdbeIncrBgPopulate(IncrMerger *pIncr){
  1628   1676     int rc;
  1629   1677     assert( pIncr->pThread==0 );
  1630         -  if( pIncr->pTask->pSorter->bUseThreads==0 ){
         1678  +  if( pIncr->bUseThread==0 ){
  1631   1679       rc = vdbeIncrPopulate(pIncr);
  1632         -  }else{
         1680  +  }
         1681  +#if SQLITE_MAX_WORKER_THREADS>0
         1682  +  else{
  1633   1683       void *pCtx = (void*)pIncr;
  1634   1684       rc = sqlite3ThreadCreate(&pIncr->pThread, vdbeIncrPopulateThreadMain, pCtx);
  1635   1685     }
         1686  +#endif
  1636   1687     return rc;
  1637   1688   }
  1638   1689   
  1639   1690   static int vdbeIncrSwap(IncrMerger *pIncr){
  1640   1691     int rc = SQLITE_OK;
  1641         -  
  1642         -  if( pIncr->pThread ){
  1643         -    void *pRet;
  1644         -    rc = sqlite3ThreadJoin(pIncr->pThread, &pRet);
  1645         -    if( rc==SQLITE_OK ) rc = SQLITE_PTR_TO_INT(pRet);
  1646         -    pIncr->pThread = 0;
  1647         -  }
  1648   1692   
  1649         -  if( rc==SQLITE_OK ){
  1650         -    SorterFile f0 = pIncr->aFile[0];
         1693  +  if( pIncr->bUseThread ){
         1694  +#if SQLITE_MAX_WORKER_THREADS>0
         1695  +    if( pIncr->pThread ){
         1696  +      void *pRet;
         1697  +      assert( pIncr->bUseThread );
         1698  +      rc = sqlite3ThreadJoin(pIncr->pThread, &pRet);
         1699  +      if( rc==SQLITE_OK ) rc = SQLITE_PTR_TO_INT(pRet);
         1700  +      pIncr->pThread = 0;
         1701  +    }
         1702  +#endif
         1703  +
         1704  +    if( rc==SQLITE_OK ){
         1705  +      SorterFile f0 = pIncr->aFile[0];
         1706  +      pIncr->aFile[0] = pIncr->aFile[1];
         1707  +      pIncr->aFile[1] = f0;
         1708  +    }
         1709  +
         1710  +    if( rc==SQLITE_OK ){
         1711  +      if( pIncr->aFile[0].iEof==pIncr->iStartOff ){
         1712  +        pIncr->bEof = 1;
         1713  +      }else{
         1714  +        rc = vdbeIncrBgPopulate(pIncr);
         1715  +      }
         1716  +    }
         1717  +  }else{
         1718  +    rc = vdbeIncrPopulate(pIncr);
  1651   1719       pIncr->aFile[0] = pIncr->aFile[1];
  1652         -    pIncr->aFile[1] = f0;
  1653         -
  1654         -    if( pIncr->aFile[0].iEof==0 ){
         1720  +    if( pIncr->aFile[0].iEof==pIncr->iStartOff ){
  1655   1721         pIncr->bEof = 1;
  1656         -    }else{
  1657         -      rc = vdbeIncrBgPopulate(pIncr);
  1658   1722       }
  1659   1723     }
  1660   1724   
  1661   1725     return rc;
  1662   1726   }
  1663   1727   
  1664   1728   static void vdbeIncrFree(IncrMerger *pIncr){
  1665         -  if( pIncr->pThread ){
  1666         -    void *pRet;
  1667         -    sqlite3ThreadJoin(pIncr->pThread, &pRet);
  1668         -  }
  1669         -  if( pIncr->aFile[0].pFd ) sqlite3OsCloseFree(pIncr->aFile[0].pFd);
  1670         -  if( pIncr->aFile[1].pFd ) sqlite3OsCloseFree(pIncr->aFile[1].pFd);
  1671         -  vdbeMergeEngineFree(pIncr->pMerger);
  1672         -  sqlite3_free(pIncr);
         1729  +  if( pIncr ){
         1730  +#if SQLITE_MAX_WORKER_THREADS>0
         1731  +    if( pIncr->pThread ){
         1732  +      void *pRet;
         1733  +      sqlite3ThreadJoin(pIncr->pThread, &pRet);
         1734  +    }
         1735  +    if( pIncr->bUseThread ){
         1736  +      if( pIncr->aFile[0].pFd ) sqlite3OsCloseFree(pIncr->aFile[0].pFd);
         1737  +      if( pIncr->aFile[1].pFd ) sqlite3OsCloseFree(pIncr->aFile[1].pFd);
         1738  +    }
         1739  +#endif
         1740  +    vdbeMergeEngineFree(pIncr->pMerger);
         1741  +    sqlite3_free(pIncr);
         1742  +  }
         1743  +}
         1744  +
         1745  +static IncrMerger *vdbeIncrNew(SortSubtask *pTask, MergeEngine *pMerger){
         1746  +  IncrMerger *pIncr = sqlite3_malloc(sizeof(IncrMerger));
         1747  +  if( pIncr ){
         1748  +    memset(pIncr, 0, sizeof(IncrMerger));
         1749  +    pIncr->pMerger = pMerger;
         1750  +    pIncr->pTask = pTask;
         1751  +    pIncr->mxSz = MAX(pTask->pSorter->mxKeysize+9,pTask->pSorter->mxPmaSize/2);
         1752  +    pTask->file2.iEof += pIncr->mxSz;
         1753  +
         1754  +#if 0
         1755  +    /* Open the two temp files. */
         1756  +    rc = vdbeSorterOpenTempFile(pTask->db->pVfs, &pIncr->aFile[0].pFd);
         1757  +    if( rc==SQLITE_OK ){
         1758  +      rc = vdbeSorterOpenTempFile(pTask->db->pVfs, &pIncr->aFile[1].pFd);
         1759  +    }
         1760  +    if( rc!=SQLITE_OK ){
         1761  +      vdbeIncrFree(pIncr);
         1762  +      pIncr = 0;
         1763  +    }
         1764  +#endif
         1765  +  }
         1766  +  return pIncr;
         1767  +}
         1768  +
         1769  +static void vdbeIncrSetThreads(IncrMerger *pIncr, int bUseThread){
         1770  +  if( bUseThread ){
         1771  +    pIncr->bUseThread = 1;
         1772  +    pIncr->pTask->file2.iEof -= pIncr->mxSz;
         1773  +  }
         1774  +}
         1775  +
         1776  +static int vdbeIncrInit2(PmaReader *pIter){
         1777  +  int rc = SQLITE_OK;
         1778  +  IncrMerger *pIncr = pIter->pIncr;
         1779  +  if( pIncr ){
         1780  +    SortSubtask *pTask = pIncr->pTask;
         1781  +    int i;
         1782  +    MergeEngine *pMerger = pIncr->pMerger;
         1783  +
         1784  +    for(i=0; rc==SQLITE_OK && i<pMerger->nTree; i++){
         1785  +      rc = vdbeIncrInit2(&pMerger->aIter[i]);
         1786  +    }
         1787  +    for(i=pMerger->nTree-1; rc==SQLITE_OK && i>0; i--){
         1788  +      rc = vdbeSorterDoCompare(pIncr->pTask, pMerger, i);
         1789  +    }
         1790  +
         1791  +    /* Set up the required files for pIncr */
         1792  +    if( rc==SQLITE_OK ){
         1793  +      if( pIncr->bUseThread==0 ){
         1794  +        if( pTask->file2.pFd==0 ){
         1795  +          rc = vdbeSorterOpenTempFile(pTask->db->pVfs, &pTask->file2.pFd);
         1796  +          assert( pTask->file2.iEof>0 );
         1797  +          if( rc==SQLITE_OK ){
         1798  +            vdbeSorterExtendFile(pTask->db,pTask->file2.pFd,pTask->file2.iEof);
         1799  +            pTask->file2.iEof = 0;
         1800  +          }
         1801  +        }
         1802  +        if( rc==SQLITE_OK ){
         1803  +          pIncr->aFile[1].pFd = pTask->file2.pFd;
         1804  +          pIncr->iStartOff = pTask->file2.iEof;
         1805  +          pTask->file2.iEof += pIncr->mxSz;
         1806  +        }
         1807  +      }else{
         1808  +        rc = vdbeSorterOpenTempFile(pTask->db->pVfs, &pIncr->aFile[0].pFd);
         1809  +        if( rc==SQLITE_OK ){
         1810  +          rc = vdbeSorterOpenTempFile(pTask->db->pVfs, &pIncr->aFile[1].pFd);
         1811  +        }
         1812  +      }
         1813  +    }
         1814  +
         1815  +    if( rc==SQLITE_OK && pIncr->bUseThread ){
         1816  +      rc = vdbeIncrBgPopulate(pIncr);
         1817  +    }
         1818  +
         1819  +    if( rc==SQLITE_OK ){
         1820  +      rc = vdbePmaReaderNext(pIter);
         1821  +    }
         1822  +  }
         1823  +  return rc;
         1824  +}
         1825  +
         1826  +/*
         1827  +** Allocate a new MergeEngine object to merge the contents of nPMA level-0
         1828  +** PMAs from pTask->file. If no error occurs, set *ppOut to point to
         1829  +** the new object and return SQLITE_OK. Or, if an error does occur, set *ppOut
         1830  +** to NULL and return an SQLite error code.
         1831  +**
         1832  +** When this function is called, *piOffset is set to the offset of the
         1833  +** first PMA to read from pTask->file. Assuming no error occurs, it is 
         1834  +** set to the offset immediately following the last byte of the last
         1835  +** PMA before returning. If an error does occur, then the final value of
         1836  +** *piOffset is undefined.
         1837  +*/
         1838  +static int vdbeMergeEngineLevel0(
         1839  +  SortSubtask *pTask,             /* Sorter task to read from */
         1840  +  int nPMA,                       /* Number of PMAs to read */
         1841  +  i64 *piOffset,                  /* IN/OUT: Read offset in pTask->file */
         1842  +  MergeEngine **ppOut             /* OUT: New merge-engine */
         1843  +){
         1844  +  MergeEngine *pNew;              /* Merge engine to return */
         1845  +  i64 iOff = *piOffset;
         1846  +  int i;
         1847  +  int rc = SQLITE_OK;
         1848  +
         1849  +  *ppOut = pNew = vdbeMergeEngineNew(nPMA);
         1850  +  if( pNew==0 ) rc = SQLITE_NOMEM;
         1851  +
         1852  +  for(i=0; i<nPMA && rc==SQLITE_OK; i++){
         1853  +    i64 nDummy;
         1854  +    PmaReader *pIter = &pNew->aIter[i];
         1855  +    rc = vdbePmaReaderInit(pTask, &pTask->file, iOff, pIter, &nDummy);
         1856  +    iOff = pIter->iEof;
         1857  +  }
         1858  +
         1859  +  if( rc!=SQLITE_OK ){
         1860  +    vdbeMergeEngineFree(pNew);
         1861  +    *ppOut = 0;
         1862  +  }
         1863  +  *piOffset = iOff;
         1864  +  return rc;
         1865  +}
         1866  +
         1867  +typedef struct IncrBuilder IncrBuilder;
         1868  +struct IncrBuilder {
         1869  +  int nPMA;                     /* Number of iterators used so far */
         1870  +  MergeEngine *pMerger;         /* Merge engine to populate. */
         1871  +};
         1872  +
         1873  +static int vdbeAddToBuilder(
         1874  +  SortSubtask *pTask,
         1875  +  IncrBuilder *pBuilder, 
         1876  +  MergeEngine *pMerger
         1877  +){
         1878  +  int rc = SQLITE_OK;
         1879  +  IncrMerger *pIncr;
         1880  +
         1881  +  assert( pMerger );
         1882  +  if( pBuilder->nPMA==SORTER_MAX_MERGE_COUNT ){
         1883  +    rc = vdbeAddToBuilder(pTask, &pBuilder[1], pBuilder->pMerger);
         1884  +    pBuilder->pMerger = 0;
         1885  +    pBuilder->nPMA = 0;
         1886  +  }
         1887  +
         1888  +  if( rc==SQLITE_OK && pBuilder->pMerger==0 ){
         1889  +    pBuilder->pMerger = vdbeMergeEngineNew(SORTER_MAX_MERGE_COUNT);
         1890  +    if( pBuilder->pMerger==0 ) rc = SQLITE_NOMEM;
         1891  +  }
         1892  +
         1893  +  if( rc==SQLITE_OK ){
         1894  +    pIncr = vdbeIncrNew(pTask, pMerger);
         1895  +    if( pIncr==0 ) rc = SQLITE_NOMEM;
         1896  +    pBuilder->pMerger->aIter[pBuilder->nPMA++].pIncr = pIncr;
         1897  +  }
         1898  +
         1899  +  if( rc!=SQLITE_OK ){
         1900  +    vdbeMergeEngineFree(pMerger);
         1901  +  }
         1902  +
         1903  +  return rc;
  1673   1904   }
  1674   1905   
  1675   1906   /*
  1676   1907   ** Populate iterator *pIter so that it may be used to iterate through all 
  1677         -** keys stored in subtask pTask using the incremental merge method.
         1908  +** keys stored in all PMAs created by this sorter.
  1678   1909   */
  1679   1910   static int vdbePmaReaderIncrInit(VdbeSorter *pSorter, PmaReader *pIter){
  1680   1911     SortSubtask *pTask0 = &pSorter->aTask[0];
         1912  +  MergeEngine *pMain = 0;
         1913  +  sqlite3 *db = pTask0->db;
  1681   1914     int rc = SQLITE_OK;
  1682         -  MergeEngine *pMerger = 0;
  1683         -  IncrMerger *pIncr = 0;
  1684         -  int i;
  1685         -  int nPMA = 0;
         1915  +  int iTask;
  1686   1916   
  1687         -  for(i=0; i<pSorter->nTask; i++){
  1688         -    nPMA += pSorter->aTask[i].nPMA;
         1917  +  IncrBuilder *aMerge;
         1918  +  const int nMerge = 32;
         1919  +  aMerge = sqlite3DbMallocZero(db, sizeof(aMerge[0])*nMerge);
         1920  +  if( aMerge==0 ) return SQLITE_NOMEM;
         1921  +
         1922  +  if( pSorter->nTask>1 ){
         1923  +    pMain = vdbeMergeEngineNew(pSorter->nTask);
         1924  +    if( pMain==0 ) rc = SQLITE_NOMEM;
  1689   1925     }
  1690         -  pMerger = vdbeMergeEngineNew(nPMA);
  1691         -  if( pMerger==0 ){
  1692         -    rc = SQLITE_NOMEM;
  1693         -  }else{
  1694         -    int iIter = 0;
         1926  +
         1927  +  for(iTask=0; iTask<pSorter->nTask && rc==SQLITE_OK; iTask++){
         1928  +    MergeEngine *pRoot = 0;
  1695   1929       int iPMA;
  1696         -    for(i=0; i<pSorter->nTask; i++){
  1697         -      i64 iReadOff = 0;
  1698         -      SortSubtask *pTask = &pSorter->aTask[i];
  1699         -      for(iPMA=0; iPMA<pTask->nPMA; iPMA++){
  1700         -        i64 nDummy = 0;
  1701         -        PmaReader *pIter = &pMerger->aIter[iIter++];
  1702         -        rc = vdbePmaReaderInit(pTask, &pTask->file, iReadOff, pIter, &nDummy);
  1703         -        iReadOff = pIter->iEof;
         1930  +    i64 iReadOff = 0;
         1931  +    SortSubtask *pTask = &pSorter->aTask[iTask];
         1932  +    if( pTask->nPMA==0 ) continue;
         1933  +    for(iPMA=0; iPMA<pTask->nPMA; iPMA += SORTER_MAX_MERGE_COUNT){
         1934  +      MergeEngine *pMerger = 0;
         1935  +      int nReader = MIN(pTask->nPMA - iPMA, SORTER_MAX_MERGE_COUNT);
         1936  +
         1937  +      rc = vdbeMergeEngineLevel0(pTask, nReader, &iReadOff, &pMerger);
         1938  +      if( rc!=SQLITE_OK ) break;
         1939  +
         1940  +      if( iPMA==0 ){
         1941  +        pRoot = pMerger;
         1942  +      }else{
         1943  +        if( pRoot ){
         1944  +          rc = vdbeAddToBuilder(pTask, &aMerge[0], pRoot);
         1945  +          pRoot = 0;
         1946  +          if( rc!=SQLITE_OK ){
         1947  +            vdbeMergeEngineFree(pMerger);
         1948  +            break;
         1949  +          }
         1950  +        }
         1951  +        rc = vdbeAddToBuilder(pTask, &aMerge[0], pMerger);
         1952  +      }
         1953  +    }
         1954  +
         1955  +    if( pRoot==0 ){
         1956  +      int i;
         1957  +      for(i=0; rc==SQLITE_OK && i<nMerge; i++){
         1958  +        if( aMerge[i].pMerger ){
         1959  +          if( pRoot ){
         1960  +            rc = vdbeAddToBuilder(pTask, &aMerge[i], pRoot);
         1961  +            if( rc!=SQLITE_OK ) break;
         1962  +          }
         1963  +          pRoot = aMerge[i].pMerger;
         1964  +          aMerge[i].pMerger = 0;
         1965  +        }
  1704   1966         }
  1705   1967       }
  1706         -    for(i=pMerger->nTree-1; rc==SQLITE_OK && i>0; i--){
  1707         -      rc = vdbeSorterDoCompare(pTask0, pMerger, i);
         1968  +
         1969  +    if( rc==SQLITE_OK ){
         1970  +      if( pMain==0 ){
         1971  +        pMain = pRoot;
         1972  +      }else{
         1973  +        IncrMerger *pNew = vdbeIncrNew(pTask, pRoot);
         1974  +        pMain->aIter[iTask].pIncr = pNew;
         1975  +        if( pNew==0 ) rc = SQLITE_NOMEM;
         1976  +      }
         1977  +      memset(aMerge, 0, nMerge*sizeof(aMerge[0]));
  1708   1978       }
  1709   1979     }
  1710   1980   
  1711   1981     if( rc==SQLITE_OK ){
  1712         -    pIncr = (IncrMerger*)sqlite3_malloc(sizeof(IncrMerger));
  1713         -    if( pIncr==0 ){
  1714         -      rc = SQLITE_NOMEM;
  1715         -    }else{
  1716         -      memset(pIncr, 0, sizeof(IncrMerger));
  1717         -      pIncr->mxSz = (pSorter->mxPmaSize / 2);
  1718         -      pIncr->pMerger = pMerger;
  1719         -      pIncr->pTask = pTask0;
         1982  +    SortSubtask *pLast = &pSorter->aTask[pSorter->nTask-1];
         1983  +
         1984  +    rc = vdbeSortAllocUnpacked(pLast);
         1985  +    if( rc==SQLITE_OK ){
         1986  +      pIter->pIncr = vdbeIncrNew(pLast, pMain);
         1987  +      if( pIter->pIncr==0 ){
         1988  +        rc = SQLITE_NOMEM;
         1989  +      }else{
         1990  +        vdbeIncrSetThreads(pIter->pIncr, pSorter->bUseThreads);
         1991  +        for(iTask=0; iTask<(pSorter->nTask-1); iTask++){
         1992  +          IncrMerger *pIncr;
         1993  +          if( (pIncr = pMain->aIter[iTask].pIncr) ){
         1994  +            vdbeIncrSetThreads(pIncr, pSorter->bUseThreads);
         1995  +            assert( pIncr->pTask!=pLast );
         1996  +          }
         1997  +        }
         1998  +      }
  1720   1999       }
  1721   2000     }
  1722         -
  1723         -  /* Open the two temp files. */
  1724   2001     if( rc==SQLITE_OK ){
  1725         -    rc = vdbeSorterOpenTempFile(pTask0->db->pVfs, &pIncr->aFile[0].pFd);
  1726         -  }
  1727         -  if( rc==SQLITE_OK ){
  1728         -    rc = vdbeSorterOpenTempFile(pTask0->db->pVfs, &pIncr->aFile[1].pFd);
         2002  +    rc = vdbeIncrInit2(pIter);
  1729   2003     }
  1730   2004   
  1731         -  /* Launch a background thread to populate aFile[1]. */
  1732         -  if( rc==SQLITE_OK ){
  1733         -    rc = vdbeIncrBgPopulate(pIncr);
  1734         -  }
  1735         -
  1736         -  pIter->pIncr = pIncr;
  1737         -  if( rc==SQLITE_OK ){
  1738         -    rc = vdbePmaReaderNext(pIter);
  1739         -  }
         2005  +  sqlite3_free(aMerge);
  1740   2006     return rc;
  1741   2007   }
  1742   2008   
  1743   2009   
  1744   2010   /*
  1745   2011   ** Once the sorter has been populated by calls to sqlite3VdbeSorterWrite,
  1746   2012   ** this function is called to prepare for iterating through the records

Changes to test/sort2.test.

    11     11   # This file implements regression tests for SQLite library. 
    12     12   #
    13     13   
    14     14   set testdir [file dirname $argv0]
    15     15   source $testdir/tester.tcl
    16     16   set testprefix sort2
    17     17   
    18         -db close
    19         -sqlite3_shutdown
    20         -sqlite3_config_worker_threads 7
    21         -reset_db
    22         -
    23         -do_execsql_test 1 {
    24         -  PRAGMA cache_size = 5;
    25         -  WITH r(x,y) AS (
    26         -    SELECT 1, randomblob(100)
    27         -    UNION ALL
    28         -    SELECT x+1, randomblob(100) FROM r
    29         -    LIMIT 100000
    30         -  )
    31         -  SELECT count(x), length(y) FROM r GROUP BY (x%5)
           18  +foreach {tn script} {
           19  +  1 { }
           20  +  2 {
           21  +    catch { db close }
           22  +    sqlite3_shutdown
           23  +    sqlite3_config_worker_threads 7
           24  +    reset_db
           25  +  }
    32     26   } {
    33         -  20000 100 20000 100 20000 100 20000 100 20000 100
    34         -}
           27  +
           28  +  eval $script
           29  +
           30  +  do_execsql_test $tn.1 {
           31  +    PRAGMA cache_size = 5;
           32  +    WITH r(x,y) AS (
           33  +      SELECT 1, randomblob(100)
           34  +      UNION ALL
           35  +      SELECT x+1, randomblob(100) FROM r
           36  +      LIMIT 100000
           37  +    )
           38  +    SELECT count(x), length(y) FROM r GROUP BY (x%5)
           39  +  } {
           40  +    20000 100 20000 100 20000 100 20000 100 20000 100
           41  +  }
    35     42   
    36         -do_execsql_test 2.1 {
    37         -  CREATE TABLE t1(a, b);
    38         -  WITH r(x,y) AS (
    39         -    SELECT 1, randomblob(100)
    40         -    UNION ALL
    41         -    SELECT x+1, randomblob(100) FROM r
    42         -    LIMIT 10000
    43         -  ) INSERT INTO t1 SELECT * FROM r;
           43  +  do_execsql_test $tn.2.1 {
           44  +    CREATE TABLE t1(a, b);
           45  +    WITH r(x,y) AS (
           46  +      SELECT 1, randomblob(100)
           47  +      UNION ALL
           48  +      SELECT x+1, randomblob(100) FROM r
           49  +      LIMIT 10000
           50  +    ) INSERT INTO t1 SELECT * FROM r;
           51  +  }
           52  +  
           53  +  do_execsql_test $tn.2.2 {
           54  +    CREATE UNIQUE INDEX i1 ON t1(b, a);
           55  +  }
           56  +  
           57  +  do_execsql_test $tn.2.3 {
           58  +    CREATE UNIQUE INDEX i2 ON t1(a);
           59  +  }
           60  +  
           61  +  do_execsql_test $tn.2.4 { PRAGMA integrity_check } {ok}
           62  +  
           63  +  breakpoint
           64  +  do_execsql_test $tn.3 {
           65  +    PRAGMA cache_size = 5;
           66  +    WITH r(x,y) AS (
           67  +      SELECT 1, randomblob(100)
           68  +      UNION ALL
           69  +      SELECT x+1, randomblob(100) FROM r
           70  +      LIMIT 1000000
           71  +    )
           72  +    SELECT count(x), length(y) FROM r GROUP BY (x%5)
           73  +  } {
           74  +    200000 100 200000 100 200000 100 200000 100 200000 100
           75  +  }
           76  +  
           77  +  db close
           78  +  sqlite3_shutdown
           79  +  sqlite3_config_worker_threads 0
           80  +  sqlite3_initialize
           81  +
    44     82   }
    45     83   
    46         -do_execsql_test 2.2 {
    47         -  CREATE UNIQUE INDEX i1 ON t1(b, a);
    48         -}
    49         -
    50         -do_execsql_test 2.3 {
    51         -  CREATE UNIQUE INDEX i2 ON t1(a);
    52         -}
    53         -
    54         -do_execsql_test 2.4 { PRAGMA integrity_check } {ok}
    55         -
    56         -db close
    57         -sqlite3_shutdown
    58         -sqlite3_config_worker_threads 0
    59         -sqlite3_initialize
    60     84   finish_test
    61     85