• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

taosdata / TDengine / #4829

30 Oct 2025 09:25AM UTC coverage: 49.734% (-11.3%) from 61.071%
#4829

push

travis-ci

web-flow
Merge pull request #33435 from taosdata/3.0

merge 3.0

123072 of 323930 branches covered (37.99%)

Branch coverage included in aggregate %.

7 of 25 new or added lines in 3 files covered. (28.0%)

35232 existing lines in 327 files now uncovered.

172062 of 269495 relevant lines covered (63.85%)

70709785.06 hits per line

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

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

16
#include "executorInt.h"
17
#include "filter.h"
18
#include "function.h"
19
#include "operator.h"
20
#include "query.h"
21
#include "querytask.h"
22
#include "tdatablock.h"
23
#include "thash.h"
24
#include "tmsg.h"
25
#include "tref.h"
26
#include "trpc.h"
27

28
typedef struct SFetchRspHandleWrapper {
29
  uint32_t exchangeId;
30
  int32_t  sourceIndex;
31
  int64_t  seqId;
32
} SFetchRspHandleWrapper;
33

34
typedef struct SSourceDataInfo {
35
  int32_t            index;
36
  int64_t            seqId;
37
  SRWLatch           lock;
38
  SRetrieveTableRsp* pRsp;
39
  uint64_t           totalRows;
40
  int64_t            startTime;
41
  int32_t            code;
42
  EX_SOURCE_STATUS   status;
43
  const char*        taskId;
44
  SArray*            pSrcUidList;
45
  int32_t            srcOpType;
46
  bool               tableSeq;
47
  char*              decompBuf;
48
  int32_t            decompBufSize;
49
  SOrgTbInfo*        colMap;
50
  bool               isVtbRefScan;
51
  bool               isVtbTagScan;
52
  STimeWindow        window;
53
  bool               fetchSent; // need reset
54
} SSourceDataInfo;
55

56
static void destroyExchangeOperatorInfo(void* param);
57
static void freeBlock(void* pParam);
58
static void freeSourceDataInfo(void* param);
59
static void setAllSourcesCompleted(SOperatorInfo* pOperator);
60

61
static int32_t loadRemoteDataCallback(void* param, SDataBuf* pMsg, int32_t code);
62
static int32_t doSendFetchDataRequest(SExchangeInfo* pExchangeInfo, SExecTaskInfo* pTaskInfo, int32_t sourceIndex);
63
static int32_t getCompletedSources(const SArray* pArray, int32_t* pRes);
64
static int32_t prepareConcurrentlyLoad(SOperatorInfo* pOperator);
65
static int32_t seqLoadRemoteData(SOperatorInfo* pOperator);
66
static int32_t prepareLoadRemoteData(SOperatorInfo* pOperator);
67
static int32_t handleLimitOffset(SOperatorInfo* pOperator, SLimitInfo* pLimitInfo, SSDataBlock* pBlock,
68
                                 bool holdDataInBuf);
69
static int32_t doExtractResultBlocks(SExchangeInfo* pExchangeInfo, SSourceDataInfo* pDataInfo);
70

71
static int32_t exchangeWait(SOperatorInfo* pOperator, SExchangeInfo* pExchangeInfo);
72

73

74
static void streamConcurrentlyLoadRemoteData(SOperatorInfo* pOperator, SExchangeInfo* pExchangeInfo,
3,065,604✔
75
                                           SExecTaskInfo* pTaskInfo) {
76
  int32_t code = 0;
3,065,604✔
77
  int32_t lino = 0;
3,065,604✔
78
  int64_t startTs = taosGetTimestampUs();  
3,066,227✔
79
  int32_t  totalSources = (int32_t)taosArrayGetSize(pExchangeInfo->pSourceDataInfo);
3,066,227✔
80
  int32_t completed = 0;
3,066,227✔
81
  code = getCompletedSources(pExchangeInfo->pSourceDataInfo, &completed);
3,065,919✔
82
  if (code != TSDB_CODE_SUCCESS) {
3,065,912!
83
    pTaskInfo->code = code;
×
84
    T_LONG_JMP(pTaskInfo->env, code);
×
85
  }
86
  if (completed == totalSources) {
3,065,912✔
87
    qDebug("%s no load since all sources completed, completed:%d, totalSources:%d", pTaskInfo->id.str, completed, totalSources);
712,729✔
88
    setAllSourcesCompleted(pOperator);
712,729✔
89
    return;
712,729✔
90
  }
91

92
  SSourceDataInfo* pDataInfo = NULL;
2,353,183✔
93

94
  while (1) {
1,414,384✔
95
    if (pExchangeInfo->current < 0) {
3,767,567✔
96
      qDebug("current %d and all sources complted, totalSources:%d", pExchangeInfo->current, totalSources);
12,906!
97
      setAllSourcesCompleted(pOperator);
12,906✔
98
      return;
12,906✔
99
    }
100
    
101
    if (pExchangeInfo->current >= totalSources) {
3,754,346✔
102
      completed = 0;
1,581,941✔
103
      code = getCompletedSources(pExchangeInfo->pSourceDataInfo, &completed);
1,581,941✔
104
      if (code != TSDB_CODE_SUCCESS) {
1,581,941!
105
        pTaskInfo->code = code;
×
106
        T_LONG_JMP(pTaskInfo->env, code);
×
107
      }
108
      if (completed == totalSources) {
1,581,941✔
109
        qDebug("stop to load since all sources complted, completed:%d, totalSources:%d", completed, totalSources);
1,318,163✔
110
        setAllSourcesCompleted(pOperator);
1,318,163✔
111
        return;
1,318,163✔
112
      }
113
      
114
      pExchangeInfo->current = 0;
263,778✔
115
    }
116

117
    qDebug("%s start stream exchange %p idx:%d fetch", GET_TASKID(pTaskInfo), pExchangeInfo, pExchangeInfo->current);
2,436,498✔
118

119
    SSourceDataInfo* pDataInfo = taosArrayGet(pExchangeInfo->pSourceDataInfo, pExchangeInfo->current);
2,436,498✔
120
    if (!pDataInfo) {
2,436,498!
121
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
122
      pTaskInfo->code = terrno;
×
123
      T_LONG_JMP(pTaskInfo->env, pTaskInfo->code);
×
124
    }
125

126
    if (pDataInfo->status == EX_SOURCE_DATA_EXHAUSTED) {
2,436,498✔
127
      pExchangeInfo->current++;
1,485✔
128
      continue;
1,485✔
129
    }
130

131
    pDataInfo->status = EX_SOURCE_DATA_NOT_READY;
2,435,013✔
132

133
    code = doSendFetchDataRequest(pExchangeInfo, pTaskInfo, pExchangeInfo->current);
2,435,328✔
134
    if (code != TSDB_CODE_SUCCESS) {
2,435,328!
135
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
×
136
      pTaskInfo->code = code;
×
137
      T_LONG_JMP(pTaskInfo->env, pTaskInfo->code);
×
138
    }
139

140
    while (true) {
321✔
141
      code = exchangeWait(pOperator, pExchangeInfo);
2,435,649✔
142
      if (code != TSDB_CODE_SUCCESS || isTaskKilled(pTaskInfo)) {
2,435,649!
143
        T_LONG_JMP(pTaskInfo->env, pTaskInfo->code);
321!
144
      }
145

146
      int64_t currSeqId = atomic_load_64(&pExchangeInfo->seqId);
2,435,328✔
147
      if (pDataInfo->seqId != currSeqId) {
2,435,328✔
148
        qDebug("%s seq rsp reqId %" PRId64 " mismatch with exchange %p curr seqId %" PRId64 ", ignore it", 
321!
149
            GET_TASKID(pTaskInfo), pDataInfo->seqId, pExchangeInfo, currSeqId);
150
        taosMemoryFreeClear(pDataInfo->pRsp);
321!
151
        continue;
321✔
152
      }
153

154
      break;
2,435,007✔
155
    }
156

157
    SDownstreamSourceNode* pSource = taosArrayGet(pExchangeInfo->pSources, pExchangeInfo->current);
2,435,007✔
158
    if (!pSource) {
2,435,007!
159
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
×
160
      pTaskInfo->code = terrno;
×
161
      T_LONG_JMP(pTaskInfo->env, pTaskInfo->code);
×
162
    }
163

164
    if (pDataInfo->code != TSDB_CODE_SUCCESS) {
2,435,007!
UNCOV
165
      qError("%s vgId:%d, clientId:0x%" PRIx64 " taskID:0x%" PRIx64 " execId:%d error happens, code:%s",
×
166
             GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->clientId, pSource->taskId, pSource->execId,
167
             tstrerror(pDataInfo->code));
UNCOV
168
      pTaskInfo->code = pDataInfo->code;
×
UNCOV
169
      T_LONG_JMP(pTaskInfo->env, code);
×
170
    }
171

172
    SRetrieveTableRsp*   pRsp = pDataInfo->pRsp;
2,435,007✔
173
    SLoadRemoteDataInfo* pLoadInfo = &pExchangeInfo->loadInfo;
2,435,007✔
174

175
    if (pRsp->numOfRows == 0) {
2,435,007✔
176
      qDebug("exhausted %p,%s vgId:%d, clientId:0x%" PRIx64 " taskID:0x%" PRIx64
1,412,899✔
177
             " execId:%d idx %d of total completed, rowsOfSource:%" PRIu64 ", totalRows:%" PRIu64 " try next", pDataInfo,
178
             GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->clientId, pSource->taskId, pSource->execId,
179
             pExchangeInfo->current + 1, pDataInfo->totalRows, pLoadInfo->totalRows);
180

181
      pDataInfo->status = EX_SOURCE_DATA_EXHAUSTED;
1,412,899✔
182
      if (pDataInfo->isVtbRefScan || pDataInfo->isVtbTagScan) {
1,412,899!
183
        pExchangeInfo->current = -1;
13,221✔
184
      } else {
185
        pExchangeInfo->current += 1;
1,399,678✔
186
      }
187
      taosMemoryFreeClear(pDataInfo->pRsp);
1,412,899!
188
      continue;
1,412,899✔
189
    }
190

191
    code = doExtractResultBlocks(pExchangeInfo, pDataInfo);
1,022,108✔
192
    TAOS_CHECK_EXIT(code);
1,022,108!
193

194
    SRetrieveTableRsp* pRetrieveRsp = pDataInfo->pRsp;
1,022,108✔
195
    if (pRsp->completed == 1) {
1,022,108✔
196
      qDebug("exhausted %p,%s fetch msg rsp from vgId:%d, clientId:0x%" PRIx64 " taskId:0x%" PRIx64 " execId:%d numOfRows:%" PRId64
740,672✔
197
             ", rowsOfSource:%" PRIu64 ", totalRows:%" PRIu64 ", totalBytes:%" PRIu64 " try next %d/%d", pDataInfo,
198
             GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->clientId, pSource->taskId, pSource->execId,
199
             pRetrieveRsp->numOfRows, pDataInfo->totalRows, pLoadInfo->totalRows, pLoadInfo->totalSize,
200
             pExchangeInfo->current + 1, totalSources);
201

202
      pDataInfo->status = EX_SOURCE_DATA_EXHAUSTED;
740,672✔
203
      if (pDataInfo->isVtbRefScan) {
740,672!
204
        pExchangeInfo->current = -1;
×
205
        taosMemoryFreeClear(pDataInfo->pRsp);
×
206
        continue;
×
207
      }
208
    } else {
209
      qDebug("%s fetch msg rsp from vgId:%d, clientId:0x%" PRIx64 " taskId:0x%" PRIx64 " execId:%d idx:%d numOfRows:%" PRId64
281,436✔
210
             ", totalRows:%" PRIu64 ", totalBytes:%" PRIu64,
211
             GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->clientId, pSource->taskId, pSource->execId,
212
             pExchangeInfo->current, pRetrieveRsp->numOfRows, pLoadInfo->totalRows, pLoadInfo->totalSize);
213
    }
214

215
    updateLoadRemoteInfo(pLoadInfo, pRetrieveRsp->numOfRows, pRetrieveRsp->compLen, startTs, pOperator);
1,022,108✔
216
    pDataInfo->totalRows += pRetrieveRsp->numOfRows;
1,022,108✔
217

218
    pExchangeInfo->current++;
1,022,108✔
219

220
    taosMemoryFreeClear(pDataInfo->pRsp);
1,022,108!
221
    return;
1,022,108✔
222
  }
223

224
_exit:
×
225

226
  if (code) {
×
227
    pTaskInfo->code = code;
×
228
    qError("%s failed at line %d since %s", __FUNCTION__, lino, tstrerror(code));
×
229
  }
230
}
231

232

