groupoperator.c 25.2 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
/*
 * 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"
wmmhello's avatar
wmmhello 已提交
27
#include "executorInt.h"
H
Haojun Liao 已提交
28

H
Haojun Liao 已提交
29
static int32_t* setupColumnOffset(const SSDataBlock* pBlock, int32_t rowCapacity);
H
Haojun Liao 已提交
30
static void* getCurrentDataGroupInfo(const SPartitionOperatorInfo* pInfo, SDataGroupInfo** pGroupInfo, 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);
}

wmmhello's avatar
wmmhello 已提交
40
static int32_t initGroupOptrInfo(SArray** pGroupColVals, int32_t* keyLen, char** keyBuf, const SArray* pGroupColList) {
H
Haojun Liao 已提交
41 42
  *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);
49
    (*keyLen) += pCol->bytes; // actual data + null_flag
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
  }

  int32_t nullFlagSize = sizeof(int8_t) * numOfGroupCols;
64
  (*keyLen) += nullFlagSize;
H
Haojun Liao 已提交
65

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

  return TSDB_CODE_SUCCESS;
}

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

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

H
Haojun Liao 已提交
85
    SGroupKeys* pkey = taosArrayGet(pGroupColVals, i);
H
Haojun Liao 已提交
86 87 88 89 90 91 92 93 94 95
    if (pkey->isNull && isNull) {
      continue;
    }

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

    char* val = colDataGetData(pColInfoData, rowIndex);

wmmhello's avatar
wmmhello 已提交
96 97 98 99 100 101 102 103 104
    if (pkey->type == TSDB_DATA_TYPE_JSON) {
      int32_t dataLen = getJsonValueLen(val);

      if (memcmp(pkey->pData, val, dataLen) == 0){
        continue;
      } else {
        return false;
      }
    } else if (IS_VAR_DATA_TYPE(pkey->type)) {
H
Haojun Liao 已提交
105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120
      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;
}

wmmhello's avatar
wmmhello 已提交
121
static void recordNewGroupKeys(SArray* pGroupCols, SArray* pGroupColVals, SSDataBlock* pBlock, int32_t rowIndex) {
H
Haojun Liao 已提交
122 123
  SColumnDataAgg* pColAgg = NULL;

124 125
  size_t numOfGroupCols = taosArrayGetSize(pGroupCols);

H
Haojun Liao 已提交
126
  for (int32_t i = 0; i < numOfGroupCols; ++i) {
H
Haojun Liao 已提交
127
    SColumn*         pCol = taosArrayGet(pGroupCols, i);
H
Haojun Liao 已提交
128 129 130
    SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, pCol->slotId);

    if (pBlock->pBlockAgg != NULL) {
131
      pColAgg = pBlock->pBlockAgg[pCol->slotId];  // TODO is agg data matched?
H
Haojun Liao 已提交
132 133
    }

H
Haojun Liao 已提交
134
    SGroupKeys* pkey = taosArrayGet(pGroupColVals, i);
H
Haojun Liao 已提交
135 136 137
    if (colDataIsNull(pColInfoData, pBlock->info.rows, rowIndex, pColAgg)) {
      pkey->isNull = true;
    } else {
138
      pkey->isNull = false;
H
Haojun Liao 已提交
139
      char* val = colDataGetData(pColInfoData, rowIndex);
wmmhello's avatar
wmmhello 已提交
140 141 142 143
      if (pkey->type == TSDB_DATA_TYPE_JSON) {
        int32_t dataLen = getJsonValueLen(val);
        memcpy(pkey->pData, val, dataLen);
      } else if (IS_VAR_DATA_TYPE(pkey->type)) {
H
Haojun Liao 已提交
144
        memcpy(pkey->pData, val, varDataTLen(val));
145
        ASSERT(varDataTLen(val) <= pkey->bytes);
H
Haojun Liao 已提交
146 147 148 149 150 151 152
      } else {
        memcpy(pkey->pData, val, pkey->bytes);
      }
    }
  }
}

wmmhello's avatar
wmmhello 已提交
153
static int32_t buildGroupKeys(void* pKey, const SArray* pGroupColVals) {
H
Haojun Liao 已提交
154 155 156 157 158 159 160 161 162 163 164 165 166
  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;
wmmhello's avatar
wmmhello 已提交
167 168 169 170 171
    if (pkey->type == TSDB_DATA_TYPE_JSON) {
      int32_t dataLen = getJsonValueLen(pkey->pData);
      memcpy(pStart, (pkey->pData), dataLen);
      pStart += dataLen;
    } else if (IS_VAR_DATA_TYPE(pkey->type)) {
H
Haojun Liao 已提交
172 173 174 175 176 177 178 179 180
      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 已提交
181
  return (int32_t) (pStart - (char*)pKey);
H
Haojun Liao 已提交
182 183 184 185 186
}

// 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) {
187
    if (pCtx[i].functionId == -1) {       // select count(*),key from t group by key.
H
Haojun Liao 已提交
188 189 190
      SResultRowEntryInfo* pEntryInfo = GET_RES_INFO(&pCtx[i]);

      SColumnInfoData* pColInfoData = pCtx[i].input.pData[0];
191
      // todo OPT all/all not NULL
H
Haojun Liao 已提交
192 193 194 195
      if (!colDataIsNull(pColInfoData, totalRows, rowIndex, NULL)) {
        char* dest = GET_ROWCELL_INTERBUF(pEntryInfo);
        char* data = colDataGetData(pColInfoData, rowIndex);

wmmhello's avatar
wmmhello 已提交
196 197 198 199
        if (pColInfoData->info.type == TSDB_DATA_TYPE_JSON) {
          int32_t dataLen = getJsonValueLen(data);
          memcpy(dest, data, dataLen);
        } else if (IS_VAR_DATA_TYPE(pColInfoData->info.type)) {
200 201 202 203
          varDataCopy(dest, data);
        } else {
          memcpy(dest, data, pColInfoData->info.bytes);
        }
H
Haojun Liao 已提交
204 205
      } else { // it is a NULL value
        pEntryInfo->isNullRes = 1;
H
Haojun Liao 已提交
206
      }
H
Haojun Liao 已提交
207 208

      pEntryInfo->numOfRes = 1;
H
Haojun Liao 已提交
209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230
    }
  }
}

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) {
231
      recordNewGroupKeys(pInfo->pGroupCols, pInfo->pGroupColVals, pBlock, j);
H
Haojun Liao 已提交
232 233 234 235 236
      pInfo->isInit = true;
      num++;
      continue;
    }

H
Haojun Liao 已提交
237
    bool equal = groupKeyCompare(pInfo->pGroupCols, pInfo->pGroupColVals, pBlock, j, numOfGroupCols);
H
Haojun Liao 已提交
238 239 240 241 242
    if (equal) {
      num++;
      continue;
    }

H
Haojun Liao 已提交
243
    // The first row of a new block does not belongs to the previous existed group
244
    if (j == 0) {
H
Haojun Liao 已提交
245
      num++;
246
      recordNewGroupKeys(pInfo->pGroupCols, pInfo->pGroupColVals, pBlock, j);
H
Haojun Liao 已提交
247 248 249
      continue;
    }

H
Haojun Liao 已提交
250
    len = buildGroupKeys(pInfo->keyBuf, pInfo->pGroupColVals);
251
    int32_t ret = setGroupResultOutputBuf(&(pInfo->binfo), pOperator->numOfExprs, pInfo->keyBuf, TSDB_DATA_TYPE_VARCHAR, len, 0, pInfo->aggSup.pResultBuf, pTaskInfo, &pInfo->aggSup);
H
Haojun Liao 已提交
252 253 254 255 256
    if (ret != TSDB_CODE_SUCCESS) {  // null data, too many state code
      longjmp(pTaskInfo->env, TSDB_CODE_QRY_APP_ERROR);
    }

    int32_t rowIndex = j - num;
257
    doApplyFunctions(pTaskInfo, pCtx, &w, NULL, rowIndex, num, NULL, pBlock->info.rows, pOperator->numOfExprs, TSDB_ORDER_ASC);
H
Haojun Liao 已提交
258 259

    // assign the group keys or user input constant values if required
260
    doAssignGroupKeys(pCtx, pOperator->numOfExprs, pBlock->info.rows, rowIndex);
261
    recordNewGroupKeys(pInfo->pGroupCols, pInfo->pGroupColVals, pBlock, j);
H
Haojun Liao 已提交
262 263 264 265
    num = 1;
  }

  if (num > 0) {
H
Haojun Liao 已提交
266
    len = buildGroupKeys(pInfo->keyBuf, pInfo->pGroupColVals);
H
Haojun Liao 已提交
267
    int32_t ret =
268
        setGroupResultOutputBuf(&(pInfo->binfo), pOperator->numOfExprs, pInfo->keyBuf, TSDB_DATA_TYPE_VARCHAR, len,
H
Haojun Liao 已提交
269 270 271 272 273 274
                                   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;
275
    doApplyFunctions(pTaskInfo, pCtx, &w, NULL, rowIndex, num, NULL, pBlock->info.rows, pOperator->numOfExprs, TSDB_ORDER_ASC);
276
    doAssignGroupKeys(pCtx, pOperator->numOfExprs, pBlock->info.rows, rowIndex);
H
Haojun Liao 已提交
277 278 279
  }
}

280
static SSDataBlock* hashGroupbyAggregate(SOperatorInfo* pOperator) {
H
Haojun Liao 已提交
281 282 283 284
  if (pOperator->status == OP_EXEC_DONE) {
    return NULL;
  }

285 286
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

H
Haojun Liao 已提交
287
  SGroupbyOperatorInfo* pInfo = pOperator->info;
288 289
  SSDataBlock* pRes = pInfo->binfo.pRes;

H
Haojun Liao 已提交
290
  if (pOperator->status == OP_RES_TO_RETURN) {
291
    doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf);
292 293 294

    size_t rows = pRes->info.rows;
    if (rows == 0 || !hashRemainDataInGroupInfo(&pInfo->groupResInfo)) {
295
      doSetOperatorCompleted(pOperator);
H
Haojun Liao 已提交
296
    }
297 298

    pOperator->resultInfo.totalRows += rows;
H
Haojun Liao 已提交
299
    return (pRes->info.rows == 0)? NULL:pRes;
H
Haojun Liao 已提交
300 301
  }

302 303 304
  int32_t order = TSDB_ORDER_ASC;
  int32_t scanFlag = MAIN_SCAN;

305
  int64_t st = taosGetTimestampUs();
H
Haojun Liao 已提交
306 307 308
  SOperatorInfo* downstream = pOperator->pDownstream[0];

  while (1) {
309
    SSDataBlock* pBlock = downstream->fpSet.getNextFn(downstream);
H
Haojun Liao 已提交
310 311 312 313
    if (pBlock == NULL) {
      break;
    }

314 315 316 317 318
    int32_t code = getTableScanInfo(pOperator, &order, &scanFlag);
    if (code != TSDB_CODE_SUCCESS) {
      longjmp(pTaskInfo->env, code);
    }

H
Haojun Liao 已提交
319
    // the pDataBlock are always the same one, no need to call this again
320
    setInputDataBlock(pOperator, pInfo->binfo.pCtx, pBlock, order, scanFlag, true);
321

322
    // there is an scalar expression that needs to be calculated right before apply the group aggregation.
323 324
    if (pInfo->scalarSup.pScalarExprInfo != NULL) {
      pTaskInfo->code = projectApplyFunctions(pInfo->scalarSup.pScalarExprInfo, pBlock, pBlock, pInfo->scalarSup.pScalarFuncCtx, pInfo->scalarSup.numOfScalarExpr, NULL);
325 326 327
      if (pTaskInfo->code != TSDB_CODE_SUCCESS) {
        longjmp(pTaskInfo->env, pTaskInfo->code);
      }
328 329
    }

H
Haojun Liao 已提交
330 331 332 333 334
    doHashGroupbyAgg(pOperator, pBlock);
  }

  pOperator->status = OP_RES_TO_RETURN;
  closeAllResultRows(&pInfo->binfo.resultRowInfo);
H
Haojun Liao 已提交
335

336 337 338 339 340 341 342 343 344 345 346 347 348 349
#if 0
  if(pOperator->fpSet.encodeResultRow){
    char *result = NULL;
    int32_t length = 0;
    pOperator->fpSet.encodeResultRow(pOperator, &result, &length);
    SAggSupporter* pSup = &pInfo->aggSup;
    taosHashClear(pSup->pResultRowHashTable);
    pInfo->binfo.resultRowInfo.size = 0;
    pOperator->fpSet.decodeResultRow(pOperator, result);
    if(result){
      taosMemoryFree(result);
    }
  }
#endif
350
  blockDataEnsureCapacity(pRes, pOperator->resultInfo.capacity);
351
  initGroupedResultInfo(&pInfo->groupResInfo, pInfo->aggSup.pResultRowHashTable, 0);
352

353 354
  pOperator->cost.openCost = (taosGetTimestampUs() - st) / 1000.0;

355
  while(1) {
356
    doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf);
357
    doFilter(pInfo->pCondition, pRes);
358

359
    bool hasRemain = hashRemainDataInGroupInfo(&pInfo->groupResInfo);
360
    if (!hasRemain) {
361
      doSetOperatorCompleted(pOperator);
362 363 364 365 366 367
      break;
    }

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

370 371 372 373
  size_t rows = pRes->info.rows;
  pOperator->resultInfo.totalRows += rows;

  return (rows == 0)? NULL:pRes;
H
Haojun Liao 已提交
374 375
}

H
Haojun Liao 已提交
376 377 378
SOperatorInfo* createGroupOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols,
                                       SSDataBlock* pResultBlock, SArray* pGroupColList, SNode* pCondition,
                                       SExprInfo* pScalarExprInfo, int32_t numOfScalarExpr, SExecTaskInfo* pTaskInfo) {
H
Haojun Liao 已提交
379 380 381 382 383 384
  SGroupbyOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SGroupbyOperatorInfo));
  SOperatorInfo*        pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
  if (pInfo == NULL || pOperator == NULL) {
    goto _error;
  }

385 386 387
  pInfo->pGroupCols      = pGroupColList;
  pInfo->pCondition      = pCondition;

388 389 390
  pInfo->scalarSup.pScalarExprInfo = pScalarExprInfo;
  pInfo->scalarSup.numOfScalarExpr = numOfScalarExpr;
  pInfo->scalarSup.pScalarFuncCtx  = createSqlFunctionCtx(pScalarExprInfo, numOfScalarExpr, &pInfo->scalarSup.rowCellInfoOffset);
391

H
Haojun Liao 已提交
392
  int32_t code = initGroupOptrInfo(&pInfo->pGroupColVals, &pInfo->groupKeyLen, &pInfo->keyBuf, pGroupColList);
H
Haojun Liao 已提交
393 394 395 396
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }

397 398
  initResultSizeInfo(pOperator, 4096);
  initAggInfo(&pInfo->binfo, &pInfo->aggSup, pExprInfo, numOfCols, pResultBlock, pInfo->groupKeyLen, pTaskInfo->id.str);
399 400
  initResultRowInfo(&pInfo->binfo.resultRowInfo, 8);

H
Haojun Liao 已提交
401
  pOperator->name         = "GroupbyAggOperator";
402
  pOperator->blocking     = true;
H
Haojun Liao 已提交
403
  pOperator->status       = OP_NOT_OPENED;
404
  // pOperator->operatorType = OP_Groupby;
H
Haojun Liao 已提交
405
  pOperator->pExpr        = pExprInfo;
H
Haojun Liao 已提交
406
  pOperator->numOfExprs   = numOfCols;
H
Haojun Liao 已提交
407
  pOperator->info         = pInfo;
408
  pOperator->pTaskInfo    = pTaskInfo;
H
Haojun Liao 已提交
409

410
  pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, hashGroupbyAggregate, NULL, NULL, destroyGroupOperatorInfo, aggEncodeResultRow, aggDecodeResultRow, NULL);
H
Haojun Liao 已提交
411 412 413 414
  code = appendDownstream(pOperator, &downstream, 1);
  return pOperator;

  _error:
H
Haojun Liao 已提交
415
  pTaskInfo->code = TSDB_CODE_OUT_OF_MEMORY;
H
Haojun Liao 已提交
416 417 418
  taosMemoryFreeClear(pInfo);
  taosMemoryFreeClear(pOperator);
  return NULL;
419 420
}

H
Haojun Liao 已提交
421
static void doHashPartition(SOperatorInfo* pOperator, SSDataBlock* pBlock) {
H
Haojun Liao 已提交
422
//  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
H
Haojun Liao 已提交
423 424 425 426

  SPartitionOperatorInfo* pInfo = pOperator->info;

  for (int32_t j = 0; j < pBlock->info.rows; ++j) {
427
    recordNewGroupKeys(pInfo->pGroupCols, pInfo->pGroupColVals, pBlock, j);
H
Haojun Liao 已提交
428 429
    int32_t len = buildGroupKeys(pInfo->keyBuf, pInfo->pGroupColVals);

H
Haojun Liao 已提交
430 431
    SDataGroupInfo* pGInfo = NULL;
    void *pPage = getCurrentDataGroupInfo(pInfo, &pGInfo, len);
H
Haojun Liao 已提交
432

H
Haojun Liao 已提交
433 434 435
    pGInfo->numOfRows += 1;
    if (pGInfo->groupId == 0) {
      pGInfo->groupId = calcGroupId(pInfo->keyBuf, len);
H
Haojun Liao 已提交
436 437
    }

438
    // number of rows
H
Haojun Liao 已提交
439
    int32_t* rows = (int32_t*) pPage;
H
Haojun Liao 已提交
440

441 442
    // group id

443
    size_t numOfCols = pOperator->numOfExprs;
H
Haojun Liao 已提交
444
    for(int32_t i = 0; i < numOfCols; ++i) {
445 446 447 448
      SExprInfo* pExpr = &pOperator->pExpr[i];
      int32_t slotId = pExpr->base.pParam[0].pCol->slotId;

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

H
Haojun Liao 已提交
450 451
      int32_t bytes = pColInfoData->info.bytes;
      int32_t startOffset = pInfo->columnOffset[i];
H
Haojun Liao 已提交
452

453 454
      int32_t* columnLen  = NULL;
      int32_t  contentLen = 0;
H
Haojun Liao 已提交
455 456

      if (IS_VAR_DATA_TYPE(pColInfoData->info.type)) {
wafwerar's avatar
wafwerar 已提交
457
        int32_t* offset = (int32_t*)((char*)pPage + startOffset);
458 459
        columnLen       = (int32_t*) ((char*)pPage + startOffset + sizeof(int32_t) * pInfo->rowCapacity);
        char*    data   = (char*)((char*) columnLen + sizeof(int32_t));
H
Haojun Liao 已提交
460 461 462 463

        if (colDataIsNull_s(pColInfoData, j)) {
          offset[(*rows)] = -1;
          contentLen = 0;
wmmhello's avatar
wmmhello 已提交
464 465 466 467 468 469 470 471 472 473
        } else if(pColInfoData->info.type == TSDB_DATA_TYPE_JSON){
          offset[*rows] = (*columnLen);
          char* src = colDataGetData(pColInfoData, j);
          int32_t dataLen = getJsonValueLen(src);

          memcpy(data + (*columnLen), src, dataLen);
          int32_t v = (data + (*columnLen) + dataLen - (char*)pPage);
          ASSERT(v > 0);

          contentLen = dataLen;
H
Haojun Liao 已提交
474 475 476 477
        } else {
          offset[*rows] = (*columnLen);
          char* src = colDataGetData(pColInfoData, j);
          memcpy(data + (*columnLen), src, varDataTLen(src));
478 479 480
          int32_t v = (data + (*columnLen) + varDataTLen(src) - (char*)pPage);
          ASSERT(v > 0);

H
Haojun Liao 已提交
481 482
          contentLen = varDataTLen(src);
        }
H
Haojun Liao 已提交
483
      } else {
wafwerar's avatar
wafwerar 已提交
484
        char* bitmap = (char*)pPage + startOffset;
485
        columnLen    = (int32_t*) ((char*)pPage + startOffset + BitmapLen(pInfo->rowCapacity));
H
Haojun Liao 已提交
486
        char* data   = (char*) columnLen + sizeof(int32_t);
H
Haojun Liao 已提交
487 488 489

        bool isNull = colDataIsNull_f(pColInfoData->nullbitmap, j);
        if (isNull) {
H
Haojun Liao 已提交
490
          colDataSetNull_f(bitmap, (*rows));
H
Haojun Liao 已提交
491
        } else {
H
Haojun Liao 已提交
492
          memcpy(data + (*columnLen), colDataGetData(pColInfoData, j), bytes);
H
Haojun Liao 已提交
493
        }
H
Haojun Liao 已提交
494
        contentLen = bytes;
H
Haojun Liao 已提交
495
      }
H
Haojun Liao 已提交
496 497

      (*columnLen) += contentLen;
498
      ASSERT(*columnLen >= 0);
H
Haojun Liao 已提交
499 500
    }

H
Haojun Liao 已提交
501 502
    (*rows) += 1;

H
Haojun Liao 已提交
503 504 505
    setBufPageDirty(pPage, true);
    releaseBufPage(pInfo->pBuf, pPage);
  }
H
Haojun Liao 已提交
506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529
}

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) {
530 531 532
      // release buffer
      releaseBufPage(pInfo->pBuf, pPage);

H
Haojun Liao 已提交
533 534 535 536
      // add a new page for current group
      int32_t pageId = 0;
      pPage = getNewBufPage(pInfo->pBuf, 0, &pageId);
      taosArrayPush(p->pPageList, &pageId);
537
      memset(pPage, 0, getBufPageSize(pInfo->pBuf));
H
Haojun Liao 已提交
538 539
    }
  }
H
Haojun Liao 已提交
540

H
Haojun Liao 已提交
541 542 543 544 545 546 547 548 549 550 551 552 553 554
  *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 已提交
555 556
}

H
Haojun Liao 已提交
557 558 559 560
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));

561
  offset[0] = sizeof(int32_t) + sizeof(uint64_t);  // the number of rows in current page, ref to SSDataBlock paged serialization format
H
Haojun Liao 已提交
562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580

  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;
}

5
54liuyao 已提交
581 582 583 584 585 586 587 588 589
static void clearPartitionOperator(SPartitionOperatorInfo* pInfo) {
  void *ite = NULL;
  while( (ite = taosHashIterate(pInfo->pGroupSet, ite)) != NULL ) {
    taosArrayDestroy( ((SDataGroupInfo *)ite)->pPageList);
  }
  taosHashClear(pInfo->pGroupSet);
  clearDiskbasedBuf(pInfo->pBuf);
}

H
Haojun Liao 已提交
590 591 592 593 594 595 596 597
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) {
598
      doSetOperatorCompleted(pOperator);
5
54liuyao 已提交
599
      clearPartitionOperator(pInfo);
H
Haojun Liao 已提交
600 601 602 603 604 605 606 607 608 609
      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 已提交
610
  blockDataFromBuf1(pInfo->binfo.pRes, page, pInfo->rowCapacity);
H
Haojun Liao 已提交
611 612

  pInfo->pageIndex += 1;
613
  releaseBufPage(pInfo->pBuf, page);
H
Haojun Liao 已提交
614

615
  blockDataUpdateTsWindow(pInfo->binfo.pRes, 0);
H
Haojun Liao 已提交
616
  pInfo->binfo.pRes->info.groupId = pGroupInfo->groupId;
617 618

  pOperator->resultInfo.totalRows += pInfo->binfo.pRes->info.rows;
H
Haojun Liao 已提交
619 620 621
  return pInfo->binfo.pRes;
}

622
static SSDataBlock* hashPartition(SOperatorInfo* pOperator) {
H
Haojun Liao 已提交
623 624
  if (pOperator->status == OP_EXEC_DONE) {
    return NULL;
625 626
  }

627 628 629
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

  SPartitionOperatorInfo* pInfo = pOperator->info;
H
Haojun Liao 已提交
630
  SSDataBlock* pRes = pInfo->binfo.pRes;
631

H
Haojun Liao 已提交
632
  if (pOperator->status == OP_RES_TO_RETURN) {
H
Haojun Liao 已提交
633 634
    blockDataCleanup(pRes);
    return buildPartitionResult(pOperator);
H
Haojun Liao 已提交
635 636
  }

637
  int64_t st = taosGetTimestampUs();
H
Haojun Liao 已提交
638
  SOperatorInfo* downstream = pOperator->pDownstream[0];
H
Haojun Liao 已提交
639

H
Haojun Liao 已提交
640
  while (1) {
641
    SSDataBlock* pBlock = downstream->fpSet.getNextFn(downstream);
H
Haojun Liao 已提交
642 643 644
    if (pBlock == NULL) {
      break;
    }
H
Haojun Liao 已提交
645

646 647 648 649 650 651 652 653
    // there is an scalar expression that needs to be calculated right before apply the group aggregation.
    if (pInfo->scalarSupp.pScalarExprInfo != NULL) {
      pTaskInfo->code = projectApplyFunctions(pInfo->scalarSupp.pScalarExprInfo, pBlock, pBlock, pInfo->scalarSupp.pScalarFuncCtx, pInfo->scalarSupp.numOfScalarExpr, NULL);
      if (pTaskInfo->code != TSDB_CODE_SUCCESS) {
        longjmp(pTaskInfo->env, pTaskInfo->code);
      }
    }

H
Haojun Liao 已提交
654
    doHashPartition(pOperator, pBlock);
H
Haojun Liao 已提交
655 656
  }

657 658
  pOperator->cost.openCost = (taosGetTimestampUs() - st) / 1000.0;

H
Haojun Liao 已提交
659
  pOperator->status = OP_RES_TO_RETURN;
H
Haojun Liao 已提交
660 661 662 663 664
  blockDataEnsureCapacity(pRes, 4096);
  return buildPartitionResult(pOperator);
}

static void destroyPartitionOperatorInfo(void* param, int32_t numOfOutput) {
H
Haojun Liao 已提交
665
  SPartitionOperatorInfo* pInfo = (SPartitionOperatorInfo*)param;
H
Haojun Liao 已提交
666 667
  doDestroyBasicInfo(&pInfo->binfo, numOfOutput);
  taosArrayDestroy(pInfo->pGroupCols);
wmmhello's avatar
wmmhello 已提交
668 669 670 671
  for(int i = 0; i < taosArrayGetSize(pInfo->pGroupColVals); i++){
    SGroupKeys key = *(SGroupKeys*)taosArrayGet(pInfo->pGroupColVals, i);
    taosMemoryFree(key.pData);
  }
H
Haojun Liao 已提交
672
  taosArrayDestroy(pInfo->pGroupColVals);
H
Haojun Liao 已提交
673
  taosMemoryFree(pInfo->keyBuf);
wmmhello's avatar
wmmhello 已提交
674
  taosHashCleanup(pInfo->pGroupSet);
H
Haojun Liao 已提交
675
  taosMemoryFree(pInfo->columnOffset);
H
Haojun Liao 已提交
676 677
}

678
SOperatorInfo* createPartitionOperatorInfo(SOperatorInfo* downstream, SPartitionPhysiNode* pPartNode, SExecTaskInfo* pTaskInfo) {
H
Haojun Liao 已提交
679 680
  SPartitionOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SPartitionOperatorInfo));
  SOperatorInfo*        pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
H
Haojun Liao 已提交
681
  if (pInfo == NULL || pOperator == NULL) {
H
Haojun Liao 已提交
682 683
    goto _error;
  }
684

685 686 687 688 689 690 691 692 693 694 695 696 697
  SSDataBlock* pResBlock = createResDataBlock(pPartNode->node.pOutputDataBlockDesc);

  int32_t numOfCols = 0;
  SExprInfo* pExprInfo = createExprInfo(pPartNode->pTargets, NULL, &numOfCols);

  pInfo->pGroupCols = extractPartitionColInfo(pPartNode->pPartitionKeys);

  if (pPartNode->pExprs != NULL) {
    pInfo->scalarSupp.numOfScalarExpr = 0;
    pInfo->scalarSupp.pScalarExprInfo = createExprInfo(pPartNode->pExprs, NULL, &pInfo->scalarSupp.numOfScalarExpr);
    pInfo->scalarSupp.pScalarFuncCtx = createSqlFunctionCtx(
        pInfo->scalarSupp.pScalarExprInfo, pInfo->scalarSupp.numOfScalarExpr, &pInfo->scalarSupp.rowCellInfoOffset);
  }
H
Haojun Liao 已提交
698 699 700 701 702 703 704

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

705 706
  uint32_t defaultPgsz  = 0;
  uint32_t defaultBufsz = 0;
707
  getBufferPgSize(pResBlock->info.rowSize, &defaultPgsz, &defaultBufsz);
708 709

  int32_t code = createDiskbasedBuf(&pInfo->pBuf, defaultPgsz, defaultBufsz, pTaskInfo->id.str, TD_TMP_DIR_PATH);
H
Haojun Liao 已提交
710 711 712 713
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }

714 715 716
  pInfo->rowCapacity = blockDataGetCapacityInRow(pResBlock, getBufPageSize(pInfo->pBuf));
  pInfo->columnOffset = setupColumnOffset(pResBlock, pInfo->rowCapacity);
  code = initGroupOptrInfo(&pInfo->pGroupColVals, &pInfo->groupKeyLen, &pInfo->keyBuf, pInfo->pGroupCols);
H
Haojun Liao 已提交
717 718 719
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }
H
Haojun Liao 已提交
720 721

  pOperator->name         = "PartitionOperator";
722
  pOperator->blocking     = true;
723
  pOperator->status       = OP_NOT_OPENED;
H
Haojun Liao 已提交
724
  pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_PARTITION;
725
  pInfo->binfo.pRes       = pResBlock;
726
  pOperator->numOfExprs   = numOfCols;
727
  pOperator->pExpr        = pExprInfo;
H
Haojun Liao 已提交
728
  pOperator->info         = pInfo;
729
  pOperator->pTaskInfo    = pTaskInfo;
730 731 732

  pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, hashPartition, NULL, NULL, destroyPartitionOperatorInfo,
                                         NULL, NULL, NULL);
H
Haojun Liao 已提交
733

H
Haojun Liao 已提交
734
  code = appendDownstream(pOperator, &downstream, 1);
735 736 737
  return pOperator;

  _error:
H
Haojun Liao 已提交
738
  pTaskInfo->code = TSDB_CODE_OUT_OF_MEMORY;
H
Haojun Liao 已提交
739 740
  taosMemoryFreeClear(pInfo);
  taosMemoryFreeClear(pOperator);
741
  return NULL;
H
Haojun Liao 已提交
742
}