tsort.c 25.6 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16
/*
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
 *
 * This program is free software: you can use, redistribute, and/or modify
 * it under the terms of the GNU Affero General Public License, version 3
 * or later ("AGPL"), as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */

#include "query.h"
H
Hongze Cheng 已提交
17
#include "tcommon.h"
18

H
Hongze Cheng 已提交
19
#include "tcompare.h"
H
Haojun Liao 已提交
20
#include "tdatablock.h"
S
Shengliang Guan 已提交
21
#include "tdef.h"
22 23
#include "tlosertree.h"
#include "tpagedbuf.h"
H
Haojun Liao 已提交
24
#include "tsort.h"
25 26
#include "tutil.h"

27 28 29 30 31
struct STupleHandle {
  SSDataBlock* pBlock;
  int32_t      rowIndex;
};

H
Haojun Liao 已提交
32
struct SSortHandle {
H
Hongze Cheng 已提交
33 34 35 36
  int32_t        type;
  int32_t        pageSize;
  int32_t        numOfPages;
  SDiskbasedBuf* pBuf;
37

H
Hongze Cheng 已提交
38 39
  SArray* pSortInfo;
  SArray* pOrderedSource;
40

H
Hongze Cheng 已提交
41 42 43 44
  int32_t  loops;
  uint64_t sortElapsed;
  int64_t  startTs;
  uint64_t totalElapsed;
45 46

  int32_t           sourceId;
H
Hongze Cheng 已提交
47
  SSDataBlock*      pDataBlock;
48 49 50
  SMsortComparParam cmpParam;
  int32_t           numOfCompletedSources;
  bool              opened;
H
Hongze Cheng 已提交
51
  const char*       idStr;
52 53 54
  bool              inMemSort;
  bool              needAdjust;
  STupleHandle      tupleHandle;
H
Hongze Cheng 已提交
55
  void*             param;
56
  void (*beforeFp)(SSDataBlock* pBlock, void* param);
57 58 59

  _sort_fetch_block_fn_t  fetchfp;
  _sort_merge_compar_fn_t comparFn;
H
Hongze Cheng 已提交
60
  SMultiwayMergeTreeInfo* pMergeTree;
H
Haojun Liao 已提交
61
};
62

H
Hongze Cheng 已提交
63
static int32_t msortComparFn(const void* pLeft, const void* pRight, void* param);
64

65 66
SSDataBlock* tsortGetSortedDataBlock(const SSortHandle* pSortHandle) {
  return createOneDataBlock(pSortHandle->pDataBlock, false);
67 68 69 70 71 72 73
}

/**
 *
 * @param type
 * @return
 */
H
Hongze Cheng 已提交
74 75
SSortHandle* tsortCreateSortHandle(SArray* pSortInfo, int32_t type, int32_t pageSize, int32_t numOfPages,
                                   SSDataBlock* pBlock, const char* idstr) {
wafwerar's avatar
wafwerar 已提交
76
  SSortHandle* pSortHandle = taosMemoryCalloc(1, sizeof(SSortHandle));
77

H
Hongze Cheng 已提交
78 79
  pSortHandle->type = type;
  pSortHandle->pageSize = pageSize;
80
  pSortHandle->numOfPages = numOfPages;
H
Hongze Cheng 已提交
81 82
  pSortHandle->pSortInfo = pSortInfo;
  pSortHandle->loops = 0;
83 84 85 86

  if (pBlock != NULL) {
    pSortHandle->pDataBlock = createOneDataBlock(pBlock, false);
  }
H
Haojun Liao 已提交
87

H
Hongze Cheng 已提交
88
  pSortHandle->pOrderedSource = taosArrayInit(4, POINTER_BYTES);
H
Haojun Liao 已提交
89
  pSortHandle->cmpParam.orderInfo = pSortInfo;
90
  pSortHandle->cmpParam.cmpGroupId = false;
91

H
Haojun Liao 已提交
92
  tsortSetComparFp(pSortHandle, msortComparFn);
93 94

  if (idstr != NULL) {
95
    pSortHandle->idStr = strdup(idstr);
96 97 98 99 100
  }

  return pSortHandle;
}

101
static int32_t sortComparCleanup(SMsortComparParam* cmpParam) {
H
Hongze Cheng 已提交
102 103 104
  for (int32_t i = 0; i < cmpParam->numOfSources; ++i) {
    SSortSource* pSource =
        cmpParam->pSources[i];  // NOTICE: pSource may be SGenericSource *, if it is SORT_MULTISOURCE_MERGE
wmmhello's avatar
wmmhello 已提交
105 106 107 108 109 110 111 112
    blockDataDestroy(pSource->src.pBlock);
    taosMemoryFreeClear(pSource);
  }

  cmpParam->numOfSources = 0;
  return TSDB_CODE_SUCCESS;
}