233
static void concurrentlyLoadRemoteDataImpl(SOperatorInfo* pOperator, SExchangeInfo* pExchangeInfo,
55,558,794✔
234
                                           SExecTaskInfo* pTaskInfo) {
235
  int32_t code = 0;
55,558,794✔
236
  int32_t lino = 0;
55,558,794✔
237
  size_t  totalSources = taosArrayGetSize(pExchangeInfo->pSourceDataInfo);
55,558,794✔
238
  int32_t completed = 0;
55,558,610✔
239
  code = getCompletedSources(pExchangeInfo->pSourceDataInfo, &completed);
55,558,610✔
240
  if (code != TSDB_CODE_SUCCESS) {
55,558,799!
241
    pTaskInfo->code = code;
×
242
    T_LONG_JMP(pTaskInfo->env, code);
×
243
  }
244
  if (completed == totalSources) {
55,558,799✔
245
    setAllSourcesCompleted(pOperator);
17,093,082✔
246
    return;
17,093,082✔
247
  }
248

249
  SSourceDataInfo* pDataInfo = NULL;
38,465,717✔
250

251
  while (1) {
3,292,910✔
252
    qDebug("prepare wait for ready, %p, %s", pExchangeInfo, GET_TASKID(pTaskInfo));
41,758,627✔
253
    code = exchangeWait(pOperator, pExchangeInfo);
41,758,627✔
254

255
    if (code != TSDB_CODE_SUCCESS || isTaskKilled(pTaskInfo)) {
41,758,599!
256
      T_LONG_JMP(pTaskInfo->env, pTaskInfo->code);
×
257
    }
258

259
    for (int32_t i = 0; i < totalSources; ++i) {
65,056,461✔
260
      pDataInfo = taosArrayGet(pExchangeInfo->pSourceDataInfo, i);
65,056,266✔
261
      QUERY_CHECK_NULL(pDataInfo, code, lino, _exit, terrno);
65,056,266!
262
      if (pDataInfo->status == EX_SOURCE_DATA_EXHAUSTED) {
65,056,266✔
263
        continue;
17,685,303✔
264
      }
265

266
      if (pDataInfo->status != EX_SOURCE_DATA_READY) {
47,370,929✔
267
        continue;
5,612,364✔
268
      }
269

270
      int64_t currSeqId = atomic_load_64(&pExchangeInfo->seqId);
41,758,771✔
271
      if (pDataInfo->seqId != currSeqId) {
41,758,771!
272
        qDebug("concurrent rsp reqId %" PRId64 " mismatch with exchange %p curr seqId %" PRId64 ", ignore it", pDataInfo->seqId, pExchangeInfo, currSeqId);
×
273
        taosMemoryFreeClear(pDataInfo->pRsp);
×
274
        break;
×
275
      }
276

277
      if (pDataInfo->code != TSDB_CODE_SUCCESS) {
41,758,796!
UNCOV
278
        code = pDataInfo->code;
×
UNCOV
279
        TAOS_CHECK_EXIT(code);
×
280
      }
281

282
      tmemory_barrier();
41,758,796✔
283
      SRetrieveTableRsp*     pRsp = pDataInfo->pRsp;
41,758,796✔
284
      SDownstreamSourceNode* pSource = taosArrayGet(pExchangeInfo->pSources, pDataInfo->index);
41,758,793✔
285
      QUERY_CHECK_NULL(pSource, code, lino, _exit, terrno);
41,758,796!
286

287
      // todo
288
      SLoadRemoteDataInfo* pLoadInfo = &pExchangeInfo->loadInfo;
41,758,796✔
289
      if (pRsp->numOfRows == 0) {
41,758,915✔
290
        if (NULL != pDataInfo->pSrcUidList && (!pDataInfo->isVtbRefScan)) {
7,848,466!
291
          pDataInfo->status = EX_SOURCE_DATA_NOT_READY;
×
292
          code = doSendFetchDataRequest(pExchangeInfo, pTaskInfo, i);
×
293
          if (code != TSDB_CODE_SUCCESS) {
×
294
            taosMemoryFreeClear(pDataInfo->pRsp);
×
295
            TAOS_CHECK_EXIT(code);
×
296
          }
297
        } else {
298
          pDataInfo->status = EX_SOURCE_DATA_EXHAUSTED;
7,848,466✔
299
          qDebug("exhausted %p,%s vgId:%d, clientId:0x%" PRIx64 " taskId:0x%" PRIx64
7,848,466✔
300
                 " execId:%d index:%d completed, rowsOfSource:%" PRIu64 ", totalRows:%" PRIu64 ", try next %d/%" PRIzu, pDataInfo,
301
                 GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->clientId, pSource->taskId, pSource->execId, i,
302
                 pDataInfo->totalRows, pExchangeInfo->loadInfo.totalRows, i + 1, totalSources);
303
          taosMemoryFreeClear(pDataInfo->pRsp);
7,848,466!
304
        }
305
        break;
7,848,466✔
306
      }
307

308
      TAOS_CHECK_EXIT(doExtractResultBlocks(pExchangeInfo, pDataInfo));
33,909,893!
309

310
      SRetrieveTableRsp* pRetrieveRsp = pDataInfo->pRsp;
33,910,449✔
311
      updateLoadRemoteInfo(pLoadInfo, pRetrieveRsp->numOfRows, pRetrieveRsp->compLen, pDataInfo->startTime, pOperator);
33,910,449✔
312
      pDataInfo->totalRows += pRetrieveRsp->numOfRows;
33,910,330✔
313

314
      if (pRsp->completed == 1) {
33,910,449✔
315
        pDataInfo->status = EX_SOURCE_DATA_EXHAUSTED;
26,912,824✔
316
        qDebug("exhausted %p,%s fetch msg rsp from vgId:%d, clientId:0x%" PRIx64 " taskId:0x%" PRIx64
26,912,824✔
317
               " execId:%d index:%d completed, blocks:%d, numOfRows:%" PRId64 ", rowsOfSource:%" PRIu64
318
               ", totalRows:%" PRIu64 ", total:%.2f Kb, try next %d/%" PRIzu, pDataInfo,
319
               GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->clientId, pSource->taskId, pSource->execId, i,
320
               pRsp->numOfBlocks, pRsp->numOfRows, pDataInfo->totalRows, pLoadInfo->totalRows,
321
               pLoadInfo->totalSize / 1024.0, i + 1, totalSources);
322
      } else {
323
        qDebug("%s fetch msg rsp from vgId:%d, clientId:0x%" PRIx64 " taskId:0x%" PRIx64
6,997,625!
324
               " execId:%d blocks:%d, numOfRows:%" PRId64 ", totalRows:%" PRIu64 ", total:%.2f Kb",
325
               GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->clientId, pSource->taskId, pSource->execId,
326
               pRsp->numOfBlocks, pRsp->numOfRows, pLoadInfo->totalRows, pLoadInfo->totalSize / 1024.0);
327
      }
328

329
      taosMemoryFreeClear(pDataInfo->pRsp);
33,910,449!
330

331
      if ((pDataInfo->status != EX_SOURCE_DATA_EXHAUSTED || NULL != pDataInfo->pSrcUidList) && !pDataInfo->isVtbRefScan && !pDataInfo->isVtbTagScan) {
33,910,449!
332
        pDataInfo->status = EX_SOURCE_DATA_NOT_READY;
6,997,625✔
333
        code = doSendFetchDataRequest(pExchangeInfo, pTaskInfo, i);
6,997,625✔
334
        if (code != TSDB_CODE_SUCCESS) {
6,997,625!
335
          taosMemoryFreeClear(pDataInfo->pRsp);
×
336
          TAOS_CHECK_EXIT(code);
×
337
        }
338
      }
339
      
340
      return;
33,910,449✔
341
    }  // end loop
342

343
    int32_t complete1 = 0;
7,848,661✔
344
    code = getCompletedSources(pExchangeInfo->pSourceDataInfo, &complete1);
7,848,466✔
345
    if (code != TSDB_CODE_SUCCESS) {
7,848,466!
346
      pTaskInfo->code = code;
×
347
      T_LONG_JMP(pTaskInfo->env, code);
×
348
    }
349
    if (complete1 == totalSources) {
7,848,466✔
350
      qDebug("all sources are completed, %s", GET_TASKID(pTaskInfo));
4,555,556✔
351
      return;
4,555,556✔
352
    }
353
  }
354

UNCOV
355
_exit:
×
356

UNCOV
357
  if (code) {
×
UNCOV
358
    pTaskInfo->code = code;
×
UNCOV
359
    qError("%s failed at line %d since %s", __FUNCTION__, lino, tstrerror(code));
×
360
  }
361
}
362

363
static SSDataBlock* doLoadRemoteDataImpl(SOperatorInfo* pOperator) {
247,499,583✔
364
  int32_t        code = TSDB_CODE_SUCCESS;
247,499,583✔
365
  SExchangeInfo* pExchangeInfo = pOperator->info;
247,499,583✔
366
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
247,500,333✔
367

368
  size_t totalSources = taosArrayGetSize(pExchangeInfo->pSources);
247,500,313✔
369

370
  SLoadRemoteDataInfo* pLoadInfo = &pExchangeInfo->loadInfo;
247,500,514✔
371
  if (pOperator->status == OP_EXEC_DONE) {
247,500,641!
372
    qDebug("%s all %" PRIzu " source(s) are exhausted, total rows:%" PRIu64 " bytes:%" PRIu64 ", elapsed:%.2f ms",
×
373
           GET_TASKID(pTaskInfo), totalSources, pLoadInfo->totalRows, pLoadInfo->totalSize,
374
           pLoadInfo->totalElapsed / 1000.0);
375
    return NULL;
×
376
  }
377

378
  // we have buffered retrieved datablock, return it directly
379
  SSDataBlock* p = NULL;
247,500,457✔
380
  if (taosArrayGetSize(pExchangeInfo->pResultBlockList) > 0) {
247,500,785✔
381
    p = taosArrayGetP(pExchangeInfo->pResultBlockList, 0);
172,007,424✔
382
    taosArrayRemove(pExchangeInfo->pResultBlockList, 0);
172,007,424✔
383
  }
384

385
  if (p != NULL) {
247,500,204✔
386
    void* tmp = taosArrayPush(pExchangeInfo->pRecycledBlocks, &p);
172,007,424✔
387
    if (!tmp) {
172,007,424!
388
      code = terrno;
×
389
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
×
390
      pTaskInfo->code = code;
×
391
      T_LONG_JMP(pTaskInfo->env, code);
×
392
    }
393
    return p;
172,007,424✔
394
  } else {
395
    if (pExchangeInfo->seqLoadData) {
75,492,780!
396
      code = seqLoadRemoteData(pOperator);
16,868,047✔
397
      if (code != TSDB_CODE_SUCCESS) {
16,868,047!
398
        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
×
399
        pTaskInfo->code = code;
×
400
        T_LONG_JMP(pTaskInfo->env, code);
×
401
      }
402
    } else if (IS_STREAM_MODE(pOperator->pTaskInfo))   {
58,624,855✔
403
      streamConcurrentlyLoadRemoteData(pOperator, pExchangeInfo, pTaskInfo);
3,065,604✔
404
    } else {
405
      concurrentlyLoadRemoteDataImpl(pOperator, pExchangeInfo, pTaskInfo);
55,559,087✔
406
    }
407
    if (TSDB_CODE_SUCCESS != pOperator->pTaskInfo->code) {
75,493,040!
UNCOV
408
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(pOperator->pTaskInfo->code));
×
UNCOV
409
      T_LONG_JMP(pTaskInfo->env, pOperator->pTaskInfo->code);
×
410
    }
411
    
412
    if (taosArrayGetSize(pExchangeInfo->pResultBlockList) == 0) {
75,492,898✔
413
      qDebug("empty resultBlockList");
28,138,775✔
414
      return NULL;
28,138,775✔
415
    } else {
416
      p = taosArrayGetP(pExchangeInfo->pResultBlockList, 0);
47,354,265✔
417
      taosArrayRemove(pExchangeInfo->pResultBlockList, 0);
47,354,265✔
418
      void* tmp = taosArrayPush(pExchangeInfo->pRecycledBlocks, &p);
47,354,265✔
419
      if (!tmp) {
47,354,265!
420
        code = terrno;
×
421
        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
×
422
        pTaskInfo->code = code;
×
423
        T_LONG_JMP(pTaskInfo->env, code);
×
424
      }
425

426
      qDebug("block with rows:%" PRId64 " loaded", p->info.rows);
47,354,265✔
427
      return p;
47,354,265✔
428
    }
429
}
430
}
431

432
static int32_t loadRemoteDataNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {
246,473,633✔
433
  int32_t        code = TSDB_CODE_SUCCESS;
246,473,633✔
434
  int32_t        lino = 0;
246,473,633✔
435
  SExchangeInfo* pExchangeInfo = pOperator->info;
246,473,633✔
436
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
246,474,508✔
437

438
  qDebug("%s start to load from exchange %p", pTaskInfo->id.str, pExchangeInfo);
246,474,756✔
439

440
  code = pOperator->fpSet._openFn(pOperator);
246,475,065✔
441
  QUERY_CHECK_CODE(code, lino, _end);
246,474,447!
442

443
  if (pOperator->status == OP_EXEC_DONE) {
246,474,447✔
444
    (*ppRes) = NULL;
7,440✔
445
    return code;
7,440✔
446
  }
447

448
  while (1) {
1,032,971✔
449
    SSDataBlock* pBlock = doLoadRemoteDataImpl(pOperator);
247,500,000✔
450
    if (pBlock == NULL) {
247,500,464✔
451
      (*ppRes) = NULL;
28,138,775✔
452
      return code;
28,138,775✔
453
    }
454

455
    code = doFilter(pBlock, pOperator->exprSupp.pFilterInfo, NULL, NULL);
219,361,689✔
456
    QUERY_CHECK_CODE(code, lino, _end);
219,361,689!
457

458
    if (blockDataGetNumOfRows(pBlock) == 0) {
219,361,689!
UNCOV
459
      qDebug("rows 0 block got, continue next load");
×
UNCOV
460
      continue;
×
461
    }
462

463
    SLimitInfo* pLimitInfo = &pExchangeInfo->limitInfo;
219,361,689✔
464
    if (hasLimitOffsetInfo(pLimitInfo)) {
219,361,689✔
465
      int32_t status = handleLimitOffset(pOperator, pLimitInfo, pBlock, false);
1,990,639✔
466
      if (status == PROJECT_RETRIEVE_CONTINUE) {
1,990,639✔
467
        qDebug("limit retrieve continue");
1,032,971✔
468
        continue;
1,032,971✔
469
      } else if (status == PROJECT_RETRIEVE_DONE) {
957,668!
470
        if (pBlock->info.rows == 0) {
957,668!
471
          setOperatorCompleted(pOperator);
×
472
          (*ppRes) = NULL;
×
473
          return code;
×
474
        } else {
475
          (*ppRes) = pBlock;
957,668✔
476
          return code;
957,668✔
477
        }
478
      }
479
    } else {
480
      (*ppRes) = pBlock;
217,370,906✔
481
      qDebug("block with rows %" PRId64 " returned in exechange", pBlock->info.rows);
217,371,050✔
482
      return code;
217,370,906✔
483
    }
484
  }
485

486
_end:
×
487

488
  if (code != TSDB_CODE_SUCCESS) {
×
489
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
490
    pTaskInfo->code = code;
×
491
    T_LONG_JMP(pTaskInfo->env, code);
×
492
  } else {
493
    qDebug("empty block returned in exchange");
×
494
  }
495
  
496
  (*ppRes) = NULL;
×
497
  return code;
×
498
}
499

