groupoperator.c 21.0 KB
Newer Older
H
Haojun Liao 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27
/*
 * 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 "os.h"
#include "function.h"
#include "tname.h"

#include "tdatablock.h"
#include "tmsg.h"

#include "executorimpl.h"
#include "tcompare.h"
#include "thash.h"
#include "ttypes.h"

H
Haojun Liao 已提交
28
static int32_t* setupColumnOffset(const SSDataBlock* pBlock, int32_t rowCapacity);
H
Haojun Liao 已提交
29 30
static void* getCurrentDataGroupInfo(const SPartitionOperatorInfo* pInfo, SDataGroupInfo** pGroupInfo, int32_t len);
static uint64_t calcGroupId(char* pData, int32_t len);
H
Haojun Liao 已提交
31

H
Haojun Liao 已提交
32
static void destroyGroupOperatorInfo(void* param, int32_t numOfOutput) {
H
Haojun Liao 已提交
33 34 35 36 37 38 39
  SGroupbyOperatorInfo* pInfo = (SGroupbyOperatorInfo*)param;
  doDestroyBasicInfo(&pInfo->binfo, numOfOutput);
  taosMemoryFreeClear(pInfo->keyBuf);
  taosArrayDestroy(pInfo->pGroupCols);
  taosArrayDestroy(pInfo->pGroupColVals);
}

H
Haojun Liao 已提交
40 41 42
static int32_t initGroupOptrInfo(SArray** pGroupColVals, int32_t* keyLen, char** keyBuf, const SArray* pGroupColList) {
  *pGroupColVals = taosArrayInit(4, sizeof(SGroupKeys));
  if ((*pGroupColVals) == NULL) {
H
Haojun Liao 已提交
43 44 45 46 47 48
    return TSDB_CODE_OUT_OF_MEMORY;
  }

  int32_t numOfGroupCols = taosArrayGetSize(pGroupColList);
  for (int32_t i = 0; i < numOfGroupCols; ++i) {
    SColumn* pCol = taosArrayGet(pGroupColList, i);
H
Haojun Liao 已提交
49
    (*keyLen) += pCol->bytes;
H
Haojun Liao 已提交
50

51
    SGroupKeys key = {0};
H
Haojun Liao 已提交
52 53
    key.bytes  = pCol->bytes;
    key.type   = pCol->type;
H
Haojun Liao 已提交
54
    key.isNull = false;
H
Haojun Liao 已提交
55
    key.pData  = taosMemoryCalloc(1, pCol->bytes);
H
Haojun Liao 已提交
56 57 58 59
    if (key.pData == NULL) {
      return TSDB_CODE_OUT_OF_MEMORY;
    }

H
Haojun Liao 已提交
60
    taosArrayPush((*pGroupColVals), &key);
H
Haojun Liao 已提交
61 62 63 64
  }

  int32_t nullFlagSize = sizeof(int8_t) * numOfGroupCols;

H
Haojun Liao 已提交
65 66
  (*keyBuf) = taosMemoryCalloc(1, (*keyLen) + nullFlagSize);
  if ((*keyBuf) == NULL) {
H
Haojun Liao 已提交
67 68 69 70 71 72
    return TSDB_CODE_OUT_OF_MEMORY;
  }

  return TSDB_CODE_SUCCESS;
}

H
Haojun Liao 已提交
73
static bool groupKeyCompare(SArray* pGroupCols, SArray* pGroupColVals, SSDataBlock* pBlock, int32_t rowIndex, int32_t numOfGroupCols) {
H
Haojun Liao 已提交
74 75
  SColumnDataAgg* pColAgg = NULL;
  for (int32_t i = 0; i < numOfGroupCols; ++i) {
H
Haojun Liao 已提交
76
    SColumn*         pCol = taosArrayGet(pGroupCols, i);
H
Haojun Liao 已提交
77 78 79 80 81 82 83
    SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, pCol->slotId);
    if (pBlock->pBlockAgg != NULL) {
      pColAgg = &pBlock->pBlockAgg[pCol->slotId];  // TODO is agg data matched?
    }

    bool isNull = colDataIsNull(pColInfoData, pBlock->info.rows, rowIndex, pColAgg);

H
Haojun Liao 已提交
84
    SGroupKeys* pkey = taosArrayGet(pGroupColVals, i);
H
Haojun Liao 已提交
85 86 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
    if (pkey->isNull && isNull) {
      continue;
    }

    if (isNull || pkey->isNull) {
      return false;
    }

    char* val = colDataGetData(pColInfoData, rowIndex);

    if (IS_VAR_DATA_TYPE(pkey->type)) {
      int32_t len = varDataLen(val);
      if (len == varDataLen(pkey->pData) && memcmp(varDataVal(pkey->pData), varDataVal(val), len) == 0) {
        continue;
      } else {
        return false;
      }
    } else {
      if (memcmp(pkey->pData, val, pkey->bytes) != 0) {
        return false;
      }
    }
  }

  return true;
}

H
Haojun Liao 已提交
112
static void recordNewGroupKeys(SArray* pGroupCols, SArray* pGroupColVals, SSDataBlock* pBlock, int32_t rowIndex, int32_t numOfGroupCols) {
H
Haojun Liao 已提交
113 114 115
  SColumnDataAgg* pColAgg = NULL;

  for (int32_t i = 0; i < numOfGroupCols; ++i) {
H
Haojun Liao 已提交
116
    SColumn*         pCol = taosArrayGet(pGroupCols, i);
H
Haojun Liao 已提交
117 118 119 120 121 122
    SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, pCol->slotId);

    if (pBlock->pBlockAgg != NULL) {
      pColAgg = &pBlock->pBlockAgg[pCol->slotId];  // TODO is agg data matched?
    }

H
Haojun Liao 已提交
123
    SGroupKeys* pkey = taosArrayGet(pGroupColVals, i);
H
Haojun Liao 已提交
124 125 126
    if (colDataIsNull(pColInfoData, pBlock->info.rows, rowIndex, pColAgg)) {
      pkey->isNull = true;
    } else {
127
      pkey->isNull = false;
H
Haojun Liao 已提交
128 129 130 131 132 133 134 135 136 137
      char* val = colDataGetData(pColInfoData, rowIndex);
      if (IS_VAR_DATA_TYPE(pkey->type)) {
        memcpy(pkey->pData, val, varDataTLen(val));
      } else {
        memcpy(pkey->pData, val, pkey->bytes);
      }
    }
  }
}

H
Haojun Liao 已提交
138
static int32_t buildGroupKeys(void* pKey, const SArray* pGroupColVals) {
H
Haojun Liao 已提交
139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161
  ASSERT(pKey != NULL);
  size_t numOfGroupCols = taosArrayGetSize(pGroupColVals);

  char* isNull = (char*)pKey;
  char* pStart = (char*)pKey + sizeof(int8_t) * numOfGroupCols;
  for (int32_t i = 0; i < numOfGroupCols; ++i) {
    SGroupKeys* pkey = taosArrayGet(pGroupColVals, i);
    if (pkey->isNull) {
      isNull[i] = 1;
      continue;
    }

    isNull[i] = 0;
    if (IS_VAR_DATA_TYPE(pkey->type)) {
      varDataCopy(pStart, pkey->pData);
      pStart += varDataTLen(pkey->pData);
      ASSERT(varDataTLen(pkey->pData) <= pkey->bytes);
    } else {
      memcpy(pStart, pkey->pData, pkey->bytes);
      pStart += pkey->bytes;
    }
  }

H
Haojun Liao 已提交
162
  return (int32_t) (pStart - (char*)pKey);
H
Haojun Liao 已提交
163 164 165 166 167 168 169 170 171 172 173 174 175 176
}

// assign the group keys or user input constant values if required
static void doAssignGroupKeys(SqlFunctionCtx* pCtx, int32_t numOfOutput, int32_t totalRows, int32_t rowIndex) {
  for (int32_t i = 0; i < numOfOutput; ++i) {
    if (pCtx[i].functionId == -1) {
      SResultRowEntryInfo* pEntryInfo = GET_RES_INFO(&pCtx[i]);

      SColumnInfoData* pColInfoData = pCtx[i].input.pData[0];
      if (!colDataIsNull(pColInfoData, totalRows, rowIndex, NULL)) {
        char* dest = GET_ROWCELL_INTERBUF(pEntryInfo);
        char* data = colDataGetData(pColInfoData, rowIndex);

        memcpy(dest, data, pColInfoData->info.bytes);
H
Haojun Liao 已提交
177 178
      } else { // it is a NULL value
        pEntryInfo->isNullRes = 1;
H
Haojun Liao 已提交
179
      }
H
Haojun Liao 已提交
180 181

      pEntryInfo->numOfRes = 1;
H
Haojun Liao 已提交
182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203
    }
  }
}

static void doHashGroupbyAgg(SOperatorInfo* pOperator, SSDataBlock* pBlock) {
  SExecTaskInfo*        pTaskInfo = pOperator->pTaskInfo;
  SGroupbyOperatorInfo* pInfo = pOperator->info;

  SqlFunctionCtx* pCtx = pInfo->binfo.pCtx;
  int32_t         numOfGroupCols = taosArrayGetSize(pInfo->pGroupCols);
  //  if (type == TSDB_DATA_TYPE_FLOAT || type == TSDB_DATA_TYPE_DOUBLE) {
  // qError("QInfo:0x%"PRIx64" group by not supported on double/float columns, abort", GET_TASKID(pRuntimeEnv));
  //    return;
  //  }

  int32_t     len = 0;
  STimeWindow w = TSWINDOW_INITIALIZER;

  int32_t num = 0;
  for (int32_t j = 0; j < pBlock->info.rows; ++j) {
    // Compare with the previous row of this column, and do not set the output buffer again if they are identical.
    if (!pInfo->isInit) {
H
Haojun Liao 已提交
204
      recordNewGroupKeys(pInfo->pGroupCols, pInfo->pGroupColVals, pBlock, j, numOfGroupCols);
H
Haojun Liao 已提交
205 206 207 208 209
      pInfo->isInit = true;
      num++;
      continue;
    }

H
Haojun Liao 已提交
210
    bool equal = groupKeyCompare(pInfo->pGroupCols, pInfo->pGroupColVals, pBlock, j, numOfGroupCols);
H
Haojun Liao 已提交
211 212 213 214 215
    if (equal) {
      num++;
      continue;
    }

H
Haojun Liao 已提交
216
    // The first row of a new block does not belongs to the previous existed group
H
Haojun Liao 已提交
217 218
    if (!equal && j == 0) {
      num++;
H
Haojun Liao 已提交
219
      recordNewGroupKeys(pInfo->pGroupCols, pInfo->pGroupColVals, pBlock, j, numOfGroupCols);
H
Haojun Liao 已提交
220 221 222
      continue;
    }

H
Haojun Liao 已提交
223
    len = buildGroupKeys(pInfo->keyBuf, pInfo->pGroupColVals);
H
Haojun Liao 已提交
224 225 226 227 228 229 230 231 232 233
    int32_t ret = setGroupResultOutputBuf_rv(&(pInfo->binfo), pOperator->numOfOutput, pInfo->keyBuf, TSDB_DATA_TYPE_VARCHAR, len, 0, pInfo->aggSup.pResultBuf, pTaskInfo, &pInfo->aggSup);
    if (ret != TSDB_CODE_SUCCESS) {  // null data, too many state code
      longjmp(pTaskInfo->env, TSDB_CODE_QRY_APP_ERROR);
    }

    int32_t rowIndex = j - num;
    doApplyFunctions(pCtx, &w, NULL, rowIndex, num, NULL, pBlock->info.rows, pOperator->numOfOutput, TSDB_ORDER_ASC);

    // assign the group keys or user input constant values if required
    doAssignGroupKeys(pCtx, pOperator->numOfOutput, pBlock->info.rows, rowIndex);
H
Haojun Liao 已提交
234
    recordNewGroupKeys(pInfo->pGroupCols, pInfo->pGroupColVals, pBlock, j, numOfGroupCols);
H
Haojun Liao 已提交
235 236 237 238
    num = 1;
  }

  if (num > 0) {
H
Haojun Liao 已提交
239
    len = buildGroupKeys(pInfo->keyBuf, pInfo->pGroupColVals);
H
Haojun Liao 已提交
240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258
    int32_t ret =
        setGroupResultOutputBuf_rv(&(pInfo->binfo), pOperator->numOfOutput, pInfo->keyBuf, TSDB_DATA_TYPE_VARCHAR, len,
                                   0, pInfo->aggSup.pResultBuf, pTaskInfo, &pInfo->aggSup);
    if (ret != TSDB_CODE_SUCCESS) {
      longjmp(pTaskInfo->env, TSDB_CODE_QRY_APP_ERROR);
    }

    int32_t rowIndex = pBlock->info.rows - num;
    doApplyFunctions(pCtx, &w, NULL, rowIndex, num, NULL, pBlock->info.rows, pOperator->numOfOutput, TSDB_ORDER_ASC);
    doAssignGroupKeys(pCtx, pOperator->numOfOutput, pBlock->info.rows, rowIndex);
  }
}

static SSDataBlock* hashGroupbyAggregate(SOperatorInfo* pOperator, bool* newgroup) {
  if (pOperator->status == OP_EXEC_DONE) {
    return NULL;
  }

  SGroupbyOperatorInfo* pInfo = pOperator->info;
259 260
  SSDataBlock* pRes = pInfo->binfo.pRes;

H
Haojun Liao 已提交
261
  if (pOperator->status == OP_RES_TO_RETURN) {
262 263
    toSDatablock(&pInfo->groupResInfo, pInfo->aggSup.pResultBuf, pRes, pInfo->binfo.capacity, pInfo->binfo.rowCellInfoOffset);
    if (pRes->info.rows == 0 || !hasRemainDataInCurrentGroup(&pInfo->groupResInfo)) {
H
Haojun Liao 已提交
264 265
      pOperator->status = OP_EXEC_DONE;
    }
H
Haojun Liao 已提交
266
    return (pRes->info.rows == 0)? NULL:pRes;
H
Haojun Liao 已提交
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
  }

  int32_t        order = TSDB_ORDER_ASC;
  SOperatorInfo* downstream = pOperator->pDownstream[0];

  while (1) {
    publishOperatorProfEvent(downstream, QUERY_PROF_BEFORE_OPERATOR_EXEC);
    SSDataBlock* pBlock = downstream->getNextFn(downstream, newgroup);
    publishOperatorProfEvent(downstream, QUERY_PROF_AFTER_OPERATOR_EXEC);
    if (pBlock == NULL) {
      break;
    }

    // the pDataBlock are always the same one, no need to call this again
    setInputDataBlock(pOperator, pInfo->binfo.pCtx, pBlock, order);
    //    setTagValue(pOperator, pRuntimeEnv->current->pTable, pInfo->binfo.pCtx, pOperator->numOfOutput);
    doHashGroupbyAgg(pOperator, pBlock);
  }

  pOperator->status = OP_RES_TO_RETURN;
  closeAllResultRows(&pInfo->binfo.resultRowInfo);

  finalizeMultiTupleQueryResult(pInfo->binfo.pCtx, pOperator->numOfOutput, pInfo->aggSup.pResultBuf,
                                &pInfo->binfo.resultRowInfo, pInfo->binfo.rowCellInfoOffset);
  //  if (!stableQuery) { // finalize include the update of result rows
  //    finalizeQueryResult(pInfo->binfo.pCtx, pOperator->numOfOutput);
  //  } else {
  //    updateNumOfRowsInResultRows(pInfo->binfo.pCtx, pOperator->numOfOutput, &pInfo->binfo.resultRowInfo,
  //    pInfo->binfo.rowCellInfoOffset);
  //  }

298
  blockDataEnsureCapacity(pRes, pInfo->binfo.capacity);
H
Haojun Liao 已提交
299
  initGroupResInfo(&pInfo->groupResInfo, &pInfo->binfo.resultRowInfo);
300 301

  while(1) {
H
Haojun Liao 已提交
302
    toSDatablock(&pInfo->groupResInfo, pInfo->aggSup.pResultBuf, pRes, pInfo->binfo.capacity, pInfo->binfo.rowCellInfoOffset);
303 304 305 306 307 308 309 310 311 312 313
    doFilter(pInfo->pCondition, pRes);

    bool hasRemain = hasRemainDataInCurrentGroup(&pInfo->groupResInfo);
    if (!hasRemain) {
      pOperator->status = OP_EXEC_DONE;
      break;
    }

    if (pRes->info.rows > 0) {
      break;
    }
H
Haojun Liao 已提交
314 315
  }

H
Haojun Liao 已提交
316
  return (pRes->info.rows == 0)? NULL:pRes;
H
Haojun Liao 已提交
317 318
}

319
SOperatorInfo* createGroupOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResultBlock, SArray* pGroupColList, SNode* pCondition, SExecTaskInfo* pTaskInfo,
H
Haojun Liao 已提交
320 321 322 323 324 325 326 327
                                       const STableGroupInfo* pTableGroupInfo) {
  SGroupbyOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SGroupbyOperatorInfo));
  SOperatorInfo*        pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
  if (pInfo == NULL || pOperator == NULL) {
    goto _error;
  }

  pInfo->pGroupCols = pGroupColList;
328
  pInfo->pCondition = pCondition;
H
Haojun Liao 已提交
329 330 331
  initAggInfo(&pInfo->binfo, &pInfo->aggSup, pExprInfo, numOfCols, 4096, pResultBlock, pTaskInfo->id.str);
  initResultRowInfo(&pInfo->binfo.resultRowInfo, 8);

H
Haojun Liao 已提交
332
  int32_t code = initGroupOptrInfo(&pInfo->pGroupColVals, &pInfo->groupKeyLen, &pInfo->keyBuf, pGroupColList);
H
Haojun Liao 已提交
333 334 335 336 337 338 339 340 341 342 343
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }

  pOperator->name         = "GroupbyAggOperator";
  pOperator->blockingOptr = true;
  pOperator->status       = OP_NOT_OPENED;
  //  pOperator->operatorType = OP_Groupby;
  pOperator->pExpr        = pExprInfo;
  pOperator->numOfOutput  = numOfCols;
  pOperator->info         = pInfo;
344
  pOperator->pTaskInfo    = pTaskInfo;
H
Haojun Liao 已提交
345 346
  pOperator->_openFn      = operatorDummyOpenFn;
  pOperator->getNextFn    = hashGroupbyAggregate;
H
Haojun Liao 已提交
347
  pOperator->closeFn      = destroyGroupOperatorInfo;
H
Haojun Liao 已提交
348 349 350 351 352

  code = appendDownstream(pOperator, &downstream, 1);
  return pOperator;

  _error:
H
Haojun Liao 已提交
353
  pTaskInfo->code = TSDB_CODE_OUT_OF_MEMORY;
H
Haojun Liao 已提交
354 355 356
  taosMemoryFreeClear(pInfo);
  taosMemoryFreeClear(pOperator);
  return NULL;
357 358
}

H
Haojun Liao 已提交
359
static void doHashPartition(SOperatorInfo* pOperator, SSDataBlock* pBlock) {
H
Haojun Liao 已提交
360
//  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
H
Haojun Liao 已提交
361 362 363 364 365 366 367 368

  SPartitionOperatorInfo* pInfo = pOperator->info;

  int32_t numOfGroupCols = taosArrayGetSize(pInfo->pGroupCols);
  for (int32_t j = 0; j < pBlock->info.rows; ++j) {
    recordNewGroupKeys(pInfo->pGroupCols, pInfo->pGroupColVals, pBlock, j, numOfGroupCols);
    int32_t len = buildGroupKeys(pInfo->keyBuf, pInfo->pGroupColVals);

H
Haojun Liao 已提交
369 370
    SDataGroupInfo* pGInfo = NULL;
    void *pPage = getCurrentDataGroupInfo(pInfo, &pGInfo, len);
H
Haojun Liao 已提交
371

H
Haojun Liao 已提交
372 373 374
    pGInfo->numOfRows += 1;
    if (pGInfo->groupId == 0) {
      pGInfo->groupId = calcGroupId(pInfo->keyBuf, len);
H
Haojun Liao 已提交
375 376
    }

H
Haojun Liao 已提交
377
    int32_t* rows = (int32_t*) pPage;
H
Haojun Liao 已提交
378

379
    size_t numOfCols = pOperator->numOfOutput;
H
Haojun Liao 已提交
380
    for(int32_t i = 0; i < numOfCols; ++i) {
381 382 383 384
      SExprInfo* pExpr = &pOperator->pExpr[i];
      int32_t slotId = pExpr->base.pParam[0].pCol->slotId;

      SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, slotId);
H
Haojun Liao 已提交
385

H
Haojun Liao 已提交
386 387
      int32_t bytes = pColInfoData->info.bytes;
      int32_t startOffset = pInfo->columnOffset[i];
H
Haojun Liao 已提交
388

H
Haojun Liao 已提交
389 390
      char* columnLen    = NULL;
      int32_t contentLen = 0;
H
Haojun Liao 已提交
391 392

      if (IS_VAR_DATA_TYPE(pColInfoData->info.type)) {
H
Haojun Liao 已提交
393 394 395 396 397 398 399 400 401 402 403 404 405
        int32_t* offset = pPage + startOffset;
        columnLen       = pPage + startOffset + sizeof(int32_t) * pInfo->rowCapacity;
        char*    data   = (char*)(columnLen + sizeof(int32_t));

        if (colDataIsNull_s(pColInfoData, j)) {
          offset[(*rows)] = -1;
          contentLen = 0;
        } else {
          offset[*rows] = (*columnLen);
          char* src = colDataGetData(pColInfoData, j);
          memcpy(data + (*columnLen), src, varDataTLen(src));
          contentLen = varDataTLen(src);
        }
H
Haojun Liao 已提交
406
      } else {
H
Haojun Liao 已提交
407 408 409
        char* bitmap = pPage + startOffset;
        columnLen    = pPage + startOffset + BitmapLen(pInfo->rowCapacity);
        char* data   = (char*) columnLen + sizeof(int32_t);
H
Haojun Liao 已提交
410 411 412

        bool isNull = colDataIsNull_f(pColInfoData->nullbitmap, j);
        if (isNull) {
H
Haojun Liao 已提交
413
          colDataSetNull_f(bitmap, (*rows));
H
Haojun Liao 已提交
414
        } else {
H
Haojun Liao 已提交
415
          memcpy(data + (*columnLen), colDataGetData(pColInfoData, j), bytes);
H
Haojun Liao 已提交
416
        }
H
Haojun Liao 已提交
417
        contentLen = bytes;
H
Haojun Liao 已提交
418
      }
H
Haojun Liao 已提交
419 420

      (*columnLen) += contentLen;
H
Haojun Liao 已提交
421 422
    }

H
Haojun Liao 已提交
423 424
    (*rows) += 1;

H
Haojun Liao 已提交
425 426 427
    setBufPageDirty(pPage, true);
    releaseBufPage(pInfo->pBuf, pPage);
  }
H
Haojun Liao 已提交
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
}

void* getCurrentDataGroupInfo(const SPartitionOperatorInfo* pInfo, SDataGroupInfo** pGroupInfo, int32_t len) {
  SDataGroupInfo* p = taosHashGet(pInfo->pGroupSet, pInfo->keyBuf, len);

  void* pPage = NULL;
  if (p == NULL) { // it is a new group
    SDataGroupInfo gi = {0};
    gi.pPageList = taosArrayInit(100, sizeof(int32_t));
    taosHashPut(pInfo->pGroupSet, pInfo->keyBuf, len, &gi, sizeof(SDataGroupInfo));

    p = taosHashGet(pInfo->pGroupSet, pInfo->keyBuf, len);

    int32_t pageId = 0;
    pPage = getNewBufPage(pInfo->pBuf, 0, &pageId);
    taosArrayPush(p->pPageList, &pageId);

    *(int32_t *) pPage = 0;
  } else {
    int32_t* curId = taosArrayGetLast(p->pPageList);
    pPage = getBufPage(pInfo->pBuf, *curId);

    int32_t *rows = (int32_t*) pPage;
    if (*rows >= pInfo->rowCapacity) {
      // add a new page for current group
      int32_t pageId = 0;
      pPage = getNewBufPage(pInfo->pBuf, 0, &pageId);
      taosArrayPush(p->pPageList, &pageId);

      *(int32_t*) pPage = 0;
    }
  }
H
Haojun Liao 已提交
460

H
Haojun Liao 已提交
461 462 463 464 465 466 467 468 469 470 471 472 473 474
  *pGroupInfo = p;
  return pPage;
}

uint64_t calcGroupId(char* pData, int32_t len) {
  T_MD5_CTX context;
  tMD5Init(&context);
  tMD5Update(&context, (uint8_t*)pData, len);
  tMD5Final(&context);

  // NOTE: only extract the initial 8 bytes of the final MD5 digest
  uint64_t id = 0;
  memcpy(&id, context.digest, sizeof(uint64_t));
  return id;
H
Haojun Liao 已提交
475 476
}

H
Haojun Liao 已提交
477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500
int32_t* setupColumnOffset(const SSDataBlock* pBlock, int32_t rowCapacity) {
  size_t numOfCols = pBlock->info.numOfCols;
  int32_t* offset = taosMemoryCalloc(pBlock->info.numOfCols, sizeof(int32_t));

  offset[0] = sizeof(int32_t);  // the number of rows in current page, ref to SSDataBlock paged serialization format

  for(int32_t i = 0; i < numOfCols - 1; ++i) {
    SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, i);

    int32_t bytes = pColInfoData->info.bytes;
    int32_t payloadLen = bytes * rowCapacity;
    
    if (IS_VAR_DATA_TYPE(pColInfoData->info.type)) {
      // offset segment + content length + payload
      offset[i + 1] = rowCapacity * sizeof(int32_t) + sizeof(int32_t) + payloadLen + offset[i];
    } else {
      // bitmap + content length + payload
      offset[i + 1] = BitmapLen(rowCapacity) + sizeof(int32_t) + payloadLen + offset[i];
    }
  }

  return offset;
}

H
Haojun Liao 已提交
501 502 503 504 505 506 507 508
static SSDataBlock* buildPartitionResult(SOperatorInfo* pOperator) {
  SPartitionOperatorInfo* pInfo = pOperator->info;

  SDataGroupInfo* pGroupInfo = pInfo->pGroupIter;
  if (pInfo->pGroupIter == NULL || pInfo->pageIndex >= taosArrayGetSize(pGroupInfo->pPageList)) {
    // try next group data
    pInfo->pGroupIter = taosHashIterate(pInfo->pGroupSet, pInfo->pGroupIter);
    if (pInfo->pGroupIter == NULL) {
H
Haojun Liao 已提交
509
      pOperator->status = OP_EXEC_DONE;
H
Haojun Liao 已提交
510 511 512 513 514 515 516 517 518 519
      return NULL;
    }

    pGroupInfo = pInfo->pGroupIter;
    pInfo->pageIndex = 0;
  }

  int32_t* pageId = taosArrayGet(pGroupInfo->pPageList, pInfo->pageIndex);
  void* page = getBufPage(pInfo->pBuf, *pageId);

H
Haojun Liao 已提交
520
  blockDataFromBuf1(pInfo->binfo.pRes, page, pInfo->rowCapacity);
H
Haojun Liao 已提交
521 522

  pInfo->pageIndex += 1;
H
Haojun Liao 已提交
523 524

  pInfo->binfo.pRes->info.groupId = pGroupInfo->groupId;
H
Haojun Liao 已提交
525 526 527 528
  return pInfo->binfo.pRes;
}

static SSDataBlock* hashPartition(SOperatorInfo* pOperator, bool* newgroup) {
H
Haojun Liao 已提交
529 530
  if (pOperator->status == OP_EXEC_DONE) {
    return NULL;
531 532
  }

H
Haojun Liao 已提交
533 534
  SGroupbyOperatorInfo* pInfo = pOperator->info;
  SSDataBlock* pRes = pInfo->binfo.pRes;
535

H
Haojun Liao 已提交
536
  if (pOperator->status == OP_RES_TO_RETURN) {
H
Haojun Liao 已提交
537 538
    blockDataCleanup(pRes);
    return buildPartitionResult(pOperator);
H
Haojun Liao 已提交
539 540
  }

H
Haojun Liao 已提交
541
  SOperatorInfo* downstream = pOperator->pDownstream[0];
H
Haojun Liao 已提交
542

H
Haojun Liao 已提交
543 544 545 546 547 548 549
  while (1) {
    publishOperatorProfEvent(downstream, QUERY_PROF_BEFORE_OPERATOR_EXEC);
    SSDataBlock* pBlock = downstream->getNextFn(downstream, newgroup);
    publishOperatorProfEvent(downstream, QUERY_PROF_AFTER_OPERATOR_EXEC);
    if (pBlock == NULL) {
      break;
    }
H
Haojun Liao 已提交
550

H
Haojun Liao 已提交
551 552
    //    setTagValue(pOperator, pRuntimeEnv->current->pTable, pInfo->binfo.pCtx, pOperator->numOfOutput);
    doHashPartition(pOperator, pBlock);
H
Haojun Liao 已提交
553 554 555
  }

  pOperator->status = OP_RES_TO_RETURN;
H
Haojun Liao 已提交
556 557 558 559 560
  blockDataEnsureCapacity(pRes, 4096);
  return buildPartitionResult(pOperator);
}

static void destroyPartitionOperatorInfo(void* param, int32_t numOfOutput) {
H
Haojun Liao 已提交
561
  SPartitionOperatorInfo* pInfo = (SPartitionOperatorInfo*)param;
H
Haojun Liao 已提交
562 563 564
  doDestroyBasicInfo(&pInfo->binfo, numOfOutput);
  taosArrayDestroy(pInfo->pGroupCols);
  taosArrayDestroy(pInfo->pGroupColVals);
H
Haojun Liao 已提交
565 566
  taosMemoryFree(pInfo->keyBuf);
  taosMemoryFree(pInfo->columnOffset);
H
Haojun Liao 已提交
567 568
}

569
SOperatorInfo* createPartitionOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResultBlock, SArray* pGroupColList,
H
Haojun Liao 已提交
570 571 572
                                           SExecTaskInfo* pTaskInfo, const STableGroupInfo* pTableGroupInfo) {
  SPartitionOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SPartitionOperatorInfo));
  SOperatorInfo*        pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
H
Haojun Liao 已提交
573
  if (pInfo == NULL || pOperator == NULL) {
H
Haojun Liao 已提交
574 575
    goto _error;
  }
576

H
Haojun Liao 已提交
577 578 579 580 581 582 583 584 585 586 587 588 589
  pInfo->pGroupCols = pGroupColList;

  _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);
  pInfo->pGroupSet = taosHashInit(100, hashFn, false, HASH_NO_LOCK);
  if (pInfo->pGroupSet == NULL) {
    goto _error;
  }

  int32_t code = createDiskbasedBuf(&pInfo->pBuf, 4096, 4096 * 256, pTaskInfo->id.str, "/tmp/");
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }

H
Haojun Liao 已提交
590 591
  pInfo->rowCapacity = blockDataGetCapacityInRow(pResultBlock, getBufPageSize(pInfo->pBuf));
  pInfo->columnOffset = setupColumnOffset(pResultBlock, pInfo->rowCapacity);
H
Haojun Liao 已提交
592 593 594 595
  code = initGroupOptrInfo(&pInfo->pGroupColVals, &pInfo->groupKeyLen, &pInfo->keyBuf, pGroupColList);
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }
H
Haojun Liao 已提交
596 597

  pOperator->name         = "PartitionOperator";
598 599
  pOperator->blockingOptr = true;
  pOperator->status       = OP_NOT_OPENED;
H
Haojun Liao 已提交
600
  pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_PARTITION;
601

H
Haojun Liao 已提交
602
  pInfo->binfo.pRes       = pResultBlock;
603 604
  pOperator->numOfOutput  = numOfCols;
  pOperator->pExpr        = pExprInfo;
H
Haojun Liao 已提交
605 606 607 608
  pOperator->info         = pInfo;
  pOperator->_openFn      = operatorDummyOpenFn;
  pOperator->getNextFn    = hashPartition;
  pOperator->closeFn      = destroyPartitionOperatorInfo;
H
Haojun Liao 已提交
609

H
Haojun Liao 已提交
610
  code = appendDownstream(pOperator, &downstream, 1);
611 612 613
  return pOperator;

  _error:
H
Haojun Liao 已提交
614
  pTaskInfo->code = TSDB_CODE_OUT_OF_MEMORY;
H
Haojun Liao 已提交
615 616
  taosMemoryFreeClear(pInfo);
  taosMemoryFreeClear(pOperator);
617
  return NULL;
H
Haojun Liao 已提交
618
}