113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128
void tsortClearOrderdSource(SArray *pOrderedSource) {
  for (size_t i = 0; i < taosArrayGetSize(pOrderedSource); i++) {
    SSortSource** pSource = taosArrayGet(pOrderedSource, i);
    if (NULL == *pSource) {
      continue;
    }
    
    if ((*pSource)->param && !(*pSource)->onlyRef) {
      taosMemoryFree((*pSource)->param);  
    }
    taosMemoryFreeClear(*pSource);
  }

  taosArrayClear(pOrderedSource);
}

H
Haojun Liao 已提交
129
void tsortDestroySortHandle(SSortHandle* pSortHandle) {
130 131 132 133
  if (pSortHandle == NULL) {
    return;
  }

H
Haojun Liao 已提交
134
  tsortClose(pSortHandle);
135 136 137 138
  if (pSortHandle->pMergeTree != NULL) {
    tMergeTreeDestroy(pSortHandle->pMergeTree);
  }

H
Haojun Liao 已提交
139
  destroyDiskbasedBuf(pSortHandle->pBuf);
wafwerar's avatar
wafwerar 已提交
140
  taosMemoryFreeClear(pSortHandle->idStr);
wmmhello's avatar
wmmhello 已提交
141
  blockDataDestroy(pSortHandle->pDataBlock);
142 143

  tsortClearOrderdSource(pSortHandle->pOrderedSource);
wmmhello's avatar
wmmhello 已提交
144
  taosArrayDestroy(pSortHandle->pOrderedSource);
wafwerar's avatar
wafwerar 已提交
145
  taosMemoryFreeClear(pSortHandle);
146 147
}

H
Haojun Liao 已提交
148
int32_t tsortAddSource(SSortHandle* pSortHandle, void* pSource) {
H
Haojun Liao 已提交
149
  taosArrayPush(pSortHandle->pOrderedSource, &pSource);
150
  return TSDB_CODE_SUCCESS;
151 152
}

H
Hongze Cheng 已提交
153 154
static int32_t doAddNewExternalMemSource(SDiskbasedBuf* pBuf, SArray* pAllSources, SSDataBlock* pBlock,
                                         int32_t* sourceId, SArray* pPageIdList) {
wmmhello's avatar
wmmhello 已提交
155
  SSortSource* pSource = taosMemoryCalloc(1, sizeof(SSortSource));
156
  if (pSource == NULL) {
H
Haojun Liao 已提交
157
    taosArrayDestroy(pPageIdList);
158 159 160 161
    return TSDB_CODE_QRY_OUT_OF_MEMORY;
  }

  pSource->src.pBlock = pBlock;
162
  pSource->pageIdList = pPageIdList;
163 164 165 166 167
  taosArrayPush(pAllSources, &pSource);

  (*sourceId) += 1;

  int32_t rowSize = blockDataGetSerialRowSize(pSource->src.pBlock);
168 169

  // The value of numOfRows must be greater than 0, which is guaranteed by the previous memory allocation
H
Hongze Cheng 已提交
170 171
  int32_t numOfRows =
      (getBufPageSize(pBuf) - blockDataGetSerialMetaSize(taosArrayGetSize(pBlock->pDataBlock))) / rowSize;
wmmhello's avatar
wmmhello 已提交
172
  ASSERT(numOfRows > 0);
H
Haojun Liao 已提交
173

174 175 176 177 178 179 180
  return blockDataEnsureCapacity(pSource->src.pBlock, numOfRows);
}

static int32_t doAddToBuf(SSDataBlock* pDataBlock, SSortHandle* pHandle) {
  int32_t start = 0;

  if (pHandle->pBuf == NULL) {
wafwerar's avatar
wafwerar 已提交
181 182 183 184 185
    if (!osTempSpaceAvailable()) {
      terrno = TSDB_CODE_NO_AVAIL_DISK;
      qError("Add to buf failed since %s", terrstr(terrno));
      return terrno;
    }
H
Hongze Cheng 已提交
186 187
    int32_t code = createDiskbasedBuf(&pHandle->pBuf, pHandle->pageSize, pHandle->numOfPages * pHandle->pageSize,
                                      "doAddToBuf", tsTempDir);
H
Haojun Liao 已提交
188
    dBufSetPrintInfo(pHandle->pBuf);
189 190 191 192 193
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }
  }

194
  SArray* pPageIdList = taosArrayInit(4, sizeof(int32_t));
H
Hongze Cheng 已提交
195
  while (start < pDataBlock->info.rows) {
196
    int32_t stop = 0;
197
    blockDataSplitRows(pDataBlock, pDataBlock->info.hasVarCol, start, &stop, pHandle->pageSize);
198 199
    SSDataBlock* p = blockDataExtractBlock(pDataBlock, start, stop - start + 1);
    if (p == NULL) {
H
Haojun Liao 已提交
200
      taosArrayDestroy(pPageIdList);
201 202 203 204
      return terrno;
    }

    int32_t pageId = -1;
H
Hongze Cheng 已提交
205
    void*   pPage = getNewBufPage(pHandle->pBuf, &pageId);
206
    if (pPage == NULL) {
wmmhello's avatar
wmmhello 已提交
207
      blockDataDestroy(p);
208 209 210
      return terrno;
    }

211 212
    taosArrayPush(pPageIdList, &pageId);

H
Hongze Cheng 已提交
213
    int32_t size = blockDataGetSize(p) + sizeof(int32_t) + taosArrayGetSize(p->pDataBlock) * sizeof(int32_t);
214 215
    assert(size <= getBufPageSize(pHandle->pBuf));

216
    blockDataToBuf(pPage, p);
217 218 219 220 221 222 223 224

    setBufPageDirty(pPage, true);
    releaseBufPage(pHandle->pBuf, pPage);

    blockDataDestroy(p);
    start = stop + 1;
  }

225
  blockDataCleanup(pDataBlock);
226

227
  SSDataBlock* pBlock = createOneDataBlock(pDataBlock, false);
228
  return doAddNewExternalMemSource(pHandle->pBuf, pHandle->pOrderedSource, pBlock, &pHandle->sourceId, pPageIdList);
229 230
}

231 232 233 234 235
static void setCurrentSourceIsDone(SSortSource* pSource, SSortHandle* pHandle) {
  pSource->src.rowIndex = -1;
  ++pHandle->numOfCompletedSources;
}

H
Hongze Cheng 已提交
236 237 238
static int32_t sortComparInit(SMsortComparParam* cmpParam, SArray* pSources, int32_t startIndex, int32_t endIndex,
                              SSortHandle* pHandle) {
  cmpParam->pSources = taosArrayGet(pSources, startIndex);
239 240 241 242
  cmpParam->numOfSources = (endIndex - startIndex + 1);

  int32_t code = 0;

H
Haojun Liao 已提交
243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258
  // multi-pass internal merge sort is required
  if (pHandle->pBuf == NULL) {
    if (!osTempSpaceAvailable()) {
      code = TSDB_CODE_NO_AVAIL_DISK;
      qError("Sort compare init failed since %s", terrstr(code));
      return code;
    }

    code = createDiskbasedBuf(&pHandle->pBuf, pHandle->pageSize, pHandle->numOfPages * pHandle->pageSize,
                              "sortComparInit", tsTempDir);
    dBufSetPrintInfo(pHandle->pBuf);
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }
  }

H
Haojun Liao 已提交
259
  if (pHandle->type == SORT_SINGLESOURCE_SORT) {
260
    for (int32_t i = 0; i < cmpParam->numOfSources; ++i) {
wmmhello's avatar
wmmhello 已提交
261
      SSortSource* pSource = cmpParam->pSources[i];
262

263
      // set current source is done
264
      if (taosArrayGetSize(pSource->pageIdList) == 0) {
265 266
        setCurrentSourceIsDone(pSource, pHandle);
        continue;
267 268
      }

D
dapan1121 已提交
269
      int32_t* pPgId = taosArrayGet(pSource->pageIdList, pSource->pageIndex);
270

D
dapan1121 已提交
271
      void* pPage = getBufPage(pHandle->pBuf, *pPgId);
272
      code = blockDataFromBuf(pSource->src.pBlock, pPage);
273 274 275 276 277 278 279 280
      if (code != TSDB_CODE_SUCCESS) {
        return code;
      }

      releaseBufPage(pHandle->pBuf, pPage);
    }
  } else {
    for (int32_t i = 0; i < cmpParam->numOfSources; ++i) {
wmmhello's avatar
wmmhello 已提交
281
      SSortSource* pSource = cmpParam->pSources[i];
282
      pSource->src.pBlock = pHandle->fetchfp(pSource->param);
283

284
      // set current source is done
285
      if (pSource->src.pBlock == NULL) {
286
        setCurrentSourceIsDone(pSource, pHandle);
287
      }
288 289 290 291 292 293
    }
  }

  return code;
}

H
Hongze Cheng 已提交
294
static void appendOneRowToDataBlock(SSDataBlock* pBlock, const SSDataBlock* pSource, int32_t* rowIndex) {
295
  for (int32_t i = 0; i < taosArrayGetSize(pBlock->pDataBlock); ++i) {
296 297 298
    SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, i);

    SColumnInfoData* pSrcColInfo = taosArrayGet(pSource->pDataBlock, i);
H
Hongze Cheng 已提交
299
    bool             isNull = colDataIsNull(pSrcColInfo, pSource->info.rows, *rowIndex, NULL);
300 301 302 303

    if (isNull) {
      colDataAppend(pColInfo, pBlock->info.rows, NULL, true);
    } else {
304
      char* pData = colDataGetData(pSrcColInfo, *rowIndex);
305 306 307 308 309 310 311 312
      colDataAppend(pColInfo, pBlock->info.rows, pData, false);
    }
  }

  pBlock->info.rows += 1;
  *rowIndex += 1;
}