500
static int32_t initDataSource(int32_t numOfSources, SExchangeInfo* pInfo, const char* id) {
26,320,446✔
501
  pInfo->pSourceDataInfo = taosArrayInit(numOfSources, sizeof(SSourceDataInfo));
26,320,446✔
502
  if (pInfo->pSourceDataInfo == NULL) {
26,320,143!
503
    return terrno;
×
504
  }
505

506
  if (pInfo->dynamicOp) {
26,320,770!
507
    return TSDB_CODE_SUCCESS;
1,851,440✔
508
  }
509

510
  int32_t len = strlen(id) + 1;
24,469,560!
511
  pInfo->pTaskId = taosMemoryCalloc(1, len);
24,469,560!
512
  if (!pInfo->pTaskId) {
24,468,972!
513
    return terrno;
×
514
  }
515
  tstrncpy(pInfo->pTaskId, id, len);
24,468,851!
516
  for (int32_t i = 0; i < numOfSources; ++i) {
60,404,444✔
517
    SSourceDataInfo dataInfo = {0};
35,934,540✔
518
    dataInfo.status = EX_SOURCE_DATA_NOT_READY;
35,934,684✔
519
    dataInfo.taskId = pInfo->pTaskId;
35,934,684✔
520
    dataInfo.index = i;
35,934,873✔
521
    SSourceDataInfo* pDs = taosArrayPush(pInfo->pSourceDataInfo, &dataInfo);
35,934,873✔
522
    if (pDs == NULL) {
35,934,496!
523
      taosArrayDestroyEx(pInfo->pSourceDataInfo, freeSourceDataInfo);
×
524
      return terrno;
×
525
    }
526
    qDebug("init source data info %d, pDs:%p, status:%d", i, pDs, pDs->status);
35,934,496✔
527
  }
528

529
  return TSDB_CODE_SUCCESS;
24,469,904✔
530
}
531

532
static int32_t initExchangeOperator(SExchangePhysiNode* pExNode, SExchangeInfo* pInfo, const char* id) {
26,320,794✔
533
  size_t numOfSources = LIST_LENGTH(pExNode->pSrcEndPoints);
26,320,794!
534

535
  if (numOfSources == 0) {
26,320,347!
536
    qError("%s invalid number: %d of sources in exchange operator", id, (int32_t)numOfSources);
×
537
    return TSDB_CODE_INVALID_PARA;
×
538
  }
539
  pInfo->pFetchRpcHandles = taosArrayInit(numOfSources, sizeof(int64_t));
26,320,347✔
540
  if (!pInfo->pFetchRpcHandles) {
26,320,936!
541
    return terrno;
×
542
  }
543
  void* ret = taosArrayReserve(pInfo->pFetchRpcHandles, numOfSources);
26,320,528✔
544
  if (!ret) {
26,320,794!
545
    return terrno;
×
546
  }
547

548
  pInfo->pSources = taosArrayInit(numOfSources, sizeof(SDownstreamSourceNode));
26,320,794✔
549
  if (pInfo->pSources == NULL) {
26,320,912!
550
    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
551
    return terrno;
×
552
  }
553

554
  if (pExNode->node.dynamicOp) {
26,320,872!
555
    pInfo->pHashSources = tSimpleHashInit(numOfSources * 2, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT));
1,851,440✔
556
    if (NULL == pInfo->pHashSources) {
1,851,440!
557
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
558
      return terrno;
×
559
    }
560
  }
561

562
  for (int32_t i = 0; i < numOfSources; ++i) {
67,281,281✔
563
    SDownstreamSourceNode* pNode = (SDownstreamSourceNode*)nodesListGetNode((SNodeList*)pExNode->pSrcEndPoints, i);
40,960,839✔
564
    if (!pNode) {
40,960,529!
565
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
566
      return terrno;
×
567
    }
568
    void* tmp = taosArrayPush(pInfo->pSources, pNode);
40,960,529✔
569
    if (!tmp) {
40,961,043!
570
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
571
      return terrno;
×
572
    }
573
    SExchangeSrcIndex idx = {.srcIdx = i, .inUseIdx = -1};
40,961,043✔
574
    int32_t           code =
575
        tSimpleHashPut(pInfo->pHashSources, &pNode->addr.nodeId, sizeof(pNode->addr.nodeId), &idx, sizeof(idx));
40,961,043✔
576
    if (pInfo->pHashSources && code != TSDB_CODE_SUCCESS) {
40,960,570!
577
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
×
578
      return code;
×
579
    }
580
  }
581

582
  initLimitInfo(pExNode->node.pLimit, pExNode->node.pSlimit, &pInfo->limitInfo);
26,320,442✔
583
  int64_t refId = taosAddRef(exchangeObjRefPool, pInfo);
26,320,944✔
584
  if (refId < 0) {
26,320,320!
585
    int32_t code = terrno;
×
586
    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
×
587
    return code;
×
588
  } else {
589
    pInfo->self = refId;
26,320,320✔
590
  }
591

592
  return initDataSource(numOfSources, pInfo, id);
26,320,256✔
593
}
594

595
int32_t resetExchangeOperState(SOperatorInfo* pOper) {
9,482,879✔
596
  SExchangeInfo* pInfo = pOper->info;
9,482,879✔
597
  SExchangePhysiNode* pPhynode = (SExchangePhysiNode*)pOper->pPhyNode;
9,483,824✔
598

599
  qDebug("%s reset exchange op:%p info:%p", pOper->pTaskInfo->id.str, pOper, pInfo);
9,482,879✔
600

601
  (void)atomic_add_fetch_64(&pInfo->seqId, 1);
9,483,818✔
602
  pOper->status = OP_NOT_OPENED;
9,484,448✔
603
  pInfo->current = 0;
9,484,448✔
604
  pInfo->loadInfo.totalElapsed = 0;
9,484,448✔
605
  pInfo->loadInfo.totalRows = 0;
9,484,448✔
606
  pInfo->loadInfo.totalSize = 0;
9,484,448✔
607
  for (int32_t i = 0; i < taosArrayGetSize(pInfo->pSourceDataInfo); ++i) {
35,350,212✔
608
    SSourceDataInfo* pDataInfo = taosArrayGet(pInfo->pSourceDataInfo, i);
25,865,764✔
609
    taosWLockLatch(&pDataInfo->lock);
25,865,764✔
610
    taosMemoryFreeClear(pDataInfo->decompBuf);
25,865,764!
611
    taosMemoryFreeClear(pDataInfo->pRsp);
25,865,764!
612

613
    pDataInfo->totalRows = 0;
25,865,764✔
614
    pDataInfo->code = 0;
25,865,764✔
615
    pDataInfo->status = EX_SOURCE_DATA_NOT_READY;
25,865,764✔
616
    pDataInfo->fetchSent = false;
25,865,764✔
617
    taosWUnLockLatch(&pDataInfo->lock);
25,865,140✔
618
  }
619

620
  if (pInfo->dynamicOp) {
9,484,448!
621
    taosArrayClearEx(pInfo->pSourceDataInfo, freeSourceDataInfo);
3,711,596✔
622
  } 
623

624
  taosArrayClearEx(pInfo->pResultBlockList, freeBlock);
9,484,448✔
625
  taosArrayClearEx(pInfo->pRecycledBlocks, freeBlock);
9,484,448✔
626

627
  blockDataCleanup(pInfo->pDummyBlock);
9,484,448✔
628

629
  void   *data = NULL;
9,484,448✔
630
  int32_t iter = 0;
9,484,448✔
631
  while ((data = tSimpleHashIterate(pInfo->pHashSources, data, &iter))) {
15,747,457✔
632
    ((SExchangeSrcIndex *)data)->inUseIdx = -1;
6,263,009✔
633
  }
634
  
635
  pInfo->limitInfo = (SLimitInfo){0};
9,483,818✔
636
  initLimitInfo(pPhynode->node.pLimit, pPhynode->node.pSlimit, &pInfo->limitInfo);
9,484,133✔
637

638
  return 0;
9,484,448✔
639
}
640

641
int32_t createExchangeOperatorInfo(void* pTransporter, SExchangePhysiNode* pExNode, SExecTaskInfo* pTaskInfo,
26,320,672✔
642
                                   SOperatorInfo** pOptrInfo) {
643
  QRY_PARAM_CHECK(pOptrInfo);
26,320,672!
644

645
  int32_t        code = 0;
26,320,672✔
646
  int32_t        lino = 0;
26,320,672✔
647
  SExchangeInfo* pInfo = taosMemoryCalloc(1, sizeof(SExchangeInfo));
26,320,672!
648
  SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
26,319,999!
649
  if (pInfo == NULL || pOperator == NULL) {
26,320,278!
650
    code = terrno;
×
651
    goto _error;
×
652
  }
653

654
  pOperator->pPhyNode = pExNode;
26,320,278✔
655
  pInfo->dynamicOp = pExNode->node.dynamicOp;
26,320,195!
656
  code = initExchangeOperator(pExNode, pInfo, GET_TASKID(pTaskInfo));
26,320,317✔
657
  QUERY_CHECK_CODE(code, lino, _error);
26,321,344!
658

659
  code = tsem_init(&pInfo->ready, 0, 0);
26,321,344✔
660
  QUERY_CHECK_CODE(code, lino, _error);
26,321,344!
661

662
  pInfo->pDummyBlock = createDataBlockFromDescNode(pExNode->node.pOutputDataBlockDesc);
26,321,344✔
663
  QUERY_CHECK_NULL(pInfo->pDummyBlock, code, lino, _error, terrno);
26,321,344!
664

665
  pInfo->pResultBlockList = taosArrayInit(64, POINTER_BYTES);
26,321,225✔
666
  QUERY_CHECK_NULL(pInfo->pResultBlockList, code, lino, _error, terrno);
26,321,082!
667
  pInfo->pRecycledBlocks = taosArrayInit(64, POINTER_BYTES);
26,321,085✔
668
  QUERY_CHECK_NULL(pInfo->pRecycledBlocks, code, lino, _error, terrno);
26,320,803!
669

670
  SExchangeOpStopInfo stopInfo = {QUERY_NODE_PHYSICAL_PLAN_EXCHANGE, pInfo->self};
26,321,083✔
671
  code = qAppendTaskStopInfo(pTaskInfo, &stopInfo);
26,321,042✔
672
  QUERY_CHECK_CODE(code, lino, _error);
26,321,225!
673

674
  pInfo->seqLoadData = pExNode->seqRecvData;
26,321,225!
675
  pInfo->dynTbname = pExNode->dynTbname;
26,321,205!
676
  if (pInfo->dynTbname) {
26,321,344!
UNCOV
677
    pInfo->seqLoadData = true;
×
678
  }
679
  pInfo->pTransporter = pTransporter;
26,321,344✔
680

681
  setOperatorInfo(pOperator, "ExchangeOperator", QUERY_NODE_PHYSICAL_PLAN_EXCHANGE, false, OP_NOT_OPENED, pInfo,
26,321,069✔
682
                  pTaskInfo);
683
  pOperator->exprSupp.numOfExprs = taosArrayGetSize(pInfo->pDummyBlock->pDataBlock);
26,321,025✔
684

685
  code = filterInitFromNode((SNode*)pExNode->node.pConditions, &pOperator->exprSupp.pFilterInfo, 0,
26,321,188✔
686
                            pTaskInfo->pStreamRuntimeInfo);
26,321,003✔
687
  QUERY_CHECK_CODE(code, lino, _error);
26,321,130!
688
  qTrace("%s exchange op:%p", __func__, pOperator);
26,321,130!
689
  pOperator->fpSet = createOperatorFpSet(prepareLoadRemoteData, loadRemoteDataNext, NULL, destroyExchangeOperatorInfo,
26,321,130✔
690
                                         optrDefaultBufFn, NULL, optrDefaultGetNextExtFn, NULL);
691
  setOperatorResetStateFn(pOperator, resetExchangeOperState);
26,320,809✔
692
  *pOptrInfo = pOperator;
26,320,802✔
693
  return TSDB_CODE_SUCCESS;
26,321,068✔
694

695
_error:
×
696
  if (code != TSDB_CODE_SUCCESS) {
×
697
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
698
    pTaskInfo->code = code;
×
699
  }
700
  if (pInfo != NULL) {
×
701
    doDestroyExchangeOperatorInfo(pInfo);
×
702
  }
703

704
  if (pOperator != NULL) {
×
705
    pOperator->info = NULL;
×
706
    destroyOperator(pOperator);
×
707
  }
708
  return code;
×
709
}
710

711
void destroyExchangeOperatorInfo(void* param) {
26,321,138✔
712
  SExchangeInfo* pExInfo = (SExchangeInfo*)param;
26,321,138✔
713
  int32_t        code = taosRemoveRef(exchangeObjRefPool, pExInfo->self);
26,321,138✔
714
  if (code != TSDB_CODE_SUCCESS) {
26,320,999!
715
    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
×
716
  }
717
}
26,320,999✔
718

