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

16
#include "os.h"
H
hjxilinx 已提交
17 18 19
#include "tutil.h"

#include "tnote.h"
H
hzcheng 已提交
20
#include "trpc.h"
S
slguan 已提交
21
#include "tscLog.h"
H
hzcheng 已提交
22
#include "tscProfile.h"
H
hjxilinx 已提交
23
#include "tscSubquery.h"
H
hzcheng 已提交
24 25
#include "tscSecondaryMerge.h"
#include "tscUtil.h"
S
slguan 已提交
26
#include "tsched.h"
H
hjxilinx 已提交
27
#include "tschemautil.h"
H
hjxilinx 已提交
28
#include "tsclient.h"
H
hzcheng 已提交
29

30 31
static void tscProcessFetchRow(SSchedMsg *pMsg);
static void tscAsyncQueryRowsForNextVnode(void *param, TAOS_RES *tres, int numOfRows);
H
hzcheng 已提交
32 33 34 35

static void tscProcessAsyncRetrieveImpl(void *param, TAOS_RES *tres, int numOfRows, void (*fp)());

/*
36 37 38
 * Proxy function to perform sequentially query&retrieve operation.
 * If sql queries upon a super table and two-stage merge procedure is not involved (when employ the projection
 * query), it will sequentially query&retrieve data for all vnodes
H
hzcheng 已提交
39
 */
40 41
static void tscAsyncFetchRowsProxy(void *param, TAOS_RES *tres, int numOfRows);
static void tscAsyncFetchSingleRowProxy(void *param, TAOS_RES *tres, int numOfRows);
H
hzcheng 已提交
42

H
[td-99]  
hjxilinx 已提交
43
void doAsyncQuery(STscObj* pObj, SSqlObj* pSql, void (*fp)(), void* param, const char* sqlstr, size_t sqlLen) {
H
hzcheng 已提交
44 45
  SSqlCmd *pCmd = &pSql->cmd;
  SSqlRes *pRes = &pSql->res;
46
  
H
hzcheng 已提交
47
  pSql->signature = pSql;
48
  pSql->param = param;
H
hzcheng 已提交
49
  pSql->pTscObj = pObj;
S
slguan 已提交
50
  pSql->maxRetry = TSDB_MAX_REPLICA_NUM;
H
hzcheng 已提交
51
  pSql->fp = fp;
52
  
S
slguan 已提交
53 54
  if (TSDB_CODE_SUCCESS != tscAllocPayload(pCmd, TSDB_DEFAULT_PAYLOAD_SIZE)) {
    tscError("failed to malloc payload");
H
[td-99]  
hjxilinx 已提交
55
    tscQueueAsyncError(fp, param, TSDB_CODE_CLI_OUT_OF_MEMORY);
H
hzcheng 已提交
56 57
    return;
  }
58
  
59
  pSql->sqlstr = realloc(pSql->sqlstr, sqlLen + 1);
H
hzcheng 已提交
60 61
  if (pSql->sqlstr == NULL) {
    tscError("%p failed to malloc sql string buffer", pSql);
H
[td-99]  
hjxilinx 已提交
62
    tscQueueAsyncError(fp, param, TSDB_CODE_CLI_OUT_OF_MEMORY);
63
    free(pCmd->payload);
H
hzcheng 已提交
64 65
    return;
  }
66
  
H
hzcheng 已提交
67 68
  pRes->qhandle = 0;
  pRes->numOfRows = 1;
69
  
S
slguan 已提交
70
  strtolower(pSql->sqlstr, sqlstr);
71
  tscDump("%p SQL: %s", pSql, pSql->sqlstr);
72
  
73
  int32_t code = tsParseSql(pSql, true);
H
hzcheng 已提交
74
  if (code == TSDB_CODE_ACTION_IN_PROGRESS) return;
75
  
H
hzcheng 已提交
76
  if (code != TSDB_CODE_SUCCESS) {
H
[td-99]  
hjxilinx 已提交
77
    pSql->res.code = code;
H
hzcheng 已提交
78 79 80
    tscQueueAsyncRes(pSql);
    return;
  }
81
  
H
hzcheng 已提交
82 83 84
  tscDoQuery(pSql);
}