H
Hongze Cheng 已提交
313 314
static int32_t adjustMergeTreeForNextTuple(SSortSource* pSource, SMultiwayMergeTreeInfo* pTree, SSortHandle* pHandle,
                                           int32_t* numOfCompleted) {
315 316 317 318 319 320 321
  /*
   * load a new SDataBlock into memory of a given intermediate data-set source,
   * since it's last record in buffer has been chosen to be processed, as the winner of loser-tree
   */
  if (pSource->src.rowIndex >= pSource->src.pBlock->info.rows) {
    pSource->src.rowIndex = 0;

322
    if (pHandle->type == SORT_SINGLESOURCE_SORT) {
H
Hongze Cheng 已提交
323
      pSource->pageIndex++;
324 325 326 327 328 329
      if (pSource->pageIndex >= taosArrayGetSize(pSource->pageIdList)) {
        (*numOfCompleted) += 1;
        pSource->src.rowIndex = -1;
        pSource->pageIndex = -1;
        pSource->src.pBlock = blockDataDestroy(pSource->src.pBlock);
      } else {
D
dapan1121 已提交
330
        int32_t* pPgId = taosArrayGet(pSource->pageIdList, pSource->pageIndex);
331

H
Hongze Cheng 已提交
332 333
        void*   pPage = getBufPage(pHandle->pBuf, *pPgId);
        int32_t code = blockDataFromBuf(pSource->src.pBlock, pPage);
334 335 336 337 338
        if (code != TSDB_CODE_SUCCESS) {
          return code;
        }

        releaseBufPage(pHandle->pBuf, pPage);
339 340
      }
    } else {
wmmhello's avatar
wmmhello 已提交
341
      pSource->src.pBlock = pHandle->fetchfp(((SSortSource*)pSource)->param);
342 343 344
      if (pSource->src.pBlock == NULL) {
        (*numOfCompleted) += 1;
        pSource->src.rowIndex = -1;
345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365
      }
    }
  }

  /*
   * Adjust loser tree otherwise, according to new candidate data
   * if the loser tree is rebuild completed, we do not need to adjust
   */
  int32_t leafNodeIndex = tMergeTreeGetAdjustIndex(pTree);

#ifdef _DEBUG_VIEW
  printf("before adjust:\t");
  tMergeTreePrint(pTree);
#endif

  tMergeTreeAdjust(pTree, leafNodeIndex);

#ifdef _DEBUG_VIEW
  printf("\nafter adjust:\t");
  tMergeTreePrint(pTree);
#endif
366
  return TSDB_CODE_SUCCESS;
367 368
}

wmmhello's avatar
wmmhello 已提交
369
static SSDataBlock* getSortedBlockDataInner(SSortHandle* pHandle, SMsortComparParam* cmpParam, int32_t capacity) {
370
  blockDataCleanup(pHandle->pDataBlock);
371

H
Hongze Cheng 已提交
372
  while (1) {
373 374 375 376 377 378
    if (cmpParam->numOfSources == pHandle->numOfCompletedSources) {
      break;
    }

    int32_t index = tMergeTreeGetChosenIndex(pHandle->pMergeTree);

H
Hongze Cheng 已提交
379
    SSortSource* pSource = (*cmpParam).pSources[index];
380 381 382 383 384 385 386 387 388 389 390 391 392
    appendOneRowToDataBlock(pHandle->pDataBlock, pSource->src.pBlock, &pSource->src.rowIndex);

    int32_t code = adjustMergeTreeForNextTuple(pSource, pHandle->pMergeTree, pHandle, &pHandle->numOfCompletedSources);
    if (code != TSDB_CODE_SUCCESS) {
      terrno = code;
      return NULL;
    }

    if (pHandle->pDataBlock->info.rows >= capacity) {
      return pHandle->pDataBlock;
    }
  }

H
Hongze Cheng 已提交
393
  return (pHandle->pDataBlock->info.rows > 0) ? pHandle->pDataBlock : NULL;
394 395
}