719
void freeBlock(void* pParam) {
100,679,059✔
720
  SSDataBlock* pBlock = *(SSDataBlock**)pParam;
100,679,059✔
721
  blockDataDestroy(pBlock);
100,678,915✔
722
}
100,678,731✔
723

724
void freeSourceDataInfo(void* p) {
36,443,768✔
725
  SSourceDataInfo* pInfo = (SSourceDataInfo*)p;
36,443,768✔
726
  taosMemoryFreeClear(pInfo->decompBuf);
36,443,768!
727
  taosMemoryFreeClear(pInfo->pRsp);
36,443,768!
728

729
  pInfo->decompBufSize = 0;
36,443,562✔
730
}
36,443,562✔
731

732
void doDestroyExchangeOperatorInfo(void* param) {
26,321,344✔
733
  if (param == NULL) {
26,321,344!
734
    return;
×
735
  }
736
  SExchangeInfo* pExInfo = (SExchangeInfo*)param;
26,321,344✔
737
  if (pExInfo->pFetchRpcHandles) {
26,321,344✔
738
    for (int32_t i = 0; i < pExInfo->pFetchRpcHandles->size; ++i) {
67,280,960✔
739
      int64_t* pRpcHandle = taosArrayGet(pExInfo->pFetchRpcHandles, i);
40,960,875✔
740
      if (*pRpcHandle > 0) {
40,960,088✔
741
        SDownstreamSourceNode* pSource = taosArrayGet(pExInfo->pSources, i);
1,661,630✔
742
        (void)asyncFreeConnById(pExInfo->pTransporter, *pRpcHandle);
1,661,630✔
743
      }
744
    }
745
    taosArrayDestroy(pExInfo->pFetchRpcHandles);
26,320,763✔
746
  }
747

748
  taosArrayDestroy(pExInfo->pSources);
26,320,912✔
749
  taosArrayDestroyEx(pExInfo->pSourceDataInfo, freeSourceDataInfo);
26,321,200✔
750

751
  taosArrayDestroyEx(pExInfo->pResultBlockList, freeBlock);
26,320,582✔
752
  taosArrayDestroyEx(pExInfo->pRecycledBlocks, freeBlock);
26,320,959✔
753

754
  blockDataDestroy(pExInfo->pDummyBlock);
26,321,103✔
755
  tSimpleHashCleanup(pExInfo->pHashSources);
26,321,103✔
756

757
  int32_t code = tsem_destroy(&pExInfo->ready);
26,320,932✔
758
  if (code != TSDB_CODE_SUCCESS) {
26,320,790!
759
    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
×
760
  }
761
  taosMemoryFreeClear(pExInfo->pTaskId);
26,320,790!
762

763
  taosMemoryFreeClear(param);
26,320,584!
764
}
765

766
int32_t loadRemoteDataCallback(void* param, SDataBuf* pMsg, int32_t code) {
62,036,459✔
767
  SFetchRspHandleWrapper* pWrapper = (SFetchRspHandleWrapper*)param;
62,036,459✔
768

769
  taosMemoryFreeClear(pMsg->pEpSet);
62,036,459!
770
  SExchangeInfo* pExchangeInfo = taosAcquireRef(exchangeObjRefPool, pWrapper->exchangeId);
62,042,295✔
771
  if (pExchangeInfo == NULL) {
62,043,218✔
772
    qWarn("failed to acquire exchange operator, since it may have been released, %p", pExchangeInfo);
7,353!
773
    taosMemoryFree(pMsg->pData);
7,353!
774
    return TSDB_CODE_SUCCESS;
7,353✔
775
  }
776

777
  int64_t currSeqId = atomic_load_64(&pExchangeInfo->seqId);
62,035,865✔
778
  if (pWrapper->seqId != currSeqId) {
62,036,502!
779
    qDebug("rsp reqId %" PRId64 " mismatch with exchange %p curr seqId %" PRId64 ", ignore it", pWrapper->seqId, pExchangeInfo, currSeqId);
×
780
    taosMemoryFree(pMsg->pData);
×
781
    code = taosReleaseRef(exchangeObjRefPool, pWrapper->exchangeId);
×
782
    if (code != TSDB_CODE_SUCCESS) {
×
783
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
×
784
    }
785
    return TSDB_CODE_SUCCESS;
×
786
  }
787

788
  int32_t          index = pWrapper->sourceIndex;
62,033,216✔
789

790
  qDebug("%s exchange %p %dth source got rsp, code:%d, rsp:%p", pExchangeInfo->pTaskId, pExchangeInfo, index, code, pMsg->pData);
62,032,450✔
791

792
  int64_t* pRpcHandle = taosArrayGet(pExchangeInfo->pFetchRpcHandles, index);
62,033,591✔
793
  if (pRpcHandle != NULL) {
62,035,410✔
794
    int32_t ret = asyncFreeConnById(pExchangeInfo->pTransporter, *pRpcHandle);
62,029,622✔
795
    if (ret != 0) {
62,037,335✔
796
      qDebug("failed to free rpc handle, code:%s, %p", tstrerror(ret), pExchangeInfo);
2,525,982✔
797
    }
798
    *pRpcHandle = -1;
62,037,335✔
799
  }
800

801
  SSourceDataInfo* pSourceDataInfo = taosArrayGet(pExchangeInfo->pSourceDataInfo, index);
62,033,557✔
802
  if (!pSourceDataInfo) {
62,035,038!
803
    return terrno;
×
804
  }
805

806
  if (0 == code && NULL == pMsg->pData) {
62,035,038!
807
    qError("invalid rsp msg, msgType:%d, len:%d", pMsg->msgType, pMsg->len);
×
808
    code = TSDB_CODE_QRY_INVALID_MSG;
×
809
  }
810

811
  taosWLockLatch(&pSourceDataInfo->lock);
62,037,618✔
812
  if (code == TSDB_CODE_SUCCESS) {
62,037,195✔
813
    pSourceDataInfo->seqId = pWrapper->seqId;
62,036,874✔
814
    pSourceDataInfo->pRsp = pMsg->pData;
62,034,529✔
815

816
    SRetrieveTableRsp* pRsp = pSourceDataInfo->pRsp;
62,025,144✔
817
    pRsp->numOfRows = htobe64(pRsp->numOfRows);
62,023,981✔
818
    pRsp->compLen = htonl(pRsp->compLen);
62,028,804✔
819
    pRsp->payloadLen = htonl(pRsp->payloadLen);
62,028,985✔
820
    pRsp->numOfCols = htonl(pRsp->numOfCols);
62,014,171✔
821
    pRsp->useconds = htobe64(pRsp->useconds);
62,028,481✔
822
    pRsp->numOfBlocks = htonl(pRsp->numOfBlocks);
62,009,159✔
823

824
    qDebug("%s fetch rsp received, index:%d, blocks:%d, rows:%" PRId64 ", %p", pSourceDataInfo->taskId, index,
62,021,682✔
825
           pRsp->numOfBlocks, pRsp->numOfRows, pExchangeInfo);
826
  } else {
827
    taosMemoryFree(pMsg->pData);
321!
828
    pSourceDataInfo->code = rpcCvtErrCode(code);
321✔
829
    if (pSourceDataInfo->code != code) {
321!
830
      qError("%s fetch rsp received, index:%d, error:%s, cvted error: %s, %p", pSourceDataInfo->taskId, index,
×
831
             tstrerror(code), tstrerror(pSourceDataInfo->code), pExchangeInfo);
832
    } else {
833
      qError("%s fetch rsp received, index:%d, error:%s, %p", pSourceDataInfo->taskId, index, tstrerror(code),
321!
834
             pExchangeInfo);
835
    }
836
  }
837

838
  tmemory_barrier();
62,022,483✔
839
  pSourceDataInfo->status = EX_SOURCE_DATA_READY;
62,022,483✔
840
  taosWUnLockLatch(&pSourceDataInfo->lock);
62,029,626✔
841
  
842
  code = tsem_post(&pExchangeInfo->ready);
62,023,777✔
843
  if (code != TSDB_CODE_SUCCESS) {
62,037,091!
844
    qError("failed to invoke post when fetch rsp is ready, code:%s, %p", tstrerror(code), pExchangeInfo);
×
845
    return code;
×
846
  }
847

848
  code = taosReleaseRef(exchangeObjRefPool, pWrapper->exchangeId);
62,037,091✔
849
  if (code != TSDB_CODE_SUCCESS) {
62,037,895!
850
    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
×
851
  }
852
  return code;
62,037,670✔
853
}
854

855
int32_t buildTableScanOperatorParam(SOperatorParam** ppRes, SArray* pUidList, int32_t srcOpType, bool tableSeq) {
51,596✔
856
  *ppRes = taosMemoryMalloc(sizeof(SOperatorParam));
51,596!
857
  if (NULL == *ppRes) {
51,596!
858
    return terrno;
×
859
  }
860

861
  STableScanOperatorParam* pScan = taosMemoryMalloc(sizeof(STableScanOperatorParam));
51,596!
862
  if (NULL == pScan) {
51,596!
863
    taosMemoryFreeClear(*ppRes);
×
864
    return terrno;
×
865
  }
866

867
  pScan->pUidList = taosArrayDup(pUidList, NULL);
51,596✔
868
  if (NULL == pScan->pUidList) {
51,596!
869
    taosMemoryFree(pScan);
×
870
    taosMemoryFreeClear(*ppRes);
×
871
    return terrno;
×
872
  }
873
  pScan->tableSeq = tableSeq;
51,596✔
874
  pScan->pOrgTbInfo = NULL;
51,596✔
875
  pScan->window.skey = INT64_MAX;
51,596✔
876
  pScan->window.ekey = INT64_MIN;
51,596✔
877

878
  (*ppRes)->opType = srcOpType;
51,596✔
879
  (*ppRes)->downstreamIdx = 0;
51,596✔
880
  (*ppRes)->value = pScan;
51,596✔
881
  (*ppRes)->pChildren = NULL;
51,596✔
882
  (*ppRes)->reUse = false;
51,596✔
883

884
  return TSDB_CODE_SUCCESS;
51,596✔
885
}
886

887
int32_t buildTableScanOperatorParamEx(SOperatorParam** ppRes, SArray* pUidList, int32_t srcOpType, SOrgTbInfo *pMap, bool tableSeq, STimeWindow *window) {
15,564,690✔
888
  int32_t                  code = TSDB_CODE_SUCCESS;
15,564,690✔
889
  int32_t                  lino = 0;
15,564,690✔
890
  STableScanOperatorParam* pScan = NULL;
15,564,690✔
891

892
  *ppRes = taosMemoryMalloc(sizeof(SOperatorParam));
15,564,690!
893
  QUERY_CHECK_NULL(*ppRes, code, lino, _return, terrno);
15,564,690!
894

895
  pScan = taosMemoryMalloc(sizeof(STableScanOperatorParam));
15,564,690!
896
  QUERY_CHECK_NULL(pScan, code, lino, _return, terrno);
15,564,690!
897

898
  pScan->pUidList = taosArrayDup(pUidList, NULL);
15,564,690✔
899
  QUERY_CHECK_NULL(pScan->pUidList, code, lino, _return, terrno);
15,564,690!
900

901
  pScan->pOrgTbInfo = taosMemoryMalloc(sizeof(SOrgTbInfo));
15,564,690!
902
  QUERY_CHECK_NULL(pScan->pOrgTbInfo, code, lino, _return, terrno);
15,564,690!
903

904
  pScan->pOrgTbInfo->vgId = pMap->vgId;
15,564,690✔
905
  tstrncpy(pScan->pOrgTbInfo->tbName, pMap->tbName, TSDB_TABLE_FNAME_LEN);
15,564,690!
906

907
  pScan->pOrgTbInfo->colMap = taosArrayDup(pMap->colMap, NULL);
15,564,690✔
908
  QUERY_CHECK_NULL(pScan->pOrgTbInfo->colMap, code, lino, _return, terrno);
15,564,690!
909

910
  pScan->tableSeq = tableSeq;
15,564,690✔
911
  pScan->window.skey = window->skey;
15,564,690✔
912
  pScan->window.ekey = window->ekey;
15,564,690✔
913

914
  (*ppRes)->opType = srcOpType;
15,564,690✔
915
  (*ppRes)->downstreamIdx = 0;
15,564,690✔
916
  (*ppRes)->value = pScan;
15,564,690✔
917
  (*ppRes)->pChildren = NULL;
15,564,690✔
918
  (*ppRes)->reUse = false;
15,564,690✔
919

920
  return code;
15,564,690✔
921
_return:
×
922
  qError("%s failed at %d, failed to build scan operator msg:%s", __FUNCTION__, lino, tstrerror(code));
×
923
  taosMemoryFreeClear(*ppRes);
×
924
  if (pScan) {
×
925
    taosArrayDestroy(pScan->pUidList);
×
926
    if (pScan->pOrgTbInfo) {
×
927
      taosArrayDestroy(pScan->pOrgTbInfo->colMap);
×
928
      taosMemoryFreeClear(pScan->pOrgTbInfo);
×
929
    }
930
    taosMemoryFree(pScan);
×
931
  }
932
  return code;
×
933
}
934