85
// TODO return the correct error code to client in tscQueueAsyncError
H
hjxilinx 已提交
86
void taos_query_a(TAOS *taos, const char *sqlstr, __async_cb_func_t fp, void *param) {
87 88 89
  STscObj *pObj = (STscObj *)taos;
  if (pObj == NULL || pObj->signature != pObj) {
    tscError("bug!!! pObj:%p", pObj);
H
[td-99]  
hjxilinx 已提交
90 91
    terrno = TSDB_CODE_DISCONNECTED;
    tscQueueAsyncError(fp, param, TSDB_CODE_DISCONNECTED);
92 93 94 95 96 97
    return;
  }
  
  int32_t sqlLen = strlen(sqlstr);
  if (sqlLen > tsMaxSQLStringLen) {
    tscError("sql string too long");
H
[td-99]  
hjxilinx 已提交
98 99
    terrno = TSDB_CODE_INVALID_SQL;
    tscQueueAsyncError(fp, param, TSDB_CODE_INVALID_SQL);
100 101 102 103 104 105 106 107
    return;
  }
  
  taosNotePrintTsc(sqlstr);
  
  SSqlObj *pSql = (SSqlObj *)calloc(1, sizeof(SSqlObj));
  if (pSql == NULL) {
    tscError("failed to malloc sqlObj");
H
[td-99]  
hjxilinx 已提交
108 109
    terrno = TSDB_CODE_CLI_OUT_OF_MEMORY;
    tscQueueAsyncError(fp, param, TSDB_CODE_CLI_OUT_OF_MEMORY);
110 111 112 113 114 115
    return;
  }
  
  doAsyncQuery(pObj, pSql, fp, param, sqlstr, sqlLen);
}

116
static void tscAsyncFetchRowsProxy(void *param, TAOS_RES *tres, int numOfRows) {
H
hzcheng 已提交
117 118 119 120 121 122 123 124
  if (tres == NULL) {
    return;
  }

  SSqlObj *pSql = (SSqlObj *)tres;
  SSqlRes *pRes = &pSql->res;
  SSqlCmd *pCmd = &pSql->cmd;

125 126 127 128
  if (numOfRows == 0) {
    if (hasMoreVnodesToTry(pSql)) { // sequentially retrieve data from remain vnodes.
      tscTryQueryNextVnode(pSql, tscAsyncQueryRowsForNextVnode);
    } else {
129 130 131 132 133 134 135 136 137
      /*
       * all available virtual node has been checked already, now we need to check
       * for the next subclause queries
       */
      if (pCmd->clauseIndex < pCmd->numOfClause - 1) {
        tscTryQueryNextClause(pSql, tscAsyncQueryRowsForNextVnode);
        return;
      }

138 139 140 141 142
      /*
       * 1. has reach the limitation
       * 2. no remain virtual nodes to be retrieved anymore
       */
      (*pSql->fetchFp)(param, pSql, 0);
H
hzcheng 已提交
143
    }
144 145 146 147 148 149
    
    return;
  }
  
  // local reducer has handle this situation during super table non-projection query.
  if (pCmd->command != TSDB_SQL_RETRIEVE_METRIC) {
150
    pRes->numOfTotalInCurrentClause += pRes->numOfRows;
H
hzcheng 已提交
151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166
  }

  (*pSql->fetchFp)(param, tres, numOfRows);
}

// actual continue retrieve function with user-specified callback function
static void tscProcessAsyncRetrieveImpl(void *param, TAOS_RES *tres, int numOfRows, void (*fp)()) {
  SSqlObj *pSql = (SSqlObj *)tres;
  if (pSql == NULL) {  // error
    tscError("sql object is NULL");
    return;
  }

  SSqlCmd *pCmd = &pSql->cmd;
  SSqlRes *pRes = &pSql->res;

167
  if ((pRes->qhandle == 0 || numOfRows != 0) && pCmd->command < TSDB_SQL_LOCAL) {
H
hzcheng 已提交
168 169 170 171 172 173
    if (pRes->qhandle == 0) {
      tscError("qhandle is NULL");
    } else {
      pRes->code = numOfRows;
    }

H
[td-99]  
hjxilinx 已提交
174
    tscQueueAsyncError(pSql->fetchFp, param, pRes->code);
H
hzcheng 已提交
175 176 177 178 179 180 181 182 183 184 185
    return;
  }

  pSql->fp = fp;
  if (pCmd->command != TSDB_SQL_RETRIEVE_METRIC && pCmd->command < TSDB_SQL_LOCAL) {
    pCmd->command = (pCmd->command > TSDB_SQL_MGMT) ? TSDB_SQL_RETRIEVE : TSDB_SQL_FETCH;
  }
  tscProcessSql(pSql);
}