H
Hongze Cheng 已提交
396 397 398
int32_t msortComparFn(const void* pLeft, const void* pRight, void* param) {
  int32_t pLeftIdx = *(int32_t*)pLeft;
  int32_t pRightIdx = *(int32_t*)pRight;
399

H
Hongze Cheng 已提交
400
  SMsortComparParam* pParam = (SMsortComparParam*)param;
401

H
Hongze Cheng 已提交
402
  SArray* pInfo = pParam->orderInfo;
403

H
Hongze Cheng 已提交
404
  SSortSource* pLeftSource = pParam->pSources[pLeftIdx];
wmmhello's avatar
wmmhello 已提交
405
  SSortSource* pRightSource = pParam->pSources[pRightIdx];
406 407 408 409 410 411 412 413 414 415 416 417 418

  // this input is exhausted, set the special value to denote this
  if (pLeftSource->src.rowIndex == -1) {
    return 1;
  }

  if (pRightSource->src.rowIndex == -1) {
    return -1;
  }

  SSDataBlock* pLeftBlock = pLeftSource->src.pBlock;
  SSDataBlock* pRightBlock = pRightSource->src.pBlock;

419 420 421 422
  if (pParam->cmpGroupId) {
    if (pLeftBlock->info.groupId != pRightBlock->info.groupId) {
      return pLeftBlock->info.groupId < pRightBlock->info.groupId ? -1 : 1;
    }
423 424
  }

H
Hongze Cheng 已提交
425
  for (int32_t i = 0; i < pInfo->size; ++i) {
426
    SBlockOrderInfo* pOrder = TARRAY_GET_ELEM(pInfo, i);
H
Haojun Liao 已提交
427
    SColumnInfoData* pLeftColInfoData = TARRAY_GET_ELEM(pLeftBlock->pDataBlock, pOrder->slotId);
428

H
Hongze Cheng 已提交
429
    bool leftNull = false;
430
    if (pLeftColInfoData->hasNull) {
431 432 433
      if (pLeftBlock->pBlockAgg == NULL) {
        leftNull = colDataIsNull_s(pLeftColInfoData, pLeftSource->src.rowIndex);
      } else {
H
Hongze Cheng 已提交
434 435
        leftNull =
            colDataIsNull(pLeftColInfoData, pLeftBlock->info.rows, pLeftSource->src.rowIndex, pLeftBlock->pBlockAgg[i]);
436
      }
437 438
    }

H
Haojun Liao 已提交
439
    SColumnInfoData* pRightColInfoData = TARRAY_GET_ELEM(pRightBlock->pDataBlock, pOrder->slotId);
H
Hongze Cheng 已提交
440
    bool             rightNull = false;
441
    if (pRightColInfoData->hasNull) {
H
Haojun Liao 已提交
442
      if (pRightBlock->pBlockAgg == NULL) {
443 444
        rightNull = colDataIsNull_s(pRightColInfoData, pRightSource->src.rowIndex);
      } else {
H
Hongze Cheng 已提交
445 446
        rightNull = colDataIsNull(pRightColInfoData, pRightBlock->info.rows, pRightSource->src.rowIndex,
                                  pRightBlock->pBlockAgg[i]);
447
      }
448 449 450
    }

    if (leftNull && rightNull) {
H
Hongze Cheng 已提交
451
      continue;  // continue to next slot
452 453 454
    }

    if (rightNull) {
H
Hongze Cheng 已提交
455
      return pOrder->nullFirst ? 1 : -1;
456 457 458
    }

    if (leftNull) {
H
Hongze Cheng 已提交
459
      return pOrder->nullFirst ? -1 : 1;
460 461
    }

H
Hongze Cheng 已提交
462
    void* left1 = colDataGetData(pLeftColInfoData, pLeftSource->src.rowIndex);
463
    void* right1 = colDataGetData(pRightColInfoData, pRightSource->src.rowIndex);
464

465
    __compar_fn_t fn = getKeyComparFunc(pLeftColInfoData->info.type, pOrder->order);
466

467 468 469 470 471
    int ret = fn(left1, right1);
    if (ret == 0) {
      continue;
    } else {
      return ret;
472 473
    }
  }
474
  return 0;
475 476 477 478
}