935
int32_t buildTagScanOperatorParam(SOperatorParam** ppRes, SArray* pUidList, int32_t srcOpType) {
1,235,466✔
936
  int32_t                  code = TSDB_CODE_SUCCESS;
1,235,466✔
937
  int32_t                  lino = 0;
1,235,466✔
938
  STagScanOperatorParam*   pScan = NULL;
1,235,466✔
939

940
  *ppRes = taosMemoryMalloc(sizeof(SOperatorParam));
1,235,466!
941
  QUERY_CHECK_NULL(*ppRes, code, lino, _return, terrno);
1,235,466!
942

943
  pScan = taosMemoryMalloc(sizeof(STagScanOperatorParam));
1,235,466!
944
  QUERY_CHECK_NULL(pScan, code, lino, _return, terrno);
1,235,466!
945
  pScan->vcUid = *(tb_uid_t*)taosArrayGet(pUidList, 0);
1,235,466✔
946

947
  (*ppRes)->opType = srcOpType;
1,235,466✔
948
  (*ppRes)->downstreamIdx = 0;
1,235,466✔
949
  (*ppRes)->value = pScan;
1,235,466✔
950
  (*ppRes)->pChildren = NULL;
1,235,466✔
951
  (*ppRes)->reUse = false;
1,235,466✔
952

953
  return code;
1,235,466✔
954
_return:
×
955
  qError("%s failed at %d, failed to build scan operator msg:%s", __FUNCTION__, lino, tstrerror(code));
×
956
  taosMemoryFreeClear(*ppRes);
×
957
  if (pScan) {
×
958
    taosMemoryFree(pScan);
×
959
  }
960
  return code;
×
961
}
962

963
static int32_t getCurrentWinCalcTimeRange(SStreamRuntimeFuncInfo* pRuntimeInfo, STimeWindow* pTimeRange) {
442,713✔
964
  if (!pRuntimeInfo || !pTimeRange) {
442,713!
965
    return TSDB_CODE_INTERNAL_ERROR;
×
966
  }
967

968
  SSTriggerCalcParam* pParam = taosArrayGet(pRuntimeInfo->pStreamPesudoFuncVals, pRuntimeInfo->curIdx);
443,028✔
969
  if (!pParam) {
443,028!
970
    return TSDB_CODE_INTERNAL_ERROR;
×
971
  }
972

973
  switch (pRuntimeInfo->triggerType) {
443,028✔
974
    case STREAM_TRIGGER_SLIDING:
291,072✔
975
      // Unable to distinguish whether there is an interval, all use wstart/wend
976
      // and the results are equal to those of prevTs/currentTs, using the same address of union.
977
      pTimeRange->skey = pParam->wstart;  // is equal to wstart
291,072✔
978
      pTimeRange->ekey = pParam->wend;    // is equal to wend
291,072✔
979
      break;
291,072✔
980
    case STREAM_TRIGGER_PERIOD:
594✔
981
      pTimeRange->skey = pParam->prevLocalTime;
594✔
982
      pTimeRange->ekey = pParam->triggerTime;
594✔
983
      break;
594✔
984
    default:
151,362✔
985
      pTimeRange->skey = pParam->wstart;
151,362✔
986
      pTimeRange->ekey = pParam->wend;
151,362✔
987
      break;
151,362✔
988
  }
989

990
  return TSDB_CODE_SUCCESS;
443,028✔
991
}
992

993
int32_t doSendFetchDataRequest(SExchangeInfo* pExchangeInfo, SExecTaskInfo* pTaskInfo, int32_t sourceIndex) {
62,121,226✔
994
  int32_t          code = TSDB_CODE_SUCCESS;
62,121,226✔
995
  int32_t          lino = 0;
62,121,226✔
996
  SSourceDataInfo* pDataInfo = taosArrayGet(pExchangeInfo->pSourceDataInfo, sourceIndex);
62,121,226✔
997
  if (!pDataInfo) {
62,121,468!
998
    return terrno;
×
999
  }
1000

1001
  if (EX_SOURCE_DATA_NOT_READY != pDataInfo->status) {
62,121,468!
1002
    return TSDB_CODE_SUCCESS;
×
1003
  }
1004

1005
  pDataInfo->status = EX_SOURCE_DATA_STARTED;
62,121,491✔
1006
  SDownstreamSourceNode* pSource = taosArrayGet(pExchangeInfo->pSources, pDataInfo->index);
62,121,666✔
1007
  if (!pSource) {
62,120,765!
1008
    return terrno;
×
1009
  }
1010

1011
  pDataInfo->startTime = taosGetTimestampUs();
62,121,702✔
1012
  size_t totalSources = taosArrayGetSize(pExchangeInfo->pSources);
62,121,320✔
1013

1014
  SFetchRspHandleWrapper* pWrapper = taosMemoryCalloc(1, sizeof(SFetchRspHandleWrapper));
62,120,810!
1015
  QUERY_CHECK_NULL(pWrapper, code, lino, _end, terrno);
62,121,362!
1016
  pWrapper->exchangeId = pExchangeInfo->self;
62,121,362✔
1017
  pWrapper->sourceIndex = sourceIndex;
62,122,067✔
1018
  pWrapper->seqId = pExchangeInfo->seqId;
62,121,911✔
1019

1020
  if (pSource->localExec) {
62,121,617!
1021
    SDataBuf pBuf = {0};
6,630✔
1022
    int32_t  code = (*pTaskInfo->localFetch.fp)(pTaskInfo->localFetch.handle, pSource->sId, pTaskInfo->id.queryId,
6,630✔
1023
                                               pSource->clientId, pSource->taskId, 0, pSource->execId, &pBuf.pData,
1024
                                               pTaskInfo->localFetch.explainRes);
1025
    code = loadRemoteDataCallback(pWrapper, &pBuf, code);
6,630✔
1026
    taosMemoryFree(pWrapper);
6,630!
1027
    QUERY_CHECK_CODE(code, lino, _end);
6,630!
1028
  } else {
1029
    bool needStreamPesudoFuncVals = true;
62,113,535✔
1030
    SResFetchReq req = {0};
62,113,535✔
1031
    req.header.vgId = pSource->addr.nodeId;
62,114,084✔
1032
    req.sId = pSource->sId;
62,114,053✔
1033
    req.clientId = pSource->clientId;
62,114,550✔
1034
    req.taskId = pSource->taskId;
62,115,128✔
1035
    req.queryId = pTaskInfo->id.queryId;
62,114,506✔
1036
    req.execId = pSource->execId;
62,113,571✔
1037
    if (pTaskInfo->pStreamRuntimeInfo) {
62,113,999✔
1038
      req.dynTbname = pExchangeInfo->dynTbname;
2,465,562!
1039
      req.execId = pTaskInfo->pStreamRuntimeInfo->execId;
2,465,247✔
1040
      req.pStRtFuncInfo = &pTaskInfo->pStreamRuntimeInfo->funcInfo;
2,465,247✔
1041

1042
      if (pSource->fetchMsgType == TDMT_STREAM_FETCH_FROM_RUNNER) {
2,465,562✔
1043
        qDebug("%s stream fetch from runner, execId:%d, %p", GET_TASKID(pTaskInfo), req.execId, pTaskInfo->pStreamRuntimeInfo);
23,463!
1044
      } else if (pSource->fetchMsgType == TDMT_STREAM_FETCH_FROM_CACHE) {
2,441,784✔
1045
        code = getCurrentWinCalcTimeRange(req.pStRtFuncInfo, &req.pStRtFuncInfo->curWindow);
442,713✔
1046
        QUERY_CHECK_CODE(code, lino, _end);
443,028!
1047
        needStreamPesudoFuncVals = false;
443,028✔
1048
        qDebug("%s stream fetch from cache, execId:%d, curWinIdx:%d, time range:[%" PRId64 ", %" PRId64 "]",
443,028✔
1049
               GET_TASKID(pTaskInfo), req.execId, req.pStRtFuncInfo->curIdx, req.pStRtFuncInfo->curWindow.skey,
1050
               req.pStRtFuncInfo->curWindow.ekey);
1051
      }
1052
      if (!pDataInfo->fetchSent) {
2,465,562!
1053
        req.reset = pDataInfo->fetchSent = true;
2,197,032!
1054
      }
1055
    }
1056
    if (pDataInfo->isVtbRefScan) {
62,114,000!
1057
      code = buildTableScanOperatorParamEx(&req.pOpParam, pDataInfo->pSrcUidList, pDataInfo->srcOpType, pDataInfo->colMap, pDataInfo->tableSeq, &pDataInfo->window);
15,564,690!
1058
      taosArrayDestroy(pDataInfo->colMap->colMap);
15,564,690✔
1059
      taosMemoryFreeClear(pDataInfo->colMap);
15,564,690!
1060
      taosArrayDestroy(pDataInfo->pSrcUidList);
15,564,690✔
1061
      pDataInfo->pSrcUidList = NULL;
15,564,690✔
1062
      if (TSDB_CODE_SUCCESS != code) {
15,564,690!
1063
        pTaskInfo->code = code;
×
1064
        taosMemoryFree(pWrapper);
×
1065
        return pTaskInfo->code;
×
1066
      }
1067
    } else if (pDataInfo->isVtbTagScan) {
46,549,769!
1068
      code = buildTagScanOperatorParam(&req.pOpParam, pDataInfo->pSrcUidList, pDataInfo->srcOpType);
1,235,466✔
1069
      taosArrayDestroy(pDataInfo->pSrcUidList);
1,235,466✔
1070
      pDataInfo->pSrcUidList = NULL;
1,235,466✔
1071
      if (TSDB_CODE_SUCCESS != code) {
1,235,466!
1072
        pTaskInfo->code = code;
×
1073
        taosMemoryFree(pWrapper);
×
1074
        return pTaskInfo->code;
×
1075
      }
1076
    } else {
1077
      if (pDataInfo->pSrcUidList) {
45,313,913✔
1078
        code = buildTableScanOperatorParam(&req.pOpParam, pDataInfo->pSrcUidList, pDataInfo->srcOpType, pDataInfo->tableSeq);
51,596!
1079
        taosArrayDestroy(pDataInfo->pSrcUidList);
51,596✔
1080
        pDataInfo->pSrcUidList = NULL;
51,596✔
1081
        if (TSDB_CODE_SUCCESS != code) {
51,596!
1082
          pTaskInfo->code = code;
×
1083
          taosMemoryFree(pWrapper);
×
1084
          return pTaskInfo->code;
×
1085
        }
1086
      }
1087
    }
1088

1089
    int32_t msgSize = tSerializeSResFetchReq(NULL, 0, &req, needStreamPesudoFuncVals);
62,114,432✔
1090
    if (msgSize < 0) {
62,112,493!
1091
      pTaskInfo->code = msgSize;
×
1092
      taosMemoryFree(pWrapper);
×
1093
      freeOperatorParam(req.pOpParam, OP_GET_PARAM);
×
1094
      return pTaskInfo->code;
×
1095
    }
1096

1097
    void* msg = taosMemoryCalloc(1, msgSize);
62,112,493!
1098
    if (NULL == msg) {
62,111,605!
1099
      pTaskInfo->code = terrno;
×
1100
      taosMemoryFree(pWrapper);
×
1101
      freeOperatorParam(req.pOpParam, OP_GET_PARAM);
×
1102
      return pTaskInfo->code;
×
1103
    }
1104

1105
    msgSize = tSerializeSResFetchReq(msg, msgSize, &req, needStreamPesudoFuncVals);
62,111,605✔
1106
    if (msgSize < 0) {
62,113,311!
1107
      pTaskInfo->code = msgSize;
×
1108
      taosMemoryFree(pWrapper);
×
1109
      taosMemoryFree(msg);
×
1110
      freeOperatorParam(req.pOpParam, OP_GET_PARAM);
×
1111
      return pTaskInfo->code;
×
1112
    }
1113

1114
    freeOperatorParam(req.pOpParam, OP_GET_PARAM);
62,113,311✔
1115

1116
    qDebug("%s build fetch msg and send to vgId:%d, ep:%s, clientId:0x%" PRIx64 " taskId:0x%" PRIx64
62,113,782✔
1117
           ", seqId:%" PRId64 ", execId:%d, %p, %d/%" PRIzu,
1118
           GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->addr.epSet.eps[0].fqdn, pSource->clientId,
1119
           pSource->taskId, pExchangeInfo->seqId, pSource->execId, pExchangeInfo, sourceIndex, totalSources);
1120

1121
    // send the fetch remote task result reques
1122
    SMsgSendInfo* pMsgSendInfo = taosMemoryCalloc(1, sizeof(SMsgSendInfo));
62,114,973!
1123
    if (NULL == pMsgSendInfo) {
62,114,223!
1124
      taosMemoryFreeClear(msg);
×
1125
      taosMemoryFree(pWrapper);
×
1126
      qError("%s prepare message %d failed", GET_TASKID(pTaskInfo), (int32_t)sizeof(SMsgSendInfo));
×
1127
      pTaskInfo->code = terrno;
×
1128
      return pTaskInfo->code;
×
1129
    }
1130

1131
    pMsgSendInfo->param = pWrapper;
62,114,223✔
1132
    pMsgSendInfo->paramFreeFp = taosAutoMemoryFree;
62,115,038✔
1133
    pMsgSendInfo->msgInfo.pData = msg;
62,115,346✔
1134
    pMsgSendInfo->msgInfo.len = msgSize;
62,115,187✔
1135
    pMsgSendInfo->msgType = pSource->fetchMsgType;
62,115,075✔
1136
    pMsgSendInfo->fp = loadRemoteDataCallback;
62,115,368✔
1137
    pMsgSendInfo->requestId = pTaskInfo->id.queryId;
62,114,890✔
1138

1139
    int64_t transporterId = 0;
62,114,475✔
1140
    void* poolHandle = NULL;
62,115,117✔
1141
    code = asyncSendMsgToServer(pExchangeInfo->pTransporter, &pSource->addr.epSet, &transporterId, pMsgSendInfo);
62,115,117✔
1142
    QUERY_CHECK_CODE(code, lino, _end);
62,115,634!
1143
    int64_t* pRpcHandle = taosArrayGet(pExchangeInfo->pFetchRpcHandles, sourceIndex);
62,115,634✔
1144
    *pRpcHandle = transporterId;
62,115,634✔
1145
  }
1146

1147
_end:
62,122,264✔
1148
  if (code != TSDB_CODE_SUCCESS) {
62,122,264!
1149
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
1150
  }
1151
  return code;
62,122,120✔
1152
}
1153

1154
void updateLoadRemoteInfo(SLoadRemoteDataInfo* pInfo, int64_t numOfRows, int32_t dataLen, int64_t startTs,
48,096,582✔
1155
                          SOperatorInfo* pOperator) {
1156
  pInfo->totalRows += numOfRows;
48,096,582✔
1157
  pInfo->totalSize += dataLen;
48,096,438✔
1158
  pInfo->totalElapsed += (taosGetTimestampUs() - startTs);
48,096,463✔
1159
  pOperator->resultInfo.totalRows += numOfRows;
48,096,582✔
1160
}
48,096,463✔
1161

1162
int32_t extractDataBlockFromFetchRsp(SSDataBlock* pRes, char* pData, SArray* pColList, char** pNextStart) {
230,221,102✔
1163
  int32_t      code = TSDB_CODE_SUCCESS;
230,221,102✔
1164
  int32_t      lino = 0;
230,221,102✔
1165
  SSDataBlock* pBlock = NULL;
230,221,102✔
1166
  if (pColList == NULL) {  // data from other sources
230,221,102✔
1167
    blockDataCleanup(pRes);
229,478,785✔
1168
    code = blockDecodeInternal(pRes, pData, (const char**)pNextStart);
229,478,929✔
1169
    if (code) {
229,478,929!
1170
      return code;
×
1171
    }
1172
  } else {  // extract data according to pColList
1173
    char* pStart = pData;
742,317✔
1174

1175
    int32_t numOfCols = htonl(*(int32_t*)pStart);
742,317✔
1176
    pStart += sizeof(int32_t);
742,317✔
1177

1178
    // todo refactor:extract method
1179
    SSysTableSchema* pSchema = (SSysTableSchema*)pStart;
742,317✔
1180
    for (int32_t i = 0; i < numOfCols; ++i) {
10,119,463✔
1181
      SSysTableSchema* p = (SSysTableSchema*)pStart;
9,377,146✔
1182

1183
      p->colId = htons(p->colId);
9,377,146✔
1184
      p->bytes = htonl(p->bytes);
9,377,146✔
1185
      pStart += sizeof(SSysTableSchema);
9,377,146✔
1186
    }
1187

1188
    pBlock = NULL;
742,317✔
1189
    code = createDataBlock(&pBlock);
742,317✔
1190
    QUERY_CHECK_CODE(code, lino, _end);
742,317!
1191

1192
    for (int32_t i = 0; i < numOfCols; ++i) {
10,119,463✔
1193
      SColumnInfoData idata = createColumnInfoData(pSchema[i].type, pSchema[i].bytes, pSchema[i].colId);
9,377,004✔
1194
      code = blockDataAppendColInfo(pBlock, &idata);
9,377,004✔
1195
      QUERY_CHECK_CODE(code, lino, _end);
9,377,146!
1196
    }
1197

1198
    code = blockDecodeInternal(pBlock, pStart, NULL);
742,459✔
1199
    QUERY_CHECK_CODE(code, lino, _end);
742,317!
1200

1201
    code = blockDataEnsureCapacity(pRes, pBlock->info.rows);
742,317✔
1202
    QUERY_CHECK_CODE(code, lino, _end);
742,317!
1203

1204
    // data from mnode
1205
    pRes->info.dataLoad = 1;
742,317✔
1206
    pRes->info.rows = pBlock->info.rows;
742,317✔
1207
    pRes->info.scanFlag = MAIN_SCAN;
742,317✔
1208
    code = relocateColumnData(pRes, pColList, pBlock->pDataBlock, false);
742,317✔
1209
    QUERY_CHECK_CODE(code, lino, _end);
742,317!
1210

1211
    blockDataDestroy(pBlock);
742,317✔
1212
    pBlock = NULL;
742,317✔
1213
  }
1214

1215
_end:
230,221,246✔
1216
  if (code != TSDB_CODE_SUCCESS) {
230,220,905!
1217
    blockDataDestroy(pBlock);
×
1218
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
1219
  }
1220
  return code;
230,220,905✔
1221
}
1222

1223
void setAllSourcesCompleted(SOperatorInfo* pOperator) {
23,583,219✔
1224
  SExchangeInfo* pExchangeInfo = pOperator->info;
23,583,219✔
1225
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
23,583,219✔
1226

1227
  SLoadRemoteDataInfo* pLoadInfo = &pExchangeInfo->loadInfo;
23,583,219✔
1228
  size_t               totalSources = taosArrayGetSize(pExchangeInfo->pSources);
23,583,219✔
1229
  qDebug("%s all %" PRIzu " sources are exhausted, total rows: %" PRIu64 ", %.2f Kb, elapsed:%.2f ms",
23,583,097✔
1230
         GET_TASKID(pTaskInfo), totalSources, pLoadInfo->totalRows, pLoadInfo->totalSize / 1024.0,
1231
         pLoadInfo->totalElapsed / 1000.0);
1232

1233
  setOperatorCompleted(pOperator);
23,583,097✔
1234
}
23,582,953✔
1235

1236
int32_t getCompletedSources(const SArray* pArray, int32_t* pRes) {
68,054,954✔
1237
  int32_t code = TSDB_CODE_SUCCESS;
68,054,954✔
1238
  int32_t lino = 0;
68,054,954✔
1239
  size_t  total = taosArrayGetSize(pArray);
68,054,954✔
1240

1241
  int32_t completed = 0;
68,054,661✔
1242
  for (int32_t k = 0; k < total; ++k) {
194,397,361✔
1243
    SSourceDataInfo* p = taosArrayGet(pArray, k);
126,342,825✔
1244
    QUERY_CHECK_NULL(p, code, lino, _end, terrno);
126,343,284!
1245
    if (p->status == EX_SOURCE_DATA_EXHAUSTED) {
126,343,284✔
1246
      qDebug("source %d is completed, info:%p %p", k, pArray, p);
58,145,768✔
1247
      completed += 1;
58,145,768✔
1248
    }
1249
  }
1250

1251
  *pRes = completed;
68,054,536✔
1252
_end:
68,054,978✔
1253
  if (code != TSDB_CODE_SUCCESS) {
68,054,978!
1254
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
1255
  }
1256
  return code;
68,055,311✔
1257
}
1258

1259
int32_t prepareConcurrentlyLoad(SOperatorInfo* pOperator) {
24,399,277✔
1260
  SExchangeInfo* pExchangeInfo = pOperator->info;
24,399,277✔
1261
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
24,399,693✔
1262

1263
  size_t  totalSources = taosArrayGetSize(pExchangeInfo->pSourceDataInfo);
24,399,260✔
1264
  int64_t startTs = taosGetTimestampUs();
24,399,153✔
1265

1266
  // Asynchronously send all fetch requests to all sources.
1267
  for (int32_t i = 0; i < totalSources; ++i) {
60,226,615✔
1268
    int32_t code = doSendFetchDataRequest(pExchangeInfo, pTaskInfo, i);
35,825,385✔
1269
    if (code != TSDB_CODE_SUCCESS) {
35,827,313!
UNCOV
1270
      pTaskInfo->code = code;
×
1271
      return code;
×
1272
    }
1273
  }
1274

1275
  int64_t endTs = taosGetTimestampUs();
24,399,705✔
1276
  qDebug("%s send all fetch requests to %" PRIzu " sources completed, elapsed:%.2fms", GET_TASKID(pTaskInfo),
24,399,705✔
1277
         totalSources, (endTs - startTs) / 1000.0);
1278

1279
  pOperator->status = OP_RES_TO_RETURN;
24,399,705✔
1280
  pOperator->cost.openCost = taosGetTimestampUs() - startTs;
24,399,084✔
1281
  if (isTaskKilled(pTaskInfo)) {
24,399,439!
1282
    T_LONG_JMP(pTaskInfo->env, pTaskInfo->code);
×
1283
  }
1284

1285
  return TSDB_CODE_SUCCESS;
24,399,705✔
1286
}
1287

1288
int32_t doExtractResultBlocks(SExchangeInfo* pExchangeInfo, SSourceDataInfo* pDataInfo) {
47,353,830✔
1289
  int32_t            code = TSDB_CODE_SUCCESS;
47,353,830✔
1290
  int32_t            lino = 0;
47,353,830✔
1291
  SRetrieveTableRsp* pRetrieveRsp = pDataInfo->pRsp;
47,353,830✔
1292
  SSDataBlock*       pb = NULL;
47,353,949✔
1293

1294
  char* pNextStart = pRetrieveRsp->data;
47,353,949✔
1295
  char* pStart = pNextStart;
47,354,027✔
1296

1297
  int32_t index = 0;
47,354,027✔
1298

1299
  if (pRetrieveRsp->compressed) {  // decompress the data
47,354,027!
1300
    if (pDataInfo->decompBuf == NULL) {
×
1301
      pDataInfo->decompBuf = taosMemoryMalloc(pRetrieveRsp->payloadLen);
×
1302
      QUERY_CHECK_NULL(pDataInfo->decompBuf, code, lino, _end, terrno);
×
1303
      pDataInfo->decompBufSize = pRetrieveRsp->payloadLen;
×
1304
    } else {
1305
      if (pDataInfo->decompBufSize < pRetrieveRsp->payloadLen) {
×
1306
        char* p = taosMemoryRealloc(pDataInfo->decompBuf, pRetrieveRsp->payloadLen);
×
1307
        QUERY_CHECK_NULL(p, code, lino, _end, terrno);
×
1308
        if (p != NULL) {
×
1309
          pDataInfo->decompBuf = p;
×
1310
          pDataInfo->decompBufSize = pRetrieveRsp->payloadLen;
×
1311
        }
1312
      }
1313
    }
1314
  }
1315

1316
  while (index++ < pRetrieveRsp->numOfBlocks) {
276,832,562✔
1317
    pStart = pNextStart;
229,478,178✔
1318

1319
    if (taosArrayGetSize(pExchangeInfo->pRecycledBlocks) > 0) {
229,478,178✔
1320
      pb = *(SSDataBlock**)taosArrayPop(pExchangeInfo->pRecycledBlocks);
128,799,989✔
1321
      blockDataCleanup(pb);
128,799,989✔
1322
    } else {
1323
      code = createOneDataBlock(pExchangeInfo->pDummyBlock, false, &pb);
100,679,059✔
1324
      QUERY_CHECK_NULL(pb, code, lino, _end, code);
100,678,940!
1325
    }
1326

1327
    int32_t compLen = *(int32_t*)pStart;
229,478,929✔
1328
    pStart += sizeof(int32_t);
229,478,929✔
1329

1330
    int32_t rawLen = *(int32_t*)pStart;
229,478,929✔
1331
    pStart += sizeof(int32_t);
229,478,810✔
1332
    QUERY_CHECK_CONDITION((compLen <= rawLen && compLen != 0), code, lino, _end, TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR);
229,478,808!
1333

1334
    pNextStart = pStart + compLen;
229,478,808✔
1335
    if (pRetrieveRsp->compressed && (compLen < rawLen)) {
229,478,929!
1336
      int32_t t = tsDecompressString(pStart, compLen, 1, pDataInfo->decompBuf, rawLen, ONE_STAGE_COMP, NULL, 0);
×
1337
      QUERY_CHECK_CONDITION((t == rawLen), code, lino, _end, TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR);
×
1338
      pStart = pDataInfo->decompBuf;
×
1339
    }
1340

1341
    code = extractDataBlockFromFetchRsp(pb, pStart, NULL, &pStart);
229,479,048✔
1342
    if (code != 0) {
229,478,469!
1343
      taosMemoryFreeClear(pDataInfo->pRsp);
×
1344
      goto _end;
×
1345
    }
1346

1347
    void* tmp = taosArrayPush(pExchangeInfo->pResultBlockList, &pb);
229,478,469✔
1348
    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);
229,478,929!
1349
    qDebug("%dth block added to resultBlockList, rows:%" PRId64, index, pb->info.rows);
229,478,929✔
1350
    pb = NULL;
229,479,048✔
1351
  }
1352

1353
_end:
47,354,265✔
1354
  if (code != TSDB_CODE_SUCCESS) {
47,354,265!
1355
    blockDataDestroy(pb);
×
1356
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
1357
  }
1358
  return code;
47,354,265✔
1359
}
1360