/*
186 187 188
 * retrieve callback for fetch rows proxy.
 * The below two functions both serve as the callback function of query virtual node.
 * query callback first, and then followed by retrieve callback
H
hzcheng 已提交
189
 */
190 191
static void tscAsyncQueryRowsForNextVnode(void *param, TAOS_RES *tres, int numOfRows) {
  // query completed, continue to retrieve
192
  tscProcessAsyncRetrieveImpl(param, tres, numOfRows, tscAsyncFetchRowsProxy);
H
hzcheng 已提交
193 194
}

195 196
void tscAsyncQuerySingleRowForNextVnode(void *param, TAOS_RES *tres, int numOfRows) {
  // query completed, continue to retrieve
197
  tscProcessAsyncRetrieveImpl(param, tres, numOfRows, tscAsyncFetchSingleRowProxy);
H
hzcheng 已提交
198 199 200 201 202 203
}

void taos_fetch_rows_a(TAOS_RES *taosa, void (*fp)(void *, TAOS_RES *, int), void *param) {
  SSqlObj *pSql = (SSqlObj *)taosa;
  if (pSql == NULL || pSql->signature != pSql) {
    tscError("sql object is NULL");
H
[td-99]  
hjxilinx 已提交
204
    tscQueueAsyncError(fp, param, TSDB_CODE_DISCONNECTED);
H
hzcheng 已提交
205 206 207 208 209 210 211 212
    return;
  }

  SSqlRes *pRes = &pSql->res;
  SSqlCmd *pCmd = &pSql->cmd;

  if (pRes->qhandle == 0) {
    tscError("qhandle is NULL");
H
[td-99]  
hjxilinx 已提交
213
    tscQueueAsyncError(fp, param, TSDB_CODE_INVALID_QHANDLE);
H
hzcheng 已提交
214 215 216 217 218
    return;
  }

  // user-defined callback function is stored in fetchFp
  pSql->fetchFp = fp;
219
  pSql->fp = tscAsyncFetchRowsProxy;
H
hzcheng 已提交
220 221

  pSql->param = param;
S
slguan 已提交
222
  tscResetForNextRetrieve(pRes);
H
hjxilinx 已提交
223 224 225 226 227 228 229 230 231 232
  
  // handle the sub queries of join query
  if (pCmd->command == TSDB_SQL_METRIC_JOIN_RETRIEVE) {
    tscFetchDatablockFromSubquery(pSql);
  } else {
    if (pCmd->command != TSDB_SQL_RETRIEVE_METRIC && pCmd->command < TSDB_SQL_LOCAL) {
      pCmd->command = (pCmd->command > TSDB_SQL_MGMT) ? TSDB_SQL_RETRIEVE : TSDB_SQL_FETCH;
    }
  
    tscProcessSql(pSql);
H
hzcheng 已提交
233 234 235 236 237 238 239
  }
}

void taos_fetch_row_a(TAOS_RES *taosa, void (*fp)(void *, TAOS_RES *, TAOS_ROW), void *param) {
  SSqlObj *pSql = (SSqlObj *)taosa;
  if (pSql == NULL || pSql->signature != pSql) {
    tscError("sql object is NULL");
H
[td-99]  
hjxilinx 已提交
240
    tscQueueAsyncError(fp, param, TSDB_CODE_DISCONNECTED);
H
hzcheng 已提交
241 242 243 244 245 246 247 248
    return;
  }

  SSqlRes *pRes = &pSql->res;
  SSqlCmd *pCmd = &pSql->cmd;

  if (pRes->qhandle == 0) {
    tscError("qhandle is NULL");
H
[td-99]  
hjxilinx 已提交
249
    tscQueueAsyncError(fp, param, TSDB_CODE_INVALID_QHANDLE);
H
hzcheng 已提交
250 251 252 253 254
    return;
  }

  pSql->fetchFp = fp;
  pSql->param = param;
255
  
H
hzcheng 已提交
256
  if (pRes->row >= pRes->numOfRows) {
S
slguan 已提交
257
    tscResetForNextRetrieve(pRes);
258
    pSql->fp = tscAsyncFetchSingleRowProxy;
259 260 261 262 263
    
    if (pCmd->command != TSDB_SQL_RETRIEVE_METRIC && pCmd->command < TSDB_SQL_LOCAL) {
      pCmd->command = (pCmd->command > TSDB_SQL_MGMT) ? TSDB_SQL_RETRIEVE : TSDB_SQL_FETCH;
    }
    
H
hzcheng 已提交
264 265 266 267 268 269 270 271 272 273 274
    tscProcessSql(pSql);
  } else {
    SSchedMsg schedMsg;
    schedMsg.fp = tscProcessFetchRow;
    schedMsg.ahandle = pSql;
    schedMsg.thandle = pRes->tsrow;
    schedMsg.msg = NULL;
    taosScheduleTask(tscQhandle, &schedMsg);
  }
}