static int32_t doInternalMergeSort(SSortHandle* pHandle) {
  size_t numOfSources = taosArrayGetSize(pHandle->pOrderedSource);
H
Haojun Liao 已提交
479 480 481
  if (numOfSources == 0) {
    return 0;
  }
482 483 484 485 486

  // Calculate the I/O counts to complete the data sort.
  double sortPass = floorl(log2(numOfSources) / log2(pHandle->numOfPages));

  pHandle->totalElapsed = taosGetTimestampUs() - pHandle->startTs;
487 488 489 490 491 492 493

  if (sortPass > 0) {
    size_t s = pHandle->pBuf ? getTotalBufSize(pHandle->pBuf) : 0;
    qDebug("%s %d rounds mergesort required to complete the sort, first-round sorted data size:%" PRIzu
           ", sort elapsed:%" PRId64 ", total elapsed:%" PRId64,
           pHandle->idStr, (int32_t)(sortPass + 1), s, pHandle->sortElapsed, pHandle->totalElapsed);
  } else {
H
Hongze Cheng 已提交
494 495
    qDebug("%s ordered source:%" PRIzu ", available buf:%d, no need internal sort", pHandle->idStr, numOfSources,
           pHandle->numOfPages);
496
  }
497

H
Haojun Liao 已提交
498
  int32_t numOfRows = blockDataGetCapacityInRow(pHandle->pDataBlock, pHandle->pageSize);
wmmhello's avatar
wmmhello 已提交
499
  blockDataEnsureCapacity(pHandle->pDataBlock, numOfRows);
500

501 502 503
  // the initial pass + sortPass + final mergePass
  pHandle->loops = sortPass + 2;

504
  size_t numOfSorted = taosArrayGetSize(pHandle->pOrderedSource);
H
Hongze Cheng 已提交
505
  for (int32_t t = 0; t < sortPass; ++t) {
506 507 508 509 510 511 512 513
    int64_t st = taosGetTimestampUs();

    SArray* pResList = taosArrayInit(4, POINTER_BYTES);

    int32_t numOfInputSources = pHandle->numOfPages;
    int32_t sortGroup = (numOfSorted + numOfInputSources - 1) / numOfInputSources;

    // Only *numOfInputSources* can be loaded into buffer to perform the external sort.
H
Hongze Cheng 已提交
514
    for (int32_t i = 0; i < sortGroup; ++i) {
515 516 517 518 519 520 521 522 523 524 525
      pHandle->sourceId += 1;

      int32_t end = (i + 1) * numOfInputSources - 1;
      if (end > numOfSorted - 1) {
        end = numOfSorted - 1;
      }

      pHandle->cmpParam.numOfSources = end - i * numOfInputSources + 1;

      int32_t code = sortComparInit(&pHandle->cmpParam, pHandle->pOrderedSource, i * numOfInputSources, end, pHandle);
      if (code != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
526
        taosArrayDestroy(pResList);
527 528 529
        return code;
      }

H
Hongze Cheng 已提交
530 531
      code =
          tMergeTreeCreate(&pHandle->pMergeTree, pHandle->cmpParam.numOfSources, &pHandle->cmpParam, pHandle->comparFn);
532
      if (code != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
533
        taosArrayDestroy(pResList);
534 535 536
        return code;
      }

537
      SArray* pPageIdList = taosArrayInit(4, sizeof(int32_t));
538
      while (1) {
wmmhello's avatar
wmmhello 已提交
539
        SSDataBlock* pDataBlock = getSortedBlockDataInner(pHandle, &pHandle->cmpParam, numOfRows);
540 541 542 543 544
        if (pDataBlock == NULL) {
          break;
        }

        int32_t pageId = -1;
H
Hongze Cheng 已提交
545
        void*   pPage = getNewBufPage(pHandle->pBuf, &pageId);
546
        if (pPage == NULL) {
H
Haojun Liao 已提交
547 548
          taosArrayDestroy(pResList);
          taosArrayDestroy(pPageIdList);
549 550 551
          return terrno;
        }

552 553
        taosArrayPush(pPageIdList, &pageId);

H
Hongze Cheng 已提交
554 555
        int32_t size =
            blockDataGetSize(pDataBlock) + sizeof(int32_t) + taosArrayGetSize(pDataBlock->pDataBlock) * sizeof(int32_t);
556 557
        assert(size <= getBufPageSize(pHandle->pBuf));

558
        blockDataToBuf(pPage, pDataBlock);
559 560 561 562

        setBufPageDirty(pPage, true);
        releaseBufPage(pHandle->pBuf, pPage);

563
        blockDataCleanup(pDataBlock);
564 565
      }

566
      sortComparCleanup(&pHandle->cmpParam);
567 568 569
      tMergeTreeDestroy(pHandle->pMergeTree);
      pHandle->numOfCompletedSources = 0;

570
      SSDataBlock* pBlock = createOneDataBlock(pHandle->pDataBlock, false);
571
      code = doAddNewExternalMemSource(pHandle->pBuf, pResList, pBlock, &pHandle->sourceId, pPageIdList);
H
Haojun Liao 已提交
572 573
      if (code != TSDB_CODE_SUCCESS) {
        taosArrayDestroy(pResList);
574 575 576 577
        return code;
      }
    }

578
    tsortClearOrderdSource(pHandle->pOrderedSource);
579 580 581 582 583 584 585 586 587
    taosArrayAddAll(pHandle->pOrderedSource, pResList);
    taosArrayDestroy(pResList);

    numOfSorted = taosArrayGetSize(pHandle->pOrderedSource);

    int64_t el = taosGetTimestampUs() - st;
    pHandle->totalElapsed += el;

    SDiskbasedBufStatis statis = getDBufStatis(pHandle->pBuf);
H
Hongze Cheng 已提交
588 589
    qDebug("%s %d round mergesort, elapsed:%" PRId64 " readDisk:%.2f Kb, flushDisk:%.2f Kb", pHandle->idStr, t + 1, el,
           statis.loadBytes / 1024.0, statis.flushBytes / 1024.0);
590

H
Haojun Liao 已提交
591 592
    if (pHandle->type == SORT_MULTISOURCE_MERGE) {
      pHandle->type = SORT_SINGLESOURCE_SORT;
593 594 595 596 597 598 599 600
      pHandle->comparFn = msortComparFn;
    }
  }

  pHandle->cmpParam.numOfSources = taosArrayGetSize(pHandle->pOrderedSource);
  return 0;
}

601 602 603 604 605
// get sort page size
int32_t getProperSortPageSize(size_t rowSize, uint32_t numOfCols) {
  uint32_t pgSize = rowSize * 4 + blockDataGetSerialMetaSize(numOfCols);
  if (pgSize < DEFAULT_PAGESIZE) {
    return DEFAULT_PAGESIZE;
606 607 608 609 610
  }

  return pgSize;
}

611
static int32_t createInitialSources(SSortHandle* pHandle) {
612
  size_t sortBufSize = pHandle->numOfPages * pHandle->pageSize;
H
Haojun Liao 已提交
613 614

  if (pHandle->type == SORT_SINGLESOURCE_SORT) {
615 616 617 618 619
    SSortSource** pSource = taosArrayGet(pHandle->pOrderedSource, 0);
    SSortSource* source = *pSource;
    *pSource = NULL;
    
    tsortClearOrderdSource(pHandle->pOrderedSource);
620

621
    while (1) {
622
      SSDataBlock* pBlock = pHandle->fetchfp(source->param);
623 624 625 626
      if (pBlock == NULL) {
        break;
      }

627
      if (pHandle->pDataBlock == NULL) {
628 629
        uint32_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);
        pHandle->pageSize = getProperSortPageSize(blockDataGetRowSize(pBlock), numOfCols);
630 631

        // todo, number of pages are set according to the total available sort buffer
632 633
        pHandle->numOfPages = 1024;
        sortBufSize = pHandle->numOfPages * pHandle->pageSize;
634
        pHandle->pDataBlock = createOneDataBlock(pBlock, false);
635 636
      }

637 638 639
      if (pHandle->beforeFp != NULL) {
        pHandle->beforeFp(pBlock, pHandle->param);
      }
640

641 642
      int32_t code = blockDataMerge(pHandle->pDataBlock, pBlock);
      if (code != 0) {
643 644 645 646
        if (source->param && !source->onlyRef) {
          taosMemoryFree(source->param);
        }
        taosMemoryFree(source);
647 648
        return code;
      }
649

650 651 652 653
      size_t size = blockDataGetSize(pHandle->pDataBlock);
      if (size > sortBufSize) {
        // Perform the in-memory sort and then flush data in the buffer into disk.
        int64_t p = taosGetTimestampUs();
wmmhello's avatar
wmmhello 已提交
654 655
        code = blockDataSort(pHandle->pDataBlock, pHandle->pSortInfo);
        if (code != 0) {
656 657 658 659
          if (source->param && !source->onlyRef) {
            taosMemoryFree(source->param);
          }
          taosMemoryFree(source);
wmmhello's avatar
wmmhello 已提交
660 661
          return code;
        }
662

663 664
        int64_t el = taosGetTimestampUs() - p;
        pHandle->sortElapsed += el;
665

666
        doAddToBuf(pHandle->pDataBlock, pHandle);
667 668 669
      }
    }

670 671 672 673 674
    if (source->param && !source->onlyRef) {
      taosMemoryFree(source->param);
    }
    taosMemoryFree(source);

675
    if (pHandle->pDataBlock != NULL && pHandle->pDataBlock->info.rows > 0) {
676 677 678
      size_t size = blockDataGetSize(pHandle->pDataBlock);

      // Perform the in-memory sort and then flush data in the buffer into disk.
679 680
      int64_t p = taosGetTimestampUs();

wmmhello's avatar
wmmhello 已提交
681 682 683 684
      int32_t code = blockDataSort(pHandle->pDataBlock, pHandle->pSortInfo);
      if (code != 0) {
        return code;
      }
685

686 687 688
      int64_t el = taosGetTimestampUs() - p;
      pHandle->sortElapsed += el;

689
      // All sorted data can fit in memory, external memory sort is not needed. Return to directly
690
      if (size <= sortBufSize && pHandle->pBuf == NULL) {
691 692 693
        pHandle->cmpParam.numOfSources = 1;
        pHandle->inMemSort = true;

694
        pHandle->loops = 1;
H
Hongze Cheng 已提交
695
        pHandle->tupleHandle.rowIndex = -1;
696 697 698 699 700 701
        pHandle->tupleHandle.pBlock = pHandle->pDataBlock;
        return 0;
      } else {
        doAddToBuf(pHandle->pDataBlock, pHandle);
      }
    }
H
Haojun Liao 已提交
702 703 704 705 706
  }

  return TSDB_CODE_SUCCESS;
}

H
Haojun Liao 已提交
707
int32_t tsortOpen(SSortHandle* pHandle) {
H
Haojun Liao 已提交
708 709 710 711 712 713 714 715 716 717
  if (pHandle->opened) {
    return 0;
  }

  if (pHandle->fetchfp == NULL || pHandle->comparFn == NULL) {
    return -1;
  }

  pHandle->opened = true;

718
  int32_t code = createInitialSources(pHandle);
H
Haojun Liao 已提交
719 720
  if (code != TSDB_CODE_SUCCESS) {
    return code;
721 722 723
  }

  // do internal sort
H
Haojun Liao 已提交
724
  code = doInternalMergeSort(pHandle);
725 726 727 728 729
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }

  int32_t numOfSources = taosArrayGetSize(pHandle->pOrderedSource);
H
Haojun Liao 已提交
730 731 732 733
  if (pHandle->pBuf != NULL) {
    ASSERT(numOfSources <= getNumOfInMemBufPages(pHandle->pBuf));
  }

H
Haojun Liao 已提交
734 735 736 737
  if (numOfSources == 0) {
    return 0;
  }

738 739 740 741 742
  code = sortComparInit(&pHandle->cmpParam, pHandle->pOrderedSource, 0, numOfSources - 1, pHandle);
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }

743
  return tMergeTreeCreate(&pHandle->pMergeTree, pHandle->cmpParam.numOfSources, &pHandle->cmpParam, pHandle->comparFn);
744 745
}

H
Haojun Liao 已提交
746
int32_t tsortClose(SSortHandle* pHandle) {
747
  // do nothing
748
  return TSDB_CODE_SUCCESS;
749 750
}

H
Hongze Cheng 已提交
751 752
int32_t tsortSetFetchRawDataFp(SSortHandle* pHandle, _sort_fetch_block_fn_t fetchFp, void (*fp)(SSDataBlock*, void*),
                               void* param) {
753 754 755
  pHandle->fetchfp = fetchFp;
  pHandle->beforeFp = fp;
  pHandle->param = param;
756
  return TSDB_CODE_SUCCESS;
757 758
}

H
Haojun Liao 已提交
759
int32_t tsortSetComparFp(SSortHandle* pHandle, _sort_merge_compar_fn_t fp) {
760
  pHandle->comparFn = fp;
761
  return TSDB_CODE_SUCCESS;
762 763
}

764 765 766 767 768
int32_t tsortSetCompareGroupId(SSortHandle* pHandle, bool compareGroupId) {
  pHandle->cmpParam.cmpGroupId = compareGroupId;
  return TSDB_CODE_SUCCESS;
}

H
Haojun Liao 已提交
769
STupleHandle* tsortNextTuple(SSortHandle* pHandle) {
770 771 772 773
  if (pHandle->cmpParam.numOfSources == pHandle->numOfCompletedSources) {
    return NULL;
  }

774
  // All the data are hold in the buffer, no external sort is invoked.
775 776 777 778 779 780 781 782 783 784
  if (pHandle->inMemSort) {
    pHandle->tupleHandle.rowIndex += 1;
    if (pHandle->tupleHandle.rowIndex == pHandle->pDataBlock->info.rows) {
      pHandle->numOfCompletedSources = 1;
      return NULL;
    }

    return &pHandle->tupleHandle;
  }

H
Hongze Cheng 已提交
785 786
  int32_t      index = tMergeTreeGetChosenIndex(pHandle->pMergeTree);
  SSortSource* pSource = pHandle->cmpParam.pSources[index];
787 788 789 790 791 792 793 794 795 796 797 798 799

  if (pHandle->needAdjust) {
    int32_t code = adjustMergeTreeForNextTuple(pSource, pHandle->pMergeTree, pHandle, &pHandle->numOfCompletedSources);
    if (code != TSDB_CODE_SUCCESS) {
      terrno = code;
      return NULL;
    }
  }

  if (pHandle->cmpParam.numOfSources == pHandle->numOfCompletedSources) {
    return NULL;
  }

800
  // Get the adjusted value after the loser tree is updated.
801 802 803 804 805 806 807 808 809 810 811 812 813 814
  index = tMergeTreeGetChosenIndex(pHandle->pMergeTree);
  pSource = pHandle->cmpParam.pSources[index];

  assert(pSource->src.pBlock != NULL);

  pHandle->tupleHandle.rowIndex = pSource->src.rowIndex;
  pHandle->tupleHandle.pBlock = pSource->src.pBlock;

  pHandle->needAdjust = true;
  pSource->src.rowIndex += 1;

  return &pHandle->tupleHandle;
}

H
Haojun Liao 已提交
815
bool tsortIsNullVal(STupleHandle* pVHandle, int32_t colIndex) {
816
  SColumnInfoData* pColInfoSrc = taosArrayGet(pVHandle->pBlock->pDataBlock, colIndex);
817
  return colDataIsNull_s(pColInfoSrc, pVHandle->rowIndex);
818 819
}

H
Haojun Liao 已提交
820
void* tsortGetValue(STupleHandle* pVHandle, int32_t colIndex) {
821
  SColumnInfoData* pColInfo = TARRAY_GET_ELEM(pVHandle->pBlock->pDataBlock, colIndex);
822 823 824 825 826
  if (pColInfo->pData == NULL) {
    return NULL;
  } else {
    return colDataGetData(pColInfo, pVHandle->rowIndex);
  }
827
}
828

H
Hongze Cheng 已提交
829
uint64_t tsortGetGroupId(STupleHandle* pVHandle) { return pVHandle->pBlock->info.groupId; }
S
slzhou 已提交
830

831 832 833 834
SSortExecInfo tsortGetSortExecInfo(SSortHandle* pHandle) {
  SSortExecInfo info = {0};

  info.sortBuffer = pHandle->pageSize * pHandle->numOfPages;
H
Hongze Cheng 已提交
835 836
  info.sortMethod = pHandle->inMemSort ? SORT_QSORT_T : SORT_SPILLED_MERGE_SORT_T;
  info.loops = pHandle->loops;
837 838 839 840 841 842 843 844 845

  if (pHandle->pBuf != NULL) {
    SDiskbasedBufStatis st = getDBufStatis(pHandle->pBuf);
    info.writeBytes = st.flushBytes;
    info.readBytes = st.loadBytes;
  }

  return info;
}