1361
int32_t seqLoadRemoteData(SOperatorInfo* pOperator) {
16,868,047✔
1362
  SExchangeInfo* pExchangeInfo = pOperator->info;
16,868,047✔
1363
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
16,868,047✔
1364

1365
  int32_t code = 0;
16,868,047✔
1366
  size_t  totalSources = taosArrayGetSize(pExchangeInfo->pSources);
16,868,047✔
1367
  int64_t startTs = taosGetTimestampUs();
16,868,047✔
1368

1369
  int32_t vgId = 0;
16,868,047✔
1370
  if (pExchangeInfo->dynTbname) {
16,868,047!
UNCOV
1371
    SArray* vals = pTaskInfo->pStreamRuntimeInfo->funcInfo.pStreamPartColVals;
×
UNCOV
1372
    for (int32_t i = 0; i < taosArrayGetSize(vals); ++i) {
×
UNCOV
1373
      SStreamGroupValue* pValue = taosArrayGet(vals, i);
×
UNCOV
1374
      if (pValue != NULL && pValue->isTbname) {
×
UNCOV
1375
        vgId = pValue->vgId;
×
UNCOV
1376
        break;
×
1377
      }
1378
    }
1379
  }
1380

1381
  while (1) {
4,439,848✔
1382
    if (pExchangeInfo->current >= totalSources) {
21,307,895✔
1383
      setAllSourcesCompleted(pOperator);
4,446,339✔
1384
      return TSDB_CODE_SUCCESS;
4,446,339✔
1385
    }
1386

1387
    SDownstreamSourceNode* pSource = taosArrayGet(pExchangeInfo->pSources, pExchangeInfo->current);
16,861,556✔
1388
    if (!pSource) {
16,861,556!
1389
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
×
1390
      pTaskInfo->code = terrno;
×
1391
      T_LONG_JMP(pTaskInfo->env, pTaskInfo->code);
×
1392
    }
1393

1394
    if (vgId != 0 && pSource->addr.nodeId != vgId){
16,861,556!
UNCOV
1395
      pExchangeInfo->current += 1;
×
UNCOV
1396
      continue;
×
1397
    }
1398

1399
    SSourceDataInfo* pDataInfo = taosArrayGet(pExchangeInfo->pSourceDataInfo, pExchangeInfo->current);
16,861,556✔
1400
    if (!pDataInfo) {
16,861,556!
1401
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
1402
      pTaskInfo->code = terrno;
×
1403
      T_LONG_JMP(pTaskInfo->env, pTaskInfo->code);
×
1404
    }
1405
    pDataInfo->status = EX_SOURCE_DATA_NOT_READY;
16,861,556✔
1406

1407
    code = doSendFetchDataRequest(pExchangeInfo, pTaskInfo, pExchangeInfo->current);
16,861,556✔
1408
    if (code != TSDB_CODE_SUCCESS) {
16,861,556!
1409
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
×
1410
      pTaskInfo->code = code;
×
1411
      T_LONG_JMP(pTaskInfo->env, pTaskInfo->code);
×
1412
    }
1413

UNCOV
1414
    while (true) {
×
1415
      code = exchangeWait(pOperator, pExchangeInfo);
16,861,556✔
1416
      if (code != TSDB_CODE_SUCCESS || isTaskKilled(pTaskInfo)) {
16,861,556!
UNCOV
1417
        T_LONG_JMP(pTaskInfo->env, pTaskInfo->code);
×
1418
      }
1419

1420
      int64_t currSeqId = atomic_load_64(&pExchangeInfo->seqId);
16,861,556✔
1421
      if (pDataInfo->seqId != currSeqId) {
16,861,556!
UNCOV
1422
        qDebug("seq rsp reqId %" PRId64 " mismatch with exchange %p curr seqId %" PRId64 ", ignore it", pDataInfo->seqId, pExchangeInfo, currSeqId);
×
UNCOV
1423
        taosMemoryFreeClear(pDataInfo->pRsp);
×
UNCOV
1424
        continue;
×
1425
      }
1426

1427
      break;
16,861,556✔
1428
    }
1429

1430
    if (pDataInfo->code != TSDB_CODE_SUCCESS) {
16,861,556!
1431
      qError("%s vgId:%d, clientId:0x%" PRIx64 " taskID:0x%" PRIx64 " execId:%d error happens, code:%s",
×
1432
             GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->clientId, pSource->taskId, pSource->execId,
1433
             tstrerror(pDataInfo->code));
1434
      pOperator->pTaskInfo->code = pDataInfo->code;
×
1435
      return pOperator->pTaskInfo->code;
×
1436
    }
1437

1438
    SRetrieveTableRsp*   pRsp = pDataInfo->pRsp;
16,861,556✔
1439
    SLoadRemoteDataInfo* pLoadInfo = &pExchangeInfo->loadInfo;
16,861,556✔
1440

1441
    if (pRsp->numOfRows == 0) {
16,861,556✔
1442
      qDebug("exhausted %p,%s vgId:%d, clientId:0x%" PRIx64 " taskID:0x%" PRIx64
4,439,848✔
1443
             " execId:%d %d of total completed, rowsOfSource:%" PRIu64 ", totalRows:%" PRIu64 " try next", pDataInfo,
1444
             GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->clientId, pSource->taskId, pSource->execId,
1445
             pExchangeInfo->current + 1, pDataInfo->totalRows, pLoadInfo->totalRows);
1446

1447
      pDataInfo->status = EX_SOURCE_DATA_EXHAUSTED;
4,439,848✔
1448
      if (pDataInfo->isVtbRefScan || pDataInfo->isVtbTagScan) {
4,439,848!
1449
        pExchangeInfo->current = totalSources;
4,438,412✔
1450
      } else {
1451
        pExchangeInfo->current += 1;
1,436✔
1452
      }
1453
      taosMemoryFreeClear(pDataInfo->pRsp);
4,439,848!
1454
      continue;
4,439,848✔
1455
    }
1456

1457
    code = doExtractResultBlocks(pExchangeInfo, pDataInfo);
12,421,708✔
1458
    if (code != TSDB_CODE_SUCCESS) {
12,421,708!
1459
      goto _error;
×
1460
    }
1461

1462
    SRetrieveTableRsp* pRetrieveRsp = pDataInfo->pRsp;
12,421,708✔
1463
    if (pRsp->completed == 1) {
12,421,708✔
1464
      qDebug("exhausted %p,%s fetch msg rsp from vgId:%d, clientId:0x%" PRIx64 " taskId:0x%" PRIx64 " execId:%d numOfRows:%" PRId64
18,138!
1465
             ", rowsOfSource:%" PRIu64 ", totalRows:%" PRIu64 ", totalBytes:%" PRIu64 " try next %d/%" PRIzu, pDataInfo,
1466
             GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->clientId, pSource->taskId, pSource->execId,
1467
             pRetrieveRsp->numOfRows, pDataInfo->totalRows, pLoadInfo->totalRows, pLoadInfo->totalSize,
1468
             pExchangeInfo->current + 1, totalSources);
1469

1470
      pDataInfo->status = EX_SOURCE_DATA_EXHAUSTED;
18,138✔
1471
      if (pDataInfo->isVtbRefScan) {
18,138!
1472
        pExchangeInfo->current = totalSources;
×
1473
      } else {
1474
        pExchangeInfo->current += 1;
18,138✔
1475
      }
1476
    } else {
1477
      qDebug("%s fetch msg rsp from vgId:%d, clientId:0x%" PRIx64 " taskId:0x%" PRIx64 " execId:%d numOfRows:%" PRId64
12,403,570✔
1478
             ", totalRows:%" PRIu64 ", totalBytes:%" PRIu64,
1479
             GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->clientId, pSource->taskId, pSource->execId,
1480
             pRetrieveRsp->numOfRows, pLoadInfo->totalRows, pLoadInfo->totalSize);
1481
    }
1482
    if (pExchangeInfo->dynamicOp && pExchangeInfo->seqLoadData) {
12,421,708!
1483
      taosArrayClear(pExchangeInfo->pSourceDataInfo);
12,335,932✔
1484
    }
1485
    updateLoadRemoteInfo(pLoadInfo, pRetrieveRsp->numOfRows, pRetrieveRsp->compLen, startTs, pOperator);
12,421,708✔
1486
    pDataInfo->totalRows += pRetrieveRsp->numOfRows;
12,421,708✔
1487

1488
    taosMemoryFreeClear(pDataInfo->pRsp);
12,421,708!
1489
    return TSDB_CODE_SUCCESS;
12,421,708✔
1490
  }
1491

1492
_error:
×
1493
  pTaskInfo->code = code;
×
1494
  return code;
×
1495
}
1496

1497
void clearVtbScanDataInfo(void* pItem) {
4,004,039✔
1498
  SSourceDataInfo *pInfo = (SSourceDataInfo *)pItem;
4,004,039✔
1499
  if (pInfo->colMap) {
4,004,039!
1500
    taosArrayDestroy(pInfo->colMap->colMap);
×
1501
    taosMemoryFreeClear(pInfo->colMap);
×
1502
  }
1503
  taosArrayDestroy(pInfo->pSrcUidList);
4,004,039✔
1504
}
4,004,039✔
1505

1506
int32_t addSingleExchangeSource(SOperatorInfo* pOperator, SExchangeOperatorBasicParam* pBasicParam) {
16,851,752✔
1507
  SExchangeInfo*     pExchangeInfo = pOperator->info;
16,851,752✔
1508
  SExchangeSrcIndex* pIdx = tSimpleHashGet(pExchangeInfo->pHashSources, &pBasicParam->vgId, sizeof(pBasicParam->vgId));
16,851,752✔
1509
  if (NULL == pIdx) {
16,851,752✔
1510
    if (pBasicParam->isNewDeployed) {
5,871!
1511
      SDownstreamSourceNode *pNode = NULL;
5,871✔
1512
      int32_t code = nodesCloneNode((SNode*)&pBasicParam->newDeployedSrc, (SNode**)&pNode);
5,871✔
1513
      if (code != TSDB_CODE_SUCCESS) {
5,871!
1514
        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
×
1515
        return code;
×
1516
      }
1517

1518
      void* tmp = taosArrayPush(pExchangeInfo->pSources, pNode);
5,871✔
1519
      if (!tmp) {
5,871!
1520
        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
1521
        return terrno;
×
1522
      }
1523
      SExchangeSrcIndex idx = {.srcIdx = taosArrayGetSize(pExchangeInfo->pSources) - 1, .inUseIdx = -1};
5,871✔
1524
      code =
1525
          tSimpleHashPut(pExchangeInfo->pHashSources, &pNode->addr.nodeId, sizeof(pNode->addr.nodeId), &idx, sizeof(idx));
5,871✔
1526
      if (pExchangeInfo->pHashSources && code != TSDB_CODE_SUCCESS) {
5,871!
1527
        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
×
1528
        return code;
×
1529
      }
1530
      pIdx = tSimpleHashGet(pExchangeInfo->pHashSources, &pBasicParam->vgId, sizeof(pBasicParam->vgId));
5,871✔
1531
      if (pIdx == NULL) {
5,871!
1532
        qError("No exchange source for vgId: %d", pBasicParam->vgId);
×
1533
        return TSDB_CODE_INVALID_PARA;
×
1534
      }
1535
    } else {
1536
      qError("No exchange source for vgId: %d", pBasicParam->vgId);
×
1537
      return TSDB_CODE_INVALID_PARA;
×
1538
    }
1539
  }
1540

1541
  qDebug("start to add single exchange source");
16,851,752✔
1542

1543
  if (pBasicParam->isVtbRefScan) {
16,851,752!
1544
    SSourceDataInfo dataInfo = {0};
15,564,690✔
1545
    dataInfo.status = EX_SOURCE_DATA_NOT_READY;
15,564,690✔
1546
    dataInfo.taskId = pExchangeInfo->pTaskId;
15,564,690✔
1547
    dataInfo.index = pIdx->srcIdx;
15,564,690✔
1548
    dataInfo.window = pBasicParam->window;
15,564,690✔
1549
    dataInfo.colMap = taosMemoryMalloc(sizeof(SOrgTbInfo));
15,564,690!
1550
    dataInfo.colMap->vgId = pBasicParam->colMap->vgId;
15,564,690✔
1551
    tstrncpy(dataInfo.colMap->tbName, pBasicParam->colMap->tbName, TSDB_TABLE_FNAME_LEN);
15,564,690!
1552
    dataInfo.colMap->colMap = taosArrayDup(pBasicParam->colMap->colMap, NULL);
15,564,690✔
1553

1554
    dataInfo.pSrcUidList = taosArrayDup(pBasicParam->uidList, NULL);
15,564,690✔
1555
    if (dataInfo.pSrcUidList == NULL) {
15,564,690!
1556
      return terrno;
×
1557
    }
1558

1559
    dataInfo.isVtbRefScan = pBasicParam->isVtbRefScan;
15,564,690!
1560
    dataInfo.isVtbTagScan = pBasicParam->isVtbTagScan;
15,564,690!
1561
    dataInfo.srcOpType = pBasicParam->srcOpType;
15,564,690✔
1562
    dataInfo.tableSeq = pBasicParam->tableSeq;
15,564,690!
1563

1564
    taosArrayClearEx(pExchangeInfo->pSourceDataInfo, clearVtbScanDataInfo);
15,564,690✔
1565
    void* tmp = taosArrayPush(pExchangeInfo->pSourceDataInfo, &dataInfo);
15,564,690✔
1566
    if (!tmp) {
15,564,690!
1567
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
1568
      return terrno;
×
1569
    }
1570
  } else if (pBasicParam->isVtbTagScan) {
1,287,062!
1571
    SSourceDataInfo dataInfo = {0};
1,235,466✔
1572
    dataInfo.status = EX_SOURCE_DATA_NOT_READY;
1,235,466✔
1573
    dataInfo.taskId = pExchangeInfo->pTaskId;
1,235,466✔
1574
    dataInfo.index = pIdx->srcIdx;
1,235,466✔
1575
    dataInfo.window = pBasicParam->window;
1,235,466✔
1576
    dataInfo.colMap = NULL;
1,235,466✔
1577

1578
    dataInfo.pSrcUidList = taosArrayDup(pBasicParam->uidList, NULL);
1,235,466✔
1579
    if (dataInfo.pSrcUidList == NULL) {
1,235,466!
1580
      return terrno;
×
1581
    }
1582

1583
    dataInfo.isVtbRefScan = pBasicParam->isVtbRefScan;
1,235,466!
1584
    dataInfo.isVtbTagScan = pBasicParam->isVtbTagScan;
1,235,466!
1585
    dataInfo.srcOpType = pBasicParam->srcOpType;
1,235,466✔
1586
    dataInfo.tableSeq = pBasicParam->tableSeq;
1,235,466!
1587

1588
    taosArrayClearEx(pExchangeInfo->pSourceDataInfo, clearVtbScanDataInfo);
1,235,466✔
1589
    void* tmp = taosArrayPush(pExchangeInfo->pSourceDataInfo, &dataInfo);
1,235,466✔
1590
    if (!tmp) {
1,235,466!
1591
      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
1592
      return terrno;
×
1593
    }
1594
  } else {
1595
    if (pIdx->inUseIdx < 0) {
51,596✔
1596
      SSourceDataInfo dataInfo = {0};
48,518✔
1597
      dataInfo.status = EX_SOURCE_DATA_NOT_READY;
48,518✔
1598
      dataInfo.taskId = pExchangeInfo->pTaskId;
48,518✔
1599
      dataInfo.index = pIdx->srcIdx;
48,518✔
1600
      if (pBasicParam->isVtbRefScan) {
48,518!
1601
        dataInfo.window = pBasicParam->window;
×
1602
        dataInfo.colMap = taosMemoryMalloc(sizeof(SOrgTbInfo));
×
1603
        dataInfo.colMap->vgId = pBasicParam->colMap->vgId;
×
1604
        tstrncpy(dataInfo.colMap->tbName, pBasicParam->colMap->tbName, TSDB_TABLE_FNAME_LEN);
×
1605
        dataInfo.colMap->colMap = taosArrayDup(pBasicParam->colMap->colMap, NULL);
×
1606
      }
1607

1608
      dataInfo.pSrcUidList = taosArrayDup(pBasicParam->uidList, NULL);
48,518✔
1609
      if (dataInfo.pSrcUidList == NULL) {
48,518!
1610
        return terrno;
×
1611
      }
1612

1613
      dataInfo.isVtbRefScan = pBasicParam->isVtbRefScan;
48,518!
1614
      dataInfo.isVtbTagScan = pBasicParam->isVtbTagScan;
48,518!
1615
      dataInfo.srcOpType = pBasicParam->srcOpType;
48,518✔
1616
      dataInfo.tableSeq = pBasicParam->tableSeq;
48,518!
1617

1618
      void* tmp = taosArrayPush(pExchangeInfo->pSourceDataInfo, &dataInfo);
48,518✔
1619
      if (!tmp) {
48,518!
1620
        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
×
1621
        return terrno;
×
1622
      }
1623
      pIdx->inUseIdx = taosArrayGetSize(pExchangeInfo->pSourceDataInfo) - 1;
48,518✔
1624
    } else {
1625
      SSourceDataInfo* pDataInfo = taosArrayGet(pExchangeInfo->pSourceDataInfo, pIdx->inUseIdx);
3,078✔
1626
      if (!pDataInfo) {
3,078!
1627
        return terrno;
×
1628
      }
1629
      if (pDataInfo->status == EX_SOURCE_DATA_EXHAUSTED) {
3,078!
1630
        pDataInfo->status = EX_SOURCE_DATA_NOT_READY;
3,078✔
1631
      }
1632

1633
      if (pBasicParam->isVtbRefScan) {
3,078!
1634
        pDataInfo->window = pBasicParam->window;
×
1635
        if (!pDataInfo->colMap) {
×
1636
          pDataInfo->colMap = taosMemoryMalloc(sizeof(SOrgTbInfo));
×
1637
        }
1638
        pDataInfo->colMap->vgId = pBasicParam->colMap->vgId;
×
1639
        tstrncpy(pDataInfo->colMap->tbName, pBasicParam->colMap->tbName, TSDB_TABLE_FNAME_LEN);
×
1640
        pDataInfo->colMap->colMap = taosArrayDup(pBasicParam->colMap->colMap, NULL);
×
1641
      }
1642

1643
      pDataInfo->pSrcUidList = taosArrayDup(pBasicParam->uidList, NULL);
3,078✔
1644
      if (pDataInfo->pSrcUidList == NULL) {
3,078!
1645
        return terrno;
×
1646
      }
1647

1648
      pDataInfo->isVtbRefScan = pBasicParam->isVtbRefScan;
3,078!
1649
      pDataInfo->isVtbTagScan = pBasicParam->isVtbTagScan;
3,078!
1650
      pDataInfo->srcOpType = pBasicParam->srcOpType;
3,078✔
1651
      pDataInfo->tableSeq = pBasicParam->tableSeq;
3,078!
1652
    }
1653
  }
1654

1655
  return TSDB_CODE_SUCCESS;
16,851,752✔
1656
}
1657