275
void tscAsyncFetchSingleRowProxy(void *param, TAOS_RES *tres, int numOfRows) {
H
hzcheng 已提交
276 277 278 279
  SSqlObj *pSql = (SSqlObj *)tres;
  SSqlRes *pRes = &pSql->res;
  SSqlCmd *pCmd = &pSql->cmd;

280
  SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(pCmd, pCmd->clauseIndex);
281
  
H
hzcheng 已提交
282
  if (numOfRows == 0) {
283 284 285
    if (hasMoreVnodesToTry(pSql)) {     // sequentially retrieve data from remain vnodes.
      tscTryQueryNextVnode(pSql, tscAsyncQuerySingleRowForNextVnode);
    } else {
H
hzcheng 已提交
286
      /*
287 288
       * 1. has reach the limitation
       * 2. no remain virtual nodes to be retrieved anymore
H
hzcheng 已提交
289 290 291
       */
      (*pSql->fetchFp)(pSql->param, pSql, NULL);
    }
292
    return;
H
hzcheng 已提交
293
  }
294
  
295
  for (int i = 0; i < pCmd->numOfCols; ++i){
H
hjxilinx 已提交
296 297 298
    SFieldSupInfo* pSup = taosArrayGet(pQueryInfo->fieldsInfo.pSupportInfo, i);
    if (pSup->pSqlExpr != NULL) {
//      pRes->tsrow[i] = TSC_GET_RESPTR_BASE(pRes, pQueryInfo, i) + pSup->pSqlExpr->resBytes * pRes->row;
299 300 301 302 303
    } else {
      //todo add
    }
  }
  
304 305 306
  pRes->row++;

  (*pSql->fetchFp)(pSql->param, pSql, pSql->res.tsrow);
H
hzcheng 已提交
307 308 309 310 311 312
}

void tscProcessFetchRow(SSchedMsg *pMsg) {
  SSqlObj *pSql = (SSqlObj *)pMsg->ahandle;
  SSqlRes *pRes = &pSql->res;
  SSqlCmd *pCmd = &pSql->cmd;
313
  
H
hjxilinx 已提交
314
  SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(pCmd, pCmd->clauseIndex);
H
hzcheng 已提交
315

316
  for (int i = 0; i < pCmd->numOfCols; ++i) {
H
hjxilinx 已提交
317 318 319 320
    SFieldSupInfo* pSup = taosArrayGet(pQueryInfo->fieldsInfo.pSupportInfo, i);

    if (pSup->pSqlExpr != NULL) {
      pRes->tsrow[i] = tscGetResultColumnChr(pRes, pQueryInfo, i);
321
    } else {
H
hjxilinx 已提交
322
//      todo add
323
    }
324 325
  }
  
H
hzcheng 已提交
326 327 328 329 330 331 332 333 334 335 336 337 338
  pRes->row++;
  (*pSql->fetchFp)(pSql->param, pSql, pRes->tsrow);
}

void tscProcessAsyncRes(SSchedMsg *pMsg) {
  SSqlObj *pSql = (SSqlObj *)pMsg->ahandle;
  SSqlCmd *pCmd = &pSql->cmd;
  SSqlRes *pRes = &pSql->res;

  void *taosres = pSql;

  // pCmd may be released, so cache pCmd->command
  int cmd = pCmd->command;
H
[TD-98]  
hjxilinx 已提交
339
  int code = pRes->code;
H
hzcheng 已提交
340 341

  // in case of async insert, restore the user specified callback function
H
hjxilinx 已提交
342
  bool shouldFree = tscShouldBeFreed(pSql);
H
hzcheng 已提交
343 344 345 346 347 348 349 350 351

  if (cmd == TSDB_SQL_INSERT) {
    assert(pSql->fp != NULL);
    pSql->fp = pSql->fetchFp;
  }

  (*pSql->fp)(pSql->param, taosres, code);

  if (shouldFree) {
352
    tscTrace("%p sqlObj is automatically freed in async res", pSql);
sangshuduo's avatar
sangshuduo 已提交
353
    tscFreeSqlObj(pSql);
H
hzcheng 已提交
354 355 356
  }
}

H
[td-99]  
hjxilinx 已提交
357
static void tscProcessAsyncError(SSchedMsg *pMsg) {
H
hzcheng 已提交
358
  void (*fp)() = pMsg->ahandle;
H
[td-99]  
hjxilinx 已提交
359
  (*fp)(pMsg->thandle, NULL, *(int32_t*)pMsg->msg);
H
hzcheng 已提交
360 361
}

H
[td-99]  
hjxilinx 已提交
362 363 364 365
void tscQueueAsyncError(void(*fp), void *param, int32_t code) {
  int32_t* c = malloc(sizeof(int32_t));
  *c = code;
  
H
hzcheng 已提交
366 367 368 369
  SSchedMsg schedMsg;
  schedMsg.fp = tscProcessAsyncError;
  schedMsg.ahandle = fp;
  schedMsg.thandle = param;
H
[td-99]  
hjxilinx 已提交
370
  schedMsg.msg = c;
H
hzcheng 已提交
371 372 373 374 375 376 377 378
  taosScheduleTask(tscQhandle, &schedMsg);
}

void tscQueueAsyncRes(SSqlObj *pSql) {
  if (pSql == NULL || pSql->signature != pSql) {
    tscTrace("%p SqlObj is freed, not add into queue async res", pSql);
    return;
  } else {
H
[td-32]  
hjxilinx 已提交
379
    tscError("%p add into queued async res, code:%s", pSql, tstrerror(pSql->res.code));
H
hzcheng 已提交
380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408
  }

  SSchedMsg schedMsg;
  schedMsg.fp = tscProcessAsyncRes;
  schedMsg.ahandle = pSql;
  schedMsg.thandle = (void *)1;
  schedMsg.msg = NULL;
  taosScheduleTask(tscQhandle, &schedMsg);
}

void tscProcessAsyncFree(SSchedMsg *pMsg) {
  SSqlObj *pSql = (SSqlObj *)pMsg->ahandle;
  tscTrace("%p sql is freed", pSql);
  taos_free_result(pSql);
}

void tscQueueAsyncFreeResult(SSqlObj *pSql) {
  tscTrace("%p sqlObj put in queue to async free", pSql);

  SSchedMsg schedMsg;
  schedMsg.fp = tscProcessAsyncFree;
  schedMsg.ahandle = pSql;
  schedMsg.thandle = (void *)1;
  schedMsg.msg = NULL;
  taosScheduleTask(tscQhandle, &schedMsg);
}

int tscSendMsgToServer(SSqlObj *pSql);

H
hjxilinx 已提交
409
void tscTableMetaCallBack(void *param, TAOS_RES *res, int code) {
H
hzcheng 已提交
410 411 412 413 414 415
  SSqlObj *pSql = (SSqlObj *)param;
  if (pSql == NULL || pSql->signature != pSql) return;

  SSqlCmd *pCmd = &pSql->cmd;
  SSqlRes *pRes = &pSql->res;

H
[td-32]  
hjxilinx 已提交
416 417
  if (code != TSDB_CODE_SUCCESS) {
    pRes->code = code;
H
hzcheng 已提交
418 419 420 421 422
    tscQueueAsyncRes(pSql);
    return;
  }

  if (pSql->pStream == NULL) {
H
hjxilinx 已提交
423
    // check if it is a sub-query of super table query first, if true, enter another routine
H
hjxilinx 已提交
424
    SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, pCmd->clauseIndex);
425 426
  
    if ((pQueryInfo->type & TSDB_QUERY_TYPE_STABLE_SUBQUERY) == TSDB_QUERY_TYPE_STABLE_SUBQUERY) {
H
hjxilinx 已提交
427
      STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
H
hjxilinx 已提交
428
      assert((tscGetNumOfTags(pTableMetaInfo->pTableMeta) != 0) && pTableMetaInfo->vgroupIndex >= 0 && pSql->param != NULL);
H
hzcheng 已提交
429 430 431

      SRetrieveSupport *trs = (SRetrieveSupport *)pSql->param;
      SSqlObj *         pParObj = trs->pParentSqlObj;
432
      
H
hjxilinx 已提交
433
      assert(pParObj->signature == pParObj && trs->subqueryIndex == pTableMetaInfo->vgroupIndex &&
H
hjxilinx 已提交
434
          tscGetNumOfTags(pTableMetaInfo->pTableMeta) != 0);
S
slguan 已提交
435

H
hjxilinx 已提交
436
      tscTrace("%p get metricMeta during super table query successfully", pSql);
437
      
H
hjxilinx 已提交
438
      code = tscGetTableMeta(pSql, pTableMetaInfo);
H
hzcheng 已提交
439 440 441 442
      pRes->code = code;

      if (code == TSDB_CODE_ACTION_IN_PROGRESS) return;

H
hjxilinx 已提交
443
      code = tscGetSTableVgroupInfo(pSql, 0);
H
hzcheng 已提交
444 445 446 447
      pRes->code = code;

      if (code == TSDB_CODE_ACTION_IN_PROGRESS) return;
    } else {  // normal async query continues
448
      if (pCmd->parseFinished) {
449
        tscTrace("%p re-send data to vnode in table Meta callback since sql parsed completed", pSql);
L
[#1083]  
lihui 已提交
450
        
451
        STableMetaInfo* pTableMetaInfo = tscGetTableMetaInfoFromCmd(pCmd, pCmd->clauseIndex, 0);
H
hjxilinx 已提交
452
        code = tscGetTableMeta(pSql, pTableMetaInfo);
L
[#1083]  
lihui 已提交
453 454
        assert(code == TSDB_CODE_SUCCESS);
      
H
hjxilinx 已提交
455
        if (pTableMetaInfo->pTableMeta) {
456 457
          // todo update the submit message according to the new table meta
          // 1. table uid, 2. ip address
L
[#1083]  
lihui 已提交
458 459 460 461
          code = tscSendMsgToServer(pSql);
          if (code == TSDB_CODE_SUCCESS) return;
        }
      } else {
H
hjxilinx 已提交
462
        code = tsParseSql(pSql, false);
L
[#1083]  
lihui 已提交
463 464
        if (code == TSDB_CODE_ACTION_IN_PROGRESS) return;
      }
H
hzcheng 已提交
465
    }
S
slguan 已提交
466

H
hzcheng 已提交
467
  } else {  // stream computing
468
    STableMetaInfo *pTableMetaInfo = tscGetTableMetaInfoFromCmd(pCmd, pCmd->clauseIndex, 0);
H
hjxilinx 已提交
469
    code = tscGetTableMeta(pSql, pTableMetaInfo);
H
hzcheng 已提交
470 471 472 473
    pRes->code = code;

    if (code == TSDB_CODE_ACTION_IN_PROGRESS) return;

H
hjxilinx 已提交
474
    if (code == TSDB_CODE_SUCCESS && UTIL_TABLE_IS_SUPERTABLE(pTableMetaInfo)) {
H
hjxilinx 已提交
475
      code = tscGetSTableVgroupInfo(pSql, pCmd->clauseIndex);
H
hzcheng 已提交
476 477 478 479 480 481
      pRes->code = code;

      if (code == TSDB_CODE_ACTION_IN_PROGRESS) return;
    }
  }

482 483
  if (code != TSDB_CODE_SUCCESS) {
    pSql->res.code = code;
H
hzcheng 已提交
484 485 486 487 488 489 490 491
    tscQueueAsyncRes(pSql);
    return;
  }

  if (pSql->pStream) {
    tscTrace("%p stream:%p meta is updated, start new query, command:%d", pSql, pSql->pStream, pSql->cmd.command);
    /*
     * NOTE:
H
hjxilinx 已提交
492
     * transfer the sql function for super table query before get meter/metric meta,
493
     * since in callback functions, only tscProcessSql(pStream->pSql) is executed!
H
hzcheng 已提交
494
     */
H
hjxilinx 已提交
495
    SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, pCmd->clauseIndex);
496
    
H
hjxilinx 已提交
497
    tscTansformSQLFuncForSTableQuery(pQueryInfo);
H
hzcheng 已提交
498 499
    tscIncStreamExecutionCount(pSql->pStream);
  } else {
H
hjxilinx 已提交
500
    tscTrace("%p get tableMeta successfully", pSql);
H
hzcheng 已提交
501 502 503 504
  }

  tscDoQuery(pSql);
}