#3660

Committed 15 Mar 2025 09:06AM UTC coverage: 62.039% (-1.3%) from 63.314%

Build # #3660

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

feat(stream): support stream processing for virtual tables (#30144)

* enh: add client processing

* enh: add mnode vtables processing

* enh: add mnode vtable processing

* enh: add normal child vtable support

* fix: compile issues

* fix: compile issues

* fix: create stream issues

* fix: multi stream scan issue

* fix: remove debug info

* fix: agg task and task level issues

* fix: correct task output type

* fix: split vtablescan from agg

* fix: memory leak issues

* fix: add limitations

* Update 09-error-code.md

* Update 09-error-code.md

* fix: remove usless case

* feat(stream): extract original table data in source scan task

Implemented functionality in the source task to extract data
corresponding to the virtual table from the original table using WAL.
The extracted data is then sent to the downstream merge task for further
processing.

* feat(stream): multi-way merge using loser tree in virtual merge task

Implemented multi-way merge in the merge task using a loser tree to
combine data from multiple original table into a single virtual table.
The merged virtual table data is then pushed downstream for further
processing.  Introduced memory limit handling during the merge process
with configurable behavior when the memory limit is reached.

* fix(test): remove useless cases

---------

Co-authored-by: dapan1121 <wpan@taosdata.com>
Co-authored-by: Pan Wei <72057773+dapan1121@users.noreply.github.com>

Run Details

154078 of 317582 branches covered (48.52%)

Branch coverage included in aggregate %.

313 of 2391 new or added lines in 34 files covered. (13.09%)

26134 existing lines in 205 files now uncovered.

240261 of 318051 relevant lines covered (75.54%)

16655189.27 hits per line

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

57.07

/source/client/src/clientMain.c

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

#include "catalog.h"
#include "clientInt.h"
#include "clientLog.h"
#include "clientMonitor.h"
#include "clientStmt.h"
#include "clientStmt2.h"
#include "functionMgt.h"
#include "os.h"
#include "query.h"
#include "scheduler.h"
#include "tcompare.h"
#include "tconv.h"
#include "tdatablock.h"
#include "tglobal.h"
#include "tmsg.h"
#include "tref.h"
#include "trpc.h"
#include "version.h"

#define TSC_VAR_NOT_RELEASE 1
#define TSC_VAR_RELEASED    0

#ifdef TAOSD_INTEGRATED
extern void shellStopDaemon();
#endif

static int32_t sentinel = TSC_VAR_NOT_RELEASE;
static int32_t createParseContext(const SRequestObj *pRequest, SParseContext **pCxt, SSqlCallbackWrapper *pWrapper);

int taos_options(TSDB_OPTION option, const void *arg, ...) {
  if (arg == NULL) {
    return TSDB_CODE_INVALID_PARA;
  }
  static int32_t lock = 0;

  for (int i = 1; atomic_val_compare_exchange_32(&lock, 0, 1) != 0; ++i) {
    if (i % 1000 == 0) {
      (void)sched_yield();
    }
  }

  int ret = taos_options_imp(option, (const char *)arg);
  atomic_store_32(&lock, 0);
  return ret;
}

#if !defined(WINDOWS) && !defined(TD_ASTRA)
static void freeTz(void *p) {
  timezone_t tz = *(timezone_t *)p;
  tzfree(tz);
}

int32_t tzInit() {
  pTimezoneMap = taosHashInit(0, MurmurHash3_32, false, HASH_ENTRY_LOCK);
  if (pTimezoneMap == NULL) {
    return terrno;
  }
  taosHashSetFreeFp(pTimezoneMap, freeTz);

  pTimezoneNameMap = taosHashInit(0, taosIntHash_64, false, HASH_ENTRY_LOCK);
  if (pTimezoneNameMap == NULL) {
    return terrno;
  }
  return 0;
}

void tzCleanup() {
  taosHashCleanup(pTimezoneMap);
  taosHashCleanup(pTimezoneNameMap);
}

static timezone_t setConnnectionTz(const char *val) {
  timezone_t  tz = NULL;
  timezone_t *tmp = taosHashGet(pTimezoneMap, val, strlen(val));
  if (tmp != NULL && *tmp != NULL) {
    tz = *tmp;
    goto END;
  }

  tscDebug("set timezone to %s", val);
  tz = tzalloc(val);
  if (tz == NULL) {
    tscWarn("%s unknown timezone %s change to UTC", __func__, val);
    tz = tzalloc("UTC");
    if (tz == NULL) {
      tscError("%s set timezone UTC error", __func__);
      terrno = TAOS_SYSTEM_ERROR(ERRNO);
      goto END;
    }
  }
  int32_t code = taosHashPut(pTimezoneMap, val, strlen(val), &tz, sizeof(timezone_t));
  if (code != 0) {
    tscError("%s put timezone to tz map error:%d", __func__, code);
    tzfree(tz);
    tz = NULL;
    goto END;
  }

  time_t tx1 = taosGetTimestampSec();
  char   output[TD_TIMEZONE_LEN] = {0};
  code = taosFormatTimezoneStr(tx1, val, tz, output);
  if (code == 0) {
    code = taosHashPut(pTimezoneNameMap, &tz, sizeof(timezone_t), output, strlen(output) + 1);
  }
  if (code != 0) {
    tscError("failed to put timezone %s to map", val);
  }

END:
  return tz;
}
#endif

static int32_t setConnectionOption(TAOS *taos, TSDB_OPTION_CONNECTION option, const char *val) {
  if (taos == NULL) {
    return terrno = TSDB_CODE_INVALID_PARA;
  }

#ifdef WINDOWS
  if (option == TSDB_OPTION_CONNECTION_TIMEZONE) {
    return terrno = TSDB_CODE_NOT_SUPPORTTED_IN_WINDOWS;
  }
#endif

  if (option < TSDB_OPTION_CONNECTION_CLEAR || option >= TSDB_MAX_OPTIONS_CONNECTION) {
    return terrno = TSDB_CODE_INVALID_PARA;
  }

  int32_t code = taos_init();
  // initialize global config
  if (code != 0) {
    return terrno = code;
  }

  STscObj *pObj = acquireTscObj(*(int64_t *)taos);
  if (NULL == pObj) {
    tscError("invalid parameter for %s", __func__);
    return terrno;
  }

  if (option == TSDB_OPTION_CONNECTION_CLEAR) {
    val = NULL;
  }

#ifndef DISALLOW_NCHAR_WITHOUT_ICONV
  if (option == TSDB_OPTION_CONNECTION_CHARSET || option == TSDB_OPTION_CONNECTION_CLEAR) {
    if (val != NULL) {
      if (!taosValidateEncodec(val)) {
        code = terrno;
        goto END;
      }
      void *tmp = taosConvInit(val);
      if (tmp == NULL) {
        code = terrno;
        goto END;
      }
      pObj->optionInfo.charsetCxt = tmp;
    } else {
      pObj->optionInfo.charsetCxt = NULL;
    }
  }
#endif
  if (option == TSDB_OPTION_CONNECTION_TIMEZONE || option == TSDB_OPTION_CONNECTION_CLEAR) {
#if !defined(WINDOWS) && !defined(TD_ASTRA)
    if (val != NULL) {
      if (val[0] == 0) {
        val = "UTC";
      }
      timezone_t tz = setConnnectionTz(val);
      if (tz == NULL) {
        code = terrno;
        goto END;
      }
      pObj->optionInfo.timezone = tz;
    } else {
      pObj->optionInfo.timezone = NULL;
    }
#endif
  }

  if (option == TSDB_OPTION_CONNECTION_USER_APP || option == TSDB_OPTION_CONNECTION_CLEAR) {
    if (val != NULL) {
      tstrncpy(pObj->optionInfo.userApp, val, sizeof(pObj->optionInfo.userApp));
    } else {
      pObj->optionInfo.userApp[0] = 0;
    }
  }

  if (option == TSDB_OPTION_CONNECTION_USER_IP || option == TSDB_OPTION_CONNECTION_CLEAR) {
    if (val != NULL) {
      pObj->optionInfo.userIp = taosInetAddr(val);
      if (pObj->optionInfo.userIp == INADDR_NONE) {
        code = TSDB_CODE_INVALID_PARA;
        goto END;
      }
    } else {
      pObj->optionInfo.userIp = INADDR_NONE;
    }
  }

END:
  releaseTscObj(*(int64_t *)taos);
  return terrno = code;
}

int taos_options_connection(TAOS *taos, TSDB_OPTION_CONNECTION option, const void *arg, ...) {
  return setConnectionOption(taos, option, (const char *)arg);
}

// this function may be called by user or system, or by both simultaneously.
void taos_cleanup(void) {
  tscDebug("start to cleanup client environment");
  if (atomic_val_compare_exchange_32(&sentinel, TSC_VAR_NOT_RELEASE, TSC_VAR_RELEASED) != TSC_VAR_NOT_RELEASE) {
    return;
  }

  monitorClose();
  tscStopCrashReport();

  hbMgrCleanUp();

  catalogDestroy();
  schedulerDestroy();

  fmFuncMgtDestroy();
  qCleanupKeywordsTable();

  if (TSDB_CODE_SUCCESS != cleanupTaskQueue()) {
    tscWarn("failed to cleanup task queue");
  }

#if !defined(WINDOWS) && !defined(TD_ASTRA)
  tzCleanup();
#endif
  tmqMgmtClose();

  int32_t id = clientReqRefPool;
  clientReqRefPool = -1;
  taosCloseRef(id);

  id = clientConnRefPool;
  clientConnRefPool = -1;
  taosCloseRef(id);

  nodesDestroyAllocatorSet();
  cleanupAppInfo();
  rpcCleanup();
  tscDebug("rpc cleanup");

  taosConvDestroy();
  DestroyRegexCache();
#ifdef TAOSD_INTEGRATED
  shellStopDaemon();
#endif
  tscInfo("all local resources released");
  taosCleanupCfg();
#ifndef TAOSD_INTEGRATED
  taosCloseLog();
#endif
}

static setConfRet taos_set_config_imp(const char *config) {
  setConfRet ret = {SET_CONF_RET_SUCC, {0}};
  // TODO: need re-implementation
  return ret;
}

setConfRet taos_set_config(const char *config) {
  // TODO  pthread_mutex_lock(&setConfMutex);
  setConfRet ret = taos_set_config_imp(config);
  //  pthread_mutex_unlock(&setConfMutex);
  return ret;
}

TAOS *taos_connect(const char *ip, const char *user, const char *pass, const char *db, uint16_t port) {
  tscInfo("try to connect to %s:%u, user:%s db:%s", ip, port, user, db);
  if (user == NULL) {
    user = TSDB_DEFAULT_USER;
  }

  if (pass == NULL) {
    pass = TSDB_DEFAULT_PASS;
  }

  STscObj *pObj = NULL;
  int32_t  code = taos_connect_internal(ip, user, pass, NULL, db, port, CONN_TYPE__QUERY, &pObj);
  if (TSDB_CODE_SUCCESS == code) {
    int64_t *rid = taosMemoryCalloc(1, sizeof(int64_t));
    if (NULL == rid) {
      tscError("out of memory when taos connect to %s:%u, user:%s db:%s", ip, port, user, db);
      return NULL;
    }
    *rid = pObj->id;
    return (TAOS *)rid;
  } else {
    terrno = code;
  }

  return NULL;
}

int taos_set_notify_cb(TAOS *taos, __taos_notify_fn_t fp, void *param, int type) {
  if (taos == NULL) {
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

  STscObj *pObj = acquireTscObj(*(int64_t *)taos);
  if (NULL == pObj) {
    terrno = TSDB_CODE_TSC_DISCONNECTED;
    tscError("invalid parameter for %s", __func__);
    return terrno;
  }

  switch (type) {
    case TAOS_NOTIFY_PASSVER: {
      TSC_ERR_RET(taosThreadMutexLock(&pObj->mutex));
      pObj->passInfo.fp = fp;
      pObj->passInfo.param = param;
      TSC_ERR_RET(taosThreadMutexUnlock(&pObj->mutex));
      break;
    }
    case TAOS_NOTIFY_WHITELIST_VER: {
      TSC_ERR_RET(taosThreadMutexLock(&pObj->mutex));
      pObj->whiteListInfo.fp = fp;
      pObj->whiteListInfo.param = param;
      TSC_ERR_RET(taosThreadMutexUnlock(&pObj->mutex));
      break;
    }
    case TAOS_NOTIFY_USER_DROPPED: {
      TSC_ERR_RET(taosThreadMutexLock(&pObj->mutex));
      pObj->userDroppedInfo.fp = fp;
      pObj->userDroppedInfo.param = param;
      TSC_ERR_RET(taosThreadMutexUnlock(&pObj->mutex));
      break;
    }
    default: {
      terrno = TSDB_CODE_INVALID_PARA;
      releaseTscObj(*(int64_t *)taos);
      return terrno;
    }
  }

  releaseTscObj(*(int64_t *)taos);
  return 0;
}

typedef struct SFetchWhiteListInfo {
  int64_t                     connId;
  __taos_async_whitelist_fn_t userCbFn;
  void                       *userParam;
} SFetchWhiteListInfo;

int32_t fetchWhiteListCallbackFn(void *param, SDataBuf *pMsg, int32_t code) {
  SFetchWhiteListInfo *pInfo = (SFetchWhiteListInfo *)param;
  TAOS                *taos = &pInfo->connId;
  if (code != TSDB_CODE_SUCCESS) {
    pInfo->userCbFn(pInfo->userParam, code, taos, 0, NULL);
    taosMemoryFree(pMsg->pData);
    taosMemoryFree(pMsg->pEpSet);
    taosMemoryFree(pInfo);
    return code;
  }

  SGetUserWhiteListRsp wlRsp;
  if (TSDB_CODE_SUCCESS != tDeserializeSGetUserWhiteListRsp(pMsg->pData, pMsg->len, &wlRsp)) {
    taosMemoryFree(pMsg->pData);
    taosMemoryFree(pMsg->pEpSet);
    taosMemoryFree(pInfo);
    tFreeSGetUserWhiteListRsp(&wlRsp);
    return terrno;
  }

  uint64_t *pWhiteLists = taosMemoryMalloc(wlRsp.numWhiteLists * sizeof(uint64_t));
  if (pWhiteLists == NULL) {
    taosMemoryFree(pMsg->pData);
    taosMemoryFree(pMsg->pEpSet);
    taosMemoryFree(pInfo);
    tFreeSGetUserWhiteListRsp(&wlRsp);
    return terrno;
  }

  for (int i = 0; i < wlRsp.numWhiteLists; ++i) {
    pWhiteLists[i] = ((uint64_t)wlRsp.pWhiteLists[i].mask << 32) | wlRsp.pWhiteLists[i].ip;
  }

  pInfo->userCbFn(pInfo->userParam, code, taos, wlRsp.numWhiteLists, pWhiteLists);

  taosMemoryFree(pWhiteLists);
  taosMemoryFree(pMsg->pData);
  taosMemoryFree(pMsg->pEpSet);
  taosMemoryFree(pInfo);
  tFreeSGetUserWhiteListRsp(&wlRsp);
  return code;
}

void taos_fetch_whitelist_a(TAOS *taos, __taos_async_whitelist_fn_t fp, void *param) {
  if (NULL == taos) {
    fp(param, TSDB_CODE_INVALID_PARA, taos, 0, NULL);
    return;
  }

  int64_t connId = *(int64_t *)taos;

  STscObj *pTsc = acquireTscObj(connId);
  if (NULL == pTsc) {
    fp(param, TSDB_CODE_TSC_DISCONNECTED, taos, 0, NULL);
    return;
  }

  SGetUserWhiteListReq req;
  (void)memcpy(req.user, pTsc->user, TSDB_USER_LEN);
  int32_t msgLen = tSerializeSGetUserWhiteListReq(NULL, 0, &req);
  if (msgLen < 0) {
    fp(param, TSDB_CODE_INVALID_PARA, taos, 0, NULL);
    releaseTscObj(connId);
    return;
  }

  void *pReq = taosMemoryMalloc(msgLen);
  if (pReq == NULL) {
    fp(param, terrno, taos, 0, NULL);
    releaseTscObj(connId);
    return;
  }

  if (tSerializeSGetUserWhiteListReq(pReq, msgLen, &req) < 0) {
    fp(param, TSDB_CODE_INVALID_PARA, taos, 0, NULL);
    taosMemoryFree(pReq);
    releaseTscObj(connId);
    return;
  }

  SFetchWhiteListInfo *pParam = taosMemoryMalloc(sizeof(SFetchWhiteListInfo));
  if (pParam == NULL) {
    fp(param, terrno, taos, 0, NULL);
    taosMemoryFree(pReq);
    releaseTscObj(connId);
    return;
  }

  pParam->connId = connId;
  pParam->userCbFn = fp;
  pParam->userParam = param;
  SMsgSendInfo *pSendInfo = taosMemoryCalloc(1, sizeof(SMsgSendInfo));
  if (pSendInfo == NULL) {
    fp(param, terrno, taos, 0, NULL);
    taosMemoryFree(pParam);
    taosMemoryFree(pReq);
    releaseTscObj(connId);
    return;
  }

  pSendInfo->msgInfo = (SDataBuf){.pData = pReq, .len = msgLen, .handle = NULL};
  pSendInfo->requestId = generateRequestId();
  pSendInfo->requestObjRefId = 0;
  pSendInfo->param = pParam;
  pSendInfo->fp = fetchWhiteListCallbackFn;
  pSendInfo->msgType = TDMT_MND_GET_USER_WHITELIST;

  SEpSet epSet = getEpSet_s(&pTsc->pAppInfo->mgmtEp);
  if (TSDB_CODE_SUCCESS != asyncSendMsgToServer(pTsc->pAppInfo->pTransporter, &epSet, NULL, pSendInfo)) {
    tscWarn("failed to async send msg to server");
  }
  releaseTscObj(connId);
  return;
}

void taos_close_internal(void *taos) {
  if (taos == NULL) {
    return;
  }

  STscObj *pTscObj = (STscObj *)taos;
  tscDebug("connObj:0x%" PRIx64 ", try to close connection, numOfReq:%d", pTscObj->id, pTscObj->numOfReqs);

  if (TSDB_CODE_SUCCESS != taosRemoveRef(clientConnRefPool, pTscObj->id)) {
    tscError("connObj:0x%" PRIx64 ", failed to remove ref from conn pool", pTscObj->id);
  }
}

void taos_close(TAOS *taos) {
  if (taos == NULL) {
    return;
  }

  STscObj *pObj = acquireTscObj(*(int64_t *)taos);
  if (NULL == pObj) {
    taosMemoryFree(taos);
    return;
  }

  taos_close_internal(pObj);
  releaseTscObj(*(int64_t *)taos);
  taosMemoryFree(taos);
}

int taos_errno(TAOS_RES *res) {
  if (res == NULL || TD_RES_TMQ_RAW(res) || TD_RES_TMQ_META(res) || TD_RES_TMQ_BATCH_META(res)) {
    return terrno;
  }

  if (TD_RES_TMQ(res) || TD_RES_TMQ_METADATA(res)) {
    return 0;
  }

  return ((SRequestObj *)res)->code;
}

const char *taos_errstr(TAOS_RES *res) {
  if (res == NULL || TD_RES_TMQ_RAW(res) || TD_RES_TMQ_META(res) || TD_RES_TMQ_BATCH_META(res)) {
    return (const char *)tstrerror(terrno);
  }

  if (TD_RES_TMQ(res) || TD_RES_TMQ_METADATA(res)) {
    return "success";
  }

  SRequestObj *pRequest = (SRequestObj *)res;
  if (NULL != pRequest->msgBuf && (strlen(pRequest->msgBuf) > 0 || pRequest->code == TSDB_CODE_RPC_FQDN_ERROR)) {
    return pRequest->msgBuf;
  } else {
    return (const char *)tstrerror(pRequest->code);
  }
}

void taos_free_result(TAOS_RES *res) {
  if (NULL == res) {
    return;
  }

  tscTrace("res:%p, will be freed", res);

  if (TD_RES_QUERY(res)) {
    SRequestObj *pRequest = (SRequestObj *)res;
    tscDebug("QID:0x%" PRIx64 ", call taos_free_result to free query", pRequest->requestId);
    destroyRequest(pRequest);
    return;
  }

  SMqRspObj *pRsp = (SMqRspObj *)res;
  if (TD_RES_TMQ(res)) {
    tDeleteMqDataRsp(&pRsp->dataRsp);
    doFreeReqResultInfo(&pRsp->resInfo);
  } else if (TD_RES_TMQ_METADATA(res)) {
    tDeleteSTaosxRsp(&pRsp->dataRsp);
    doFreeReqResultInfo(&pRsp->resInfo);
  } else if (TD_RES_TMQ_META(res)) {
    tDeleteMqMetaRsp(&pRsp->metaRsp);
  } else if (TD_RES_TMQ_BATCH_META(res)) {
    tDeleteMqBatchMetaRsp(&pRsp->batchMetaRsp);
  } else if (TD_RES_TMQ_RAW(res)) {
    tDeleteMqRawDataRsp(&pRsp->dataRsp);
  }
  taosMemoryFree(pRsp);
}

void taos_kill_query(TAOS *taos) {
  if (NULL == taos) {
    return;
  }

  int64_t  rid = *(int64_t *)taos;
  STscObj *pTscObj = acquireTscObj(rid);
  if (pTscObj) {
    stopAllRequests(pTscObj->pRequests);
  }
  releaseTscObj(rid);
}

int taos_field_count(TAOS_RES *res) {
  if (res == NULL || TD_RES_TMQ_RAW(res) || TD_RES_TMQ_META(res) || TD_RES_TMQ_BATCH_META(res)) {
    return 0;
  }

  SReqResultInfo *pResInfo = tscGetCurResInfo(res);
  return pResInfo->numOfCols;
}

int taos_num_fields(TAOS_RES *res) { return taos_field_count(res); }

TAOS_FIELD *taos_fetch_fields(TAOS_RES *res) {
  if (taos_num_fields(res) == 0 || TD_RES_TMQ_RAW(res) || TD_RES_TMQ_META(res) || TD_RES_TMQ_BATCH_META(res)) {
    return NULL;
  }

  SReqResultInfo *pResInfo = tscGetCurResInfo(res);
  return pResInfo->userFields;
}

TAOS_RES *taos_query(TAOS *taos, const char *sql) { return taosQueryImpl(taos, sql, false, TD_REQ_FROM_APP); }
TAOS_RES *taos_query_with_reqid(TAOS *taos, const char *sql, int64_t reqid) {
  return taosQueryImplWithReqid(taos, sql, false, reqid);
}

TAOS_FIELD_E *taos_fetch_fields_e(TAOS_RES *res) {
  if (taos_num_fields(res) == 0 || TD_RES_TMQ_META(res) || TD_RES_TMQ_BATCH_META(res)) {
    return NULL;
  }
  SReqResultInfo* pResInfo = tscGetCurResInfo(res);
  return pResInfo->fields;
}

TAOS_ROW taos_fetch_row(TAOS_RES *res) {
  if (res == NULL) {
    return NULL;
  }

  if (TD_RES_QUERY(res)) {
    SRequestObj *pRequest = (SRequestObj *)res;
    if (pRequest->type == TSDB_SQL_RETRIEVE_EMPTY_RESULT || pRequest->type == TSDB_SQL_INSERT ||
        pRequest->code != TSDB_CODE_SUCCESS || taos_num_fields(res) == 0 || pRequest->killed) {
      return NULL;
    }

    if (pRequest->inCallback) {
      tscError("can not call taos_fetch_row before query callback ends.");
      terrno = TSDB_CODE_TSC_INVALID_OPERATION;
      return NULL;
    }

    return doAsyncFetchRows(pRequest, true, true);
  } else if (TD_RES_TMQ(res) || TD_RES_TMQ_METADATA(res)) {
    SMqRspObj      *msg = ((SMqRspObj *)res);
    SReqResultInfo *pResultInfo = NULL;
    if (msg->resIter == -1) {
      if (tmqGetNextResInfo(res, true, &pResultInfo) != 0) {
        return NULL;
      }
    } else {
      pResultInfo = tmqGetCurResInfo(res);
    }

    if (pResultInfo->current < pResultInfo->numOfRows) {
      doSetOneRowPtr(pResultInfo, false);
      pResultInfo->current += 1;
      return pResultInfo->row;
    } else {
      if (tmqGetNextResInfo(res, true, &pResultInfo) != 0) {
        return NULL;
      }

      doSetOneRowPtr(pResultInfo, false);
      pResultInfo->current += 1;
      return pResultInfo->row;
    }
  } else if (TD_RES_TMQ_META(res) || TD_RES_TMQ_BATCH_META(res)) {
    return NULL;
  } else {
    tscError("invalid result passed to taos_fetch_row");
    terrno = TSDB_CODE_TMQ_INVALID_DATA;
    return NULL;
  }
}

int taos_print_row(char *str, TAOS_ROW row, TAOS_FIELD *fields, int num_fields) {
  return taos_print_row_with_size(str, INT32_MAX, row, fields, num_fields);
}
int taos_print_row_with_size(char *str, uint32_t size, TAOS_ROW row, TAOS_FIELD *fields, int num_fields) {
  int32_t len = 0;
  for (int i = 0; i < num_fields; ++i) {
    if (i > 0 && len < size - 1) {
      str[len++] = ' ';
    }

    if (row[i] == NULL) {
      len += tsnprintf(str + len, size - len, "%s", TSDB_DATA_NULL_STR);
      continue;
    }

    switch (fields[i].type) {
      case TSDB_DATA_TYPE_TINYINT:
        len += tsnprintf(str + len, size - len, "%d", *((int8_t *)row[i]));
        break;

      case TSDB_DATA_TYPE_UTINYINT:
        len += tsnprintf(str + len, size - len, "%u", *((uint8_t *)row[i]));
        break;

      case TSDB_DATA_TYPE_SMALLINT:
        len += tsnprintf(str + len, size - len, "%d", *((int16_t *)row[i]));
        break;

      case TSDB_DATA_TYPE_USMALLINT:
        len += tsnprintf(str + len, size - len, "%u", *((uint16_t *)row[i]));
        break;

      case TSDB_DATA_TYPE_INT:
        len += tsnprintf(str + len, size - len, "%d", *((int32_t *)row[i]));
        break;

      case TSDB_DATA_TYPE_UINT:
        len += tsnprintf(str + len, size - len, "%u", *((uint32_t *)row[i]));
        break;

      case TSDB_DATA_TYPE_BIGINT:
        len += tsnprintf(str + len, size - len, "%" PRId64, *((int64_t *)row[i]));
        break;

      case TSDB_DATA_TYPE_UBIGINT:
        len += tsnprintf(str + len, size - len, "%" PRIu64, *((uint64_t *)row[i]));
        break;

      case TSDB_DATA_TYPE_FLOAT: {
        float fv = 0;
        fv = GET_FLOAT_VAL(row[i]);
        len += snprintf(str + len, size - len, "%.*g", FLT_DIG, fv);
      } break;

      case TSDB_DATA_TYPE_DOUBLE: {
        double dv = 0;
        dv = GET_DOUBLE_VAL(row[i]);
        len += snprintf(str + len, size - len, "%.*g", DBL_DIG, dv);
      } break;

      case TSDB_DATA_TYPE_VARBINARY: {
        void    *data = NULL;
        uint32_t tmp = 0;
        int32_t  charLen = varDataLen((char *)row[i] - VARSTR_HEADER_SIZE);
        if (taosAscii2Hex(row[i], charLen, &data, &tmp) < 0) {
          break;
        }
        uint32_t copyLen = TMIN(size - len - 1, tmp);
        (void)memcpy(str + len, data, copyLen);
        len += copyLen;
        taosMemoryFree(data);
      } break;
      case TSDB_DATA_TYPE_BINARY:
      case TSDB_DATA_TYPE_NCHAR:
      case TSDB_DATA_TYPE_GEOMETRY: {
        int32_t charLen = varDataLen((char *)row[i] - VARSTR_HEADER_SIZE);
        if (fields[i].type == TSDB_DATA_TYPE_BINARY || fields[i].type == TSDB_DATA_TYPE_VARBINARY ||
            fields[i].type == TSDB_DATA_TYPE_GEOMETRY) {
          if (charLen > fields[i].bytes || charLen < 0) {
            tscError("taos_print_row error binary. charLen:%d, fields[i].bytes:%d", charLen, fields[i].bytes);
            break;
          }
        } else {
          if (charLen > fields[i].bytes * TSDB_NCHAR_SIZE || charLen < 0) {
            tscError("taos_print_row error. charLen:%d, fields[i].bytes:%d", charLen, fields[i].bytes);
            break;
          }
        }

        uint32_t copyLen = TMIN(size - len - 1, charLen);
        (void)memcpy(str + len, row[i], copyLen);
        len += copyLen;
      } break;

      case TSDB_DATA_TYPE_TIMESTAMP:
        len += tsnprintf(str + len, size - len, "%" PRId64, *((int64_t *)row[i]));
        break;

      case TSDB_DATA_TYPE_BOOL:
        len += tsnprintf(str + len, size - len, "%d", *((int8_t *)row[i]));
        break;
      case TSDB_DATA_TYPE_DECIMAL64:
      case TSDB_DATA_TYPE_DECIMAL: {
        uint32_t decimalLen = strlen(row[i]);
        uint32_t copyLen = TMIN(size - len - 1, decimalLen);
        (void)memcpy(str + len, row[i], copyLen);
        len += copyLen;
      } break;
      default:
        break;
    }

    if (len >= size - 1) {
      break;
    }
  }
  if (len < size) {
    str[len] = 0;
  }

  return len;
}

int *taos_fetch_lengths(TAOS_RES *res) {
  if (res == NULL || TD_RES_TMQ_RAW(res) || TD_RES_TMQ_META(res) || TD_RES_TMQ_BATCH_META(res)) {
    return NULL;
  }

  SReqResultInfo *pResInfo = tscGetCurResInfo(res);
  return pResInfo->length;
}

TAOS_ROW *taos_result_block(TAOS_RES *res) {
  if (res == NULL || TD_RES_TMQ_RAW(res) || TD_RES_TMQ_META(res) || TD_RES_TMQ_BATCH_META(res)) {
    terrno = TSDB_CODE_INVALID_PARA;
    return NULL;
  }

  if (taos_is_update_query(res)) {
    return NULL;
  }

  SReqResultInfo *pResInfo = tscGetCurResInfo(res);
  return &pResInfo->row;
}

// todo intergrate with tDataTypes
const char *taos_data_type(int type) {
  switch (type) {
    case TSDB_DATA_TYPE_NULL:
      return "TSDB_DATA_TYPE_NULL";
    case TSDB_DATA_TYPE_BOOL:
      return "TSDB_DATA_TYPE_BOOL";
    case TSDB_DATA_TYPE_TINYINT:
      return "TSDB_DATA_TYPE_TINYINT";
    case TSDB_DATA_TYPE_SMALLINT:
      return "TSDB_DATA_TYPE_SMALLINT";
    case TSDB_DATA_TYPE_INT:
      return "TSDB_DATA_TYPE_INT";
    case TSDB_DATA_TYPE_BIGINT:
      return "TSDB_DATA_TYPE_BIGINT";
    case TSDB_DATA_TYPE_FLOAT:
      return "TSDB_DATA_TYPE_FLOAT";
    case TSDB_DATA_TYPE_DOUBLE:
      return "TSDB_DATA_TYPE_DOUBLE";
    case TSDB_DATA_TYPE_VARCHAR:
      return "TSDB_DATA_TYPE_VARCHAR";
      //    case TSDB_DATA_TYPE_BINARY:          return "TSDB_DATA_TYPE_VARCHAR";
    case TSDB_DATA_TYPE_TIMESTAMP:
      return "TSDB_DATA_TYPE_TIMESTAMP";
    case TSDB_DATA_TYPE_NCHAR:
      return "TSDB_DATA_TYPE_NCHAR";
    case TSDB_DATA_TYPE_JSON:
      return "TSDB_DATA_TYPE_JSON";
    case TSDB_DATA_TYPE_GEOMETRY:
      return "TSDB_DATA_TYPE_GEOMETRY";
    case TSDB_DATA_TYPE_UTINYINT:
      return "TSDB_DATA_TYPE_UTINYINT";
    case TSDB_DATA_TYPE_USMALLINT:
      return "TSDB_DATA_TYPE_USMALLINT";
    case TSDB_DATA_TYPE_UINT:
      return "TSDB_DATA_TYPE_UINT";
    case TSDB_DATA_TYPE_UBIGINT:
      return "TSDB_DATA_TYPE_UBIGINT";
    case TSDB_DATA_TYPE_VARBINARY:
      return "TSDB_DATA_TYPE_VARBINARY";
    case TSDB_DATA_TYPE_DECIMAL:
      return "TSDB_DATA_TYPE_DECIMAL";
    case TSDB_DATA_TYPE_BLOB:
      return "TSDB_DATA_TYPE_BLOB";
    case TSDB_DATA_TYPE_MEDIUMBLOB:
      return "TSDB_DATA_TYPE_MEDIUMBLOB";
    default:
      return "UNKNOWN";
  }
}

const char *taos_get_client_info() { return td_version; }

// return int32_t
int taos_affected_rows(TAOS_RES *res) {
  if (res == NULL || TD_RES_TMQ_RAW(res) || TD_RES_TMQ(res) || TD_RES_TMQ_META(res) || TD_RES_TMQ_METADATA(res) ||
      TD_RES_TMQ_BATCH_META(res)) {
    return 0;
  }

  SRequestObj    *pRequest = (SRequestObj *)res;
  SReqResultInfo *pResInfo = &pRequest->body.resInfo;
  return (int)pResInfo->numOfRows;
}

// return int64_t
int64_t taos_affected_rows64(TAOS_RES *res) {
  if (res == NULL || TD_RES_TMQ_RAW(res) || TD_RES_TMQ(res) || TD_RES_TMQ_META(res) || TD_RES_TMQ_METADATA(res) ||
      TD_RES_TMQ_BATCH_META(res)) {
    return 0;
  }

  SRequestObj    *pRequest = (SRequestObj *)res;
  SReqResultInfo *pResInfo = &pRequest->body.resInfo;
  return pResInfo->numOfRows;
}

int taos_result_precision(TAOS_RES *res) {
  if (res == NULL || TD_RES_TMQ_RAW(res) || TD_RES_TMQ_META(res) || TD_RES_TMQ_BATCH_META(res)) {
    return TSDB_TIME_PRECISION_MILLI;
  }

  if (TD_RES_QUERY(res)) {
    SRequestObj *pRequest = (SRequestObj *)res;
    return pRequest->body.resInfo.precision;
  } else if (TD_RES_TMQ(res) || TD_RES_TMQ_METADATA(res)) {
    SReqResultInfo *info = tmqGetCurResInfo(res);
    return info->precision;
  }
  return TSDB_TIME_PRECISION_MILLI;
}

int taos_select_db(TAOS *taos, const char *db) {
  STscObj *pObj = acquireTscObj(*(int64_t *)taos);
  if (pObj == NULL) {
    releaseTscObj(*(int64_t *)taos);
    terrno = TSDB_CODE_TSC_DISCONNECTED;
    return TSDB_CODE_TSC_DISCONNECTED;
  }

  if (db == NULL || strlen(db) == 0) {
    releaseTscObj(*(int64_t *)taos);
    tscError("invalid parameter for %s", db == NULL ? "db is NULL" : "db is empty");
    terrno = TSDB_CODE_TSC_INVALID_INPUT;
    return terrno;
  }

  char sql[256] = {0};
  (void)snprintf(sql, tListLen(sql), "use %s", db);

  TAOS_RES *pRequest = taos_query(taos, sql);
  int32_t   code = taos_errno(pRequest);

  taos_free_result(pRequest);
  releaseTscObj(*(int64_t *)taos);
  return code;
}

void taos_stop_query(TAOS_RES *res) {
  if (res == NULL || TD_RES_TMQ_RAW(res) || TD_RES_TMQ(res) || TD_RES_TMQ_META(res) || TD_RES_TMQ_METADATA(res) ||
      TD_RES_TMQ_BATCH_META(res)) {
    return;
  }

  stopAllQueries((SRequestObj *)res);
}

bool taos_is_null(TAOS_RES *res, int32_t row, int32_t col) {
  if (res == NULL || TD_RES_TMQ_RAW(res) || TD_RES_TMQ_META(res) || TD_RES_TMQ_BATCH_META(res)) {
    return true;
  }
  SReqResultInfo *pResultInfo = tscGetCurResInfo(res);
  if (col >= pResultInfo->numOfCols || col < 0 || row >= pResultInfo->numOfRows || row < 0) {
    return true;
  }

  SResultColumn *pCol = &pResultInfo->pCol[col];
  if (IS_VAR_DATA_TYPE(pResultInfo->fields[col].type)) {
    return (pCol->offset[row] == -1);
  } else {
    return colDataIsNull_f(pCol->nullbitmap, row);
  }
}

bool taos_is_update_query(TAOS_RES *res) { return taos_num_fields(res) == 0; }

int taos_fetch_block(TAOS_RES *res, TAOS_ROW *rows) {
  int32_t numOfRows = 0;
  /*int32_t code = */ terrno = taos_fetch_block_s(res, &numOfRows, rows);
  return numOfRows;
}

int taos_fetch_block_s(TAOS_RES *res, int *numOfRows, TAOS_ROW *rows) {
  if (res == NULL || TD_RES_TMQ_RAW(res) || TD_RES_TMQ_META(res) || TD_RES_TMQ_BATCH_META(res)) {
    return 0;
  }

  if (TD_RES_QUERY(res)) {
    SRequestObj *pRequest = (SRequestObj *)res;

    (*rows) = NULL;
    (*numOfRows) = 0;

    if (pRequest->type == TSDB_SQL_RETRIEVE_EMPTY_RESULT || pRequest->type == TSDB_SQL_INSERT ||
        pRequest->code != TSDB_CODE_SUCCESS || taos_num_fields(res) == 0) {
      return pRequest->code;
    }

    (void)doAsyncFetchRows(pRequest, false, true);

    // TODO refactor
    SReqResultInfo *pResultInfo = &pRequest->body.resInfo;
    pResultInfo->current = pResultInfo->numOfRows;

    (*rows) = pResultInfo->row;
    (*numOfRows) = pResultInfo->numOfRows;
    return pRequest->code;
  } else if (TD_RES_TMQ(res) || TD_RES_TMQ_METADATA(res)) {
    SReqResultInfo *pResultInfo = NULL;
    int32_t         code = tmqGetNextResInfo(res, true, &pResultInfo);
    if (code != 0) return code;

    pResultInfo->current = pResultInfo->numOfRows;
    (*rows) = pResultInfo->row;
    (*numOfRows) = pResultInfo->numOfRows;
    return 0;
  } else {
    tscError("taos_fetch_block_s invalid res type");
    return TSDB_CODE_TMQ_INVALID_DATA;
  }
}

int taos_fetch_raw_block(TAOS_RES *res, int *numOfRows, void **pData) {
  *numOfRows = 0;
  *pData = NULL;

  if (res == NULL || TD_RES_TMQ_RAW(res) || TD_RES_TMQ_META(res) || TD_RES_TMQ_BATCH_META(res)) {
    return 0;
  }

  if (TD_RES_TMQ(res) || TD_RES_TMQ_METADATA(res)) {
    SReqResultInfo *pResultInfo = NULL;
    int32_t         code = tmqGetNextResInfo(res, false, &pResultInfo);
    if (code != 0) {
      (*numOfRows) = 0;
      return 0;
    }

    pResultInfo->current = pResultInfo->numOfRows;
    (*numOfRows) = pResultInfo->numOfRows;
    (*pData) = (void *)pResultInfo->pData;
    return 0;
  }

  SRequestObj *pRequest = (SRequestObj *)res;

  if (pRequest->type == TSDB_SQL_RETRIEVE_EMPTY_RESULT || pRequest->type == TSDB_SQL_INSERT ||
      pRequest->code != TSDB_CODE_SUCCESS || taos_num_fields(res) == 0) {
    return pRequest->code;
  }

  (void)doAsyncFetchRows(pRequest, false, false);

  SReqResultInfo *pResultInfo = &pRequest->body.resInfo;

  pResultInfo->current = pResultInfo->numOfRows;
  (*numOfRows) = pResultInfo->numOfRows;
  (*pData) = (void *)pResultInfo->pData;

  return pRequest->code;
}

int *taos_get_column_data_offset(TAOS_RES *res, int columnIndex) {
  if (res == NULL || TD_RES_TMQ_RAW(res) || TD_RES_TMQ_META(res) || TD_RES_TMQ_BATCH_META(res)) {
    return 0;
  }

  int32_t numOfFields = taos_num_fields(res);
  if (columnIndex < 0 || columnIndex >= numOfFields || numOfFields == 0) {
    return 0;
  }

  SReqResultInfo *pResInfo = tscGetCurResInfo(res);
  TAOS_FIELD     *pField = &pResInfo->userFields[columnIndex];
  if (!IS_VAR_DATA_TYPE(pField->type)) {
    return 0;
  }

  return pResInfo->pCol[columnIndex].offset;
}

int taos_is_null_by_column(TAOS_RES *res, int columnIndex, bool result[], int *rows) {
  if (res == NULL || result == NULL || rows == NULL || *rows <= 0 || columnIndex < 0 || TD_RES_TMQ_META(res) ||
      TD_RES_TMQ_RAW(res) || TD_RES_TMQ_BATCH_META(res)) {
    return TSDB_CODE_INVALID_PARA;
  }

  int32_t numOfFields = taos_num_fields(res);
  if (columnIndex >= numOfFields || numOfFields == 0) {
    return TSDB_CODE_INVALID_PARA;
  }

  SReqResultInfo *pResInfo = tscGetCurResInfo(res);
  TAOS_FIELD     *pField = &pResInfo->userFields[columnIndex];
  SResultColumn  *pCol = &pResInfo->pCol[columnIndex];

  if (*rows > pResInfo->numOfRows) {
    *rows = pResInfo->numOfRows;
  }
  if (IS_VAR_DATA_TYPE(pField->type)) {
    for (int i = 0; i < *rows; i++) {
      if (pCol->offset[i] == -1) {
        result[i] = true;
      } else {
        result[i] = false;
      }
    }
  } else {
    for (int i = 0; i < *rows; i++) {
      if (colDataIsNull_f(pCol->nullbitmap, i)) {
        result[i] = true;
      } else {
        result[i] = false;
      }
    }
  }
  return 0;
}

int taos_validate_sql(TAOS *taos, const char *sql) {
  TAOS_RES *pObj = taosQueryImpl(taos, sql, true, TD_REQ_FROM_APP);

  int code = taos_errno(pObj);

  taos_free_result(pObj);
  return code;
}

void taos_reset_current_db(TAOS *taos) {
  STscObj *pTscObj = acquireTscObj(*(int64_t *)taos);
  if (pTscObj == NULL) {
    terrno = TSDB_CODE_TSC_DISCONNECTED;
    return;
  }

  resetConnectDB(pTscObj);

  releaseTscObj(*(int64_t *)taos);
}

const char *taos_get_server_info(TAOS *taos) {
  STscObj *pTscObj = acquireTscObj(*(int64_t *)taos);
  if (pTscObj == NULL) {
    terrno = TSDB_CODE_TSC_DISCONNECTED;
    return NULL;
  }

  releaseTscObj(*(int64_t *)taos);

  return pTscObj->sDetailVer;
}

int taos_get_current_db(TAOS *taos, char *database, int len, int *required) {
  STscObj *pTscObj = acquireTscObj(*(int64_t *)taos);
  if (pTscObj == NULL) {
    return TSDB_CODE_TSC_DISCONNECTED;
  }

  int code = TSDB_CODE_SUCCESS;
  (void)taosThreadMutexLock(&pTscObj->mutex);
  if (database == NULL || len <= 0) {
    if (required != NULL) *required = strlen(pTscObj->db) + 1;
    TSC_ERR_JRET(TSDB_CODE_INVALID_PARA);
  } else if (len < strlen(pTscObj->db) + 1) {
    tstrncpy(database, pTscObj->db, len);
    if (required) *required = strlen(pTscObj->db) + 1;
    TSC_ERR_JRET(TSDB_CODE_INVALID_PARA);
  } else {
    tstrncpy(database, pTscObj->db, len);
    code = 0;
  }
_return:
  (void)taosThreadMutexUnlock(&pTscObj->mutex);
  releaseTscObj(*(int64_t *)taos);
  return code;
}

void destorySqlCallbackWrapper(SSqlCallbackWrapper *pWrapper) {
  if (NULL == pWrapper) {
    return;
  }
  destoryCatalogReq(pWrapper->pCatalogReq);
  taosMemoryFree(pWrapper->pCatalogReq);
  qDestroyParseContext(pWrapper->pParseCtx);
  taosMemoryFree(pWrapper);
}

void destroyCtxInRequest(SRequestObj *pRequest) {
  schedulerFreeJob(&pRequest->body.queryJob, 0);
  qDestroyQuery(pRequest->pQuery);
  pRequest->pQuery = NULL;
  destorySqlCallbackWrapper(pRequest->pWrapper);
  pRequest->pWrapper = NULL;
}

static void doAsyncQueryFromAnalyse(SMetaData *pResultMeta, void *param, int32_t code) {
  SSqlCallbackWrapper *pWrapper = (SSqlCallbackWrapper *)param;
  SRequestObj         *pRequest = pWrapper->pRequest;
  SQuery              *pQuery = pRequest->pQuery;

  qDebug("req:0x%" PRIx64 ", start to semantic analysis, QID:0x%" PRIx64, pRequest->self, pRequest->requestId);

  int64_t analyseStart = taosGetTimestampUs();
  pRequest->metric.ctgCostUs = analyseStart - pRequest->metric.ctgStart;
  pWrapper->pParseCtx->parseOnly = pRequest->parseOnly;

  if (TSDB_CODE_SUCCESS == code) {
    code = qAnalyseSqlSemantic(pWrapper->pParseCtx, pWrapper->pCatalogReq, pResultMeta, pQuery);
  }

  pRequest->metric.analyseCostUs += taosGetTimestampUs() - analyseStart;

  if (pRequest->parseOnly) {
    (void)memcpy(&pRequest->parseMeta, pResultMeta, sizeof(*pResultMeta));
    (void)memset(pResultMeta, 0, sizeof(*pResultMeta));
  }

  handleQueryAnslyseRes(pWrapper, pResultMeta, code);
}

int32_t cloneCatalogReq(SCatalogReq **ppTarget, SCatalogReq *pSrc) {
  int32_t      code = TSDB_CODE_SUCCESS;
  SCatalogReq *pTarget = taosMemoryCalloc(1, sizeof(SCatalogReq));
  if (pTarget == NULL) {
    code = terrno;
  } else {
    pTarget->pDbVgroup = taosArrayDup(pSrc->pDbVgroup, NULL);
    pTarget->pDbCfg = taosArrayDup(pSrc->pDbCfg, NULL);
    pTarget->pDbInfo = taosArrayDup(pSrc->pDbInfo, NULL);
    pTarget->pTableMeta = taosArrayDup(pSrc->pTableMeta, NULL);
    pTarget->pTableHash = taosArrayDup(pSrc->pTableHash, NULL);
    pTarget->pUdf = taosArrayDup(pSrc->pUdf, NULL);
    pTarget->pIndex = taosArrayDup(pSrc->pIndex, NULL);
    pTarget->pUser = taosArrayDup(pSrc->pUser, NULL);
    pTarget->pTableIndex = taosArrayDup(pSrc->pTableIndex, NULL);
    pTarget->pTableCfg = taosArrayDup(pSrc->pTableCfg, NULL);
    pTarget->pTableTag = taosArrayDup(pSrc->pTableTag, NULL);
    pTarget->pView = taosArrayDup(pSrc->pView, NULL);
    pTarget->pTableTSMAs = taosArrayDup(pSrc->pTableTSMAs, NULL);
    pTarget->pTSMAs = taosArrayDup(pSrc->pTSMAs, NULL);
    pTarget->qNodeRequired = pSrc->qNodeRequired;
    pTarget->dNodeRequired = pSrc->dNodeRequired;
    pTarget->svrVerRequired = pSrc->svrVerRequired;
    pTarget->forceUpdate = pSrc->forceUpdate;
    pTarget->cloned = true;

    *ppTarget = pTarget;
  }

  return code;
}

void handleSubQueryFromAnalyse(SSqlCallbackWrapper *pWrapper, SMetaData *pResultMeta, SNode *pRoot) {
  SRequestObj         *pNewRequest = NULL;
  SSqlCallbackWrapper *pNewWrapper = NULL;
  int32_t              code = buildPreviousRequest(pWrapper->pRequest, pWrapper->pRequest->sqlstr, &pNewRequest);
  if (code) {
    handleQueryAnslyseRes(pWrapper, pResultMeta, code);
    return;
  }

  pNewRequest->pQuery = NULL;
  code = nodesMakeNode(QUERY_NODE_QUERY, (SNode **)&pNewRequest->pQuery);
  if (pNewRequest->pQuery) {
    pNewRequest->pQuery->pRoot = pRoot;
    pRoot = NULL;
    pNewRequest->pQuery->execStage = QUERY_EXEC_STAGE_ANALYSE;
  }
  if (TSDB_CODE_SUCCESS == code) {
    code = prepareAndParseSqlSyntax(&pNewWrapper, pNewRequest, false);
  }
  if (TSDB_CODE_SUCCESS == code) {
    code = cloneCatalogReq(&pNewWrapper->pCatalogReq, pWrapper->pCatalogReq);
  }
  if (TSDB_CODE_SUCCESS == code) {
    doAsyncQueryFromAnalyse(pResultMeta, pNewWrapper, code);
    nodesDestroyNode(pRoot);
  } else {
    handleQueryAnslyseRes(pWrapper, pResultMeta, code);
    return;
  }
}

void handleQueryAnslyseRes(SSqlCallbackWrapper *pWrapper, SMetaData *pResultMeta, int32_t code) {
  SRequestObj *pRequest = pWrapper->pRequest;
  SQuery      *pQuery = pRequest->pQuery;

  if (code == TSDB_CODE_SUCCESS && pQuery->pPrevRoot) {
    SNode *prevRoot = pQuery->pPrevRoot;
    pQuery->pPrevRoot = NULL;
    handleSubQueryFromAnalyse(pWrapper, pResultMeta, prevRoot);
    return;
  }

  if (code == TSDB_CODE_SUCCESS) {
    pRequest->stableQuery = pQuery->stableQuery;
    if (pQuery->pRoot) {
      pRequest->stmtType = pQuery->pRoot->type;
    }

    if (pQuery->haveResultSet) {
      code = setResSchemaInfo(&pRequest->body.resInfo, pQuery->pResSchema, pQuery->numOfResCols, pQuery->pResExtSchema, pRequest->isStmtBind);
      setResPrecision(&pRequest->body.resInfo, pQuery->precision);
    }
  }

  if (code == TSDB_CODE_SUCCESS) {
    TSWAP(pRequest->dbList, (pQuery)->pDbList);
    TSWAP(pRequest->tableList, (pQuery)->pTableList);
    TSWAP(pRequest->targetTableList, (pQuery)->pTargetTableList);

    launchAsyncQuery(pRequest, pQuery, pResultMeta, pWrapper);
  } else {
    destorySqlCallbackWrapper(pWrapper);
    pRequest->pWrapper = NULL;
    qDestroyQuery(pRequest->pQuery);
    pRequest->pQuery = NULL;

    if (NEED_CLIENT_HANDLE_ERROR(code)) {
      tscDebug("req:0x%" PRIx64 ", client retry to handle the error, code:%d - %s, tryCount:%d, QID:0x%" PRIx64,
               pRequest->self, code, tstrerror(code), pRequest->retry, pRequest->requestId);
      restartAsyncQuery(pRequest, code);
      return;
    }

    // return to app directly
    tscError("req:0x%" PRIx64 ", error occurs, code:%s, return to user app, QID:0x%" PRIx64, pRequest->self, tstrerror(code),
             pRequest->requestId);
    pRequest->code = code;
    returnToUser(pRequest);
  }
}

static int32_t getAllMetaAsync(SSqlCallbackWrapper *pWrapper, catalogCallback fp) {
  SRequestConnInfo conn = {.pTrans = pWrapper->pParseCtx->pTransporter,
                           .requestId = pWrapper->pParseCtx->requestId,
                           .requestObjRefId = pWrapper->pParseCtx->requestRid,
                           .mgmtEps = pWrapper->pParseCtx->mgmtEpSet};

  pWrapper->pRequest->metric.ctgStart = taosGetTimestampUs();

  return catalogAsyncGetAllMeta(pWrapper->pParseCtx->pCatalog, &conn, pWrapper->pCatalogReq, fp, pWrapper,
                                &pWrapper->pRequest->body.queryJob);
}

static void doAsyncQueryFromParse(SMetaData *pResultMeta, void *param, int32_t code);

static int32_t phaseAsyncQuery(SSqlCallbackWrapper *pWrapper) {
  int32_t code = TSDB_CODE_SUCCESS;
  switch (pWrapper->pRequest->pQuery->execStage) {
    case QUERY_EXEC_STAGE_PARSE: {
      // continue parse after get metadata
      code = getAllMetaAsync(pWrapper, doAsyncQueryFromParse);
      break;
    }
    case QUERY_EXEC_STAGE_ANALYSE: {
      // analysis after get metadata
      code = getAllMetaAsync(pWrapper, doAsyncQueryFromAnalyse);
      break;
    }
    case QUERY_EXEC_STAGE_SCHEDULE: {
      launchAsyncQuery(pWrapper->pRequest, pWrapper->pRequest->pQuery, NULL, pWrapper);
      break;
    }
    default:
      break;
  }
  return code;
}

static void doAsyncQueryFromParse(SMetaData *pResultMeta, void *param, int32_t code) {
  SSqlCallbackWrapper *pWrapper = (SSqlCallbackWrapper *)param;
  SRequestObj         *pRequest = pWrapper->pRequest;
  SQuery              *pQuery = pRequest->pQuery;

  pRequest->metric.ctgCostUs += taosGetTimestampUs() - pRequest->metric.ctgStart;
  qDebug("req:0x%" PRIx64 ", start to continue parse, QID:0x%" PRIx64 ", code:%s", pRequest->self, pRequest->requestId,
         tstrerror(code));

  if (code == TSDB_CODE_SUCCESS) {
    // pWrapper->pCatalogReq->forceUpdate = false;
    code = qContinueParseSql(pWrapper->pParseCtx, pWrapper->pCatalogReq, pResultMeta, pQuery);
  }

  if (TSDB_CODE_SUCCESS == code) {
    code = phaseAsyncQuery(pWrapper);
  }

  if (TSDB_CODE_SUCCESS != code) {
    tscError("req:0x%" PRIx64 ", error happens, code:%d - %s, QID:0x%" PRIx64, pWrapper->pRequest->self, code,
             tstrerror(code), pWrapper->pRequest->requestId);
    destorySqlCallbackWrapper(pWrapper);
    pRequest->pWrapper = NULL;
    terrno = code;
    pRequest->code = code;
    doRequestCallback(pRequest, code);
  }
}

void continueInsertFromCsv(SSqlCallbackWrapper *pWrapper, SRequestObj *pRequest) {
  int32_t code = qParseSqlSyntax(pWrapper->pParseCtx, &pRequest->pQuery, pWrapper->pCatalogReq);
  if (TSDB_CODE_SUCCESS == code) {
    code = phaseAsyncQuery(pWrapper);
  }

  if (TSDB_CODE_SUCCESS != code) {
    tscError("req:0x%" PRIx64 ", error happens, code:%d - %s, QID:0x%" PRIx64, pWrapper->pRequest->self, code,
             tstrerror(code), pWrapper->pRequest->requestId);
    destorySqlCallbackWrapper(pWrapper);
    pRequest->pWrapper = NULL;
    terrno = code;
    pRequest->code = code;
    doRequestCallback(pRequest, code);
  }
}

void taos_query_a(TAOS *taos, const char *sql, __taos_async_fn_t fp, void *param) {
  int64_t connId = *(int64_t *)taos;
  tscDebug("taos_query_a start with sql:%s", sql);
  taosAsyncQueryImpl(connId, sql, fp, param, false, TD_REQ_FROM_APP);
  tscDebug("taos_query_a end with sql:%s", sql);
}

void taos_query_a_with_reqid(TAOS *taos, const char *sql, __taos_async_fn_t fp, void *param, int64_t reqid) {
  int64_t connId = *(int64_t *)taos;
  taosAsyncQueryImplWithReqid(connId, sql, fp, param, false, reqid);
}

int32_t createParseContext(const SRequestObj *pRequest, SParseContext **pCxt, SSqlCallbackWrapper *pWrapper) {
  const STscObj *pTscObj = pRequest->pTscObj;

  *pCxt = taosMemoryCalloc(1, sizeof(SParseContext));
  if (*pCxt == NULL) {
    return terrno;
  }

  **pCxt = (SParseContext){.requestId = pRequest->requestId,
                           .requestRid = pRequest->self,
                           .acctId = pTscObj->acctId,
                           .db = pRequest->pDb,
                           .topicQuery = false,
                           .pSql = pRequest->sqlstr,
                           .sqlLen = pRequest->sqlLen,
                           .pMsg = pRequest->msgBuf,
                           .msgLen = ERROR_MSG_BUF_DEFAULT_SIZE,
                           .pTransporter = pTscObj->pAppInfo->pTransporter,
                           .pStmtCb = NULL,
                           .pUser = pTscObj->user,
                           .pEffectiveUser = pRequest->effectiveUser,
                           .isSuperUser = (0 == strcmp(pTscObj->user, TSDB_DEFAULT_USER)),
                           .enableSysInfo = pTscObj->sysInfo,
                           .async = true,
                           .svrVer = pTscObj->sVer,
                           .nodeOffline = (pTscObj->pAppInfo->onlineDnodes < pTscObj->pAppInfo->totalDnodes),
                           .allocatorId = pRequest->allocatorRefId,
                           .parseSqlFp = clientParseSql,
                           .parseSqlParam = pWrapper,
                           .setQueryFp = setQueryRequest,
                           .timezone = pTscObj->optionInfo.timezone,
                           .charsetCxt = pTscObj->optionInfo.charsetCxt,
                           .streamRunHistory = pRequest->streamRunHistory};
  int8_t biMode = atomic_load_8(&((STscObj *)pTscObj)->biMode);
  (*pCxt)->biMode = biMode;
  return TSDB_CODE_SUCCESS;
}

int32_t prepareAndParseSqlSyntax(SSqlCallbackWrapper **ppWrapper, SRequestObj *pRequest, bool updateMetaForce) {
  int32_t              code = TSDB_CODE_SUCCESS;
  STscObj             *pTscObj = pRequest->pTscObj;
  SSqlCallbackWrapper *pWrapper = taosMemoryCalloc(1, sizeof(SSqlCallbackWrapper));
  if (pWrapper == NULL) {
    code = terrno;
  } else {
    pWrapper->pRequest = pRequest;
    pRequest->pWrapper = pWrapper;
    *ppWrapper = pWrapper;
  }

  if (TSDB_CODE_SUCCESS == code) {
    code = createParseContext(pRequest, &pWrapper->pParseCtx, pWrapper);
  }

  if (TSDB_CODE_SUCCESS == code) {
    pWrapper->pParseCtx->mgmtEpSet = getEpSet_s(&pTscObj->pAppInfo->mgmtEp);
    code = catalogGetHandle(pTscObj->pAppInfo->clusterId, &pWrapper->pParseCtx->pCatalog);
  }

  if (TSDB_CODE_SUCCESS == code && NULL == pRequest->pQuery) {
    int64_t syntaxStart = taosGetTimestampUs();

    pWrapper->pCatalogReq = taosMemoryCalloc(1, sizeof(SCatalogReq));
    if (pWrapper->pCatalogReq == NULL) {
      code = terrno;
    } else {
      pWrapper->pCatalogReq->forceUpdate = updateMetaForce;
      TSC_ERR_RET(qnodeRequired(pRequest, &pWrapper->pCatalogReq->qNodeRequired));
      code = qParseSqlSyntax(pWrapper->pParseCtx, &pRequest->pQuery, pWrapper->pCatalogReq);
    }

    pRequest->metric.parseCostUs += taosGetTimestampUs() - syntaxStart;
  }

  return code;
}

void doAsyncQuery(SRequestObj *pRequest, bool updateMetaForce) {
  SSqlCallbackWrapper *pWrapper = NULL;
  int32_t              code = TSDB_CODE_SUCCESS;

  if (pRequest->retry++ > REQUEST_TOTAL_EXEC_TIMES) {
    code = pRequest->prevCode;
    terrno = code;
    pRequest->code = code;
    tscDebug("req:0x%" PRIx64 ", call sync query cb with code:%s", pRequest->self, tstrerror(code));
    doRequestCallback(pRequest, code);
    return;
  }

  if (TSDB_CODE_SUCCESS == code) {
    code = prepareAndParseSqlSyntax(&pWrapper, pRequest, updateMetaForce);
  }

  if (TSDB_CODE_SUCCESS == code) {
    pRequest->stmtType = pRequest->pQuery->pRoot->type;
    code = phaseAsyncQuery(pWrapper);
  }

  if (TSDB_CODE_SUCCESS != code) {
    tscError("req:0x%" PRIx64 ", error happens, code:%d - %s, QID:0x%" PRIx64, pRequest->self, code, tstrerror(code),
             pRequest->requestId);
    destorySqlCallbackWrapper(pWrapper);
    pRequest->pWrapper = NULL;
    qDestroyQuery(pRequest->pQuery);
    pRequest->pQuery = NULL;

    if (NEED_CLIENT_HANDLE_ERROR(code)) {
      tscDebug("req:0x%" PRIx64 ", client retry to handle the error, code:%d - %s, tryCount:%d, QID:0x%" PRIx64,
               pRequest->self, code, tstrerror(code), pRequest->retry, pRequest->requestId);
      code = refreshMeta(pRequest->pTscObj, pRequest);
      if (code != 0) {
        tscWarn("req:0x%" PRIx64 ", refresh meta failed, code:%d - %s, QID:0x%" PRIx64, pRequest->self, code, tstrerror(code),
                pRequest->requestId);
      }
      pRequest->prevCode = code;
      doAsyncQuery(pRequest, true);
      return;
    }

    terrno = code;
    pRequest->code = code;
    doRequestCallback(pRequest, code);
  }
}

void restartAsyncQuery(SRequestObj *pRequest, int32_t code) {
  tscInfo("restart request:%s p:%p", pRequest->sqlstr, pRequest);
  SRequestObj *pUserReq = pRequest;
  (void)acquireRequest(pRequest->self);
  while (pUserReq) {
    if (pUserReq->self == pUserReq->relation.userRefId || pUserReq->relation.userRefId == 0) {
      break;
    } else {
      int64_t nextRefId = pUserReq->relation.nextRefId;
      (void)releaseRequest(pUserReq->self);
      if (nextRefId) {
        pUserReq = acquireRequest(nextRefId);
      }
    }
  }
  bool hasSubRequest = pUserReq != pRequest || pRequest->relation.prevRefId != 0;
  if (pUserReq) {
    destroyCtxInRequest(pUserReq);
    pUserReq->prevCode = code;
    (void)memset(&pUserReq->relation, 0, sizeof(pUserReq->relation));
  } else {
    tscError("User req is missing");
    (void)removeFromMostPrevReq(pRequest);
    return;
  }
  if (hasSubRequest)
    (void)removeFromMostPrevReq(pRequest);
  else
    (void)releaseRequest(pUserReq->self);
  doAsyncQuery(pUserReq, true);
}

typedef struct SAsyncFetchParam {
  SRequestObj      *pReq;
  __taos_async_fn_t fp;
  void             *param;
} SAsyncFetchParam;

static int32_t doAsyncFetch(void *pParam) {
  SAsyncFetchParam *param = pParam;
  taosAsyncFetchImpl(param->pReq, param->fp, param->param);
  taosMemoryFree(param);
  return TSDB_CODE_SUCCESS;
}

void taos_fetch_rows_a(TAOS_RES *res, __taos_async_fn_t fp, void *param) {
  if (res == NULL || fp == NULL) {
    tscError("taos_fetch_rows_a invalid paras");
    return;
  }
  if (!TD_RES_QUERY(res)) {
    tscError("taos_fetch_rows_a res is NULL");
    fp(param, res, TSDB_CODE_APP_ERROR);
    return;
  }

  SRequestObj *pRequest = res;
  if (TSDB_SQL_RETRIEVE_EMPTY_RESULT == pRequest->type) {
    fp(param, res, 0);
    return;
  }

  SAsyncFetchParam *pParam = taosMemoryCalloc(1, sizeof(SAsyncFetchParam));
  if (!pParam) {
    fp(param, res, terrno);
    return;
  }
  pParam->pReq = pRequest;
  pParam->fp = fp;
  pParam->param = param;
  int32_t code = taosAsyncExec(doAsyncFetch, pParam, NULL);
  if (TSDB_CODE_SUCCESS != code) {
    taosMemoryFree(pParam);
    fp(param, res, code);
    return;
  }
}

void taos_fetch_raw_block_a(TAOS_RES *res, __taos_async_fn_t fp, void *param) {
  if (res == NULL || fp == NULL) {
    tscError("taos_fetch_raw_block_a invalid paras");
    return;
  }
  if (!TD_RES_QUERY(res)) {
    tscError("taos_fetch_raw_block_a res is NULL");
    return;
  }
  SRequestObj    *pRequest = res;
  SReqResultInfo *pResultInfo = &pRequest->body.resInfo;

  // set the current block is all consumed
  pResultInfo->convertUcs4 = false;

  // it is a local executed query, no need to do async fetch
  taos_fetch_rows_a(pRequest, fp, param);
}

const void *taos_get_raw_block(TAOS_RES *res) {
  if (res == NULL) {
    tscError("taos_get_raw_block invalid paras");
    return NULL;
  }
  if (!TD_RES_QUERY(res)) {
    tscError("taos_get_raw_block res is NULL");
    return NULL;
  }
  SRequestObj *pRequest = res;

  return pRequest->body.resInfo.pData;
}

int taos_get_db_route_info(TAOS *taos, const char *db, TAOS_DB_ROUTE_INFO *dbInfo) {
  if (NULL == taos) {
    terrno = TSDB_CODE_TSC_DISCONNECTED;
    return terrno;
  }

  if (NULL == db || NULL == dbInfo) {
    tscError("invalid input param, db:%p, dbInfo:%p", db, dbInfo);
    terrno = TSDB_CODE_TSC_INVALID_INPUT;
    return terrno;
  }

  int64_t      connId = *(int64_t *)taos;
  SRequestObj *pRequest = NULL;
  char        *sql = "taos_get_db_route_info";
  int32_t      code = buildRequest(connId, sql, strlen(sql), NULL, false, &pRequest, 0);
  if (code != TSDB_CODE_SUCCESS) {
    terrno = code;
    return terrno;
  }

  STscObj  *pTscObj = pRequest->pTscObj;
  SCatalog *pCtg = NULL;
  code = catalogGetHandle(pTscObj->pAppInfo->clusterId, &pCtg);
  if (code != TSDB_CODE_SUCCESS) {
    goto _return;
  }

  SRequestConnInfo conn = {
      .pTrans = pTscObj->pAppInfo->pTransporter, .requestId = pRequest->requestId, .requestObjRefId = pRequest->self};

  conn.mgmtEps = getEpSet_s(&pTscObj->pAppInfo->mgmtEp);

  char dbFName[TSDB_DB_FNAME_LEN] = {0};
  (void)snprintf(dbFName, sizeof(dbFName), "%d.%s", pTscObj->acctId, db);

  code = catalogGetDBVgInfo(pCtg, &conn, dbFName, dbInfo);
  if (code) {
    goto _return;
  }

_return:

  terrno = code;

  destroyRequest(pRequest);
  return code;
}

int taos_get_table_vgId(TAOS *taos, const char *db, const char *table, int *vgId) {
  if (NULL == taos) {
    terrno = TSDB_CODE_TSC_DISCONNECTED;
    return terrno;
  }

  if (NULL == db || NULL == table || NULL == vgId) {
    tscError("invalid input param, db:%p, table:%p, vgId:%p", db, table, vgId);
    terrno = TSDB_CODE_TSC_INVALID_INPUT;
    return terrno;
  }

  int64_t      connId = *(int64_t *)taos;
  SRequestObj *pRequest = NULL;
  char        *sql = "taos_get_table_vgId";
  int32_t      code = buildRequest(connId, sql, strlen(sql), NULL, false, &pRequest, 0);
  if (code != TSDB_CODE_SUCCESS) {
    return terrno;
  }

  pRequest->syncQuery = true;

  STscObj  *pTscObj = pRequest->pTscObj;
  SCatalog *pCtg = NULL;
  code = catalogGetHandle(pTscObj->pAppInfo->clusterId, &pCtg);
  if (code != TSDB_CODE_SUCCESS) {
    goto _return;
  }

  SRequestConnInfo conn = {
      .pTrans = pTscObj->pAppInfo->pTransporter, .requestId = pRequest->requestId, .requestObjRefId = pRequest->self};

  conn.mgmtEps = getEpSet_s(&pTscObj->pAppInfo->mgmtEp);

  SName tableName = {0};
  toName(pTscObj->acctId, db, table, &tableName);

  SVgroupInfo vgInfo;
  code = catalogGetTableHashVgroup(pCtg, &conn, &tableName, &vgInfo);
  if (code) {
    goto _return;
  }

  *vgId = vgInfo.vgId;

_return:

  terrno = code;

  destroyRequest(pRequest);
  return code;
}

int taos_get_tables_vgId(TAOS *taos, const char *db, const char *table[], int tableNum, int *vgId) {
  if (NULL == taos) {
    terrno = TSDB_CODE_TSC_DISCONNECTED;
    return terrno;
  }

  if (NULL == db || NULL == table || NULL == vgId || tableNum <= 0) {
    tscError("invalid input param, db:%p, table:%p, vgId:%p, tbNum:%d", db, table, vgId, tableNum);
    terrno = TSDB_CODE_TSC_INVALID_INPUT;
    return terrno;
  }

  int64_t      connId = *(int64_t *)taos;
  SRequestObj *pRequest = NULL;
  char        *sql = "taos_get_table_vgId";
  int32_t      code = buildRequest(connId, sql, strlen(sql), NULL, false, &pRequest, 0);
  if (code != TSDB_CODE_SUCCESS) {
    return terrno;
  }

  pRequest->syncQuery = true;

  STscObj  *pTscObj = pRequest->pTscObj;
  SCatalog *pCtg = NULL;
  code = catalogGetHandle(pTscObj->pAppInfo->clusterId, &pCtg);
  if (code != TSDB_CODE_SUCCESS) {
    goto _return;
  }

  SRequestConnInfo conn = {
      .pTrans = pTscObj->pAppInfo->pTransporter, .requestId = pRequest->requestId, .requestObjRefId = pRequest->self};

  conn.mgmtEps = getEpSet_s(&pTscObj->pAppInfo->mgmtEp);

  code = catalogGetTablesHashVgId(pCtg, &conn, pTscObj->acctId, db, table, tableNum, vgId);
  if (code) {
    goto _return;
  }

_return:

  terrno = code;

  destroyRequest(pRequest);
  return code;
}

int taos_load_table_info(TAOS *taos, const char *tableNameList) {
  if (NULL == taos) {
    terrno = TSDB_CODE_TSC_DISCONNECTED;
    return terrno;
  }

  int64_t       connId = *(int64_t *)taos;
  const int32_t MAX_TABLE_NAME_LENGTH = 12 * 1024 * 1024;  // 12MB list
  int32_t       code = 0;
  SRequestObj  *pRequest = NULL;
  SCatalogReq   catalogReq = {0};

  if (NULL == tableNameList) {
    return TSDB_CODE_SUCCESS;
  }

  int32_t length = (int32_t)strlen(tableNameList);
  if (0 == length) {
    return TSDB_CODE_SUCCESS;
  } else if (length > MAX_TABLE_NAME_LENGTH) {
    tscError("tableNameList too long, length:%d, maximum allowed:%d", length, MAX_TABLE_NAME_LENGTH);
    return TSDB_CODE_TSC_INVALID_OPERATION;
  }

  char *sql = "taos_load_table_info";
  code = buildRequest(connId, sql, strlen(sql), NULL, false, &pRequest, 0);
  if (code != TSDB_CODE_SUCCESS) {
    terrno = code;
    goto _return;
  }

  pRequest->syncQuery = true;

  STscObj *pTscObj = pRequest->pTscObj;
  code = transferTableNameList(tableNameList, pTscObj->acctId, pTscObj->db, &catalogReq.pTableMeta);
  if (code) {
    goto _return;
  }

  SCatalog *pCtg = NULL;
  code = catalogGetHandle(pTscObj->pAppInfo->clusterId, &pCtg);
  if (code != TSDB_CODE_SUCCESS) {
    goto _return;
  }

  SRequestConnInfo conn = {
      .pTrans = pTscObj->pAppInfo->pTransporter, .requestId = pRequest->requestId, .requestObjRefId = pRequest->self};

  conn.mgmtEps = getEpSet_s(&pTscObj->pAppInfo->mgmtEp);

  code = catalogAsyncGetAllMeta(pCtg, &conn, &catalogReq, syncCatalogFn, pRequest->body.interParam, NULL);
  if (code) {
    goto _return;
  }

  SSyncQueryParam *pParam = pRequest->body.interParam;
  code = tsem_wait(&pParam->sem);
  if (code) {
    tscError("tsem wait failed, code:%d - %s", code, tstrerror(code));
    goto _return;
  }
_return:
  destoryCatalogReq(&catalogReq);
  destroyRequest(pRequest);
  return code;
}

TAOS_STMT *taos_stmt_init(TAOS *taos) {
  STscObj *pObj = acquireTscObj(*(int64_t *)taos);
  if (NULL == pObj) {
    tscError("invalid parameter for %s", __FUNCTION__);
    terrno = TSDB_CODE_TSC_DISCONNECTED;
    return NULL;
  }

  TAOS_STMT *pStmt = stmtInit(pObj, 0, NULL);
  if (NULL == pStmt) {
    tscError("stmt init failed, errcode:%s", terrstr());
  }
  releaseTscObj(*(int64_t *)taos);

  return pStmt;
}

TAOS_STMT *taos_stmt_init_with_reqid(TAOS *taos, int64_t reqid) {
  STscObj *pObj = acquireTscObj(*(int64_t *)taos);
  if (NULL == pObj) {
    tscError("invalid parameter for %s", __FUNCTION__);
    terrno = TSDB_CODE_TSC_DISCONNECTED;
    return NULL;
  }

  TAOS_STMT *pStmt = stmtInit(pObj, reqid, NULL);
  if (NULL == pStmt) {
    tscError("stmt init failed, errcode:%s", terrstr());
  }
  releaseTscObj(*(int64_t *)taos);

  return pStmt;
}

TAOS_STMT *taos_stmt_init_with_options(TAOS *taos, TAOS_STMT_OPTIONS *options) {
  STscObj *pObj = acquireTscObj(*(int64_t *)taos);
  if (NULL == pObj) {
    tscError("invalid parameter for %s", __FUNCTION__);
    terrno = TSDB_CODE_TSC_DISCONNECTED;
    return NULL;
  }

  TAOS_STMT *pStmt = stmtInit(pObj, options->reqId, options);
  if (NULL == pStmt) {
    tscError("stmt init failed, errcode:%s", terrstr());
  }
  releaseTscObj(*(int64_t *)taos);

  return pStmt;
}

int taos_stmt_prepare(TAOS_STMT *stmt, const char *sql, unsigned long length) {
  if (stmt == NULL || sql == NULL) {
    tscError("NULL parameter for %s", __FUNCTION__);
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

  return stmtPrepare(stmt, sql, length);
}

int taos_stmt_set_tbname_tags(TAOS_STMT *stmt, const char *name, TAOS_MULTI_BIND *tags) {
  if (stmt == NULL || name == NULL) {
    tscError("NULL parameter for %s", __FUNCTION__);
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

  int32_t code = stmtSetTbName(stmt, name);
  if (code) {
    return code;
  }

  if (tags) {
    return stmtSetTbTags(stmt, tags);
  }

  return TSDB_CODE_SUCCESS;
}

int taos_stmt_set_tbname(TAOS_STMT *stmt, const char *name) {
  if (stmt == NULL || name == NULL) {
    tscError("NULL parameter for %s", __FUNCTION__);
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

  return stmtSetTbName(stmt, name);
}

int taos_stmt_set_tags(TAOS_STMT *stmt, TAOS_MULTI_BIND *tags) {
  if (stmt == NULL || tags == NULL) {
    tscError("NULL parameter for %s", __FUNCTION__);
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

  return stmtSetTbTags(stmt, tags);
}

int taos_stmt_set_sub_tbname(TAOS_STMT *stmt, const char *name) { return taos_stmt_set_tbname(stmt, name); }

int taos_stmt_get_tag_fields(TAOS_STMT *stmt, int *fieldNum, TAOS_FIELD_E **fields) {
  if (stmt == NULL || NULL == fieldNum) {
    tscError("NULL parameter for %s", __FUNCTION__);
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

  return stmtGetTagFields(stmt, fieldNum, fields);
}

int taos_stmt_get_col_fields(TAOS_STMT *stmt, int *fieldNum, TAOS_FIELD_E **fields) {
  if (stmt == NULL || NULL == fieldNum) {
    tscError("NULL parameter for %s", __FUNCTION__);
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

  return stmtGetColFields(stmt, fieldNum, fields);
}

// let stmt to reclaim TAOS_FIELD_E that was allocated by `taos_stmt_get_tag_fields`/`taos_stmt_get_col_fields`
void taos_stmt_reclaim_fields(TAOS_STMT *stmt, TAOS_FIELD_E *fields) {
  (void)stmt;
  if (!fields) return;
  taosMemoryFree(fields);
}

int taos_stmt_bind_param(TAOS_STMT *stmt, TAOS_MULTI_BIND *bind) {
  if (stmt == NULL || bind == NULL) {
    tscError("NULL parameter for %s", __FUNCTION__);
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

  if (bind->num > 1) {
    tscError("invalid bind number %d for %s", bind->num, __FUNCTION__);
    terrno = TSDB_CODE_TSC_STMT_BIND_NUMBER_ERROR;
    return terrno;
  }

  return stmtBindBatch(stmt, bind, -1);
}

int taos_stmt_bind_param_batch(TAOS_STMT *stmt, TAOS_MULTI_BIND *bind) {
  if (stmt == NULL || bind == NULL) {
    tscError("NULL parameter for %s", __FUNCTION__);
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

  if (bind->num <= 0 || bind->num > INT16_MAX) {
    tscError("invalid bind num %d", bind->num);
    terrno = TSDB_CODE_TSC_STMT_BIND_NUMBER_ERROR;
    return terrno;
  }

  int32_t insert = 0;
  int32_t code = stmtIsInsert(stmt, &insert);
  if (TSDB_CODE_SUCCESS != code) {
    tscError("stmt insert failed, errcode:%s", tstrerror(code));
    return code;
  }
  if (0 == insert && bind->num > 1) {
    tscError("only one row data allowed for query");
    terrno = TSDB_CODE_TSC_STMT_BIND_NUMBER_ERROR;
    return terrno;
  }

  return stmtBindBatch(stmt, bind, -1);
}

int taos_stmt_bind_single_param_batch(TAOS_STMT *stmt, TAOS_MULTI_BIND *bind, int colIdx) {
  if (stmt == NULL || bind == NULL) {
    tscError("NULL parameter for %s", __FUNCTION__);
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

  if (colIdx < 0) {
    tscError("invalid bind column idx %d", colIdx);
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

  int32_t insert = 0;
  int32_t code = stmtIsInsert(stmt, &insert);
  if (TSDB_CODE_SUCCESS != code) {
    tscError("stmt insert failed, errcode:%s", tstrerror(code));
    return code;
  }
  if (0 == insert && bind->num > 1) {
    tscError("only one row data allowed for query");
    terrno = TSDB_CODE_TSC_STMT_BIND_NUMBER_ERROR;
    return terrno;
  }

  return stmtBindBatch(stmt, bind, colIdx);
}

int taos_stmt_add_batch(TAOS_STMT *stmt) {
  if (stmt == NULL) {
    tscError("NULL parameter for %s", __FUNCTION__);
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

  return stmtAddBatch(stmt);
}

int taos_stmt_execute(TAOS_STMT *stmt) {
  if (stmt == NULL) {
    tscError("NULL parameter for %s", __FUNCTION__);
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

  return stmtExec(stmt);
}

int taos_stmt_is_insert(TAOS_STMT *stmt, int *insert) {
  if (stmt == NULL || insert == NULL) {
    tscError("NULL parameter for %s", __FUNCTION__);
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

  return stmtIsInsert(stmt, insert);
}

int taos_stmt_num_params(TAOS_STMT *stmt, int *nums) {
  if (stmt == NULL || nums == NULL) {
    tscError("NULL parameter for %s", __FUNCTION__);
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

  return stmtGetParamNum(stmt, nums);
}

int taos_stmt_get_param(TAOS_STMT *stmt, int idx, int *type, int *bytes) {
  if (stmt == NULL || type == NULL || NULL == bytes || idx < 0) {
    tscError("invalid parameter for %s", __FUNCTION__);
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

  return stmtGetParam(stmt, idx, type, bytes);
}

TAOS_RES *taos_stmt_use_result(TAOS_STMT *stmt) {
  if (stmt == NULL) {
    tscError("NULL parameter for %s", __FUNCTION__);
    terrno = TSDB_CODE_INVALID_PARA;
    return NULL;
  }

  return stmtUseResult(stmt);
}

char *taos_stmt_errstr(TAOS_STMT *stmt) { return (char *)stmtErrstr(stmt); }

int taos_stmt_affected_rows(TAOS_STMT *stmt) {
  if (stmt == NULL) {
    tscError("NULL parameter for %s", __FUNCTION__);
    terrno = TSDB_CODE_INVALID_PARA;
    return 0;
  }

  return stmtAffectedRows(stmt);
}

int taos_stmt_affected_rows_once(TAOS_STMT *stmt) {
  if (stmt == NULL) {
    tscError("NULL parameter for %s", __FUNCTION__);
    terrno = TSDB_CODE_INVALID_PARA;
    return 0;
  }

  return stmtAffectedRowsOnce(stmt);
}

int taos_stmt_close(TAOS_STMT *stmt) {
  if (stmt == NULL) {
    tscError("NULL parameter for %s", __FUNCTION__);
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

  return stmtClose(stmt);
}

TAOS_STMT2 *taos_stmt2_init(TAOS *taos, TAOS_STMT2_OPTION *option) {
  if (NULL == taos) {
    tscError("NULL parameter for %s", __FUNCTION__);
    terrno = TSDB_CODE_INVALID_PARA;
    return NULL;
  }
  STscObj *pObj = acquireTscObj(*(int64_t *)taos);
  if (NULL == pObj) {
    tscError("invalid parameter for %s", __FUNCTION__);
    terrno = TSDB_CODE_TSC_DISCONNECTED;
    return NULL;
  }

  TAOS_STMT2 *pStmt = stmtInit2(pObj, option);

  releaseTscObj(*(int64_t *)taos);

  return pStmt;
}

int taos_stmt2_prepare(TAOS_STMT2 *stmt, const char *sql, unsigned long length) {
  if (stmt == NULL || sql == NULL) {
    tscError("NULL parameter for %s", __FUNCTION__);
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

  return stmtPrepare2(stmt, sql, length);
}

int taos_stmt2_bind_param(TAOS_STMT2 *stmt, TAOS_STMT2_BINDV *bindv, int32_t col_idx) {
  if (stmt == NULL) {
    tscError("NULL parameter for %s", __FUNCTION__);
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

  STscStmt2 *pStmt = (STscStmt2 *)stmt;
  if (atomic_load_8((int8_t *)&pStmt->asyncBindParam.asyncBindNum) > 1) {
    tscError("async bind param is still working, please try again later");
    return TSDB_CODE_TSC_STMT_API_ERROR;
  }

  if (pStmt->options.asyncExecFn && !pStmt->execSemWaited) {
    if (tsem_wait(&pStmt->asyncExecSem) != 0) {
      tscError("wait asyncExecSem failed");
    }
    pStmt->execSemWaited = true;
  }

  SSHashObj *hashTbnames = tSimpleHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_VARCHAR));
  if (NULL == hashTbnames) {
    tscError("stmt2 bind failed, %s", tstrerror(terrno));
    return terrno;
  }

  int32_t code = TSDB_CODE_SUCCESS;
  for (int i = 0; i < bindv->count; ++i) {
    if (bindv->tbnames && bindv->tbnames[i]) {
      if (pStmt->sql.stbInterlaceMode) {
        if (tSimpleHashGet(hashTbnames, bindv->tbnames[i], strlen(bindv->tbnames[i])) != NULL) {
          code = terrno = TSDB_CODE_PAR_TBNAME_DUPLICATED;
          tscError("stmt2 bind failed, %s %s", tstrerror(terrno), bindv->tbnames[i]);
          goto out;
        }

        code = tSimpleHashPut(hashTbnames, bindv->tbnames[i], strlen(bindv->tbnames[i]), NULL, 0);
        if (code) {
          goto out;
        }
      }

      code = stmtSetTbName2(stmt, bindv->tbnames[i]);
      if (code) {
        goto out;
      }
    }

    SVCreateTbReq *pCreateTbReq = NULL;
    if (bindv->tags && bindv->tags[i]) {
      code = stmtSetTbTags2(stmt, bindv->tags[i], &pCreateTbReq);
    } else if (pStmt->sql.autoCreateTbl || pStmt->bInfo.needParse) {
      code = stmtCheckTags2(stmt, &pCreateTbReq);
    } else {
      pStmt->sql.autoCreateTbl = false;
    }

    if (code) {
      goto out;
    }

    if (bindv->bind_cols && bindv->bind_cols[i]) {
      TAOS_STMT2_BIND *bind = bindv->bind_cols[i];

      if (bind->num <= 0 || bind->num > INT16_MAX) {
        tscError("invalid bind num %d", bind->num);
        code = terrno = TSDB_CODE_TSC_STMT_BIND_NUMBER_ERROR;
        goto out;
      }

      int32_t insert = 0;
      (void)stmtIsInsert2(stmt, &insert);
      if (0 == insert && bind->num > 1) {
        tscError("only one row data allowed for query");
        code = terrno = TSDB_CODE_TSC_STMT_BIND_NUMBER_ERROR;
        goto out;
      }

      code = stmtBindBatch2(stmt, bind, col_idx, pCreateTbReq);
      if (TSDB_CODE_SUCCESS != code) {
        goto out;
      }
    }
  }

out:
  tSimpleHashCleanup(hashTbnames);

  return code;
}

int taos_stmt2_bind_param_a(TAOS_STMT2 *stmt, TAOS_STMT2_BINDV *bindv, int32_t col_idx, __taos_async_fn_t fp,
                            void *param) {
  if (stmt == NULL || bindv == NULL || fp == NULL) {
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

  STscStmt2 *pStmt = (STscStmt2 *)stmt;

  ThreadArgs *args = (ThreadArgs *)taosMemoryMalloc(sizeof(ThreadArgs));
  args->stmt = stmt;
  args->bindv = bindv;
  args->col_idx = col_idx;
  args->fp = fp;
  args->param = param;

  (void)taosThreadMutexLock(&(pStmt->asyncBindParam.mutex));
  if (atomic_load_8((int8_t *)&pStmt->asyncBindParam.asyncBindNum) > 0) {
    (void)taosThreadMutexUnlock(&(pStmt->asyncBindParam.mutex));
    tscError("async bind param is still working, please try again later");
    return TSDB_CODE_TSC_STMT_API_ERROR;
  }
  (void)atomic_add_fetch_8(&pStmt->asyncBindParam.asyncBindNum, 1);
  (void)taosThreadMutexUnlock(&(pStmt->asyncBindParam.mutex));

  int code_s = taosStmt2AsyncBind(stmtAsyncBindThreadFunc, (void *)args);
  if (code_s != TSDB_CODE_SUCCESS) {
    (void)taosThreadMutexLock(&(pStmt->asyncBindParam.mutex));
    (void)taosThreadCondSignal(&(pStmt->asyncBindParam.waitCond));
    (void)atomic_sub_fetch_8(&pStmt->asyncBindParam.asyncBindNum, 1);
    (void)taosThreadMutexUnlock(&(pStmt->asyncBindParam.mutex));
    tscError("async bind failed, code:%d , %s", code_s, tstrerror(code_s));
  }

  return code_s;
}

int taos_stmt2_exec(TAOS_STMT2 *stmt, int *affected_rows) {
  if (stmt == NULL) {
    tscError("NULL parameter for %s", __FUNCTION__);
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

  return stmtExec2(stmt, affected_rows);
}

int taos_stmt2_close(TAOS_STMT2 *stmt) {
  if (stmt == NULL) {
    tscError("NULL parameter for %s", __FUNCTION__);
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

  return stmtClose2(stmt);
}

int taos_stmt2_is_insert(TAOS_STMT2 *stmt, int *insert) {
  if (stmt == NULL || insert == NULL) {
    tscError("NULL parameter for %s", __FUNCTION__);
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

  return stmtIsInsert2(stmt, insert);
}

int taos_stmt2_get_fields(TAOS_STMT2 *stmt, int *count, TAOS_FIELD_ALL **fields) {
  if (stmt == NULL || count == NULL) {
    tscError("NULL parameter for %s", __FUNCTION__);
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

  STscStmt2 *pStmt = (STscStmt2 *)stmt;
  if (pStmt->sql.type == 0) {
    int isInsert = 0;
    (void)stmtIsInsert2(stmt, &isInsert);
    if (!isInsert) {
      pStmt->sql.type = STMT_TYPE_QUERY;
    }
  }
  if (pStmt->sql.type == STMT_TYPE_QUERY) {
    return stmtGetParamNum2(stmt, count);
  }

  return stmtGetStbColFields2(stmt, count, fields);
}

DLL_EXPORT void taos_stmt2_free_fields(TAOS_STMT2 *stmt, TAOS_FIELD_ALL *fields) {
  (void)stmt;
  if (!fields) return;
  taosMemoryFree(fields);
}

TAOS_RES *taos_stmt2_result(TAOS_STMT2 *stmt) {
  if (stmt == NULL) {
    tscError("NULL parameter for %s", __FUNCTION__);
    terrno = TSDB_CODE_INVALID_PARA;
    return NULL;
  }

  return stmtUseResult2(stmt);
}

char *taos_stmt2_error(TAOS_STMT2 *stmt) { return (char *)stmtErrstr2(stmt); }

int taos_set_conn_mode(TAOS *taos, int mode, int value) {
  if (taos == NULL) {
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

  STscObj *pObj = acquireTscObj(*(int64_t *)taos);
  if (NULL == pObj) {
    terrno = TSDB_CODE_TSC_DISCONNECTED;
    tscError("invalid parameter for %s", __func__);
    return terrno;
  }
  switch (mode) {
    case TAOS_CONN_MODE_BI:
      atomic_store_8(&pObj->biMode, value);
      break;
    default:
      tscError("not supported mode.");
      return TSDB_CODE_INVALID_PARA;
  }
  return 0;
}

char *getBuildInfo() { return td_buildinfo; }

taosdata / TDengine / #3660

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

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