1658
int32_t addDynamicExchangeSource(SOperatorInfo* pOperator) {
16,838,190✔
1659
  SExchangeInfo*               pExchangeInfo = pOperator->info;
16,838,190✔
1660
  int32_t                      code = TSDB_CODE_SUCCESS;
16,838,190✔
1661
  SExchangeOperatorBasicParam* pBasicParam = NULL;
16,838,190✔
1662
  SExchangeOperatorParam*      pParam = (SExchangeOperatorParam*)pOperator->pOperatorGetParam->value;
16,838,190✔
1663
  if (pParam->multiParams) {
16,838,190!
1664
    SExchangeOperatorBatchParam* pBatch = (SExchangeOperatorBatchParam*)pOperator->pOperatorGetParam->value;
31,878✔
1665
    int32_t                      iter = 0;
31,878✔
1666
    while (NULL != (pBasicParam = tSimpleHashIterate(pBatch->pBatchs, pBasicParam, &iter))) {
77,318✔
1667
      code = addSingleExchangeSource(pOperator, pBasicParam);
45,440✔
1668
      if (code) {
45,440!
1669
        return code;
×
1670
      }
1671
    }
1672
  } else {
1673
    pBasicParam = &pParam->basic;
16,806,312✔
1674
    code = addSingleExchangeSource(pOperator, pBasicParam);
16,806,312✔
1675
  }
1676

1677
  freeOperatorParam(pOperator->pOperatorGetParam, OP_GET_PARAM);
16,838,190✔
1678
  pOperator->pOperatorGetParam = NULL;
16,838,190✔
1679

1680
  return code;
16,838,190✔
1681
}
1682

1683
int32_t prepareLoadRemoteData(SOperatorInfo* pOperator) {
257,259,838✔
1684
  SExchangeInfo* pExchangeInfo = pOperator->info;
257,259,838✔
1685
  int32_t        code = TSDB_CODE_SUCCESS;
257,260,380✔
1686
  int32_t        lino = 0;
257,260,380✔
1687
  
1688
  if ((OPTR_IS_OPENED(pOperator) && !pExchangeInfo->dynamicOp) ||
257,260,380!
1689
      (pExchangeInfo->dynamicOp && NULL == pOperator->pOperatorGetParam)) {
43,502,661!
1690
    qDebug("skip prepare, opened:%d, dynamicOp:%d, getParam:%p", OPTR_IS_OPENED(pOperator), pExchangeInfo->dynamicOp, pOperator->pOperatorGetParam);
214,019,817!
1691
    return TSDB_CODE_SUCCESS;
214,019,303✔
1692
  }
1693

1694
  if (pExchangeInfo->dynamicOp) {
43,239,934!
1695
    code = addDynamicExchangeSource(pOperator);
16,838,190✔
1696
    QUERY_CHECK_CODE(code, lino, _end);
16,838,190!
1697
  }
1698

1699
  if (pOperator->status == OP_NOT_OPENED && (pExchangeInfo->dynamicOp && pExchangeInfo->seqLoadData) || IS_STREAM_MODE(pOperator->pTaskInfo)) {
43,240,597!
1700
    pExchangeInfo->current = 0;
18,007,700✔
1701
  }
1702

1703
  int64_t st = taosGetTimestampUs();
43,240,401✔
1704

1705
  if (!IS_STREAM_MODE(pOperator->pTaskInfo) && !pExchangeInfo->seqLoadData) {
43,240,401!
1706
    code = prepareConcurrentlyLoad(pOperator);
24,398,609✔
1707
    QUERY_CHECK_CODE(code, lino, _end);
24,399,705!
1708
    pExchangeInfo->openedTs = taosGetTimestampUs();
24,399,849✔
1709
  }
1710

1711
  OPTR_SET_OPENED(pOperator);
43,241,429✔
1712
  pOperator->cost.openCost = (taosGetTimestampUs() - st) / 1000.0;
43,241,335✔
1713

1714
  qDebug("%s prepare load complete", pOperator->pTaskInfo->id.str);
43,240,543✔
1715

1716
_end:
4,917,209✔
1717
  if (code != TSDB_CODE_SUCCESS) {
43,240,548!
1718
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
1719
    pOperator->pTaskInfo->code = code;
×
1720
    T_LONG_JMP(pOperator->pTaskInfo->env, code);
×
1721
  }
1722
  return TSDB_CODE_SUCCESS;
43,240,548✔
1723
}
1724

1725
int32_t handleLimitOffset(SOperatorInfo* pOperator, SLimitInfo* pLimitInfo, SSDataBlock* pBlock, bool holdDataInBuf) {
1,990,639✔
1726
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
1,990,639✔
1727

1728
  if (pLimitInfo->remainGroupOffset > 0) {
1,990,639!
1729
    if (pLimitInfo->currentGroupId == 0) {  // it is the first group
×
1730
      pLimitInfo->currentGroupId = pBlock->info.id.groupId;
×
1731
      blockDataCleanup(pBlock);
×
1732
      return PROJECT_RETRIEVE_CONTINUE;
×
1733
    } else if (pLimitInfo->currentGroupId != pBlock->info.id.groupId) {
×
1734
      // now it is the data from a new group
1735
      pLimitInfo->remainGroupOffset -= 1;
×
1736

1737
      // ignore data block in current group
1738
      if (pLimitInfo->remainGroupOffset > 0) {
×
1739
        blockDataCleanup(pBlock);
×
1740
        return PROJECT_RETRIEVE_CONTINUE;
×
1741
      }
1742
    }
1743

1744
    // set current group id of the project operator
1745
    pLimitInfo->currentGroupId = pBlock->info.id.groupId;
×
1746
  }
1747

1748
  // here check for a new group data, we need to handle the data of the previous group.
1749
  if (pLimitInfo->currentGroupId != 0 && pLimitInfo->currentGroupId != pBlock->info.id.groupId) {
1,990,639✔
1750
    pLimitInfo->numOfOutputGroups += 1;
37,078✔
1751
    if ((pLimitInfo->slimit.limit > 0) && (pLimitInfo->slimit.limit <= pLimitInfo->numOfOutputGroups)) {
37,078!
1752
      pOperator->status = OP_EXEC_DONE;
×
1753
      blockDataCleanup(pBlock);
×
1754

1755
      return PROJECT_RETRIEVE_DONE;
×
1756
    }
1757

1758
    // reset the value for a new group data
1759
    resetLimitInfoForNextGroup(pLimitInfo);
37,078✔
1760
    // existing rows that belongs to previous group.
1761
    if (pBlock->info.rows > 0) {
37,078!
1762
      return PROJECT_RETRIEVE_DONE;
37,078✔
1763
    }
1764
  }
1765

1766
  // here we reach the start position, according to the limit/offset requirements.
1767

1768
  // set current group id
1769
  pLimitInfo->currentGroupId = pBlock->info.id.groupId;
1,953,561✔
1770

1771
  bool limitReached = applyLimitOffset(pLimitInfo, pBlock, pTaskInfo);
1,953,561✔
1772
  if (pBlock->info.rows == 0) {
1,953,561✔
1773
    return PROJECT_RETRIEVE_CONTINUE;
1,032,971✔
1774
  } else {
1775
    if (limitReached && (pLimitInfo->slimit.limit > 0 && pLimitInfo->slimit.limit <= pLimitInfo->numOfOutputGroups)) {
920,590!
1776
      setOperatorCompleted(pOperator);
×
1777
      return PROJECT_RETRIEVE_DONE;
×
1778
    }
1779
  }
1780

1781
  // todo optimize performance
1782
  // If there are slimit/soffset value exists, multi-round result can not be packed into one group, since the
1783
  // they may not belong to the same group the limit/offset value is not valid in this case.
1784
  if ((!holdDataInBuf) || (pBlock->info.rows >= pOperator->resultInfo.threshold) || hasSlimitOffsetInfo(pLimitInfo)) {
920,590!
1785
    return PROJECT_RETRIEVE_DONE;
920,590✔
1786
  } else {  // not full enough, continue to accumulate the output data in the buffer.
1787
    return PROJECT_RETRIEVE_CONTINUE;
×
1788
  }
1789
}
1790

1791
static int32_t exchangeWait(SOperatorInfo* pOperator, SExchangeInfo* pExchangeInfo) {
61,055,661✔
1792
  SExecTaskInfo* pTask = pOperator->pTaskInfo;
61,055,661✔
1793
  int32_t        code = TSDB_CODE_SUCCESS;
61,056,120✔
1794
  if (pTask->pWorkerCb) {
61,056,120!
1795
    code = pTask->pWorkerCb->beforeBlocking(pTask->pWorkerCb->pPool);
61,055,971✔
1796
    if (code != TSDB_CODE_SUCCESS) {
61,056,120!
1797
      pTask->code = code;
×
1798
      return pTask->code;
×
1799
    }
1800
  }
1801

1802
  code = tsem_wait(&pExchangeInfo->ready);
61,055,976✔
1803
  if (code != TSDB_CODE_SUCCESS) {
61,055,805!
1804
    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
×
1805
    pTask->code = code;
×
1806
    return pTask->code;
×
1807
  }
1808

1809
  if (pTask->pWorkerCb) {
61,055,805!
1810
    code = pTask->pWorkerCb->afterRecoverFromBlocking(pTask->pWorkerCb->pPool);
61,056,120✔
1811
    if (code != TSDB_CODE_SUCCESS) {
61,055,804!
1812
      pTask->code = code;
×
1813
      return pTask->code;
×
1814
    }
1815
  }
1816
  return TSDB_CODE_SUCCESS;
61,055,804✔
1817
}
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc