taosdata/TDengine | Build 3621 | source/libs/executor/src/groupoperator.c | Coveralls

UNCOV

static void freeGroupKey(void* param) {

×

UNCOV

74

  SGroupKeys* pKey = (SGroupKeys*)param;

×

UNCOV

75

  taosMemoryFree(pKey->pData);

×

UNCOV

76

×

UNCOV

78

static void destroyGroupOperatorInfo(void* param) {

×

UNCOV

79

  if (param == NULL) {

×

80

    return;

×

UNCOV

82

  SGroupbyOperatorInfo* pInfo = (SGroupbyOperatorInfo*)param;

×

UNCOV

84

  cleanupBasicInfo(&pInfo->binfo);

×

UNCOV

85

  taosMemoryFreeClear(pInfo->keyBuf);

×

UNCOV

86

  taosArrayDestroy(pInfo->pGroupCols);

×

UNCOV

87

  taosArrayDestroyEx(pInfo->pGroupColVals, freeGroupKey);

×

UNCOV

88

  cleanupExprSupp(&pInfo->scalarSup);

×

UNCOV

90

  if (pInfo->pOperator != NULL) {

×

UNCOV

91

    cleanupResultInfo(pInfo->pOperator->pTaskInfo, &pInfo->pOperator->exprSupp, &pInfo->groupResInfo, &pInfo->aggSup,

×

UNCOV

93

    pInfo->pOperator = NULL;

×

UNCOV

96

  cleanupGroupResInfo(&pInfo->groupResInfo);

×

UNCOV

97

  cleanupAggSup(&pInfo->aggSup);

×

UNCOV

98

  taosMemoryFreeClear(param);

×

UNCOV

101

static int32_t initGroupOptrInfo(SArray** pGroupColVals, int32_t* keyLen, char** keyBuf, const SArray* pGroupColList) {

×

UNCOV

102

  *pGroupColVals = taosArrayInit(4, sizeof(SGroupKeys));

×

UNCOV

103

  if ((*pGroupColVals) == NULL) {

×

104

    return terrno;

×

UNCOV

107

  int32_t numOfGroupCols = taosArrayGetSize(pGroupColList);

×

UNCOV

108

  for (int32_t i = 0; i < numOfGroupCols; ++i) {

×

UNCOV

109

    SColumn* pCol = (SColumn*)taosArrayGet(pGroupColList, i);

×

UNCOV

110

    if (!pCol) {

×

111

      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));

×

112

      return terrno;

×

UNCOV

114

    (*keyLen) += pCol->bytes;  // actual data + null_flag

×

UNCOV

116

    SGroupKeys key = {0};

×

UNCOV

117

    key.bytes = pCol->bytes;

×

UNCOV

118

    key.type = pCol->type;

×

UNCOV

119

    key.isNull = false;

×

UNCOV

120

    key.pData = taosMemoryCalloc(1, pCol->bytes);

×

UNCOV

121

    if (key.pData == NULL) {

×

122

      return terrno;

×

UNCOV

125

    void* tmp = taosArrayPush((*pGroupColVals), &key);

×

UNCOV

126

    if (!tmp) {

×

127

      return terrno;

×

UNCOV

131

  int32_t nullFlagSize = sizeof(int8_t) * numOfGroupCols;

×

UNCOV

132

  (*keyLen) += nullFlagSize;

×

UNCOV

134

  (*keyBuf) = taosMemoryCalloc(1, (*keyLen));

×

UNCOV

135

  if ((*keyBuf) == NULL) {

×

136

    return terrno;

×

UNCOV

139

  return TSDB_CODE_SUCCESS;

×

UNCOV

142

static bool groupKeyCompare(SArray* pGroupCols, SArray* pGroupColVals, SSDataBlock* pBlock, int32_t rowIndex,

×

UNCOV

144

  SColumnDataAgg* pColAgg = NULL;

×

UNCOV

145

  for (int32_t i = 0; i < numOfGroupCols; ++i) {

×

UNCOV

146

    SColumn*         pCol = taosArrayGet(pGroupCols, i);

×

UNCOV

147

    SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, pCol->slotId);

×

UNCOV

148

    if (pBlock->pBlockAgg != NULL) {

×

149

      pColAgg = &pBlock->pBlockAgg[pCol->slotId];  // TODO is agg data matched?

×

UNCOV

152

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

×

UNCOV

154

    SGroupKeys* pkey = taosArrayGet(pGroupColVals, i);

×

UNCOV

155

    if (pkey->isNull && isNull) {

×

UNCOV

156

      continue;

×

UNCOV

159

    if (isNull || pkey->isNull) {

×

UNCOV

160

      return false;

×

UNCOV

163

    char* val = colDataGetData(pColInfoData, rowIndex);

×

UNCOV

165

    if (pkey->type == TSDB_DATA_TYPE_JSON) {

×

UNCOV

166

      int32_t dataLen = getJsonValueLen(val);

×

UNCOV

168

      if (memcmp(pkey->pData, val, dataLen) == 0) {

×

UNCOV

169

        continue;

×

UNCOV

171

        return false;

×

UNCOV

173

    } else if (IS_VAR_DATA_TYPE(pkey->type)) {

×

UNCOV

174

      int32_t len = varDataLen(val);

×

UNCOV

175

      if (len == varDataLen(pkey->pData) && memcmp(varDataVal(pkey->pData), varDataVal(val), len) == 0) {

×

UNCOV

176

        continue;

×

UNCOV

178

        return false;

×

UNCOV

181

      if (memcmp(pkey->pData, val, pkey->bytes) != 0) {

×

UNCOV

182

        return false;

×

UNCOV

187

  return true;

×

UNCOV

190

static void recordNewGroupKeys(SArray* pGroupCols, SArray* pGroupColVals, SSDataBlock* pBlock, int32_t rowIndex) {

×

UNCOV

191

  SColumnDataAgg* pColAgg = NULL;

×

UNCOV

193

  size_t numOfGroupCols = taosArrayGetSize(pGroupCols);

×

UNCOV

195

  for (int32_t i = 0; i < numOfGroupCols; ++i) {

×

UNCOV

196

    SColumn*         pCol = (SColumn*)taosArrayGet(pGroupCols, i);

×

UNCOV

197

    SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, pCol->slotId);

×

UNCOV

200

    if (pCol->slotId > taosArrayGetSize(pBlock->pDataBlock)) {

×

201

      continue;

×

UNCOV

204

    if (pBlock->pBlockAgg != NULL) {

×

205

      pColAgg = &pBlock->pBlockAgg[pCol->slotId];  // TODO is agg data matched?

×

UNCOV

208

    SGroupKeys* pkey = taosArrayGet(pGroupColVals, i);

×

UNCOV

209

    if (colDataIsNull(pColInfoData, pBlock->info.rows, rowIndex, pColAgg)) {

×

UNCOV

210

      pkey->isNull = true;

×

UNCOV

212

      pkey->isNull = false;

×

UNCOV

213

      char* val = colDataGetData(pColInfoData, rowIndex);

×

UNCOV

214

      if (pkey->type == TSDB_DATA_TYPE_JSON) {

×

UNCOV

215

        if (tTagIsJson(val)) {

×

UNCOV

216

          terrno = TSDB_CODE_QRY_JSON_IN_GROUP_ERROR;

×

UNCOV

217

          return;

×

UNCOV

219

        int32_t dataLen = getJsonValueLen(val);

×

UNCOV

220

        memcpy(pkey->pData, val, dataLen);

×

UNCOV

221

      } else if (IS_VAR_DATA_TYPE(pkey->type)) {

×

UNCOV

222

        memcpy(pkey->pData, val, varDataTLen(val));

×

UNCOV

224

        memcpy(pkey->pData, val, pkey->bytes);

×

UNCOV

230

static int32_t buildGroupKeys(void* pKey, const SArray* pGroupColVals) {

×

UNCOV

231

  size_t numOfGroupCols = taosArrayGetSize(pGroupColVals);

×

UNCOV

233

  char* isNull = (char*)pKey;

×

UNCOV

234

  char* pStart = (char*)pKey + sizeof(int8_t) * numOfGroupCols;

×

UNCOV

235

  for (int32_t i = 0; i < numOfGroupCols; ++i) {

×

UNCOV

236

    SGroupKeys* pkey = taosArrayGet(pGroupColVals, i);

×

UNCOV

237

    if (pkey->isNull) {

×

UNCOV

238

      isNull[i] = 1;

×

UNCOV

239

      continue;

×

UNCOV

242

    isNull[i] = 0;

×

UNCOV

243

    if (pkey->type == TSDB_DATA_TYPE_JSON) {

×

UNCOV

244

      int32_t dataLen = getJsonValueLen(pkey->pData);

×

UNCOV

245

      memcpy(pStart, (pkey->pData), dataLen);

×

UNCOV

246

      pStart += dataLen;

×

UNCOV

247

    } else if (IS_VAR_DATA_TYPE(pkey->type)) {

×

UNCOV

248

      varDataCopy(pStart, pkey->pData);

×

UNCOV

249

      pStart += varDataTLen(pkey->pData);

×

UNCOV

251

      memcpy(pStart, pkey->pData, pkey->bytes);

×

UNCOV

252

      pStart += pkey->bytes;

×

UNCOV

256

  return (int32_t)(pStart - (char*)pKey);

×

UNCOV

260

static void doAssignGroupKeys(SqlFunctionCtx* pCtx, int32_t numOfOutput, int32_t totalRows, int32_t rowIndex) {

×

UNCOV

261

  for (int32_t i = 0; i < numOfOutput; ++i) {

×

UNCOV

262

    if (pCtx[i].functionId == -1) {  // select count(*),key from t group by key.

×

UNCOV

263

      SResultRowEntryInfo* pEntryInfo = GET_RES_INFO(&pCtx[i]);

×

UNCOV

265

      SColumnInfoData* pColInfoData = pCtx[i].input.pData[0];

×

UNCOV

267

      if (!colDataIsNull(pColInfoData, totalRows, rowIndex, NULL)) {

×

UNCOV

268

        char* dest = GET_ROWCELL_INTERBUF(pEntryInfo);

×

UNCOV

269

        char* data = colDataGetData(pColInfoData, rowIndex);

×

UNCOV

271

        if (pColInfoData->info.type == TSDB_DATA_TYPE_JSON) {

×

UNCOV

272

          int32_t dataLen = getJsonValueLen(data);

×

UNCOV

273

          memcpy(dest, data, dataLen);

×

UNCOV

274

        } else if (IS_VAR_DATA_TYPE(pColInfoData->info.type)) {

×

UNCOV

275

          varDataCopy(dest, data);

×

UNCOV

277

          memcpy(dest, data, pColInfoData->info.bytes);

×

UNCOV

280

        pEntryInfo->isNullRes = 1;

×

UNCOV

283

      pEntryInfo->numOfRes = 1;

×

UNCOV

286

×

UNCOV

288

static void doHashGroupbyAgg(SOperatorInfo* pOperator, SSDataBlock* pBlock) {

×

UNCOV

289

  SExecTaskInfo*        pTaskInfo = pOperator->pTaskInfo;

×

UNCOV

290

  SGroupbyOperatorInfo* pInfo = pOperator->info;

×

UNCOV

292

  SqlFunctionCtx* pCtx = pOperator->exprSupp.pCtx;

×

UNCOV

293

  int32_t         numOfGroupCols = taosArrayGetSize(pInfo->pGroupCols);

×

UNCOV

299

  int32_t len = 0;

×

UNCOV

300

  terrno = TSDB_CODE_SUCCESS;

×

UNCOV

302

  int32_t num = 0;

×

UNCOV

303

  for (int32_t j = 0; j < pBlock->info.rows; ++j) {

×

UNCOV

305

    if (!pInfo->isInit) {

×

UNCOV

306

      recordNewGroupKeys(pInfo->pGroupCols, pInfo->pGroupColVals, pBlock, j);

×

UNCOV

307

      if (terrno != TSDB_CODE_SUCCESS) {  // group by json error

×

UNCOV

308

        T_LONG_JMP(pTaskInfo->env, terrno);

×

UNCOV

310

      pInfo->isInit = true;

×

UNCOV

311

      num++;

×

UNCOV

312

      continue;

×

UNCOV

315

    bool equal = groupKeyCompare(pInfo->pGroupCols, pInfo->pGroupColVals, pBlock, j, numOfGroupCols);

×

UNCOV

316

    if (equal) {

×

UNCOV

317

      num++;

×

UNCOV

318

      continue;

×

UNCOV

322

    if (j == 0) {

×

UNCOV

323

      num++;

×

UNCOV

324

      recordNewGroupKeys(pInfo->pGroupCols, pInfo->pGroupColVals, pBlock, j);

×

UNCOV

325

      if (terrno != TSDB_CODE_SUCCESS) {  // group by json error

×

326

        T_LONG_JMP(pTaskInfo->env, terrno);

×

UNCOV

328

      continue;

×

UNCOV

331

    len = buildGroupKeys(pInfo->keyBuf, pInfo->pGroupColVals);

×

UNCOV

332

    int32_t ret = setGroupResultOutputBuf(pOperator, &(pInfo->binfo), pOperator->exprSupp.numOfExprs, pInfo->keyBuf,

×

UNCOV

334

    if (ret != TSDB_CODE_SUCCESS) {  // null data, too many state code

×

335

      T_LONG_JMP(pTaskInfo->env, ret);

×

UNCOV

338

    int32_t rowIndex = j - num;

×

UNCOV

339

    ret = applyAggFunctionOnPartialTuples(pTaskInfo, pCtx, NULL, rowIndex, num, pBlock->info.rows,

×

UNCOV

341

    if (ret != TSDB_CODE_SUCCESS) {

×

342

      T_LONG_JMP(pTaskInfo->env, ret);

×

UNCOV

346

    doAssignGroupKeys(pCtx, pOperator->exprSupp.numOfExprs, pBlock->info.rows, rowIndex);

×

UNCOV

347

    recordNewGroupKeys(pInfo->pGroupCols, pInfo->pGroupColVals, pBlock, j);

×

UNCOV

348

    num = 1;

×

UNCOV

351

  if (num > 0) {

×

UNCOV

352

    len = buildGroupKeys(pInfo->keyBuf, pInfo->pGroupColVals);

×

UNCOV

353

    int32_t ret = setGroupResultOutputBuf(pOperator, &(pInfo->binfo), pOperator->exprSupp.numOfExprs, pInfo->keyBuf,

×

UNCOV

355

    if (ret != TSDB_CODE_SUCCESS) {

×

356

      T_LONG_JMP(pTaskInfo->env, ret);

×

UNCOV

359

    int32_t rowIndex = pBlock->info.rows - num;

×

UNCOV

360

    ret = applyAggFunctionOnPartialTuples(pTaskInfo, pCtx, NULL, rowIndex, num, pBlock->info.rows,

×

UNCOV

362

    if (ret != TSDB_CODE_SUCCESS) {

×

363

      T_LONG_JMP(pTaskInfo->env, ret);

×

UNCOV

365

    doAssignGroupKeys(pCtx, pOperator->exprSupp.numOfExprs, pBlock->info.rows, rowIndex);

×

UNCOV

367

×

UNCOV

369

bool hasRemainResultByHash(SOperatorInfo* pOperator) {

×

UNCOV

370

  SGroupbyOperatorInfo* pInfo = pOperator->info;

×

UNCOV

371

  SSHashObj*            pHashmap = pInfo->aggSup.pResultRowHashTable;

×

UNCOV

372

  return pInfo->groupResInfo.index < tSimpleHashGetSize(pHashmap);

×

UNCOV

375

void doBuildResultDatablockByHash(SOperatorInfo* pOperator, SOptrBasicInfo* pbInfo, SGroupResInfo* pGroupResInfo,

×

UNCOV

377

  SGroupbyOperatorInfo* pInfo = pOperator->info;

×

UNCOV

378

  SSHashObj*            pHashmap = pInfo->aggSup.pResultRowHashTable;

×

UNCOV

379

  SExecTaskInfo*        pTaskInfo = pOperator->pTaskInfo;

×

UNCOV

381

  SSDataBlock* pBlock = pInfo->binfo.pRes;

×

UNCOV

384

  pBlock->info.version = pTaskInfo->version;

×

UNCOV

386

  blockDataCleanup(pBlock);

×

UNCOV

387

  if (!hasRemainResultByHash(pOperator)) {

×

UNCOV

388

    return;

×

UNCOV

391

  pBlock->info.id.groupId = 0;

×

UNCOV

392

  if (!pInfo->binfo.mergeResultBlock) {

×

UNCOV

393

    doCopyToSDataBlockByHash(pTaskInfo, pBlock, &pOperator->exprSupp, pInfo->aggSup.pResultBuf, &pInfo->groupResInfo,

×

UNCOV

396

    while (hasRemainResultByHash(pOperator)) {

×

UNCOV

397

      doCopyToSDataBlockByHash(pTaskInfo, pBlock, &pOperator->exprSupp, pInfo->aggSup.pResultBuf, &pInfo->groupResInfo,

×

UNCOV

399

      if (pBlock->info.rows >= pOperator->resultInfo.threshold) {

×

UNCOV

400

        break;

×

UNCOV

402

      pBlock->info.id.groupId = 0;

×

UNCOV

406

    pBlock->info.id.groupId = 0;

×

UNCOV

410

static SSDataBlock* buildGroupResultDataBlockByHash(SOperatorInfo* pOperator) {

×

UNCOV

411

  int32_t               code = TSDB_CODE_SUCCESS;

×

UNCOV

412

  int32_t               lino = 0;

×

UNCOV

413

  SExecTaskInfo*        pTaskInfo = pOperator->pTaskInfo;

×

UNCOV

414

  SGroupbyOperatorInfo* pInfo = pOperator->info;

×

UNCOV

415

  SSDataBlock*          pRes = pInfo->binfo.pRes;

×

UNCOV

419

    doBuildResultDatablockByHash(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf);

×

UNCOV

421

    code = doFilter(pRes, pOperator->exprSupp.pFilterInfo, NULL);

×

UNCOV

422

    QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

424

    if (!hasRemainResultByHash(pOperator)) {

×

UNCOV

425

      setOperatorCompleted(pOperator);

×

UNCOV

427

      tSimpleHashCleanup(pInfo->aggSup.pResultRowHashTable);

×

UNCOV

428

      pInfo->aggSup.pResultRowHashTable = NULL;

×

UNCOV

429

      break;

×

UNCOV

431

    if (pRes->info.rows > 0) {

×

UNCOV

432

      break;

×

UNCOV

436

  pOperator->resultInfo.totalRows += pRes->info.rows;

×

UNCOV

438

_end:

×

UNCOV

439

  if (code != TSDB_CODE_SUCCESS) {

×

440

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

441

    T_LONG_JMP(pTaskInfo->env, code);

×

UNCOV

443

  return (pRes->info.rows == 0) ? NULL : pRes;

×

UNCOV

446

static int32_t hashGroupbyAggregateNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {

×

UNCOV

447

  int32_t               code = TSDB_CODE_SUCCESS;

×

UNCOV

448

  int32_t               lino = 0;

×

UNCOV

449

  SExecTaskInfo*        pTaskInfo = pOperator->pTaskInfo;

×

UNCOV

450

  SGroupbyOperatorInfo* pInfo = pOperator->info;

×

UNCOV

451

  SGroupResInfo*        pGroupResInfo = &pInfo->groupResInfo;

×

UNCOV

452

  int32_t               order = pInfo->binfo.inputTsOrder;

×

UNCOV

453

  int64_t               st = taosGetTimestampUs();

×

UNCOV

455

  QRY_PARAM_CHECK(ppRes);

×

UNCOV

456

  if (pOperator->status == OP_EXEC_DONE) {

×

UNCOV

457

    return code;

×

UNCOV

460

  if (pOperator->status == OP_RES_TO_RETURN) {

×

UNCOV

461

    (*ppRes) = buildGroupResultDataBlockByHash(pOperator);

×

UNCOV

462

    return code;

×

UNCOV

465

  while (1) {

×

UNCOV

466

    SSDataBlock* pBlock = getNextBlockFromDownstream(pOperator, 0);

×

UNCOV

467

    if (pBlock == NULL) {

×

UNCOV

468

      break;

×

UNCOV

471

    pInfo->binfo.pRes->info.scanFlag = pBlock->info.scanFlag;

×

UNCOV

474

    code = setInputDataBlock(&pOperator->exprSupp, pBlock, order, pBlock->info.scanFlag, true);

×

UNCOV

475

    QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

478

    if (pInfo->scalarSup.pExprInfo != NULL) {

×

UNCOV

479

      code = projectApplyFunctions(pInfo->scalarSup.pExprInfo, pBlock, pBlock, pInfo->scalarSup.pCtx,

×

UNCOV

481

      QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

484

    doHashGroupbyAgg(pOperator, pBlock);

×

UNCOV

487

  pOperator->status = OP_RES_TO_RETURN;

×

UNCOV

490

  if (pGroupResInfo->pRows != NULL) {

×

491

    taosArrayDestroy(pGroupResInfo->pRows);

×

UNCOV

494

  if (pGroupResInfo->pBuf) {

×

495

    taosMemoryFree(pGroupResInfo->pBuf);

×

496

    pGroupResInfo->pBuf = NULL;

×

UNCOV

499

  pGroupResInfo->index = 0;

×

UNCOV

500

  pGroupResInfo->iter = 0;

×

UNCOV

501

  pGroupResInfo->dataPos = NULL;

×

UNCOV

503

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

×

UNCOV

505

_end:

×

UNCOV

506

  if (code != TSDB_CODE_SUCCESS) {

×

507

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

508

    pTaskInfo->code = code;

×

509

    T_LONG_JMP(pTaskInfo->env, code);

×

UNCOV

511

    (*ppRes) = buildGroupResultDataBlockByHash(pOperator);

×

UNCOV

514

  return code;

×

UNCOV

517

int32_t createGroupOperatorInfo(SOperatorInfo* downstream, SAggPhysiNode* pAggNode, SExecTaskInfo* pTaskInfo,

×

UNCOV

519

  QRY_PARAM_CHECK(pOptrInfo);

×

UNCOV

521

  int32_t               code = TSDB_CODE_SUCCESS;

×

UNCOV

522

  int32_t               lino = 0;

×

UNCOV

523

  SGroupbyOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SGroupbyOperatorInfo));

×

UNCOV

524

  SOperatorInfo*        pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));

×

UNCOV

525

  if (pInfo == NULL || pOperator == NULL) {

×

526

    code = terrno;

×

527

    goto _error;

×

UNCOV

530

  pOperator->exprSupp.hasWindowOrGroup = true;

×

UNCOV

532

  SSDataBlock* pResBlock = createDataBlockFromDescNode(pAggNode->node.pOutputDataBlockDesc);

×

UNCOV

533

  if (pResBlock == NULL) {

×

534

    code = terrno;

×

535

    goto _error;

×

UNCOV

537

  initBasicInfo(&pInfo->binfo, pResBlock);

×

UNCOV

539

  pInfo->pGroupCols = NULL;

×

UNCOV

540

  code = extractColumnInfo(pAggNode->pGroupKeys, &pInfo->pGroupCols);

×

UNCOV

541

  QUERY_CHECK_CODE(code, lino, _error);

×

UNCOV

543

  int32_t    numOfScalarExpr = 0;

×

UNCOV

544

  SExprInfo* pScalarExprInfo = NULL;

×

UNCOV

545

  if (pAggNode->pExprs != NULL) {

×

UNCOV

546

    code = createExprInfo(pAggNode->pExprs, NULL, &pScalarExprInfo, &numOfScalarExpr);

×

UNCOV

547

    QUERY_CHECK_CODE(code, lino, _error);

×

UNCOV

550

  code = initExprSupp(&pInfo->scalarSup, pScalarExprInfo, numOfScalarExpr, &pTaskInfo->storageAPI.functionStore);

×

UNCOV

551

  QUERY_CHECK_CODE(code, lino, _error);

×

UNCOV

553

  initResultSizeInfo(&pOperator->resultInfo, 4096);

×

UNCOV

554

  code = blockDataEnsureCapacity(pInfo->binfo.pRes, pOperator->resultInfo.capacity);

×

UNCOV

555

  QUERY_CHECK_CODE(code, lino, _error);

×

UNCOV

557

  code = initGroupOptrInfo(&pInfo->pGroupColVals, &pInfo->groupKeyLen, &pInfo->keyBuf, pInfo->pGroupCols);

×

UNCOV

558

  QUERY_CHECK_CODE(code, lino, _error);

×

UNCOV

560

  int32_t    num = 0;

×

UNCOV

561

  SExprInfo* pExprInfo = NULL;

×

UNCOV

563

  code = createExprInfo(pAggNode->pAggFuncs, pAggNode->pGroupKeys, &pExprInfo, &num);

×

UNCOV

564

  QUERY_CHECK_CODE(code, lino, _error);

×

UNCOV

566

  code = initAggSup(&pOperator->exprSupp, &pInfo->aggSup, pExprInfo, num, pInfo->groupKeyLen, pTaskInfo->id.str,

×

UNCOV

567

                    pTaskInfo->streamInfo.pState, &pTaskInfo->storageAPI.functionStore);

×

UNCOV

568

  QUERY_CHECK_CODE(code, lino, _error);

×

UNCOV

570

  code = filterInitFromNode((SNode*)pAggNode->node.pConditions, &pOperator->exprSupp.pFilterInfo, 0);

×

UNCOV

571

  QUERY_CHECK_CODE(code, lino, _error);

×

UNCOV

573

  initResultRowInfo(&pInfo->binfo.resultRowInfo);

×

UNCOV

574

  setOperatorInfo(pOperator, "GroupbyAggOperator", 0, true, OP_NOT_OPENED, pInfo, pTaskInfo);

×

UNCOV

576

  pInfo->binfo.mergeResultBlock = pAggNode->mergeDataBlock;

×

UNCOV

577

  pInfo->binfo.inputTsOrder = pAggNode->node.inputTsOrder;

×

UNCOV

578

  pInfo->binfo.outputTsOrder = pAggNode->node.outputTsOrder;

×

UNCOV

580

  pInfo->pOperator = pOperator;

×

UNCOV

582

  pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, hashGroupbyAggregateNext, NULL, destroyGroupOperatorInfo,

×

UNCOV

584

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

×

UNCOV

585

  QUERY_CHECK_CODE(code, lino, _error);

×

UNCOV

587

  *pOptrInfo = pOperator;

×

UNCOV

588

  return TSDB_CODE_SUCCESS;

×

590

_error:

×

591

  if (pInfo != NULL) destroyGroupOperatorInfo(pInfo);

×

592

  destroyOperatorAndDownstreams(pOperator, &downstream, 1);

×

593

  pTaskInfo->code = code;

×

594

  return code;

×

597

SSDataBlock* createBlockDataNotLoaded(const SOperatorInfo* pOperator, SSDataBlock* pDataBlock) {

×

598

  int32_t code = TSDB_CODE_SUCCESS;

×

599

  int32_t lino = 0;

×

600

  if (pDataBlock == NULL) {

×

601

    return NULL;

×

604

  SSDataBlock* pDstBlock = NULL;

×

605

  code = createDataBlock(&pDstBlock);

×

606

  QUERY_CHECK_CODE(code, lino, _end);

×

608

  pDstBlock->info = pDataBlock->info;

×

609

  pDstBlock->info.id.blockId = pOperator->resultDataBlockId;

×

610

  pDstBlock->info.capacity = 0;

×

611

  pDstBlock->info.rowSize = 0;

×

613

  size_t numOfCols = pOperator->exprSupp.numOfExprs;

×

614

  if (pDataBlock->pBlockAgg) {

×

615

    pDstBlock->pBlockAgg = taosMemoryCalloc(numOfCols, sizeof(SColumnDataAgg));

×

616

    if (pDstBlock->pBlockAgg == NULL) {

×

617

      blockDataDestroy(pDstBlock);

×

618

      return NULL;

×

620

    for (int i = 0; i < numOfCols; ++i) {

×

621

      pDstBlock->pBlockAgg[i].colId = -1;

×

625

  for (int32_t i = 0; i < pOperator->exprSupp.numOfExprs; ++i) {

×

626

    SExprInfo*       pExpr = &pOperator->exprSupp.pExprInfo[i];

×

627

    int32_t          slotId = pExpr->base.pParam[0].pCol->slotId;

×

628

    SColumnInfoData* pSrc = taosArrayGet(pDataBlock->pDataBlock, slotId);

×

629

    SColumnInfoData  colInfo = {.hasNull = true, .info = pSrc->info};

×

630

    code = blockDataAppendColInfo(pDstBlock, &colInfo);

×

631

    QUERY_CHECK_CODE(code, lino, _end);

×

633

    SColumnInfoData* pDst = taosArrayGet(pDstBlock->pDataBlock, i);

×

634

    if (pDataBlock->pBlockAgg && pDataBlock->pBlockAgg[slotId].colId != -1) {

×

635

      pDstBlock->pBlockAgg[i] = pDataBlock->pBlockAgg[slotId];

×

637

      code = doEnsureCapacity(pDst, &pDstBlock->info, pDataBlock->info.rows, false);

×

638

      QUERY_CHECK_CODE(code, lino, _end);

×

640

      code = colDataAssign(pDst, pSrc, pDataBlock->info.rows, &pDataBlock->info);

×

641

      QUERY_CHECK_CODE(code, lino, _end);

×

645

_end:

×

646

  if (code != TSDB_CODE_SUCCESS) {

×

647

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

648

    blockDataDestroy(pDstBlock);

×

649

    return NULL;

×

651

  return pDstBlock;

×

UNCOV

654

static void doHashPartition(SOperatorInfo* pOperator, SSDataBlock* pBlock) {

×

UNCOV

655

  int32_t                 code = TSDB_CODE_SUCCESS;

×

UNCOV

656

  int32_t                 lino = 0;

×

UNCOV

657

  SPartitionOperatorInfo* pInfo = pOperator->info;

×

UNCOV

658

  SExecTaskInfo*          pTaskInfo = pOperator->pTaskInfo;

×

UNCOV

660

  for (int32_t j = 0; j < pBlock->info.rows; ++j) {

×

UNCOV

661

    recordNewGroupKeys(pInfo->pGroupCols, pInfo->pGroupColVals, pBlock, j);

×

UNCOV

662

    int32_t len = buildGroupKeys(pInfo->keyBuf, pInfo->pGroupColVals);

×

UNCOV

664

    SDataGroupInfo* pGroupInfo = NULL;

×

UNCOV

665

    void*           pPage = getCurrentDataGroupInfo(pInfo, &pGroupInfo, len);

×

UNCOV

666

    if (pPage == NULL) {

×

667

      T_LONG_JMP(pTaskInfo->env, terrno);

×

UNCOV

670

    pGroupInfo->numOfRows += 1;

×

UNCOV

673

    if (pGroupInfo->groupId == 0) {

×

UNCOV

674

      pGroupInfo->groupId = calcGroupId(pInfo->keyBuf, len);

×

UNCOV

677

    if (pBlock->info.dataLoad) {

×

UNCOV

679

      int32_t* rows = (int32_t*)pPage;

×

UNCOV

681

      size_t numOfCols = pOperator->exprSupp.numOfExprs;

×

UNCOV

682

      for (int32_t i = 0; i < numOfCols; ++i) {

×

UNCOV

683

        SExprInfo* pExpr = &pOperator->exprSupp.pExprInfo[i];

×

UNCOV

684

        int32_t    slotId = pExpr->base.pParam[0].pCol->slotId;

×

UNCOV

686

        SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, slotId);

×

UNCOV

688

        int32_t bytes = pColInfoData->info.bytes;

×

UNCOV

689

        int32_t startOffset = pInfo->columnOffset[i];

×

UNCOV

691

        int32_t* columnLen = NULL;

×

UNCOV

692

        int32_t  contentLen = 0;

×

UNCOV

694

        if (IS_VAR_DATA_TYPE(pColInfoData->info.type)) {

×

UNCOV

695

          int32_t* offset = (int32_t*)((char*)pPage + startOffset);

×

UNCOV

696

          columnLen = (int32_t*)((char*)pPage + startOffset + sizeof(int32_t) * pInfo->rowCapacity);

×

UNCOV

697

          char* data = (char*)((char*)columnLen + sizeof(int32_t));

×

UNCOV

699

          if (colDataIsNull_s(pColInfoData, j)) {

×

UNCOV

700

            offset[(*rows)] = -1;

×

UNCOV

701

            contentLen = 0;

×

UNCOV

702

          } else if (pColInfoData->info.type == TSDB_DATA_TYPE_JSON) {

×

UNCOV

703

            offset[*rows] = (*columnLen);

×

UNCOV

704

            char*   src = colDataGetData(pColInfoData, j);

×

UNCOV

705

            int32_t dataLen = getJsonValueLen(src);

×

UNCOV

707

            memcpy(data + (*columnLen), src, dataLen);

×

UNCOV

708

            int32_t v = (data + (*columnLen) + dataLen - (char*)pPage);

×

UNCOV

709

            QUERY_CHECK_CONDITION((v > 0), code, lino, _end, TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR);

×

UNCOV

711

            contentLen = dataLen;

×

UNCOV

713

            offset[*rows] = (*columnLen);

×

UNCOV

714

            char* src = colDataGetData(pColInfoData, j);

×

UNCOV

715

            memcpy(data + (*columnLen), src, varDataTLen(src));

×

UNCOV

716

            int32_t v = (data + (*columnLen) + varDataTLen(src) - (char*)pPage);

×

UNCOV

717

            QUERY_CHECK_CONDITION((v > 0), code, lino, _end, TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR);

×

UNCOV

719

            contentLen = varDataTLen(src);

×

UNCOV

722

          char* bitmap = (char*)pPage + startOffset;

×

UNCOV

723

          columnLen = (int32_t*)((char*)pPage + startOffset + BitmapLen(pInfo->rowCapacity));

×

UNCOV

724

          char* data = (char*)columnLen + sizeof(int32_t);

×

UNCOV

726

          bool isNull = colDataIsNull_f(pColInfoData->nullbitmap, j);

×

UNCOV

727

          if (isNull) {

×

UNCOV

728

            colDataSetNull_f(bitmap, (*rows));

×

UNCOV

730

            memcpy(data + (*columnLen), colDataGetData(pColInfoData, j), bytes);

×

UNCOV

731

            QUERY_CHECK_CONDITION(((data + (*columnLen) + bytes - (char*)pPage) <= getBufPageSize(pInfo->pBuf)), code,

×

UNCOV

734

          contentLen = bytes;

×

UNCOV

737

        (*columnLen) += contentLen;

×

UNCOV

740

      (*rows) += 1;

×

UNCOV

742

      setBufPageDirty(pPage, true);

×

UNCOV

743

      releaseBufPage(pInfo->pBuf, pPage);

×

745

      SSDataBlock* dataNotLoadBlock = createBlockDataNotLoaded(pOperator, pBlock);

×

746

      if (dataNotLoadBlock == NULL) {

×

747

        T_LONG_JMP(pTaskInfo->env, terrno);

×

749

      if (pGroupInfo->blockForNotLoaded == NULL) {

×

750

        pGroupInfo->blockForNotLoaded = taosArrayInit(0, sizeof(SSDataBlock*));

×

751

        QUERY_CHECK_NULL(pGroupInfo->blockForNotLoaded, code, lino, _end, terrno);

×

752

        pGroupInfo->offsetForNotLoaded = 0;

×

754

      dataNotLoadBlock->info.id.groupId = pGroupInfo->groupId;

×

755

      dataNotLoadBlock->info.dataLoad = 0;

×

756

      void* tmp = taosArrayPush(pGroupInfo->blockForNotLoaded, &dataNotLoadBlock);

×

757

      QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

×

758

      break;

×

UNCOV

762

_end:

×

UNCOV

763

  if (code != TSDB_CODE_SUCCESS) {

×

764

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

765

    T_LONG_JMP(pTaskInfo->env, code);

×

UNCOV

767

×

UNCOV

769

void* getCurrentDataGroupInfo(const SPartitionOperatorInfo* pInfo, SDataGroupInfo** pGroupInfo, int32_t len) {

×

UNCOV

770

  int32_t         code = TSDB_CODE_SUCCESS;

×

UNCOV

771

  int32_t         lino = 0;

×

UNCOV

772

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

×

UNCOV

774

  void* pPage = NULL;

×

UNCOV

775

  if (p == NULL) {  // it is a new group

×

UNCOV

776

    SDataGroupInfo gi = {0};

×

UNCOV

777

    gi.pPageList = taosArrayInit(100, sizeof(int32_t));

×

UNCOV

778

    QUERY_CHECK_NULL(gi.pPageList, code, lino, _end, terrno);

×

UNCOV

780

    code = taosHashPut(pInfo->pGroupSet, pInfo->keyBuf, len, &gi, sizeof(SDataGroupInfo));

×

UNCOV

781

    if (code == TSDB_CODE_DUP_KEY) {

×

782

      code = TSDB_CODE_SUCCESS;

×

UNCOV

784

    QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

786

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

×

UNCOV

788

    int32_t pageId = 0;

×

UNCOV

789

    pPage = getNewBufPage(pInfo->pBuf, &pageId);

×

UNCOV

790

    if (pPage == NULL) {

×

791

      return pPage;

×

UNCOV

794

    void* tmp = taosArrayPush(p->pPageList, &pageId);

×

UNCOV

795

    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

×

UNCOV

797

    *(int32_t*)pPage = 0;

×

UNCOV

799

    int32_t* curId = taosArrayGetLast(p->pPageList);

×

UNCOV

800

    pPage = getBufPage(pInfo->pBuf, *curId);

×

UNCOV

801

    if (pPage == NULL) {

×

802

      qError("failed to get buffer, code:%s", tstrerror(terrno));

×

803

      return pPage;

×

UNCOV

806

    int32_t* rows = (int32_t*)pPage;

×

UNCOV

807

    if (*rows >= pInfo->rowCapacity) {

×

UNCOV

809

      releaseBufPage(pInfo->pBuf, pPage);

×

UNCOV

812

      int32_t pageId = 0;

×

UNCOV

813

      pPage = getNewBufPage(pInfo->pBuf, &pageId);

×

UNCOV

814

      if (pPage == NULL) {

×

815

        qError("failed to get new buffer, code:%s", tstrerror(terrno));

×

816

        return NULL;

×

UNCOV

819

      void* tmp = taosArrayPush(p->pPageList, &pageId);

×

UNCOV

820

      QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

×

UNCOV

822

      memset(pPage, 0, getBufPageSize(pInfo->pBuf));

×

UNCOV

826

  *pGroupInfo = p;

×

UNCOV

828

_end:

×

UNCOV

829

  if (code != TSDB_CODE_SUCCESS) {

×

830

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

831

    return NULL;

×

UNCOV

834

  return pPage;

×

UNCOV

837

int32_t* setupColumnOffset(const SSDataBlock* pBlock, int32_t rowCapacity) {

×

UNCOV

838

  size_t   numOfCols = taosArrayGetSize(pBlock->pDataBlock);

×

UNCOV

839

  int32_t* offset = taosMemoryCalloc(numOfCols, sizeof(int32_t));

×

UNCOV

840

  if (!offset) {

×

841

    return NULL;

×

UNCOV

844

  offset[0] = sizeof(int32_t) +

×

UNCOV

847

  for (int32_t i = 0; i < numOfCols - 1; ++i) {

×

UNCOV

848

    SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, i);

×

UNCOV

850

    int32_t bytes = pColInfoData->info.bytes;

×

UNCOV

851

    int32_t payloadLen = bytes * rowCapacity;

×

UNCOV

853

    if (IS_VAR_DATA_TYPE(pColInfoData->info.type)) {

×

UNCOV

855

      offset[i + 1] = rowCapacity * sizeof(int32_t) + sizeof(int32_t) + payloadLen + offset[i];

×

UNCOV

858

      offset[i + 1] = BitmapLen(rowCapacity) + sizeof(int32_t) + payloadLen + offset[i];

×

UNCOV

862

  return offset;

×

UNCOV

865

static void clearPartitionOperator(SPartitionOperatorInfo* pInfo) {

×

UNCOV

866

  int32_t size = taosArrayGetSize(pInfo->sortedGroupArray);

×

UNCOV

867

  for (int32_t i = 0; i < size; i++) {

×

UNCOV

868

    SDataGroupInfo* pGp = taosArrayGet(pInfo->sortedGroupArray, i);

×

UNCOV

869

    if (pGp && pGp->blockForNotLoaded) {

×

870

      for (int32_t i = 0; i < pGp->blockForNotLoaded->size; i++) {

×

871

        SSDataBlock** pBlock = taosArrayGet(pGp->blockForNotLoaded, i);

×

872

        if (pBlock) blockDataDestroy(*pBlock);

×

874

      taosArrayClear(pGp->blockForNotLoaded);

×

875

      pGp->offsetForNotLoaded = 0;

×

UNCOV

877

    taosArrayDestroy(pGp->pPageList);

×

UNCOV

879

  taosArrayClear(pInfo->sortedGroupArray);

×

UNCOV

880

  clearDiskbasedBuf(pInfo->pBuf);

×

UNCOV

881

×

UNCOV

883

static int compareDataGroupInfo(const void* group1, const void* group2) {

×

UNCOV

884

  const SDataGroupInfo* pGroupInfo1 = group1;

×

UNCOV

885

  const SDataGroupInfo* pGroupInfo2 = group2;

×

UNCOV

887

  if (pGroupInfo1->groupId == pGroupInfo2->groupId) {

×

888

    return 0;

×

UNCOV

891

  return (pGroupInfo1->groupId < pGroupInfo2->groupId) ? -1 : 1;

×

UNCOV

894

static SSDataBlock* buildPartitionResultForNotLoadBlock(SDataGroupInfo* pGroupInfo) {

×

UNCOV

895

  if (pGroupInfo->blockForNotLoaded && pGroupInfo->offsetForNotLoaded < pGroupInfo->blockForNotLoaded->size) {

×

896

    SSDataBlock** pBlock = taosArrayGet(pGroupInfo->blockForNotLoaded, pGroupInfo->offsetForNotLoaded);

×

897

    if (!pBlock) {

×

898

      return NULL;

×

900

    pGroupInfo->offsetForNotLoaded++;

×

901

    return *pBlock;

×

UNCOV

903

  return NULL;

×

UNCOV

906

static SSDataBlock* buildPartitionResult(SOperatorInfo* pOperator) {

×

UNCOV

907

  int32_t                 code = TSDB_CODE_SUCCESS;

×

UNCOV

908

  int32_t                 lino = 0;

×

UNCOV

909

  SPartitionOperatorInfo* pInfo = pOperator->info;

×

UNCOV

910

  SExecTaskInfo*          pTaskInfo = pOperator->pTaskInfo;

×

UNCOV

912

  if (pInfo->remainRows == 0) {

×

UNCOV

913

    blockDataCleanup(pInfo->binfo.pRes);

×

UNCOV

914

    SDataGroupInfo* pGroupInfo =

×

UNCOV

915

        (pInfo->groupIndex != -1) ? taosArrayGet(pInfo->sortedGroupArray, pInfo->groupIndex) : NULL;

×

UNCOV

916

    if (pInfo->groupIndex == -1 || pInfo->pageIndex >= taosArrayGetSize(pGroupInfo->pPageList)) {

×

UNCOV

917

      if (pGroupInfo != NULL) {

×

UNCOV

918

        SSDataBlock* ret = buildPartitionResultForNotLoadBlock(pGroupInfo);

×

UNCOV

919

        if (ret != NULL) return ret;

×

UNCOV

922

      if (pInfo->groupIndex + 1 >= taosArrayGetSize(pInfo->sortedGroupArray)) {

×

UNCOV

923

        setOperatorCompleted(pOperator);

×

UNCOV

924

        clearPartitionOperator(pInfo);

×

UNCOV

925

        return NULL;

×

UNCOV

927

      ++pInfo->groupIndex;

×

UNCOV

929

      pGroupInfo = taosArrayGet(pInfo->sortedGroupArray, pInfo->groupIndex);

×

UNCOV

930

      if (pGroupInfo == NULL) {

×

931

        qError("failed to get buffer, code:%s, %s", tstrerror(terrno), GET_TASKID(pTaskInfo));

×

932

        T_LONG_JMP(pTaskInfo->env, terrno);

×

UNCOV

934

      pInfo->pageIndex = 0;

×

UNCOV

937

    int32_t* pageId = taosArrayGet(pGroupInfo->pPageList, pInfo->pageIndex);

×

UNCOV

938

    if (pageId == NULL) {

×

939

      qError("failed to get buffer, code:%s, %s", tstrerror(terrno), GET_TASKID(pTaskInfo));

×

940

      T_LONG_JMP(pTaskInfo->env, terrno);

×

UNCOV

942

    void*    page = getBufPage(pInfo->pBuf, *pageId);

×

UNCOV

943

    if (page == NULL) {

×

944

      qError("failed to get buffer, code:%s, %s", tstrerror(terrno), GET_TASKID(pTaskInfo));

×

945

      T_LONG_JMP(pTaskInfo->env, terrno);

×

UNCOV

947

    if (*(int32_t*)page == 0) {

×

948

      releaseBufPage(pInfo->pBuf, page);

×

949

      SSDataBlock* ret = buildPartitionResultForNotLoadBlock(pGroupInfo);

×

950

      if (ret != NULL) return ret;

×

951

      if (pInfo->groupIndex + 1 < taosArrayGetSize(pInfo->sortedGroupArray)) {

×

952

        pInfo->groupIndex++;

×

953

        pInfo->pageIndex = 0;

×

955

        setOperatorCompleted(pOperator);

×

956

        clearPartitionOperator(pInfo);

×

957

        return NULL;

×

959

      return buildPartitionResult(pOperator);

×

UNCOV

962

    code = blockDataEnsureCapacity(pInfo->binfo.pRes, pInfo->rowCapacity);

×

UNCOV

963

    QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

965

    code = blockDataFromBuf1(pInfo->binfo.pRes, page, pInfo->rowCapacity);

×

UNCOV

966

    QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

968

    pInfo->pageIndex += 1;

×

UNCOV

969

    releaseBufPage(pInfo->pBuf, page);

×

UNCOV

970

    pInfo->binfo.pRes->info.id.groupId = pGroupInfo->groupId;

×

UNCOV

971

    pInfo->binfo.pRes->info.dataLoad = 1;

×

UNCOV

972

    pInfo->orderedRows = 0;

×

UNCOV

973

  } else if (pInfo->pOrderInfoArr == NULL) {

×

974

    qError("Exception, remainRows not zero, but pOrderInfoArr is NULL");

×

UNCOV

977

  if (pInfo->pOrderInfoArr) {

×

UNCOV

978

    pInfo->binfo.pRes->info.rows += pInfo->remainRows;

×

UNCOV

979

    code = blockDataTrimFirstRows(pInfo->binfo.pRes, pInfo->orderedRows);

×

UNCOV

980

    QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

981

    pInfo->orderedRows = blockDataGetSortedRows(pInfo->binfo.pRes, pInfo->pOrderInfoArr);

×

UNCOV

982

    pInfo->remainRows = pInfo->binfo.pRes->info.rows - pInfo->orderedRows;

×

UNCOV

983

    pInfo->binfo.pRes->info.rows = pInfo->orderedRows;

×

UNCOV

986

  code = blockDataUpdateTsWindow(pInfo->binfo.pRes, 0);

×

UNCOV

987

  QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

989

_end:

×

UNCOV

990

  if (code != TSDB_CODE_SUCCESS) {

×

991

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

992

    T_LONG_JMP(pTaskInfo->env, code);

×

UNCOV

995

  pOperator->resultInfo.totalRows += pInfo->binfo.pRes->info.rows;

×

UNCOV

996

  return pInfo->binfo.pRes;

×

UNCOV

999

static int32_t hashPartitionNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {

×

UNCOV

1000

  if (pOperator->status == OP_EXEC_DONE) {

×

UNCOV

1001

    (*ppRes) = NULL;

×

UNCOV

1002

    return TSDB_CODE_SUCCESS;

×

UNCOV

1005

  int32_t                 code = TSDB_CODE_SUCCESS;

×

UNCOV

1006

  int32_t                 lino = 0;

×

UNCOV

1007

  SExecTaskInfo*          pTaskInfo = pOperator->pTaskInfo;

×

UNCOV

1008

  SPartitionOperatorInfo* pInfo = pOperator->info;

×

UNCOV

1009

  SSDataBlock*            pRes = pInfo->binfo.pRes;

×

UNCOV

1011

  if (pOperator->status == OP_RES_TO_RETURN) {

×

UNCOV

1012

    (*ppRes) =  buildPartitionResult(pOperator);

×

UNCOV

1013

    return code;

×

UNCOV

1016

  int64_t        st = taosGetTimestampUs();

×

UNCOV

1017

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

×

UNCOV

1019

  while (1) {

×

UNCOV

1020

    SSDataBlock* pBlock = getNextBlockFromDownstream(pOperator, 0);

×

UNCOV

1021

    if (pBlock == NULL) {

×

UNCOV

1022

      break;

×

UNCOV

1025

    pInfo->binfo.pRes->info.scanFlag = pBlock->info.scanFlag;

×

UNCOV

1027

    if (pInfo->scalarSup.pExprInfo != NULL) {

×

UNCOV

1028

      code = projectApplyFunctions(pInfo->scalarSup.pExprInfo, pBlock, pBlock, pInfo->scalarSup.pCtx,

×

UNCOV

1030

      QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1033

    terrno = TSDB_CODE_SUCCESS;

×

UNCOV

1034

    doHashPartition(pOperator, pBlock);

×

UNCOV

1035

    if (terrno != TSDB_CODE_SUCCESS) {  // group by json error

×

1036

      code = terrno;

×

1037

      QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1041

  SArray* groupArray = taosArrayInit(taosHashGetSize(pInfo->pGroupSet), sizeof(SDataGroupInfo));

×

UNCOV

1042

  QUERY_CHECK_NULL(groupArray, code, lino, _end, terrno);

×

UNCOV

1044

  void* pGroupIter = taosHashIterate(pInfo->pGroupSet, NULL);

×

UNCOV

1045

  while (pGroupIter != NULL) {

×

UNCOV

1046

    SDataGroupInfo* pGroupInfo = pGroupIter;

×

UNCOV

1047

    void*           tmp = taosArrayPush(groupArray, pGroupInfo);

×

UNCOV

1048

    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

×

UNCOV

1049

    pGroupIter = taosHashIterate(pInfo->pGroupSet, pGroupIter);

×

UNCOV

1052

  taosArraySort(groupArray, compareDataGroupInfo);

×

UNCOV

1053

  pInfo->sortedGroupArray = groupArray;

×

UNCOV

1054

  pInfo->groupIndex = -1;

×

UNCOV

1055

  taosHashClear(pInfo->pGroupSet);

×

UNCOV

1057

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

×

UNCOV

1059

  pOperator->status = OP_RES_TO_RETURN;

×

UNCOV

1060

  code = blockDataEnsureCapacity(pRes, 4096);

×

UNCOV

1061

  QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1063

_end:

×

UNCOV

1064

  if (code != TSDB_CODE_SUCCESS) {

×

1065

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

1066

    pTaskInfo->code = code;

×

1067

    T_LONG_JMP(pTaskInfo->env, code);

×

UNCOV

1070

  (*ppRes) = buildPartitionResult(pOperator);

×

UNCOV

1071

  return code;

×

UNCOV

1074

static void destroyPartitionOperatorInfo(void* param) {

×

UNCOV

1075

  SPartitionOperatorInfo* pInfo = (SPartitionOperatorInfo*)param;

×

UNCOV

1076

  cleanupBasicInfo(&pInfo->binfo);

×

UNCOV

1077

  taosArrayDestroy(pInfo->pGroupCols);

×

UNCOV

1079

  for (int i = 0; i < taosArrayGetSize(pInfo->pGroupColVals); i++) {

×

UNCOV

1080

    SGroupKeys key = *(SGroupKeys*)taosArrayGet(pInfo->pGroupColVals, i);

×

UNCOV

1081

    taosMemoryFree(key.pData);

×

UNCOV

1084

  taosArrayDestroy(pInfo->pGroupColVals);

×

UNCOV

1085

  taosMemoryFree(pInfo->keyBuf);

×

UNCOV

1087

  int32_t size = taosArrayGetSize(pInfo->sortedGroupArray);

×

UNCOV

1088

  for (int32_t i = 0; i < size; i++) {

×

UNCOV

1089

    SDataGroupInfo* pGp = taosArrayGet(pInfo->sortedGroupArray, i);

×

UNCOV

1090

    if (pGp) {

×

UNCOV

1091

      taosArrayDestroy(pGp->pPageList);

×

UNCOV

1094

  taosArrayDestroy(pInfo->sortedGroupArray);

×

UNCOV

1096

  void* pGroupIter = taosHashIterate(pInfo->pGroupSet, NULL);

×

UNCOV

1097

  while (pGroupIter != NULL) {

×

1098

    SDataGroupInfo* pGroupInfo = pGroupIter;

×

1099

    taosArrayDestroy(pGroupInfo->pPageList);

×

1100

    pGroupIter = taosHashIterate(pInfo->pGroupSet, pGroupIter);

×

UNCOV

1103

  taosHashCleanup(pInfo->pGroupSet);

×

UNCOV

1104

  taosMemoryFree(pInfo->columnOffset);

×

UNCOV

1106

  cleanupExprSupp(&pInfo->scalarSup);

×

UNCOV

1107

  destroyDiskbasedBuf(pInfo->pBuf);

×

UNCOV

1108

  taosArrayDestroy(pInfo->pOrderInfoArr);

×

UNCOV

1109

  taosMemoryFreeClear(param);

×

UNCOV

1110

×

UNCOV

1112

int32_t createPartitionOperatorInfo(SOperatorInfo* downstream, SPartitionPhysiNode* pPartNode,

×

UNCOV

1114

  QRY_PARAM_CHECK(pOptrInfo);

×

UNCOV

1116

  int32_t                 code = TSDB_CODE_SUCCESS;

×

UNCOV

1117

  int32_t                 lino = 0;

×

UNCOV

1118

  SPartitionOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SPartitionOperatorInfo));

×

UNCOV

1119

  SOperatorInfo*          pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));

×

UNCOV

1120

  if (pInfo == NULL || pOperator == NULL) {

×

1121

    pTaskInfo->code = code = terrno;

×

1122

    goto _error;

×

UNCOV

1125

  int32_t    numOfCols = 0;

×

UNCOV

1126

  SExprInfo* pExprInfo = NULL;

×

UNCOV

1127

  code = createExprInfo(pPartNode->pTargets, NULL, &pExprInfo, &numOfCols);

×

UNCOV

1128

  QUERY_CHECK_CODE(code, lino, _error);

×

UNCOV

1129

  pOperator->exprSupp.numOfExprs = numOfCols;

×

UNCOV

1130

  pOperator->exprSupp.pExprInfo = pExprInfo;

×

UNCOV

1132

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

×

UNCOV

1134

  if (pPartNode->needBlockOutputTsOrder) {

×

UNCOV

1135

    SBlockOrderInfo order = {.order = ORDER_ASC, .pColData = NULL, .nullFirst = false, .slotId = pPartNode->tsSlotId};

×

UNCOV

1136

    pInfo->pOrderInfoArr = taosArrayInit(1, sizeof(SBlockOrderInfo));

×

UNCOV

1137

    if (!pInfo->pOrderInfoArr) {

×

1138

      pTaskInfo->code = terrno;

×

1139

      goto _error;

×

UNCOV

1142

    void* tmp = taosArrayPush(pInfo->pOrderInfoArr, &order);

×

UNCOV

1143

    QUERY_CHECK_NULL(tmp, code, lino, _error, terrno);

×

UNCOV

1146

  if (pPartNode->pExprs != NULL) {

×

UNCOV

1147

    int32_t    num = 0;

×

UNCOV

1148

    SExprInfo* pExprInfo1 = NULL;

×

UNCOV

1149

    code = createExprInfo(pPartNode->pExprs, NULL, &pExprInfo1, &num);

×

UNCOV

1150

    QUERY_CHECK_CODE(code, lino, _error);

×

UNCOV

1152

    code = initExprSupp(&pInfo->scalarSup, pExprInfo1, num, &pTaskInfo->storageAPI.functionStore);

×

UNCOV

1153

    QUERY_CHECK_CODE(code, lino, _error);

×

UNCOV

1156

  _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);

×

UNCOV

1157

  pInfo->pGroupSet = taosHashInit(100, hashFn, false, HASH_NO_LOCK);

×

UNCOV

1158

  if (pInfo->pGroupSet == NULL) {

×

1159

    goto _error;

×

UNCOV

1162

  uint32_t defaultPgsz = 0;

×

UNCOV

1163

  int64_t defaultBufsz = 0;

×

UNCOV

1165

  pInfo->binfo.pRes = createDataBlockFromDescNode(pPartNode->node.pOutputDataBlockDesc);

×

UNCOV

1166

  QUERY_CHECK_NULL(pInfo->binfo.pRes, code, lino, _error, terrno);

×

UNCOV

1167

  code = getBufferPgSize(pInfo->binfo.pRes->info.rowSize, &defaultPgsz, &defaultBufsz);

×

UNCOV

1168

  if (code != TSDB_CODE_SUCCESS) {

×

1169

    goto _error;

×

UNCOV

1172

  if (!osTempSpaceAvailable()) {

×

1173

    terrno = TSDB_CODE_NO_DISKSPACE;

×

1174

    qError("Create partition operator info failed since %s, tempDir:%s", terrstr(), tsTempDir);

×

1175

    goto _error;

×

UNCOV

1178

  code = createDiskbasedBuf(&pInfo->pBuf, defaultPgsz, defaultBufsz, pTaskInfo->id.str, tsTempDir);

×

UNCOV

1179

  if (code != TSDB_CODE_SUCCESS) {

×

1180

    goto _error;

×

UNCOV

1183

  pInfo->rowCapacity =

×

UNCOV

1184

      blockDataGetCapacityInRow(pInfo->binfo.pRes, getBufPageSize(pInfo->pBuf),

×

UNCOV

1185

                                blockDataGetSerialMetaSize(taosArrayGetSize(pInfo->binfo.pRes->pDataBlock)));

×

UNCOV

1186

  if (pInfo->rowCapacity < 0) {

×

1187

    code = terrno;

×

1188

    goto _error;

×

UNCOV

1191

  pInfo->columnOffset = setupColumnOffset(pInfo->binfo.pRes, pInfo->rowCapacity);

×

UNCOV

1192

  QUERY_CHECK_NULL(pInfo->columnOffset, code, lino, _error, terrno);

×

UNCOV

1194

  code = initGroupOptrInfo(&pInfo->pGroupColVals, &pInfo->groupKeyLen, &pInfo->keyBuf, pInfo->pGroupCols);

×

UNCOV

1195

  if (code != TSDB_CODE_SUCCESS) {

×

1196

    goto _error;

×

UNCOV

1199

  setOperatorInfo(pOperator, "PartitionOperator", QUERY_NODE_PHYSICAL_PLAN_PARTITION, false, OP_NOT_OPENED, pInfo,

×

UNCOV

1202

  pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, hashPartitionNext, NULL, destroyPartitionOperatorInfo,

×

UNCOV

1205

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

×

UNCOV

1206

  if (code != TSDB_CODE_SUCCESS) {

×

1207

    goto _error;

×

UNCOV

1210

  *pOptrInfo = pOperator;

×

UNCOV

1211

  return TSDB_CODE_SUCCESS;

×

1213

_error:

×

1214

  if (pInfo != NULL) {

×

1215

    destroyPartitionOperatorInfo(pInfo);

×

1217

  pTaskInfo->code = code;

×

1218

  destroyOperatorAndDownstreams(pOperator, &downstream, 1);

×

1219

  TAOS_RETURN(code);

×

UNCOV

1222

int32_t setGroupResultOutputBuf(SOperatorInfo* pOperator, SOptrBasicInfo* binfo, int32_t numOfCols, char* pData,

×

UNCOV

1224

  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;

×

UNCOV

1225

  SResultRowInfo* pResultRowInfo = &binfo->resultRowInfo;

×

UNCOV

1226

  SqlFunctionCtx* pCtx = pOperator->exprSupp.pCtx;

×

UNCOV

1228

  SResultRow* pResultRow = doSetResultOutBufByKey(pBuf, pResultRowInfo, (char*)pData, bytes, true, groupId, pTaskInfo,

×

UNCOV

1230

  if (pResultRow == NULL || pTaskInfo->code != 0) {

×

1231

    return pTaskInfo->code;

×

UNCOV

1234

  return setResultRowInitCtx(pResultRow, pCtx, numOfCols, pOperator->exprSupp.rowEntryInfoOffset);

×

UNCOV

1237

uint64_t calGroupIdByData(SPartitionBySupporter* pParSup, SExprSupp* pExprSup, SSDataBlock* pBlock, int32_t rowId) {

×

UNCOV

1238

  if (pExprSup->pExprInfo != NULL) {

×

UNCOV

1240

        projectApplyFunctions(pExprSup->pExprInfo, pBlock, pBlock, pExprSup->pCtx, pExprSup->numOfExprs, NULL);

×

UNCOV

1241

    if (code != TSDB_CODE_SUCCESS) {

×

1242

      qError("calaculate group id error, code:%d", code);

×

UNCOV

1245

  recordNewGroupKeys(pParSup->pGroupCols, pParSup->pGroupColVals, pBlock, rowId);

×

UNCOV

1246

  int32_t  len = buildGroupKeys(pParSup->keyBuf, pParSup->pGroupColVals);

×

UNCOV

1247

  uint64_t groupId = calcGroupId(pParSup->keyBuf, len);

×

UNCOV

1248

  return groupId;

×

UNCOV

1251

static bool hasRemainPartion(SStreamPartitionOperatorInfo* pInfo) { return pInfo->parIte != NULL; }

×

UNCOV

1252

static bool hasRemainTbName(SStreamPartitionOperatorInfo* pInfo) { return pInfo->pTbNameIte != NULL; }

×

UNCOV

1254

static SSDataBlock* buildStreamPartitionResult(SOperatorInfo* pOperator) {

×

UNCOV

1255

  int32_t        code = TSDB_CODE_SUCCESS;

×

UNCOV

1256

  int32_t        lino = 0;

×

UNCOV

1257

  SStorageAPI*   pAPI = &pOperator->pTaskInfo->storageAPI;

×

UNCOV

1258

  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

×

UNCOV

1260

  SStreamPartitionOperatorInfo* pInfo = pOperator->info;

×

UNCOV

1261

  SSDataBlock*                  pDest = pInfo->binfo.pRes;

×

UNCOV

1262

  QUERY_CHECK_CONDITION((hasRemainPartion(pInfo)), code, lino, _end, TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR);

×

UNCOV

1263

  SPartitionDataInfo* pParInfo = (SPartitionDataInfo*)pInfo->parIte;

×

UNCOV

1264

  blockDataCleanup(pDest);

×

UNCOV

1265

  int32_t rows = taosArrayGetSize(pParInfo->rowIds);

×

UNCOV

1266

  code = blockDataEnsureCapacity(pDest, rows);

×

UNCOV

1267

  QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1269

  SSDataBlock* pSrc = pInfo->pInputDataBlock;

×

UNCOV

1270

  for (int32_t i = 0; i < rows; i++) {

×

UNCOV

1271

    int32_t rowIndex = *(int32_t*)taosArrayGet(pParInfo->rowIds, i);

×

UNCOV

1272

    for (int32_t j = 0; j < pOperator->exprSupp.numOfExprs; j++) {

×

UNCOV

1273

      int32_t          slotId = pOperator->exprSupp.pExprInfo[j].base.pParam[0].pCol->slotId;

×

UNCOV

1274

      SColumnInfoData* pSrcCol = taosArrayGet(pSrc->pDataBlock, slotId);

×

UNCOV

1275

      QUERY_CHECK_NULL(pSrcCol, code, lino, _end, terrno);

×

UNCOV

1276

      SColumnInfoData* pDestCol = taosArrayGet(pDest->pDataBlock, j);

×

UNCOV

1277

      QUERY_CHECK_NULL(pDestCol, code, lino, _end, terrno);

×

UNCOV

1278

      bool             isNull = colDataIsNull(pSrcCol, pSrc->info.rows, rowIndex, NULL);

×

UNCOV

1279

      char*            pSrcData = NULL;

×

UNCOV

1280

      if (!isNull) pSrcData = colDataGetData(pSrcCol, rowIndex);

×

UNCOV

1281

      code = colDataSetVal(pDestCol, pDest->info.rows, pSrcData, isNull);

×

UNCOV

1282

      QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1284

    pDest->info.rows++;

×

UNCOV

1286

  pDest->info.parTbName[0] = 0;

×

UNCOV

1287

  if (pInfo->tbnameCalSup.numOfExprs > 0) {

×

UNCOV

1288

    void*   tbname = NULL;

×

UNCOV

1289

    int32_t winCode = TSDB_CODE_SUCCESS;

×

UNCOV

1290

    code = pAPI->stateStore.streamStateGetParName(pOperator->pTaskInfo->streamInfo.pState, pParInfo->groupId, &tbname,

×

UNCOV

1292

    QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1294

    if (winCode == TSDB_CODE_SUCCESS) {

×

UNCOV

1295

      memcpy(pDest->info.parTbName, tbname, TSDB_TABLE_NAME_LEN);

×

UNCOV

1296

      pAPI->stateStore.streamStateFreeVal(tbname);

×

UNCOV

1299

  taosArrayDestroy(pParInfo->rowIds);

×

UNCOV

1300

  pParInfo->rowIds = NULL;

×

UNCOV

1301

  pDest->info.dataLoad = 1;

×

UNCOV

1303

  code = blockDataUpdateTsWindow(pDest, pInfo->tsColIndex);

×

UNCOV

1304

  QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1306

  pDest->info.id.groupId = pParInfo->groupId;

×

UNCOV

1307

  pOperator->resultInfo.totalRows += pDest->info.rows;

×

UNCOV

1308

  pInfo->parIte = taosHashIterate(pInfo->pPartitions, pInfo->parIte);

×

UNCOV

1309

  QUERY_CHECK_CONDITION((pDest->info.rows > 0), code, lino, _end, TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR);

×

UNCOV

1311

_end:

×

UNCOV

1312

  if (code != TSDB_CODE_SUCCESS) {

×

1313

    blockDataCleanup(pDest);

×

1314

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

UNCOV

1316

  printDataBlock(pDest, getStreamOpName(pOperator->operatorType), GET_TASKID(pTaskInfo));

×

UNCOV

1317

  return pDest;

×

UNCOV

1320

int32_t appendCreateTableRow(void* pState, SExprSupp* pTableSup, SExprSupp* pTagSup, uint64_t groupId,

×

UNCOV

1322

  int32_t code = TSDB_CODE_SUCCESS;

×

UNCOV

1323

  int32_t lino = 0;

×

UNCOV

1324

  void*   pValue = NULL;

×

UNCOV

1325

  int32_t winCode = TSDB_CODE_SUCCESS;

×

UNCOV

1326

  code = pAPI->streamStateGetParName(pState, groupId, &pValue, true, &winCode);

×

UNCOV

1327

  QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1328

  if (winCode != TSDB_CODE_SUCCESS) {

×

UNCOV

1329

    SSDataBlock* pTmpBlock = NULL;

×

UNCOV

1330

    code = blockCopyOneRow(pSrcBlock, rowId, &pTmpBlock);

×

UNCOV

1331

    QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1333

    memset(pTmpBlock->info.parTbName, 0, TSDB_TABLE_NAME_LEN);

×

UNCOV

1334

    pTmpBlock->info.id.groupId = groupId;

×

UNCOV

1335

    char* tbName = pSrcBlock->info.parTbName;

×

UNCOV

1336

    if (pTableSup->numOfExprs > 0) {

×

UNCOV

1337

      code = projectApplyFunctions(pTableSup->pExprInfo, pDestBlock, pTmpBlock, pTableSup->pCtx, pTableSup->numOfExprs,

×

UNCOV

1339

      QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1341

      SColumnInfoData* pTbCol = taosArrayGet(pDestBlock->pDataBlock, UD_TABLE_NAME_COLUMN_INDEX);

×

UNCOV

1342

      QUERY_CHECK_NULL(pTbCol, code, lino, _end, terrno);

×

UNCOV

1343

      memset(tbName, 0, TSDB_TABLE_NAME_LEN);

×

UNCOV

1344

      int32_t len = 0;

×

UNCOV

1345

      if (colDataIsNull_s(pTbCol, pDestBlock->info.rows - 1)) {

×

UNCOV

1346

        len = 1;

×

UNCOV

1347

        tbName[0] = 0;

×

UNCOV

1349

        void* pData = colDataGetData(pTbCol, pDestBlock->info.rows - 1);

×

UNCOV

1350

        len = TMIN(varDataLen(pData), TSDB_TABLE_NAME_LEN - 1);

×

UNCOV

1351

        memcpy(tbName, varDataVal(pData), len);

×

UNCOV

1352

        code = pAPI->streamStatePutParName(pState, groupId, tbName);

×

UNCOV

1353

        QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1355

      memcpy(pTmpBlock->info.parTbName, tbName, len);

×

UNCOV

1356

      pDestBlock->info.rows--;

×

UNCOV

1358

      void* pTbNameCol = taosArrayGet(pDestBlock->pDataBlock, UD_TABLE_NAME_COLUMN_INDEX);

×

UNCOV

1359

      QUERY_CHECK_NULL(pTbNameCol, code, lino, _end, terrno);

×

UNCOV

1360

      colDataSetNULL(pTbNameCol, pDestBlock->info.rows);

×

UNCOV

1361

      tbName[0] = 0;

×

UNCOV

1364

    if (pTagSup->numOfExprs > 0) {

×

UNCOV

1365

      code = projectApplyFunctions(pTagSup->pExprInfo, pDestBlock, pTmpBlock, pTagSup->pCtx, pTagSup->numOfExprs, NULL);

×

UNCOV

1366

      QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1367

      pDestBlock->info.rows--;

×

UNCOV

1369

      memcpy(pDestBlock->info.parTbName, pTmpBlock->info.parTbName, TSDB_TABLE_NAME_LEN);

×

UNCOV

1372

    void* pGpIdCol = taosArrayGet(pDestBlock->pDataBlock, UD_GROUPID_COLUMN_INDEX);

×

UNCOV

1373

    QUERY_CHECK_NULL(pGpIdCol, code, lino, _end, terrno);

×

UNCOV

1374

    code = colDataSetVal(pGpIdCol, pDestBlock->info.rows, (const char*)&groupId, false);

×

UNCOV

1375

    QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1376

    pDestBlock->info.rows++;

×

UNCOV

1377

    blockDataDestroy(pTmpBlock);

×

UNCOV

1379

    memcpy(pSrcBlock->info.parTbName, pValue, TSDB_TABLE_NAME_LEN);

×

UNCOV

1381

  pAPI->streamStateFreeVal(pValue);

×

UNCOV

1383

_end:

×

UNCOV

1384

  if (code != TSDB_CODE_SUCCESS) {

×

1385

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

UNCOV

1387

  return code;

×

UNCOV

1390

static int32_t buildStreamCreateTableResult(SOperatorInfo* pOperator) {

×

UNCOV

1391

  int32_t                       code = TSDB_CODE_SUCCESS;

×

UNCOV

1392

  int32_t                       lino = 0;

×

UNCOV

1393

  SExecTaskInfo*                pTask = pOperator->pTaskInfo;

×

UNCOV

1394

  SStreamPartitionOperatorInfo* pInfo = pOperator->info;

×

UNCOV

1395

  SSDataBlock*                  pSrc = pInfo->pInputDataBlock;

×

UNCOV

1396

  if ((pInfo->tbnameCalSup.numOfExprs == 0 && pInfo->tagCalSup.numOfExprs == 0) ||

×

UNCOV

1397

      taosHashGetSize(pInfo->pPartitions) == 0) {

×

UNCOV

1398

    goto _end;

×

UNCOV

1400

  blockDataCleanup(pInfo->pCreateTbRes);

×

UNCOV

1401

  code = blockDataEnsureCapacity(pInfo->pCreateTbRes, taosHashGetSize(pInfo->pPartitions));

×

UNCOV

1402

  QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1404

  if (pInfo->pTbNameIte != NULL) {

×

UNCOV

1405

    SPartitionDataInfo* pParInfo = (SPartitionDataInfo*)pInfo->pTbNameIte;

×

UNCOV

1406

    int32_t             rowId = *(int32_t*)taosArrayGet(pParInfo->rowIds, 0);

×

UNCOV

1407

    code = appendCreateTableRow(pTask->streamInfo.pState, &pInfo->tbnameCalSup, &pInfo->tagCalSup, pParInfo->groupId,

×

UNCOV

1409

    QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1410

    pInfo->pTbNameIte = taosHashIterate(pInfo->pPartitions, pInfo->pTbNameIte);

×

1413

_end:

×

UNCOV

1414

  if (code != TSDB_CODE_SUCCESS) {

×

1415

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

UNCOV

1417

  return code;

×

UNCOV

1420

static void doStreamHashPartitionImpl(SStreamPartitionOperatorInfo* pInfo, SSDataBlock* pBlock) {

×

UNCOV

1421

  int32_t code = TSDB_CODE_SUCCESS;

×

UNCOV

1422

  int32_t lino = 0;

×

UNCOV

1423

  pInfo->pInputDataBlock = pBlock;

×

UNCOV

1424

  for (int32_t i = 0; i < pBlock->info.rows; ++i) {

×

UNCOV

1425

    recordNewGroupKeys(pInfo->partitionSup.pGroupCols, pInfo->partitionSup.pGroupColVals, pBlock, i);

×

UNCOV

1426

    int32_t             keyLen = buildGroupKeys(pInfo->partitionSup.keyBuf, pInfo->partitionSup.pGroupColVals);

×

UNCOV

1428

        (SPartitionDataInfo*)taosHashGet(pInfo->pPartitions, pInfo->partitionSup.keyBuf, keyLen);

×

UNCOV

1429

    if (pParData) {

×

UNCOV

1430

      void* tmp = taosArrayPush(pParData->rowIds, &i);

×

UNCOV

1431

      QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

×

UNCOV

1433

      SPartitionDataInfo newParData = {0};

×

UNCOV

1434

      newParData.groupId = calcGroupId(pInfo->partitionSup.keyBuf, keyLen);

×

UNCOV

1435

      newParData.rowIds = taosArrayInit(64, sizeof(int32_t));

×

UNCOV

1436

      QUERY_CHECK_NULL(newParData.rowIds, code, lino, _end, terrno);

×

UNCOV

1437

      void* tmp = taosArrayPush(newParData.rowIds, &i);

×

UNCOV

1438

      QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

×

UNCOV

1441

          taosHashPut(pInfo->pPartitions, pInfo->partitionSup.keyBuf, keyLen, &newParData, sizeof(SPartitionDataInfo));

×

UNCOV

1442

      if (code == TSDB_CODE_DUP_KEY) {

×

1443

        code = TSDB_CODE_SUCCESS;

×

UNCOV

1445

      QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1449

_end:

×

UNCOV

1450

  if (code != TSDB_CODE_SUCCESS) {

×

1451

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

UNCOV

1453

×

UNCOV

1455

static int32_t doStreamHashPartitionNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {

×

UNCOV

1456

  int32_t                       code = TSDB_CODE_SUCCESS;

×

UNCOV

1457

  int32_t                       lino = 0;

×

UNCOV

1458

  SExecTaskInfo*                pTaskInfo = pOperator->pTaskInfo;

×

UNCOV

1459

  SStreamPartitionOperatorInfo* pInfo = pOperator->info;

×

UNCOV

1461

  if (pOperator->status == OP_EXEC_DONE) {

×

1462

    (*ppRes) = NULL;

×

1463

    return code;

×

UNCOV

1466

  if (hasRemainTbName(pInfo)) {

×

UNCOV

1467

    code = buildStreamCreateTableResult(pOperator);

×

UNCOV

1468

    QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1469

    if (pInfo->pCreateTbRes && pInfo->pCreateTbRes->info.rows > 0) {

×

UNCOV

1470

      (*ppRes) = pInfo->pCreateTbRes;

×

UNCOV

1471

      return code;

×

UNCOV

1475

  if (hasRemainPartion(pInfo)) {

×

UNCOV

1476

    (*ppRes) = buildStreamPartitionResult(pOperator);

×

UNCOV

1477

    return code;

×

UNCOV

1480

  int64_t        st = taosGetTimestampUs();

×

UNCOV

1481

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

×

UNCOV

1483

    pInfo->pInputDataBlock = NULL;

×

UNCOV

1484

    SSDataBlock* pBlock = getNextBlockFromDownstream(pOperator, 0);

×

UNCOV

1485

    if (pBlock == NULL) {

×

UNCOV

1486

      setOperatorCompleted(pOperator);

×

UNCOV

1487

      (*ppRes) = NULL;

×

UNCOV

1488

      return code;

×

UNCOV

1490

    printSpecDataBlock(pBlock, getStreamOpName(pOperator->operatorType), "recv", GET_TASKID(pTaskInfo));

×

UNCOV

1491

    switch (pBlock->info.type) {

×

UNCOV

1492

      case STREAM_NORMAL:

×

UNCOV

1495

        pInfo->binfo.pRes->info.type = pBlock->info.type;

×

UNCOV

1496

        break;

×

UNCOV

1497

      case STREAM_DELETE_DATA: {

×

UNCOV

1498

        code = copyDataBlock(pInfo->pDelRes, pBlock);

×

UNCOV

1499

        QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1501

        pInfo->pDelRes->info.type = STREAM_DELETE_RESULT;

×

UNCOV

1502

        printDataBlock(pInfo->pDelRes, getStreamOpName(pOperator->operatorType), GET_TASKID(pTaskInfo));

×

UNCOV

1503

        (*ppRes) = pInfo->pDelRes;

×

UNCOV

1504

        return code;

×

UNCOV

1506

      case STREAM_CREATE_CHILD_TABLE:

×

UNCOV

1511

        (*ppRes) = pBlock;

×

UNCOV

1512

        return code;

×

1514

      default:

×

1515

        code = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;

×

1516

        QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1520

    if (pInfo->scalarSup.pExprInfo != NULL) {

×

UNCOV

1521

      code = projectApplyFunctions(pInfo->scalarSup.pExprInfo, pBlock, pBlock, pInfo->scalarSup.pCtx,

×

UNCOV

1523

      QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1525

    taosHashClear(pInfo->pPartitions);

×

UNCOV

1526

    doStreamHashPartitionImpl(pInfo, pBlock);

×

UNCOV

1528

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

×

UNCOV

1530

  pInfo->parIte = taosHashIterate(pInfo->pPartitions, NULL);

×

UNCOV

1531

  pInfo->pTbNameIte = taosHashIterate(pInfo->pPartitions, NULL);

×

UNCOV

1532

  code = buildStreamCreateTableResult(pOperator);

×

UNCOV

1533

  QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1534

  if (pInfo->pCreateTbRes && pInfo->pCreateTbRes->info.rows > 0) {

×

UNCOV

1535

    (*ppRes) = pInfo->pCreateTbRes;

×

UNCOV

1536

    return code;

×

UNCOV

1538

  (*ppRes) = buildStreamPartitionResult(pOperator);

×

UNCOV

1539

  return code;

×

1541

_end:

×

1542

  if (code != TSDB_CODE_SUCCESS) {

×

1543

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

1544

    pTaskInfo->code = code;

×

1545

    T_LONG_JMP(pTaskInfo->env, code);

×

1547

  (*ppRes) = NULL;

×

1548

  return code;

×

UNCOV

1551

static void destroyStreamPartitionOperatorInfo(void* param) {

×

UNCOV

1552

  SStreamPartitionOperatorInfo* pInfo = (SStreamPartitionOperatorInfo*)param;

×

UNCOV

1553

  cleanupBasicInfo(&pInfo->binfo);

×

UNCOV

1554

  taosArrayDestroy(pInfo->partitionSup.pGroupCols);

×

UNCOV

1556

  for (int i = 0; i < taosArrayGetSize(pInfo->partitionSup.pGroupColVals); i++) {

×

UNCOV

1557

    void* tmp = taosArrayGet(pInfo->partitionSup.pGroupColVals, i);

×

UNCOV

1558

    if (!tmp) {

×

1559

      continue;

×

UNCOV

1561

    SGroupKeys key = *(SGroupKeys*)tmp;

×

UNCOV

1562

    taosMemoryFree(key.pData);

×

UNCOV

1564

  taosArrayDestroy(pInfo->partitionSup.pGroupColVals);

×

UNCOV

1566

  taosMemoryFree(pInfo->partitionSup.keyBuf);

×

UNCOV

1567

  cleanupExprSupp(&pInfo->scalarSup);

×

UNCOV

1568

  cleanupExprSupp(&pInfo->tbnameCalSup);

×

UNCOV

1569

  cleanupExprSupp(&pInfo->tagCalSup);

×

UNCOV

1570

  blockDataDestroy(pInfo->pDelRes);

×

UNCOV

1571

  taosHashCleanup(pInfo->pPartitions);

×

UNCOV

1572

  blockDataDestroy(pInfo->pCreateTbRes);

×

UNCOV

1573

  taosMemoryFreeClear(param);

×

UNCOV

1574

×

UNCOV

1576

int32_t initParDownStream(SOperatorInfo* downstream, SPartitionBySupporter* pParSup, SExprSupp* pExpr,

×

UNCOV

1578

  int32_t      code = TSDB_CODE_SUCCESS;

×

UNCOV

1579

  int32_t      lino = 0;

×

UNCOV

1580

  SStorageAPI* pAPI = &downstream->pTaskInfo->storageAPI;

×

UNCOV

1582

  if (downstream->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {

×

1583

    return code;

×

UNCOV

1586

  SStreamScanInfo* pScanInfo = downstream->info;

×

UNCOV

1587

  pScanInfo->partitionSup = *pParSup;

×

UNCOV

1588

  pScanInfo->pPartScalarSup = pExpr;

×

UNCOV

1589

  pScanInfo->pPartTbnameSup = pTbnameExpr;

×

UNCOV

1590

  for (int32_t j = 0; j < pResExprSupp->numOfExprs; j++) {

×

UNCOV

1591

    if (pScanInfo->primaryKeyIndex == pResExprSupp->pExprInfo[j].base.pParam[0].pCol->slotId) {

×

UNCOV

1592

      *pPkColIndex = j;

×

UNCOV

1595

  if (!pScanInfo->pUpdateInfo) {

×

UNCOV

1596

    code = pAPI->stateStore.updateInfoInit(60000, TSDB_TIME_PRECISION_MILLI, 0, pScanInfo->igCheckUpdate,

×

UNCOV

1597

                                           pScanInfo->pkColType, pScanInfo->pkColLen, &pScanInfo->pUpdateInfo);

×

UNCOV

1599

  return code;

×

UNCOV

1602

SSDataBlock* buildCreateTableBlock(SExprSupp* tbName, SExprSupp* tag) {

×

UNCOV

1603

  int32_t      code = TSDB_CODE_SUCCESS;

×

UNCOV

1604

  int32_t      lino = 0;

×

UNCOV

1605

  SSDataBlock* pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock));

×

UNCOV

1606

  if (!pBlock) {

×

1607

    return NULL;

×

UNCOV

1609

  pBlock->info.hasVarCol = false;

×

UNCOV

1610

  pBlock->info.id.groupId = 0;

×

UNCOV

1611

  pBlock->info.rows = 0;

×

UNCOV

1612

  pBlock->info.type = STREAM_CREATE_CHILD_TABLE;

×

UNCOV

1613

  pBlock->info.watermark = INT64_MIN;

×

UNCOV

1615

  pBlock->pDataBlock = taosArrayInit(4, sizeof(SColumnInfoData));

×

UNCOV

1616

  QUERY_CHECK_NULL(pBlock->pDataBlock, code, lino, _end, terrno);

×

UNCOV

1617

  SColumnInfoData infoData = {0};

×

UNCOV

1618

  infoData.info.type = TSDB_DATA_TYPE_VARCHAR;

×

UNCOV

1619

  if (tbName->numOfExprs > 0) {

×

UNCOV

1620

    infoData.info.bytes = tbName->pExprInfo->base.resSchema.bytes;

×

UNCOV

1622

    infoData.info.bytes = 1;

×

UNCOV

1624

  pBlock->info.rowSize += infoData.info.bytes;

×

UNCOV

1626

  void* tmp = taosArrayPush(pBlock->pDataBlock, &infoData);

×

UNCOV

1627

  QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

×

UNCOV

1629

  SColumnInfoData gpIdData = {0};

×

UNCOV

1630

  gpIdData.info.type = TSDB_DATA_TYPE_UBIGINT;

×

UNCOV

1631

  gpIdData.info.bytes = 8;

×

UNCOV

1632

  pBlock->info.rowSize += gpIdData.info.bytes;

×

UNCOV

1634

  tmp = taosArrayPush(pBlock->pDataBlock, &gpIdData);

×

UNCOV

1635

  QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

×

UNCOV

1637

  for (int32_t i = 0; i < tag->numOfExprs; i++) {

×

UNCOV

1638

    SColumnInfoData tagCol = {0};

×

UNCOV

1639

    tagCol.info.type = tag->pExprInfo[i].base.resSchema.type;

×

UNCOV

1640

    tagCol.info.bytes = tag->pExprInfo[i].base.resSchema.bytes;

×

UNCOV

1641

    tagCol.info.precision = tag->pExprInfo[i].base.resSchema.precision;

×

UNCOV

1643

    tmp = taosArrayPush(pBlock->pDataBlock, &tagCol);

×

UNCOV

1644

    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

×

UNCOV

1645

    pBlock->info.rowSize += tagCol.info.bytes;

×

UNCOV

1648

_end:

×

UNCOV

1649

  if (code != TSDB_CODE_SUCCESS) {

×

1650

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

1651

    blockDataDestroy(pBlock);

×

1652

    return NULL;

×

UNCOV

1654

  return pBlock;

×

UNCOV

1657

void freePartItem(void* ptr) {

×

UNCOV

1658

  SPartitionDataInfo* pPart = (SPartitionDataInfo*)ptr;

×

UNCOV

1659

  taosArrayDestroy(pPart->rowIds);

×

UNCOV

1660

×

UNCOV

1662

int32_t createStreamPartitionOperatorInfo(SOperatorInfo* downstream, SStreamPartitionPhysiNode* pPartNode,

×

UNCOV

1664

  QRY_PARAM_CHECK(pOptrInfo);

×

UNCOV

1666

  int32_t                       code = TSDB_CODE_SUCCESS;

×

UNCOV

1667

  int32_t                       lino = 0;

×

UNCOV

1668

  SStreamPartitionOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SStreamPartitionOperatorInfo));

×

UNCOV

1669

  SOperatorInfo*                pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));

×

UNCOV

1670

  if (pInfo == NULL || pOperator == NULL) {

×

1671

    code = terrno;

×

1672

    goto _error;

×

UNCOV

1675

  pInfo->partitionSup.pGroupCols = makeColumnArrayFromList(pPartNode->part.pPartitionKeys);

×

UNCOV

1677

  if (pPartNode->part.pExprs != NULL) {

×

UNCOV

1678

    int32_t    num = 0;

×

UNCOV

1679

    SExprInfo* pCalExprInfo = NULL;

×

UNCOV

1680

    code = createExprInfo(pPartNode->part.pExprs, NULL, &pCalExprInfo, &num);

×

UNCOV

1681

    QUERY_CHECK_CODE(code, lino, _error);

×

UNCOV

1683

    code = initExprSupp(&pInfo->scalarSup, pCalExprInfo, num, &pTaskInfo->storageAPI.functionStore);

×

UNCOV

1684

    QUERY_CHECK_CODE(code, lino, _error);

×

UNCOV

1687

  pInfo->tbnameCalSup.numOfExprs = 0;

×

UNCOV

1688

  if (pPartNode->pSubtable != NULL) {

×

UNCOV

1689

    SExprInfo* pSubTableExpr = taosMemoryCalloc(1, sizeof(SExprInfo));

×

UNCOV

1690

    QUERY_CHECK_NULL(pSubTableExpr, code, lino, _error, terrno);

×

UNCOV

1692

    pInfo->tbnameCalSup.pExprInfo = pSubTableExpr;

×

UNCOV

1693

    code = createExprFromOneNode(pSubTableExpr, pPartNode->pSubtable, 0);

×

UNCOV

1694

    QUERY_CHECK_CODE(code, lino, _error);

×

UNCOV

1696

    code = initExprSupp(&pInfo->tbnameCalSup, pSubTableExpr, 1, &pTaskInfo->storageAPI.functionStore);

×

UNCOV

1697

    QUERY_CHECK_CODE(code, lino, _error);

×

UNCOV

1700

  pInfo->tagCalSup.numOfExprs = 0;

×

UNCOV

1701

  if (pPartNode->pTags != NULL) {

×

UNCOV

1703

    SExprInfo* pTagExpr = createExpr(pPartNode->pTags, &numOfTags);

×

UNCOV

1704

    QUERY_CHECK_NULL(pTagExpr, code, lino, _error, terrno);

×

UNCOV

1706

    code = initExprSupp(&pInfo->tagCalSup, pTagExpr, numOfTags, &pTaskInfo->storageAPI.functionStore);

×

UNCOV

1707

    QUERY_CHECK_CODE(code, lino, _error);

×

UNCOV

1710

  if (pInfo->tbnameCalSup.numOfExprs != 0 || pInfo->tagCalSup.numOfExprs != 0) {

×

UNCOV

1711

    pInfo->pCreateTbRes = buildCreateTableBlock(&pInfo->tbnameCalSup, &pInfo->tagCalSup);

×

UNCOV

1712

    QUERY_CHECK_NULL(pInfo->pCreateTbRes, code, lino, _error, terrno);

×

UNCOV

1714

    pInfo->pCreateTbRes = NULL;

×

UNCOV

1717

  int32_t keyLen = 0;

×

UNCOV

1718

  code = initGroupOptrInfo(&pInfo->partitionSup.pGroupColVals, &keyLen, &pInfo->partitionSup.keyBuf,

×

UNCOV

1719

                           pInfo->partitionSup.pGroupCols);

×

UNCOV

1720

  QUERY_CHECK_CODE(code, lino, _error);

×

UNCOV

1722

  pInfo->partitionSup.needCalc = true;

×

UNCOV

1724

  pInfo->binfo.pRes = createDataBlockFromDescNode(pPartNode->part.node.pOutputDataBlockDesc);

×

UNCOV

1725

  QUERY_CHECK_NULL(pInfo->binfo.pRes, code, lino, _error, terrno);

×

UNCOV

1727

  code = blockDataEnsureCapacity(pInfo->binfo.pRes, 4096);

×

UNCOV

1728

  QUERY_CHECK_CODE(code, lino, _error);

×

UNCOV

1730

  pInfo->parIte = NULL;

×

UNCOV

1731

  pInfo->pTbNameIte = NULL;

×

UNCOV

1732

  pInfo->pInputDataBlock = NULL;

×

UNCOV

1734

  _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);

×

UNCOV

1735

  pInfo->pPartitions = taosHashInit(1024, hashFn, false, HASH_NO_LOCK);

×

UNCOV

1736

  taosHashSetFreeFp(pInfo->pPartitions, freePartItem);

×

UNCOV

1737

  pInfo->tsColIndex = 0;

×

UNCOV

1739

  code = createSpecialDataBlock(STREAM_DELETE_RESULT, &pInfo->pDelRes);

×

UNCOV

1740

  QUERY_CHECK_CODE(code, lino, _error);

×

UNCOV

1742

  int32_t    numOfCols = 0;

×

UNCOV

1743

  SExprInfo* pExprInfo = NULL;

×

UNCOV

1744

  code = createExprInfo(pPartNode->part.pTargets, NULL, &pExprInfo, &numOfCols);

×

UNCOV

1745

  QUERY_CHECK_CODE(code, lino, _error);

×

UNCOV

1747

  setOperatorInfo(pOperator, "StreamPartitionOperator", QUERY_NODE_PHYSICAL_PLAN_STREAM_PARTITION, false, OP_NOT_OPENED,

×

UNCOV

1749

  pOperator->exprSupp.numOfExprs = numOfCols;

×

UNCOV

1750

  pOperator->exprSupp.pExprInfo = pExprInfo;

×

UNCOV

1752

      createOperatorFpSet(optrDummyOpenFn, doStreamHashPartitionNext, NULL, destroyStreamPartitionOperatorInfo,

×

UNCOV

1754

  setOperatorStreamStateFn(pOperator, streamOpReleaseState, streamOpReloadState);

×

UNCOV

1756

  pInfo->basic.primaryPkIndex = -1;

×

UNCOV

1757

  code = initParDownStream(downstream, &pInfo->partitionSup, &pInfo->scalarSup, &pInfo->tbnameCalSup, &pOperator->exprSupp, &pInfo->basic.primaryPkIndex);

×

UNCOV

1758

  QUERY_CHECK_CODE(code, lino, _error);

×

UNCOV

1760

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

×

UNCOV

1761

  QUERY_CHECK_CODE(code, lino, _error);

×

UNCOV

1763

  *pOptrInfo = pOperator;

×

UNCOV

1764

  return TSDB_CODE_SUCCESS;

×

1766

_error:

×

1767

  pTaskInfo->code = code;

×

1768

  if (pInfo != NULL) destroyStreamPartitionOperatorInfo(pInfo);

×

1769

  destroyOperatorAndDownstreams(pOperator, &downstream, 1);

×

1770

  qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

1771

  return code;

×

UNCOV

1774

int32_t extractColumnInfo(SNodeList* pNodeList, SArray** pArrayRes) {

×

UNCOV

1775

  int32_t code = TSDB_CODE_SUCCESS;

×

UNCOV

1776

  int32_t lino = 0;

×

UNCOV

1777

  size_t  numOfCols = LIST_LENGTH(pNodeList);

×

UNCOV

1778

  SArray* pList = taosArrayInit(numOfCols, sizeof(SColumn));

×

UNCOV

1779

  if (pList == NULL) {

×

1780

    code = terrno;

×

1781

    (*pArrayRes) = NULL;

×

1782

    QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1785

  for (int32_t i = 0; i < numOfCols; ++i) {

×

UNCOV

1786

    STargetNode* pNode = (STargetNode*)nodesListGetNode(pNodeList, i);

×

UNCOV

1787

    QUERY_CHECK_NULL(pNode, code, lino, _end, terrno);

×

UNCOV

1789

    if (nodeType(pNode->pExpr) == QUERY_NODE_COLUMN) {

×

UNCOV

1790

      SColumnNode* pColNode = (SColumnNode*)pNode->pExpr;

×

UNCOV

1792

      SColumn c = extractColumnFromColumnNode(pColNode);

×

UNCOV

1793

      void*   tmp = taosArrayPush(pList, &c);

×

UNCOV

1794

      QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

×

1795

    } else if (nodeType(pNode->pExpr) == QUERY_NODE_VALUE) {

×

1796

      SValueNode* pValNode = (SValueNode*)pNode->pExpr;

×

1797

      SColumn     c = {0};

×

1798

      c.slotId = pNode->slotId;

×

1799

      c.colId = pNode->slotId;

×

1800

      c.type = pValNode->node.type;

×

1801

      c.bytes = pValNode->node.resType.bytes;

×

1802

      c.scale = pValNode->node.resType.scale;

×

1803

      c.precision = pValNode->node.resType.precision;

×

1805

      void* tmp = taosArrayPush(pList, &c);

×

1806

      QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

×

UNCOV

1810

  (*pArrayRes) = pList;

×

UNCOV

1812

_end:

×

UNCOV

1813

  if (code != TSDB_CODE_SUCCESS) {

×

1814

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

UNCOV

1816

  return code;

×

taosdata / TDengine / #3621

Source File
Press 'n' to go to next uncovered line, 'b' for previous

taosdata / TDengine / #3621

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous